5ce8187c65
- Add _cmd_create_arrangement_audio_pattern with 5-method fallback chain - Method 1: track.insert_arrangement_clip() [Live 12+] - Method 2: track.create_audio_clip() [Live 11+] - Method 3: arrangement_clips.add_new_clip() [Live 12+] - Method 4: Session->duplicate_clip_to_arrangement [Legacy] - Method 5: Session->Recording [Universal] - Add _cmd_duplicate_clip_to_arrangement for session-to-arrangement workflow - Update skills documentation - Verified: 3 clips created at positions [0, 4, 8] in Arrangement View Closes: Audio injection in Arrangement View
1014 lines
37 KiB
Python
1014 lines
37 KiB
Python
"""
|
|
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",
|
|
]
|