"""AbletonMCP_AI Engines Package - Architecture Integration Module. This module wires together all the engines and provides a unified interface for the MCP server to access music production capabilities. Architecture Overview: ====================== ┌─────────────────────────────────────────────────────────────────────────────┐ │ ENGINES PACKAGE │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │ │ Metadata & │ │ Arrangement │ │ Live Bridge │ │ │ │ Analysis │ │ Recording │ │ │ │ │ │ │ │ │ │ │ │ │ │ • metadata_ │ │ • arrangement_ │ │ • live_bridge │ │ │ │ store │ │ recorder │ │ │ │ │ │ • abstract_ │ │ │ │ (Connection │ │ │ │ analyzer │ │ │ │ management) │ │ │ │ │ │ │ │ │ │ │ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │ │ │ │ │ │ │ └────────────────────┼────────────────────┘ │ │ ▼ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ SENIOR ARCHITECTURE - INTELLIGENT SELECTION │ │ │ │ │ │ │ │ • intelligent_selector - Coherent kit selection with validation │ │ │ │ • coherence_scorer - Spectral coherence quality scoring │ │ │ │ • variation_engine - Section-based kit evolution │ │ │ │ • rationale_logger - Selection decision logging │ │ │ │ • preset_manager - Kit preset save/load │ │ │ │ • iteration_engine - Auto-iteration to professional grade │ │ │ │ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ ▼ │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ EXISTING ENGINES (Sprints 1-3) │ │ │ │ │ │ │ │ Sprint 1: Core Analysis Sprint 2: Pattern & Mixing │ │ │ │ • libreria_analyzer • pattern_library │ │ │ │ • embedding_engine • song_generator │ │ │ │ • reference_matcher • mixing_engine │ │ │ │ • sample_selector • workflow_engine │ │ │ │ │ │ │ │ Sprint 3: Arrangement & Harmony │ │ │ │ • arrangement_engine • harmony_engine │ │ │ │ • preset_system • production_workflow │ │ │ │ • musical_intelligence │ │ │ │ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ Key Design Principles: │ │ ─────────────────────── │ │ • Lazy loading: All numpy/librosa-dependent modules load only when used │ │ • Graceful degradation: Missing dependencies don't break the system │ │ • Singleton pattern: Shared state via init_*() functions │ │ • Compatibility layer: Maintains backward compatibility │ │ │ └─────────────────────────────────────────────────────────────────────────────┘ Usage: ====== # Quick initialization from AbletonMCP_AI.mcp_server.engines import ( init_metadata_store, init_arrangement_recorder, init_live_bridge, get_analyzer, ) # Initialize systems store = init_metadata_store() analyzer = get_analyzer(prefer_database=True) recorder = init_arrangement_recorder(song, connection) bridge = init_live_bridge(song, connection) # Initialize Intelligent Selection System from AbletonMCP_AI.mcp_server.engines import ( init_intelligent_selector, init_coherence_scorer, init_variation_engine, init_rationale_logger, init_preset_manager, init_iteration_engine, ) scorer = init_coherence_scorer() selector = init_intelligent_selector(coherence_scorer=scorer) variation = init_variation_engine() rationale = init_rationale_logger(session_id="session_001") presets = init_preset_manager() iterator = init_iteration_engine(intelligent_selector=selector, max_iterations=10) # Use existing engines (backward compatible) from AbletonMCP_AI.mcp_server.engines import ( SampleSelector, EmbeddingEngine, ReggaetonGenerator, MixingEngine, ArrangementBuilder ) Capabilities Detection: ======================= The system auto-detects available capabilities: - numpy: Required for numerical analysis - librosa: Required for audio feature extraction - sqlite3: Required for metadata database - python_version: For compatibility checks Engines gracefully degrade when dependencies are missing. Version: 1.0.0 """ from __future__ import annotations import sys import logging from typing import TYPE_CHECKING, Optional, Dict, Any from enum import Enum # Configure logging for the engines package logger = logging.getLogger(__name__) # ============================================================================= # LAZY IMPORT SYSTEM # ============================================================================= # This allows the module to be imported even when optional dependencies # (numpy, librosa) are not available. The heavy modules are only loaded # when their functionality is actually accessed. class LazyModule: """Lazy module loader that imports only when accessed.""" def __init__(self, name: str, import_path: str, fallback=None): self.name = name self.import_path = import_path self._module = None self._fallback = fallback self._error = None def _import(self): if self._module is None and self._error is None: try: parts = self.import_path.split('.') module = __import__(self.import_path, fromlist=[parts[-1]]) self._module = module logger.debug(f"Lazy loaded module: {self.import_path}") except ImportError as e: self._error = e logger.warning(f"Could not import {self.import_path}: {e}") if self._fallback: self._module = self._fallback return self._module def __getattr__(self, name: str): module = self._import() if module is None: raise ImportError( f"Module {self.import_path} not available. " f"Original error: {self._error}" ) return getattr(module, name) def __call__(self, *args, **kwargs): # If this represents a class/function, allow calling callable_ = self._import() if callable_ is None: raise ImportError( f"Module {self.import_path} not available. " f"Original error: {self._error}" ) return callable_(*args, **kwargs) def is_available(self) -> bool: """Check if the lazy-loaded module is available.""" return self._import() is not None # ============================================================================= # MODULE AVAILABILITY TRACKING # ============================================================================= class ModuleAvailability(Enum): """Tracks which modules are available.""" AVAILABLE = "available" MISSING = "missing" DEGRADED = "degraded" # Available but with limited functionality # Global registry of module availability _module_availability: Dict[str, ModuleAvailability] = {} def _mark_available(name: str, status: ModuleAvailability = ModuleAvailability.AVAILABLE): """Mark a module as available.""" _module_availability[name] = status def _mark_missing(name: str): """Mark a module as missing.""" _module_availability[name] = ModuleAvailability.MISSING def is_module_available(name: str) -> bool: """Check if a module is available.""" return _module_availability.get(name) == ModuleAvailability.AVAILABLE # ============================================================================= # EXISTING ENGINES (Sprints 1-3) - Always Available # ============================================================================= # Sprint 1: Core Analysis from .libreria_analyzer import LibreriaAnalyzer, analyze_library _mark_available("libreria_analyzer") from .embedding_engine import ( EmbeddingEngine, create_embeddings_index, find_similar_samples, find_samples_like_audio ) _mark_available("embedding_engine") from .reference_matcher import ( ReferenceMatcher, SpectralFingerprint, SampleMatch, UserSoundProfile, AudioAnalyzer, SimilarityEngine, get_matcher, get_user_profile, get_recommended_samples, analyze_reference, refresh_profile, ) _mark_available("reference_matcher") from .sample_selector import ( SampleSelector, SampleInfo, DrumKit, InstrumentGroup, get_selector, select_samples_for_track, get_drum_kit, reset_cross_generation_memory, ) _mark_available("sample_selector") # Sprint 2: Pattern & Mixing from .pattern_library import ( DembowPatterns, BassPatterns, ChordProgressions, MelodyGenerator, HumanFeel, PercussionLibrary, NoteEvent, ScaleType, get_patterns, ) _mark_available("pattern_library") from .song_generator import ( ReggaetonGenerator, SongGenerator, SongConfig, Section, TrackConfig, ClipConfig, Pattern, DeviceConfig, get_song_generator, generate_song, generate_from_reference, get_supported_styles, get_supported_structures, ) _mark_available("song_generator") from .mixing_engine import ( MixingEngine, BusManager, ReturnTrackManager, MixConfiguration, get_mixing_engine, reset_mixing_engine, DeviceManager, EQConfiguration, CompressionSettings, GainStaging, MasterChain, DeviceParameter, MixQualityChecker, DeviceInfo, QualityReport, SUPPORTED_DEVICES, EQ_PRESETS, COMP_PRESETS, MASTER_PRESETS, get_device_manager, get_eq_configuration, get_compression_settings, get_gain_staging, get_master_chain, get_device_parameter, get_quality_checker, create_standard_buses, apply_send_preset, ) _mark_available("mixing_engine") from .workflow_engine import ( ProductionWorkflow, ActionHistory, ProjectValidator, ExportManager, ActionRecord, ValidationIssue, get_workflow, ) _mark_available("workflow_engine") # Sprint 3: Arrangement & Harmony from .arrangement_engine import ( ArrangementBuilder, AutomationEngine, FXCreator, SampleProcessor, ArrangementConfig, SectionMarker, AutomationPoint, AutomationEnvelope, ArrangementClip, ArrangementSection, arrangement_to_dict, dict_to_arrangement, get_arrangement_length, create_full_arrangement, ) _mark_available("arrangement_engine") from .harmony_engine import ( ProjectAnalyzer, CounterMelodyGenerator, VariationEngine, SampleIntelligence, ) _mark_available("harmony_engine") from .preset_system import ( PresetManager, Preset, TrackPreset, MixingConfig, SampleSelectionCriteria, get_preset_manager, apply_preset_to_project, get_default_preset, list_available_presets, quick_apply_preset, create_builtin_presets, ) _mark_available("preset_system") # Sprint 4: Iteration & Quality Assurance from .iteration_engine import ( IterationEngine, ProfessionalCoherenceError, CoherenceScorer, RationaleLogger, IterationResult, IterationAttempt, IterationStatus, ITERATION_STRATEGIES, iterate_for_coherence, quick_coherence_check, ) _mark_available("iteration_engine") # Optional engines that might not exist in all installations optional_engines = [ ("musical_intelligence", "MusicalIntelligence, PhraseAnalyzer, MotifLibrary"), ("production_workflow", "ProductionOrchestrator"), ] for engine_name, exports in optional_engines: try: exec(f"from .{engine_name} import {exports}") _mark_available(engine_name) except ImportError: _mark_missing(engine_name) logger.debug(f"Optional engine {engine_name} not available") # ============================================================================= # NEW COMPONENTS (With Graceful Fallback) # ============================================================================= # Metadata and Analysis _metadata_store_loaded = False try: from .metadata_store import SampleMetadataStore, SampleFeatures _metadata_store_loaded = True _mark_available("metadata_store") except ImportError as e: _mark_missing("metadata_store") logger.debug(f"metadata_store not available: {e}") # Define placeholder classes class SampleMetadataStore: """Placeholder - metadata_store module not available.""" def __init__(self, *args, **kwargs): raise ImportError("metadata_store module not available") class SampleFeatures: """Placeholder - metadata_store module not available.""" pass # Abstract Analyzer - uses lazy loading for librosa-dependent components _abstract_analyzer_loaded = False try: from .abstract_analyzer import ( FeatureExtractor, LibrosaExtractor, DatabaseExtractor, HybridExtractor ) _abstract_analyzer_loaded = True _mark_available("abstract_analyzer") except ImportError as e: _mark_missing("abstract_analyzer") logger.debug(f"abstract_analyzer not available: {e}") # Define placeholder base classes class FeatureExtractor: """Placeholder base class for feature extraction.""" def extract(self, sample_path: str) -> Dict[str, Any]: raise NotImplementedError("FeatureExtractor not available") class LibrosaExtractor(FeatureExtractor): """Placeholder - librosa-based extraction not available.""" def __init__(self, *args, **kwargs): raise ImportError("LibrosaExtractor requires librosa and numpy") class DatabaseExtractor(FeatureExtractor): """Placeholder - database extraction not available.""" def __init__(self, *args, **kwargs): raise ImportError("DatabaseExtractor requires metadata_store") class HybridExtractor(FeatureExtractor): """Placeholder - hybrid extraction not available.""" def __init__(self, *args, **kwargs): raise ImportError("HybridExtractor requires abstract_analyzer module") # Arrangement Recording _arrangement_recorder_loaded = False try: from .arrangement_recorder import ( ArrangementRecorder, RecordingState, RecordingConfig ) _arrangement_recorder_loaded = True _mark_available("arrangement_recorder") except ImportError as e: _mark_missing("arrangement_recorder") logger.debug(f"arrangement_recorder not available: {e}") from enum import Enum class RecordingState(Enum): """Placeholder enum for recording state.""" IDLE = "idle" RECORDING = "recording" PAUSED = "paused" ERROR = "error" class RecordingConfig: """Placeholder configuration class.""" def __init__(self, *args, **kwargs): pass class ArrangementRecorder: """Placeholder - arrangement_recorder module not available.""" def __init__(self, *args, **kwargs): raise ImportError("arrangement_recorder module not available") @property def state(self) -> RecordingState: return RecordingState.IDLE # Live Bridge _live_bridge_loaded = False try: from .live_bridge import AbletonLiveBridge _live_bridge_loaded = True _mark_available("live_bridge") except ImportError as e: _mark_missing("live_bridge") logger.debug(f"live_bridge not available: {e}") class AbletonLiveBridge: """Placeholder - live_bridge module not available.""" def __init__(self, *args, **kwargs): raise ImportError("live_bridge module not available") # ============================================================================= # SENIOR ARCHITECTURE - INTELLIGENT SELECTION SYSTEM # ============================================================================= # Intelligent Sample Selector _intelligent_selector_loaded = False try: from .intelligent_selector import ( IntelligentSampleSelector, CoherenceError, select_coherent_kit, find_similar_samples as intelligent_find_similar ) _intelligent_selector_loaded = True _mark_available("intelligent_selector") except ImportError as e: _mark_missing("intelligent_selector") logger.debug(f"intelligent_selector not available: {e}") class IntelligentSampleSelector: """Placeholder - intelligent_selector module not available.""" def __init__(self, *args, **kwargs): raise ImportError("intelligent_selector module not available") class CoherenceError(Exception): """Placeholder - raised when samples lack coherence.""" pass def select_coherent_kit(*args, **kwargs): raise ImportError("intelligent_selector module not available") def intelligent_find_similar(*args, **kwargs): raise ImportError("intelligent_selector module not available") # Coherence Scorer _coherence_scorer_loaded = False try: from .coherence_scorer import ( CoherenceScorer, score_kit_coherence, is_professional_grade, MIN_COHERENCE ) _coherence_scorer_loaded = True _mark_available("coherence_scorer") except ImportError as e: _mark_missing("coherence_scorer") logger.debug(f"coherence_scorer not available: {e}") class CoherenceScorer: """Placeholder - coherence_scorer module not available.""" def __init__(self, *args, **kwargs): raise ImportError("coherence_scorer module not available") def score_kit_coherence(*args, **kwargs): raise ImportError("coherence_scorer module not available") def is_professional_grade(*args, **kwargs): return False MIN_COHERENCE = 0.7 # Variation Engine (Sample Kit Evolution) _variation_engine_loaded = False try: from .variation_engine import ( VariationEngine, SectionKit, EnergyCharacteristics, CoherenceMetrics, SECTION_PROFILES, evolve_kit_for_sections, get_section_energy_profile, validate_coherence, ) _variation_engine_loaded = True _mark_available("variation_engine") except ImportError as e: _mark_missing("variation_engine") logger.debug(f"variation_engine not available: {e}") class VariationEngine: """Placeholder - variation_engine module not available.""" def __init__(self, *args, **kwargs): raise ImportError("variation_engine module not available") class SectionKit: """Placeholder - variation_engine module not available.""" pass class EnergyCharacteristics: """Placeholder - variation_engine module not available.""" pass class CoherenceMetrics: """Placeholder - variation_engine module not available.""" pass SECTION_PROFILES = {} def evolve_kit_for_sections(*args, **kwargs): raise ImportError("variation_engine module not available") def get_section_energy_profile(*args, **kwargs): raise ImportError("variation_engine module not available") def validate_coherence(*args, **kwargs): raise ImportError("variation_engine module not available") # Rationale Logger _rationale_logger_loaded = False try: from .rationale_logger import ( RationaleLogger, log_sample_selection, log_kit_assembly, get_session_rationale ) _rationale_logger_loaded = True _mark_available("rationale_logger") except ImportError as e: _mark_missing("rationale_logger") logger.debug(f"rationale_logger not available: {e}") class RationaleLogger: """Placeholder - rationale_logger module not available.""" def __init__(self, *args, **kwargs): raise ImportError("rationale_logger module not available") def log_sample_selection(*args, **kwargs): pass def log_kit_assembly(*args, **kwargs): pass def get_session_rationale(*args, **kwargs): return {} # Preset Manager _preset_manager_loaded = False try: from .preset_manager import ( PresetManager, save_kit_preset, load_kit_preset, list_presets, KitPreset ) _preset_manager_loaded = True _mark_available("preset_manager") except ImportError as e: _mark_missing("preset_manager") logger.debug(f"preset_manager not available: {e}") class PresetManager: """Placeholder - preset_manager module not available.""" def __init__(self, *args, **kwargs): raise ImportError("preset_manager module not available") class KitPreset: """Placeholder - preset_manager module not available.""" pass def save_kit_preset(*args, **kwargs): raise ImportError("preset_manager module not available") def load_kit_preset(*args, **kwargs): raise ImportError("preset_manager module not available") def list_presets(*args, **kwargs): return [] # Iteration Engine _iteration_engine_loaded = False try: from .iteration_engine import ( IterationEngine, iterate_until_coherence, ProfessionalCoherenceError, ITERATION_STRATEGIES ) _iteration_engine_loaded = True _mark_available("iteration_engine") except ImportError as e: _mark_missing("iteration_engine") logger.debug(f"iteration_engine not available: {e}") class IterationEngine: """Placeholder - iteration_engine module not available.""" def __init__(self, *args, **kwargs): raise ImportError("iteration_engine module not available") class ProfessionalCoherenceError(Exception): """Placeholder - raised when professional coherence cannot be achieved.""" pass def iterate_until_coherence(*args, **kwargs): raise ImportError("iteration_engine module not available") ITERATION_STRATEGIES = [] # ============================================================================= # SENIOR ARCHITECTURE - NEW MODULES (Coherence System, Audio Analyzer Dual, Bus Architecture) # ============================================================================= # Coherence System coherence_system_loaded = False try: # Try relative import first (for normal Python usage) from .coherence_system import ( calculate_joint_score, update_cross_generation_memory, get_cross_generation_penalty, get_persistent_fatigue, ROLE_ACTIVITY, SECTION_DENSITY_PROFILES, get_section_role_bonus, calculate_section_appropriateness, set_palette_lock, calculate_palette_bonus, get_palette_coherence_score, calculate_comprehensive_coherence, reset_all_memory, get_coherence_memory_stats ) coherence_system_loaded = True _mark_available("coherence_system") except ImportError: # Fallback to absolute import (for Ableton runtime) try: from coherence_system import ( calculate_joint_score, update_cross_generation_memory, get_cross_generation_penalty, get_persistent_fatigue, ROLE_ACTIVITY, SECTION_DENSITY_PROFILES, get_section_role_bonus, calculate_section_appropriateness, set_palette_lock, calculate_palette_bonus, get_palette_coherence_score, calculate_comprehensive_coherence, reset_all_memory, get_coherence_memory_stats ) coherence_system_loaded = True _mark_available("coherence_system") except ImportError as e: _mark_missing("coherence_system") logger.debug(f"coherence_system not available: {e}") # Define placeholder functions def calculate_joint_score(*args, **kwargs): raise ImportError("coherence_system module not available") def update_cross_generation_memory(*args, **kwargs): pass def get_cross_generation_penalty(*args, **kwargs): return 0.0 def get_persistent_fatigue(*args, **kwargs): return 0.0 ROLE_ACTIVITY = {} SECTION_DENSITY_PROFILES = {} def get_section_role_bonus(*args, **kwargs): return 0.0 def calculate_section_appropriateness(*args, **kwargs): return 0.5 def set_palette_lock(*args, **kwargs): pass def calculate_palette_bonus(*args, **kwargs): return 0.0 def get_palette_coherence_score(*args, **kwargs): return 0.0 def calculate_comprehensive_coherence(*args, **kwargs): return 0.7 def reset_all_memory(): pass def get_coherence_memory_stats(): return {} # Audio Analyzer Dual audio_analyzer_dual_loaded = False try: # Try relative import first (for normal Python usage) from .audio_analyzer_dual import ( AudioAnalyzerDual, AudioFeatures, analyze_sample, analyze_audio, get_backend_info ) audio_analyzer_dual_loaded = True _mark_available("audio_analyzer_dual") except ImportError: # Fallback to absolute import (for Ableton runtime) try: from audio_analyzer_dual import ( AudioAnalyzerDual, AudioFeatures, analyze_sample, analyze_audio, get_backend_info ) audio_analyzer_dual_loaded = True _mark_available("audio_analyzer_dual") except ImportError as e: _mark_missing("audio_analyzer_dual") logger.debug(f"audio_analyzer_dual not available: {e}") class AudioAnalyzerDual: """Placeholder - audio_analyzer_dual module not available.""" def __init__(self, *args, **kwargs): raise ImportError("audio_analyzer_dual module not available") class AudioFeatures: """Placeholder - audio_analyzer_dual module not available.""" def __init__(self, *args, **kwargs): raise ImportError("audio_analyzer_dual module not available") def analyze_sample(*args, **kwargs): raise ImportError("audio_analyzer_dual module not available") def analyze_audio(*args, **kwargs): raise ImportError("audio_analyzer_dual module not available") def get_backend_info(): return {"available": False, "backend": "none"} # Bus Architecture bus_architecture_loaded = False try: # Try relative import first (for normal Python usage) from .bus_architecture import ( BUS_GAIN_CALIBRATION, RETURN_CONFIG, ROLE_MIX, MASTER_CHAIN, BusArchitecture, create_bus_track, create_return_track, route_track_to_bus, set_track_send, configure_bus_gain, configure_return_effect, apply_role_mix, configure_master_chain, apply_professional_mix, get_bus_config, get_return_config, get_role_mix, list_available_buses, list_available_returns, list_available_roles ) bus_architecture_loaded = True _mark_available("bus_architecture") except ImportError: # Fallback to absolute import (for Ableton runtime) try: from bus_architecture import ( BUS_GAIN_CALIBRATION, RETURN_CONFIG, ROLE_MIX, MASTER_CHAIN, BusArchitecture, create_bus_track, create_return_track, route_track_to_bus, set_track_send, configure_bus_gain, configure_return_effect, apply_role_mix, configure_master_chain, apply_professional_mix, get_bus_config, get_return_config, get_role_mix, list_available_buses, list_available_returns, list_available_roles ) bus_architecture_loaded = True _mark_available("bus_architecture") except ImportError as e: _mark_missing("bus_architecture") logger.debug(f"bus_architecture not available: {e}") BUS_GAIN_CALIBRATION = {} RETURN_CONFIG = {} ROLE_MIX = {} MASTER_CHAIN = {} class BusArchitecture: """Placeholder - bus_architecture module not available.""" def __init__(self, *args, **kwargs): raise ImportError("bus_architecture module not available") def create_bus_track(*args, **kwargs): raise ImportError("bus_architecture module not available") def create_return_track(*args, **kwargs): raise ImportError("bus_architecture module not available") def route_track_to_bus(*args, **kwargs): raise ImportError("bus_architecture module not available") def set_track_send(*args, **kwargs): raise ImportError("bus_architecture module not available") def configure_bus_gain(*args, **kwargs): raise ImportError("bus_architecture module not available") def configure_return_effect(*args, **kwargs): raise ImportError("bus_architecture module not available") def apply_role_mix(*args, **kwargs): raise ImportError("bus_architecture module not available") def configure_master_chain(*args, **kwargs): raise ImportError("bus_architecture module not available") def apply_professional_mix(*args, **kwargs): raise ImportError("bus_architecture module not available") def get_bus_config(): return {} def get_return_config(): return {} def get_role_mix(): return {} def list_available_buses(): return [] def list_available_returns(): return [] def list_available_roles(): return [] # ============================================================================= # INITIALIZATION FUNCTIONS # ============================================================================= # Singleton storage for initialized components _initialized_components: Dict[str, Any] = {} def init_metadata_store(db_path: Optional[str] = None) -> SampleMetadataStore: """ Initialize and return metadata store singleton. Args: db_path: Optional path to SQLite database. Uses default if None. Returns: SampleMetadataStore instance (cached singleton) Raises: ImportError: If metadata_store module is not available """ if "metadata_store" not in _initialized_components: if not _metadata_store_loaded: raise ImportError( "metadata_store module not available. " "Ensure metadata_store.py is present in engines/" ) store = SampleMetadataStore(db_path=db_path) _initialized_components["metadata_store"] = store logger.info(f"Initialized metadata store with db: {db_path or 'default'}") return _initialized_components["metadata_store"] def init_hybrid_extractor(db_path: Optional[str] = None) -> HybridExtractor: """ Initialize hybrid extractor with database + optional librosa. This creates a HybridExtractor that combines database-backed metadata with runtime librosa analysis when available. Args: db_path: Optional path to metadata database Returns: HybridExtractor instance Raises: ImportError: If abstract_analyzer module is not available """ if "hybrid_extractor" not in _initialized_components: if not _abstract_analyzer_loaded: raise ImportError( "abstract_analyzer module not available. " "Cannot create hybrid extractor." ) # Initialize with database extractor store = init_metadata_store(db_path) if _metadata_store_loaded else None db_extractor = DatabaseExtractor(store) if store else None # Try to add librosa extractor if available librosa_extractor = None try: librosa_extractor = LibrosaExtractor() logger.info("Librosa extractor initialized") except ImportError: logger.warning("Librosa not available - hybrid extractor will use database only") # Create hybrid hybrid = HybridExtractor( database_extractor=db_extractor, librosa_extractor=librosa_extractor ) _initialized_components["hybrid_extractor"] = hybrid logger.info("Initialized hybrid extractor") return _initialized_components["hybrid_extractor"] def init_arrangement_recorder(song, connection) -> ArrangementRecorder: """ Initialize arrangement recorder. Args: song: Ableton Live song object connection: TCP connection to Ableton Returns: ArrangementRecorder instance Raises: ImportError: If arrangement_recorder module is not available """ if "arrangement_recorder" not in _initialized_components: if not _arrangement_recorder_loaded: raise ImportError( "arrangement_recorder module not available. " "Ensure arrangement_recorder.py is present in engines/" ) recorder = ArrangementRecorder(song=song, connection=connection) _initialized_components["arrangement_recorder"] = recorder logger.info("Initialized arrangement recorder") return _initialized_components["arrangement_recorder"] def init_live_bridge(song, connection) -> AbletonLiveBridge: """ Initialize Live bridge for direct API access. Args: song: Ableton Live song object connection: TCP connection to Ableton Returns: AbletonLiveBridge instance Raises: ImportError: If live_bridge module is not available """ if "live_bridge" not in _initialized_components: if not _live_bridge_loaded: raise ImportError( "live_bridge module not available. " "Ensure live_bridge.py is present in engines/" ) bridge = AbletonLiveBridge(song=song, connection=connection) _initialized_components["live_bridge"] = bridge logger.info("Initialized Live bridge") return _initialized_components["live_bridge"] def init_intelligent_selector( coherence_scorer=None, variation_engine=None, rationale_logger=None ) -> IntelligentSampleSelector: """ Initialize intelligent sample selector with optional dependencies. Args: coherence_scorer: Optional CoherenceScorer instance variation_engine: Optional VariationEngine instance rationale_logger: Optional RationaleLogger instance Returns: IntelligentSampleSelector instance Raises: ImportError: If intelligent_selector module is not available """ if "intelligent_selector" not in _initialized_components: if not _intelligent_selector_loaded: raise ImportError( "intelligent_selector module not available. " "Ensure intelligent_selector.py is present in engines/" ) # Initialize dependencies if not provided if coherence_scorer is None and _coherence_scorer_loaded: coherence_scorer = init_coherence_scorer() if variation_engine is None and _variation_engine_loaded: variation_engine = init_variation_engine() if rationale_logger is None and _rationale_logger_loaded: rationale_logger = init_rationale_logger() selector = IntelligentSampleSelector( coherence_scorer=coherence_scorer, variation_engine=variation_engine, rationale_logger=rationale_logger ) _initialized_components["intelligent_selector"] = selector logger.info("Initialized intelligent sample selector") return _initialized_components["intelligent_selector"] def init_coherence_scorer() -> CoherenceScorer: """ Initialize coherence scorer for kit quality evaluation. Returns: CoherenceScorer instance Raises: ImportError: If coherence_scorer module is not available """ if "coherence_scorer" not in _initialized_components: if not _coherence_scorer_loaded: raise ImportError( "coherence_scorer module not available. " "Ensure coherence_scorer.py is present in engines/" ) scorer = CoherenceScorer() _initialized_components["coherence_scorer"] = scorer logger.info("Initialized coherence scorer") return _initialized_components["coherence_scorer"] def init_variation_engine() -> VariationEngine: """ Initialize variation engine for section-based kit evolution. Returns: VariationEngine instance Raises: ImportError: If variation_engine module is not available """ if "variation_engine" not in _initialized_components: if not _variation_engine_loaded: raise ImportError( "variation_engine module not available. " "Ensure variation_engine.py is present in engines/" ) engine = VariationEngine() _initialized_components["variation_engine"] = engine logger.info("Initialized variation engine") return _initialized_components["variation_engine"] def init_rationale_logger(session_id: Optional[str] = None) -> RationaleLogger: """ Initialize rationale logger for tracking selection decisions. Args: session_id: Optional session identifier for grouping logs Returns: RationaleLogger instance Raises: ImportError: If rationale_logger module is not available """ if "rationale_logger" not in _initialized_components: if not _rationale_logger_loaded: raise ImportError( "rationale_logger module not available. " "Ensure rationale_logger.py is present in engines/" ) logger_instance = RationaleLogger(session_id=session_id) _initialized_components["rationale_logger"] = logger_instance logger.info(f"Initialized rationale logger (session: {session_id or 'auto'})") return _initialized_components["rationale_logger"] def init_preset_manager(presets_dir: Optional[str] = None) -> PresetManager: """ Initialize preset manager for kit preset operations. Args: presets_dir: Optional directory path for storing presets Returns: PresetManager instance Raises: ImportError: If preset_manager module is not available """ if "preset_manager" not in _initialized_components: if not _preset_manager_loaded: raise ImportError( "preset_manager module not available. " "Ensure preset_manager.py is present in engines/" ) manager = PresetManager(presets_dir=presets_dir) _initialized_components["preset_manager"] = manager logger.info(f"Initialized preset manager (dir: {presets_dir or 'default'})") return _initialized_components["preset_manager"] def init_iteration_engine( intelligent_selector=None, coherence_scorer=None, max_iterations: int = 10 ) -> IterationEngine: """ Initialize iteration engine for coherence-based iteration. Args: intelligent_selector: Optional IntelligentSampleSelector instance coherence_scorer: Optional CoherenceScorer instance max_iterations: Maximum number of iteration attempts Returns: IterationEngine instance Raises: ImportError: If iteration_engine module is not available """ if "iteration_engine" not in _initialized_components: if not _iteration_engine_loaded: raise ImportError( "iteration_engine module not available. " "Ensure iteration_engine.py is present in engines/" ) # Initialize dependencies if not provided if intelligent_selector is None and _intelligent_selector_loaded: intelligent_selector = init_intelligent_selector(coherence_scorer=coherence_scorer) if coherence_scorer is None and _coherence_scorer_loaded: coherence_scorer = init_coherence_scorer() engine = IterationEngine( selector=intelligent_selector, scorer=coherence_scorer, max_iterations=max_iterations ) _initialized_components["iteration_engine"] = engine logger.info(f"Initialized iteration engine (max_iterations: {max_iterations})") return _initialized_components["iteration_engine"] def clear_initialized_components(): """Clear all initialized component singletons. Useful for testing.""" _initialized_components.clear() logger.info("Cleared all initialized component singletons") # ============================================================================= # CONFIGURATION DETECTION # ============================================================================= def get_system_capabilities() -> Dict[str, Any]: """ Detect available capabilities (numpy, librosa, etc.). Returns a dictionary indicating what's available in the current Python environment. This is used to auto-configure engines. Returns: Dict with keys: - numpy: bool - numpy available - librosa: bool - librosa available - sqlite3: bool - sqlite3 available - python_version: str - Python version - modules: Dict[str, str] - status of each engine module - has_advanced_analysis: bool - can do audio analysis - has_metadata_db: bool - can use metadata database """ capabilities = { "numpy": False, "librosa": False, "sqlite3": False, "python_version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", "modules": {}, "has_advanced_analysis": False, "has_metadata_db": False, } # Check numpy try: import numpy capabilities["numpy"] = True capabilities["numpy_version"] = numpy.__version__ except ImportError: pass # Check librosa try: import librosa capabilities["librosa"] = True capabilities["librosa_version"] = librosa.__version__ capabilities["has_advanced_analysis"] = True except ImportError: pass # Check sqlite3 try: import sqlite3 capabilities["sqlite3"] = True capabilities["has_metadata_db"] = True except ImportError: pass # Check all engine modules for name, status in _module_availability.items(): capabilities["modules"][name] = status.value # Check for new components specifically new_components = [ "metadata_store", "abstract_analyzer", "arrangement_recorder", "live_bridge" ] for component in new_components: if component not in capabilities["modules"]: capabilities["modules"][component] = "not_loaded" # Check for Senior Architecture - Intelligent Selection System components intelligent_selection_components = [ "intelligent_selector", "coherence_scorer", "variation_engine", "rationale_logger", "preset_manager", "iteration_engine" ] for component in intelligent_selection_components: if component not in capabilities["modules"]: capabilities["modules"][component] = "not_loaded" # Intelligent Selection System ready if all components available capabilities["has_intelligent_selection"] = all( capabilities["modules"].get(c) == "available" for c in intelligent_selection_components ) return capabilities def configure_engines_for_capabilities(): """ Auto-configure engines based on available dependencies. This function sets up the engine configuration to work optimally with whatever dependencies are available. Call this at startup to ensure engines are properly configured. Side effects: - Sets global configuration flags - Logs configuration decisions - May initialize singletons if needed """ capabilities = get_system_capabilities() logger.info("Configuring engines for system capabilities...") logger.info(f"Python version: {capabilities['python_version']}") logger.info(f"Numpy: {capabilities['numpy']}") logger.info(f"Librosa: {capabilities['librosa']}") logger.info(f"SQLite3: {capabilities['sqlite3']}") # Configure based on available modules if capabilities["has_advanced_analysis"]: logger.info("Advanced analysis available - enabling full audio processing") # Could set global flags here else: logger.warning("Advanced analysis not available - using database-only mode") if capabilities["has_metadata_db"]: logger.info("Metadata database available") # Could auto-initialize metadata store here # Log module availability available = [name for name, status in capabilities["modules"].items() if status == "available"] missing = [name for name, status in capabilities["modules"].items() if status == "missing"] logger.info(f"Available modules: {', '.join(available)}") if missing: logger.warning(f"Missing modules: {', '.join(missing)}") return capabilities # ============================================================================= # COMPATIBILITY LAYER # ============================================================================= def get_analyzer(prefer_database: bool = True) -> FeatureExtractor: """ Get appropriate analyzer based on configuration. This function provides a compatibility layer that returns the best available analyzer for the current system configuration. Priority order: 1. HybridExtractor (if available) - best of both worlds 2. DatabaseExtractor (if prefer_database and available) - fast metadata 3. LibrosaExtractor (if available) - full audio analysis 4. Basic placeholder - raises errors on use Args: prefer_database: If True, prefer database-only extraction when hybrid is not available Returns: FeatureExtractor instance appropriate for the system Raises: ImportError: If no analyzer can be created """ # Try hybrid first (best option) if _abstract_analyzer_loaded: try: return init_hybrid_extractor() except Exception as e: logger.warning(f"Could not initialize hybrid extractor: {e}") # Try database-only if prefer_database and _metadata_store_loaded: try: store = init_metadata_store() return DatabaseExtractor(store) except Exception as e: logger.warning(f"Could not initialize database extractor: {e}") # Try librosa-only if _abstract_analyzer_loaded: try: return LibrosaExtractor() except Exception as e: logger.warning(f"Could not initialize librosa extractor: {e}") # Nothing available - create placeholder that gives helpful errors logger.error("No analyzer available - returning placeholder") return FeatureExtractor() # This will raise NotImplementedError on use def get_recorder_or_placeholder(song=None, connection=None): """ Get arrangement recorder if available, or a placeholder. This is a compatibility function for code that needs to work whether or not the arrangement_recorder is available. Args: song: Ableton Live song object (optional) connection: TCP connection (optional) Returns: ArrangementRecorder or RecordingState stub """ if _arrangement_recorder_loaded and song is not None and connection is not None: return init_arrangement_recorder(song, connection) # Return a stub that has the state property class RecordingStub: state = RecordingState.IDLE def is_available(self): return False return RecordingStub() # ============================================================================= # PUBLIC API - __all__ DEFINITION # ============================================================================= __all__ = [ # ========================================================================= # NEW COMPONENTS (Metadata & Analysis) # ========================================================================= # Metadata Store "SampleMetadataStore", "SampleFeatures", # Abstract Analyzer "FeatureExtractor", "LibrosaExtractor", "DatabaseExtractor", "HybridExtractor", # Arrangement Recording "ArrangementRecorder", "RecordingState", "RecordingConfig", # Live Bridge "AbletonLiveBridge", # ========================================================================= # SENIOR ARCHITECTURE - INTELLIGENT SELECTION SYSTEM # ========================================================================= # Intelligent Selector "IntelligentSampleSelector", "CoherenceError", "select_coherent_kit", # Coherence Scorer "CoherenceScorer", "score_kit_coherence", "is_professional_grade", "MIN_COHERENCE", # Variation Engine "VariationEngine", "SECTION_PROFILES", "evolve_kit_for_section", "find_energy_variant", # Rationale Logger "RationaleLogger", "log_sample_selection", "log_kit_assembly", "get_session_rationale", # Preset Manager "PresetManager", "KitPreset", "save_kit_preset", "load_kit_preset", "list_presets", # Iteration Engine "IterationEngine", "ProfessionalCoherenceError", "CoherenceScorer", "RationaleLogger", "IterationResult", "IterationAttempt", "IterationStatus", "ITERATION_STRATEGIES", "iterate_for_coherence", "quick_coherence_check", # ========================================================================= # SENIOR ARCHITECTURE - NEW MODULES (Coherence System, Audio Analyzer Dual, Bus Architecture) # ========================================================================= # Coherence System "calculate_joint_score", "update_cross_generation_memory", "get_cross_generation_penalty", "get_persistent_fatigue", "ROLE_ACTIVITY", "SECTION_DENSITY_PROFILES", "get_section_role_bonus", "calculate_section_appropriateness", "set_palette_lock", "calculate_palette_bonus", "get_palette_coherence_score", "calculate_comprehensive_coherence", "reset_all_memory", "get_coherence_memory_stats", # Audio Analyzer Dual "AudioAnalyzerDual", "AudioFeatures", "analyze_sample", "analyze_audio", "get_backend_info", # Bus Architecture "BUS_GAIN_CALIBRATION", "RETURN_CONFIG", "ROLE_MIX", "MASTER_CHAIN", "BusArchitecture", "create_bus_track", "create_return_track", "route_track_to_bus", "set_track_send", "configure_bus_gain", "configure_return_effect", "apply_role_mix", "configure_master_chain", "apply_professional_mix", "get_bus_config", "get_return_config", "get_role_mix", "list_available_buses", "list_available_returns", "list_available_roles", # ========================================================================= # INITIALIZATION FUNCTIONS # ========================================================================= "init_metadata_store", "init_hybrid_extractor", "init_arrangement_recorder", "init_live_bridge", "init_intelligent_selector", "init_coherence_scorer", "init_variation_engine", "init_rationale_logger", "init_preset_manager", "init_iteration_engine", "clear_initialized_components", # ========================================================================= # CONFIGURATION & COMPATIBILITY # ========================================================================= "get_system_capabilities", "configure_engines_for_capabilities", "get_analyzer", "get_recorder_or_placeholder", "is_module_available", "ModuleAvailability", # ========================================================================= # SPRINT 1 - Core Analysis # ========================================================================= "LibreriaAnalyzer", "analyze_library", "EmbeddingEngine", "create_embeddings_index", "find_similar_samples", "find_samples_like_audio", "ReferenceMatcher", "SpectralFingerprint", "SampleMatch", "UserSoundProfile", "AudioAnalyzer", "SimilarityEngine", "get_matcher", "get_user_profile", "get_recommended_samples", "analyze_reference", "refresh_profile", "SampleSelector", "SampleInfo", "DrumKit", "InstrumentGroup", "get_selector", "select_samples_for_track", "get_drum_kit", "reset_cross_generation_memory", # ========================================================================= # SPRINT 2 - Pattern & Mixing # ========================================================================= "DembowPatterns", "BassPatterns", "ChordProgressions", "MelodyGenerator", "HumanFeel", "PercussionLibrary", "NoteEvent", "ScaleType", "get_patterns", "ReggaetonGenerator", "SongGenerator", "SongConfig", "Section", "TrackConfig", "ClipConfig", "Pattern", "DeviceConfig", "get_song_generator", "generate_song", "generate_from_reference", "get_supported_styles", "get_supported_structures", "MixingEngine", "BusManager", "ReturnTrackManager", "MixConfiguration", "get_mixing_engine", "reset_mixing_engine", "DeviceManager", "EQConfiguration", "CompressionSettings", "GainStaging", "MasterChain", "DeviceParameter", "MixQualityChecker", "DeviceInfo", "QualityReport", "SUPPORTED_DEVICES", "EQ_PRESETS", "COMP_PRESETS", "MASTER_PRESETS", "get_device_manager", "get_eq_configuration", "get_compression_settings", "get_gain_staging", "get_master_chain", "get_device_parameter", "get_quality_checker", "create_standard_buses", "apply_send_preset", "ProductionWorkflow", "ActionHistory", "ProjectValidator", "ExportManager", "ActionRecord", "ValidationIssue", "get_workflow", # ========================================================================= # SPRINT 3 - Arrangement & Harmony # ========================================================================= "ArrangementBuilder", "AutomationEngine", "FXCreator", "SampleProcessor", "ArrangementConfig", "SectionMarker", "AutomationPoint", "AutomationEnvelope", "ArrangementClip", "ArrangementSection", "arrangement_to_dict", "dict_to_arrangement", "get_arrangement_length", "create_full_arrangement", "ProjectAnalyzer", "CounterMelodyGenerator", "SampleIntelligence", # VariationEngine (Sample-based, from variation_engine module) # Note: This is the sample kit evolution engine. For MIDI variations, # use the methods from harmony_engine module directly. "VariationEngine", "SectionKit", "EnergyCharacteristics", "CoherenceMetrics", "SECTION_PROFILES", "evolve_kit_for_sections", "get_section_energy_profile", "validate_coherence", "PresetManager", "Preset", "TrackPreset", "MixingConfig", "SampleSelectionCriteria", "get_preset_manager", "apply_preset_to_project", "get_default_preset", "list_available_presets", "quick_apply_preset", "create_builtin_presets", ] # ============================================================================= # MODULE INITIALIZATION # ============================================================================= def _on_import(): """Run on module import to set up the package.""" # Detect capabilities but don't configure yet (let caller decide) capabilities = get_system_capabilities() # Log summary available_count = sum(1 for s in capabilities["modules"].values() if s == "available") total_count = len(capabilities["modules"]) logger.debug( f"Engines package loaded. " f"Capabilities: numpy={capabilities['numpy']}, " f"librosa={capabilities['librosa']}, " f"modules={available_count}/{total_count} available" ) # Run initialization _on_import()