- 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
1696 lines
59 KiB
Python
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()
|