Files
ableton-mcp-ai/mcp_server/engines/__init__.py
OpenCode Agent 5ce8187c65 feat: Implement senior audio injection with 5 fallback methods
- 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
2026-04-12 14:02:32 -03:00

1696 lines
59 KiB
Python

"""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()