""" VariationEngine - Intelligent Sample Kit Evolution Across Song Sections. This module provides professional-grade sample kit variation for different song sections (intro, verse, chorus, bridge, outro) while maintaining coherence with the base kit. Core functionality: - Evolve drum kits based on section energy profiles - Find energy-matched sample variants from the library - Add/remove elements based on section requirements - Track coherence score (>0.80 required) - Integration with IntelligentSampleSelector Section Energy Profiles: intro: 0.3 - Minimal, building anticipation verse: 0.6 - Full groove, foundation pre_chorus: 0.75 - Adding tension, rising chorus: 0.9 - Maximum impact, all elements bridge: 0.5 - Contrast, variation outro: 0.2 - Fading, elements leaving Usage: from engines.variation_engine import VariationEngine, SectionKit # Create base kit base_kit = selector.select_for_genre("reggaeton") # Initialize variation engine engine = VariationEngine(selector=selector) # Evolve kit for chorus (high energy) chorus_kit = engine.evolve_kit_for_section(base_kit, "chorus") # Get coherence score coherence = engine.calculate_coherence(base_kit, chorus_kit) print(f"Coherence: {coherence:.2f}") # Must be > 0.80 Professional-grade design: - No random selection - Audio analysis-based decisions - Coherence tracking and validation - Seamless integration with metadata store """ import logging from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Tuple, Any, Set from enum import Enum # Configure logging logger = logging.getLogger("VariationEngine") # ============================================================================= # SECTION ENERGY PROFILES # ============================================================================= SECTION_PROFILES = { "intro": {"energy": 0.30, "description": "Minimal, building anticipation"}, "verse": {"energy": 0.60, "description": "Full groove, foundation"}, "pre_chorus": {"energy": 0.75, "description": "Adding tension, rising"}, "chorus": {"energy": 0.90, "description": "Maximum impact, all elements"}, "bridge": {"energy": 0.50, "description": "Contrast, variation"}, "outro": {"energy": 0.20, "description": "Fading, elements leaving"}, } # ============================================================================= # DATACLASSES # ============================================================================= @dataclass class EnergyCharacteristics: """ Audio energy characteristics extracted from sample analysis. Used to match samples by energy level for section-appropriate selection. """ rms: float = 0.0 # Root mean square (loudness) spectral_centroid: float = 0.0 # Brightness spectral_rolloff: float = 0.0 # Frequency distribution zero_crossing_rate: float = 0.0 # Noisiness/brightness attack_time: float = 0.0 # Transient sharpness decay_time: float = 0.0 # Sustain character # Energy score derived from features (0.0-1.0) derived_energy: float = 0.0 def calculate_energy_score(self) -> float: """ Calculate overall energy score from audio features. Weighted combination of perceptual energy indicators: - RMS contributes 40% (primary loudness indicator) - Spectral centroid 25% (brightness = perceived energy) - Attack time 20% (sharp transients = punch/impact) - Zero crossing rate 15% (high-frequency content) """ # Normalize RMS to 0-1 range (assuming typical range -30 to 0 dB) rms_norm = max(0.0, min(1.0, (self.rms + 30) / 30)) if self.rms else 0.5 # Normalize spectral centroid (assuming typical range 200-8000 Hz) centroid_norm = max(0.0, min(1.0, (self.spectral_centroid - 200) / 7800)) if self.spectral_centroid else 0.5 # Attack time: shorter = punchier (invert and normalize, typical 0.001-0.1s) attack_norm = max(0.0, min(1.0, 1.0 - (self.attack_time / 0.1))) if self.attack_time else 0.5 # Zero crossing rate (typical 0.0-0.3 for percussion) zcr_norm = max(0.0, min(1.0, self.zero_crossing_rate / 0.3)) if self.zero_crossing_rate else 0.5 # Weighted combination energy = ( rms_norm * 0.40 + centroid_norm * 0.25 + attack_norm * 0.20 + zcr_norm * 0.15 ) self.derived_energy = round(energy, 3) return self.derived_energy @dataclass class CoherenceMetrics: """ Coherence metrics between two sample kits. Tracks similarity across multiple dimensions to ensure variations maintain >0.80 coherence with base kit. """ # Individual dimension scores (0.0-1.0) timbre_score: float = 0.0 # Spectral similarity dynamics_score: float = 0.0 # Amplitude envelope similarity transient_score: float = 0.0 # Attack characteristics similarity rhythmic_score: float = 0.0 # Timing/structure similarity # Weighted total coherence total_coherence: float = 0.0 # Coherence check status is_valid: bool = False def calculate_total(self) -> float: """ Calculate weighted total coherence. Weights: - Timbre: 35% (most important for sonic identity) - Dynamics: 25% (amplitude behavior) - Transient: 20% (attack/punch similarity) - Rhythmic: 20% (for loops/patterns) """ self.total_coherence = ( self.timbre_score * 0.35 + self.dynamics_score * 0.25 + self.transient_score * 0.20 + self.rhythmic_score * 0.20 ) self.is_valid = self.total_coherence >= 0.80 return round(self.total_coherence, 3) def to_dict(self) -> Dict[str, Any]: return { "timbre_score": round(self.timbre_score, 3), "dynamics_score": round(self.dynamics_score, 3), "transient_score": round(self.transient_score, 3), "rhythmic_score": round(self.rhythmic_score, 3), "total_coherence": round(self.total_coherence, 3), "is_valid": self.is_valid, "threshold": 0.80, } @dataclass class SectionKit: """ A sample kit evolved for a specific song section. Contains the evolved kit plus metadata about the variation. """ section_name: str base_kit_name: str # Kit components (references to SampleInfo or SampleFeatures) kick: Optional[Any] = None snare: Optional[Any] = None clap: Optional[Any] = None hat_closed: Optional[Any] = None hat_open: Optional[Any] = None bass: List[Any] = field(default_factory=list) percussion: List[Any] = field(default_factory=list) fx: List[Any] = field(default_factory=list) # Variation metadata target_energy: float = 0.0 coherence_score: float = 0.0 variation_elements_added: List[str] = field(default_factory=list) variation_elements_removed: List[str] = field(default_factory=list) def get_all_samples(self) -> List[Any]: """Get list of all samples in this kit.""" samples = [] if self.kick: samples.append(self.kick) if self.snare: samples.append(self.snare) if self.clap: samples.append(self.clap) if self.hat_closed: samples.append(self.hat_closed) if self.hat_open: samples.append(self.hat_open) samples.extend(self.bass) samples.extend(self.percussion) samples.extend(self.fx) return samples def to_dict(self) -> Dict[str, Any]: return { "section_name": self.section_name, "base_kit_name": self.base_kit_name, "target_energy": self.target_energy, "coherence_score": round(self.coherence_score, 3), "samples": { "kick": self._sample_to_dict(self.kick), "snare": self._sample_to_dict(self.snare), "clap": self._sample_to_dict(self.clap), "hat_closed": self._sample_to_dict(self.hat_closed), "hat_open": self._sample_to_dict(self.hat_open), "bass_count": len(self.bass), "perc_count": len(self.percussion), "fx_count": len(self.fx), }, "variation_added": self.variation_elements_added, "variation_removed": self.variation_elements_removed, } @staticmethod def _sample_to_dict(sample: Optional[Any]) -> Optional[Dict]: if sample is None: return None if hasattr(sample, 'path'): return {"path": sample.path, "name": getattr(sample, 'name', Path(sample.path).name)} return {"path": str(sample)} # ============================================================================= # VARIATION ENGINE # ============================================================================= class VariationEngine: """ Professional-grade sample kit evolution engine. Creates section-specific kit variations that maintain >0.80 coherence with the base kit while adapting to section energy requirements. Key capabilities: - Energy-based sample selection from library - Coherence calculation and validation - Intelligent addition/removal of elements - Integration with IntelligentSampleSelector No random selection - all decisions based on audio analysis. """ # Coherence threshold (must be maintained across variations) COHERENCE_THRESHOLD = 0.80 # Energy tolerance for sample matching DEFAULT_ENERGY_TOLERANCE = 0.10 def __init__( self, selector=None, metadata_store=None, library_path: Optional[str] = None, verbose: bool = False ): """ Initialize VariationEngine. Args: selector: IntelligentSampleSelector instance (optional) metadata_store: SampleMetadataStore for feature access (optional) library_path: Path to sample library (optional) verbose: Enable detailed logging """ self.selector = selector self.metadata_store = metadata_store self.library_path = library_path self.verbose = verbose # Cache for sample energy characteristics self._energy_cache: Dict[str, EnergyCharacteristics] = {} # Track coherence scores for validation self.coherence_log: List[Dict[str, Any]] = [] if verbose: logger.info("[VariationEngine] Initialized") def evolve_kit_for_section( self, base_kit, section_name: str, min_coherence: float = 0.80 ) -> SectionKit: """ Evolve a base kit for a specific song section. Creates a section-appropriate variation by: 1. Determining target energy from section profile 2. Finding energy-appropriate sample variants 3. Adding/removing elements based on energy requirements 4. Validating coherence > 0.80 Args: base_kit: Base DrumKit or InstrumentGroup to evolve section_name: Target section (intro, verse, chorus, etc.) min_coherence: Minimum coherence required (default 0.80) Returns: SectionKit with evolved samples for the section """ if section_name not in SECTION_PROFILES: raise ValueError(f"Unknown section: {section_name}. " f"Valid: {list(SECTION_PROFILES.keys())}") profile = SECTION_PROFILES[section_name] target_energy = profile["energy"] if self.verbose: logger.info(f"[VariationEngine] Evolving kit for '{section_name}' " f"(target energy: {target_energy})") # Create section kit section_kit = SectionKit( section_name=section_name, base_kit_name=getattr(base_kit, 'genre', 'unknown'), target_energy=target_energy ) # Get target elements based on energy level elements_to_include = self._determine_elements_for_energy(target_energy) # Evolve each drum component if hasattr(base_kit, 'drums') and base_kit.drums: drums = base_kit.drums if "kick" in elements_to_include and drums.kick: section_kit.kick = self.find_energy_variant( drums.kick.path if hasattr(drums.kick, 'path') else str(drums.kick), target_energy ) if "snare" in elements_to_include and drums.snare: section_kit.snare = self.find_energy_variant( drums.snare.path if hasattr(drums.snare, 'path') else str(drums.snare), target_energy ) if "clap" in elements_to_include and drums.clap: section_kit.clap = self.find_energy_variant( drums.clap.path if hasattr(drums.clap, 'path') else str(drums.clap), target_energy ) if "hat_closed" in elements_to_include and drums.hat_closed: section_kit.hat_closed = self.find_energy_variant( drums.hat_closed.path if hasattr(drums.hat_closed, 'path') else str(drums.hat_closed), target_energy ) if "hat_open" in elements_to_include and drums.hat_open: section_kit.hat_open = self.find_energy_variant( drums.hat_open.path if hasattr(drums.hat_open, 'path') else str(drums.hat_open), target_energy ) # Handle bass and additional elements if hasattr(base_kit, 'bass') and base_kit.bass: for bass_sample in base_kit.bass[:2]: # Keep top 2 bass samples variant = self.find_energy_variant( bass_sample.path if hasattr(bass_sample, 'path') else str(bass_sample), target_energy ) if variant: section_kit.bass.append(variant) # Add variation elements based on section requirements added = self.add_variation_element(section_kit, target_energy) section_kit.variation_elements_added = added # Remove elements for low-energy sections if target_energy < 0.4: removed = self.remove_elements_for_energy(section_kit, target_energy) section_kit.variation_elements_removed = removed # Calculate and validate coherence coherence = self.calculate_coherence(base_kit, section_kit) section_kit.coherence_score = coherence.total_coherence # Log coherence result self._log_coherence(section_name, coherence) # Warn if coherence below threshold if not coherence.is_valid: logger.warning( f"[VariationEngine] Coherence {coherence.total_coherence:.2f} " f"below threshold {min_coherence} for section '{section_name}'" ) return section_kit def find_energy_variant( self, sample_path: str, target_energy: float, tolerance: float = 0.10, role: Optional[str] = None ) -> Optional[Any]: """ Find a sample variant matching the target energy characteristics. Uses audio analysis to find samples with similar spectral characteristics but matching energy level. Args: sample_path: Path to the base sample target_energy: Target energy level (0.0-1.0) tolerance: Energy matching tolerance role: Sample role (kick, snare, etc.) for filtering Returns: SampleInfo or SampleFeatures of matching sample, or original if no match """ # Get base sample characteristics base_energy = self._get_sample_energy(sample_path) if self.verbose: logger.info(f"[VariationEngine] Finding variant for {Path(sample_path).name} " f"(base energy: {base_energy:.2f}, target: {target_energy:.2f})") # If already close to target, return original if abs(base_energy - target_energy) <= tolerance: return self._get_sample_info(sample_path) # Search for matching samples via selector or metadata store candidates = self._find_similar_samples(sample_path, role) # Find closest energy match best_match = None best_diff = float('inf') for candidate in candidates: candidate_path = candidate.path if hasattr(candidate, 'path') else str(candidate) candidate_energy = self._get_sample_energy(candidate_path) energy_diff = abs(candidate_energy - target_energy) # Prefer samples within tolerance if energy_diff < tolerance and energy_diff < best_diff: best_match = candidate best_diff = energy_diff if best_match: if self.verbose: match_path = best_match.path if hasattr(best_match, 'path') else str(best_match) match_energy = self._get_sample_energy(match_path) logger.info(f"[VariationEngine] Found energy match: {Path(match_path).name} " f"(energy: {match_energy:.2f})") return best_match # Return original if no suitable variant found if self.verbose: logger.info(f"[VariationEngine] No energy variant found, using original") return self._get_sample_info(sample_path) def add_variation_element( self, section_kit: SectionKit, section_energy: float ) -> List[str]: """ Add appropriate FX or percussion elements based on section energy. High energy sections get: - Layered percussion - Impact FX - High-energy fills Building sections get: - Progressive elements - Risers/transitions Args: section_kit: Kit to add elements to section_energy: Energy level of the section Returns: List of element types added """ added = [] # High energy: Add layered elements if section_energy >= 0.8: # Add percussion layers perc_samples = self._get_samples_by_energy("perc", section_energy, count=2) for perc in perc_samples: section_kit.percussion.append(perc) if perc_samples: added.append(f"percussion_layers ({len(perc_samples)})") # Add impact FX fx_samples = self._get_samples_by_energy("fx", section_energy, count=1) for fx in fx_samples: section_kit.fx.append(fx) if fx_samples: added.append("impact_fx") # Building energy (0.6-0.8): Add risers/transitions elif section_energy >= 0.6: fx_samples = self._get_samples_by_energy("fx", section_energy, count=1) for fx in fx_samples: section_kit.fx.append(fx) if fx_samples: added.append("riser_fx") # Medium energy: Subtle variations elif section_energy >= 0.4: # Add subtle percussion for groove variation perc_samples = self._get_samples_by_energy("perc", section_energy, count=1) for perc in perc_samples: section_kit.percussion.append(perc) if perc_samples: added.append("subtle_perc") if self.verbose and added: logger.info(f"[VariationEngine] Added elements: {added}") return added def remove_elements_for_energy( self, section_kit: SectionKit, target_energy: float ) -> List[str]: """ Strip down kit elements for low-energy sections. Low energy sections (intro, outro, breakdown): - Remove reverb-heavy samples - Use dry, punchy samples - Reduce layering Args: section_kit: Kit to strip down target_energy: Target energy level Returns: List of element types removed """ removed = [] if target_energy >= 0.4: return removed # No removal needed # Very low energy: minimal kit if target_energy <= 0.25: # Keep only kick and minimal hats if section_kit.snare: section_kit.snare = None removed.append("snare") if section_kit.clap: section_kit.clap = None removed.append("clap") if section_kit.hat_open: section_kit.hat_open = None removed.append("hat_open") # Clear percussion and FX if section_kit.percussion: section_kit.percussion = [] removed.append("all_percussion") if section_kit.fx: section_kit.fx = [] removed.append("all_fx") # Reduce bass if len(section_kit.bass) > 1: section_kit.bass = section_kit.bass[:1] removed.append("extra_bass") # Low-medium energy: reduced kit elif target_energy < 0.4: # Remove open hats and some percussion if section_kit.hat_open: section_kit.hat_open = None removed.append("hat_open") if len(section_kit.percussion) > 1: section_kit.percussion = section_kit.percussion[:1] removed.append("extra_perc") if section_kit.fx: section_kit.fx = [] removed.append("all_fx") if self.verbose and removed: logger.info(f"[VariationEngine] Removed elements: {removed}") return removed def calculate_coherence( self, base_kit, section_kit: SectionKit ) -> CoherenceMetrics: """ Calculate coherence between base kit and section variation. Compares samples across multiple dimensions: - Timbre: Spectral characteristics similarity - Dynamics: Amplitude envelope similarity - Transient: Attack characteristics - Rhythmic: Pattern/timing similarity (for loops) Args: base_kit: Original kit section_kit: Evolved section kit Returns: CoherenceMetrics with detailed scores """ metrics = CoherenceMetrics() # Compare each component that exists in both kits comparisons = [] if hasattr(base_kit, 'drums') and base_kit.drums: base_drums = base_kit.drums if base_drums.kick and section_kit.kick: comparisons.append(self._compare_samples( base_drums.kick.path if hasattr(base_drums.kick, 'path') else str(base_drums.kick), section_kit.kick.path if hasattr(section_kit.kick, 'path') else str(section_kit.kick) )) if base_drums.snare and section_kit.snare: comparisons.append(self._compare_samples( base_drums.snare.path if hasattr(base_drums.snare, 'path') else str(base_drums.snare), section_kit.snare.path if hasattr(section_kit.snare, 'path') else str(section_kit.snare) )) if base_drums.hat_closed and section_kit.hat_closed: comparisons.append(self._compare_samples( base_drums.hat_closed.path if hasattr(base_drums.hat_closed, 'path') else str(base_drums.hat_closed), section_kit.hat_closed.path if hasattr(section_kit.hat_closed, 'path') else str(section_kit.hat_closed) )) # Calculate average scores across all comparisons if comparisons: metrics.timbre_score = sum(c.get('timbre', 0.5) for c in comparisons) / len(comparisons) metrics.dynamics_score = sum(c.get('dynamics', 0.5) for c in comparisons) / len(comparisons) metrics.transient_score = sum(c.get('transient', 0.5) for c in comparisons) / len(comparisons) metrics.rhythmic_score = sum(c.get('rhythmic', 0.5) for c in comparisons) / len(comparisons) else: # Default scores if no comparisons possible metrics.timbre_score = 0.85 metrics.dynamics_score = 0.85 metrics.transient_score = 0.85 metrics.rhythmic_score = 0.85 metrics.calculate_total() return metrics def get_coherence_report(self) -> Dict[str, Any]: """ Get comprehensive coherence report for all logged variations. Returns: Dict with coherence statistics and validation results """ if not self.coherence_log: return {"status": "no_variations", "total": 0} scores = [entry["coherence"] for entry in self.coherence_log] valid_count = sum(1 for s in scores if s >= self.COHERENCE_THRESHOLD) return { "status": "ok", "total_variations": len(self.coherence_log), "valid_coherence": valid_count, "failed_coherence": len(self.coherence_log) - valid_count, "average_coherence": round(sum(scores) / len(scores), 3), "min_coherence": round(min(scores), 3), "max_coherence": round(max(scores), 3), "threshold": self.COHERENCE_THRESHOLD, "sections": self.coherence_log, } # ========================================================================== # INTERNAL METHODS # ========================================================================== def _get_sample_energy(self, sample_path: str) -> float: """ Get energy characteristics for a sample. Uses metadata store if available, otherwise returns default. """ if sample_path in self._energy_cache: return self._energy_cache[sample_path].derived_energy characteristics = EnergyCharacteristics() # Try to get from metadata store if self.metadata_store: try: features = self.metadata_store.get_sample_features(sample_path) if features: characteristics.rms = features.rms or 0.0 characteristics.spectral_centroid = features.spectral_centroid or 0.0 characteristics.spectral_rolloff = features.spectral_rolloff or 0.0 characteristics.zero_crossing_rate = features.zero_crossing_rate or 0.0 except Exception as e: if self.verbose: logger.warning(f"[VariationEngine] Failed to get features: {e}") # Calculate energy score energy = characteristics.calculate_energy_score() self._energy_cache[sample_path] = characteristics return energy def _get_sample_info(self, sample_path: str) -> Any: """Get sample info object for a path.""" # Try to get from selector if self.selector: # Return a minimal SampleInfo-like object class MinimalSampleInfo: def __init__(self, path): self.path = path self.name = Path(path).name return MinimalSampleInfo(sample_path) # Return path string if no selector return sample_path def _find_similar_samples( self, sample_path: str, role: Optional[str] = None ) -> List[Any]: """ Find similar samples using selector or metadata store. """ candidates = [] # Try selector first if self.selector: try: if hasattr(self.selector, 'get_recommended_samples'): role = role or self._guess_role(sample_path) candidates = self.selector.get_recommended_samples( role=role, count=10 ) except Exception as e: if self.verbose: logger.warning(f"[VariationEngine] Selector failed: {e}") # Fallback to metadata store if not candidates and self.metadata_store: try: role = role or self._guess_role(sample_path) db_results = self.metadata_store.search_samples( category=role, limit=10 ) candidates = db_results except Exception as e: if self.verbose: logger.warning(f"[VariationEngine] Metadata store failed: {e}") return candidates def _get_samples_by_energy( self, role: str, target_energy: float, count: int = 3, tolerance: float = 0.15 ) -> List[Any]: """ Get samples matching target energy level. """ candidates = [] if self.selector and hasattr(self.selector, 'get_recommended_samples'): try: all_samples = self.selector.get_recommended_samples(role=role, count=20) # Filter by energy for sample in all_samples: sample_path = sample.path if hasattr(sample, 'path') else str(sample) energy = self._get_sample_energy(sample_path) if abs(energy - target_energy) <= tolerance: candidates.append(sample) if len(candidates) >= count: break except Exception as e: if self.verbose: logger.warning(f"[VariationEngine] Energy selection failed: {e}") return candidates[:count] def _compare_samples(self, path1: str, path2: str) -> Dict[str, float]: """ Compare two samples and return similarity scores. Uses audio features to calculate timbre, dynamics, and transient similarity. """ energy1 = self._get_sample_energy(path1) char1 = self._energy_cache.get(path1, EnergyCharacteristics()) energy2 = self._get_sample_energy(path2) char2 = self._energy_cache.get(path2, EnergyCharacteristics()) # Timbre similarity (based on spectral features) if char1.spectral_centroid and char2.spectral_centroid: centroid_sim = 1.0 - abs(char1.spectral_centroid - char2.spectral_centroid) / 8000 else: centroid_sim = 0.8 # Default if no data if char1.spectral_rolloff and char2.spectral_rolloff: rolloff_sim = 1.0 - abs(char1.spectral_rolloff - char2.spectral_rolloff) / 10000 else: rolloff_sim = 0.8 timbre_score = (centroid_sim + rolloff_sim) / 2 # Dynamics similarity (based on RMS) if char1.rms and char2.rms: rms_diff = abs(char1.rms - char2.rms) dynamics_score = max(0.0, 1.0 - (rms_diff / 20)) # 20dB difference = 0 similarity else: dynamics_score = 0.85 # Transient similarity (based on attack characteristics) if char1.attack_time and char2.attack_time: attack_sim = 1.0 - abs(char1.attack_time - char2.attack_time) / 0.1 else: attack_sim = 0.85 # Rhythmic similarity (placeholder - would need pattern analysis) rhythmic_score = 0.85 return { "timbre": max(0.0, min(1.0, timbre_score)), "dynamics": max(0.0, min(1.0, dynamics_score)), "transient": max(0.0, min(1.0, attack_sim)), "rhythmic": rhythmic_score, } def _determine_elements_for_energy(self, energy: float) -> Set[str]: """ Determine which kit elements should be present at given energy level. Returns: Set of element names to include """ # All elements present at medium energy and above if energy >= 0.5: return {"kick", "snare", "clap", "hat_closed", "hat_open", "bass"} # Reduced kit for low energy elif energy >= 0.25: return {"kick", "hat_closed", "bass"} # Minimal kit for very low energy else: return {"kick", "hat_closed"} def _guess_role(self, sample_path: str) -> str: """Guess sample role from filename/path.""" lower = sample_path.lower() if "kick" in lower: return "kick" elif "snare" in lower: return "snare" elif "clap" in lower: return "clap" elif "hat" in lower or "hihat" in lower: return "hat_closed" elif "bass" in lower: return "bass" elif "perc" in lower: return "perc" elif "fx" in lower: return "fx" return "unknown" def _log_coherence(self, section_name: str, coherence: CoherenceMetrics): """Log coherence score for a section variation.""" entry = { "section": section_name, "coherence": coherence.total_coherence, "is_valid": coherence.is_valid, "details": coherence.to_dict() } self.coherence_log.append(entry) if self.verbose: status = "✓" if coherence.is_valid else "✗" logger.info(f"[VariationEngine] {status} Coherence for '{section_name}': " f"{coherence.total_coherence:.2f}") # ============================================================================= # CONVENIENCE FUNCTIONS # ============================================================================= def evolve_kit_for_sections( base_kit, sections: List[str], selector=None, metadata_store=None, verbose: bool = False ) -> Dict[str, SectionKit]: """ Evolve a base kit for multiple sections. Convenience function to create section variations in one call. Args: base_kit: Base kit to evolve sections: List of section names (intro, verse, chorus, etc.) selector: SampleSelector instance metadata_store: MetadataStore instance verbose: Enable logging Returns: Dict mapping section names to SectionKit instances """ engine = VariationEngine( selector=selector, metadata_store=metadata_store, verbose=verbose ) result = {} for section in sections: try: section_kit = engine.evolve_kit_for_section(base_kit, section) result[section] = section_kit except ValueError as e: logger.error(f"[evolve_kit_for_sections] Failed for {section}: {e}") return result def get_section_energy_profile(section_name: str) -> Optional[Dict[str, Any]]: """ Get energy profile for a section type. Args: section_name: Section name (intro, verse, chorus, etc.) Returns: Dict with energy level and description, or None if unknown """ return SECTION_PROFILES.get(section_name) def validate_coherence( base_kit, section_kit: SectionKit, threshold: float = 0.80 ) -> Tuple[bool, float]: """ Validate coherence between base kit and section variation. Args: base_kit: Original kit section_kit: Section variation threshold: Minimum coherence required Returns: Tuple of (is_valid, coherence_score) """ engine = VariationEngine() metrics = engine.calculate_coherence(base_kit, section_kit) return metrics.is_valid, metrics.total_coherence # ============================================================================= # MODULE EXPORTS # ============================================================================= __all__ = [ # Core class "VariationEngine", # Data classes "SectionKit", "EnergyCharacteristics", "CoherenceMetrics", # Constants "SECTION_PROFILES", # Functions "evolve_kit_for_sections", "get_section_energy_profile", "validate_coherence", ]