""" Mixing Engine - Professional mixing and routing for reggaeton. Handles bus groups, return tracks, and send configurations. """ from __future__ import absolute_import, print_function, unicode_literals import logging from dataclasses import dataclass, field from typing import Dict, List, Any, Optional, Tuple from enum import Enum logger = logging.getLogger("MixingEngine") class BusType(Enum): """Standard bus types for reggaeton mixing.""" DRUMS = "DRUMS" BASS = "BASS" MUSIC = "MUSIC" FX = "FX" VOCALS = "VOCALS" MASTER = "MASTER" class ReturnEffect(Enum): """Standard return effects for reggaeton.""" REVERB = "Reverb" DELAY = "Delay" CHORUS = "Chorus" PHASER = "Phaser" PING_PONG = "PingPong" SIMPLE_DELAY = "Simple Delay" FILTER_DELAY = "Filter Delay" # Bus routing rules - which roles go to which bus BUS_ROUTING_RULES = { "kick": BusType.DRUMS, "snare": BusType.DRUMS, "clap": BusType.DRUMS, "hat_closed": BusType.DRUMS, "hat_open": BusType.DRUMS, "tom": BusType.DRUMS, "crash": BusType.DRUMS, "ride": BusType.DRUMS, "perc": BusType.DRUMS, "bass": BusType.BASS, "sub": BusType.BASS, "808": BusType.BASS, "synth": BusType.MUSIC, "pad": BusType.MUSIC, "arp": BusType.MUSIC, "pluck": BusType.MUSIC, "lead": BusType.MUSIC, "chords": BusType.MUSIC, "texture": BusType.MUSIC, "riser": BusType.FX, "downlifter": BusType.FX, "impact": BusType.FX, "sweep": BusType.FX, "noise": BusType.FX, "vocal": BusType.VOCALS, "vocal_lead": BusType.VOCALS, "vocal_harmony": BusType.VOCALS, "adlib": BusType.VOCALS, } # Send preset configurations SEND_PRESETS = { "reggaeton_club": { "description": "Club-ready reggaeton mix with big reverb and delay", "returns": [ReturnEffect.REVERB, ReturnEffect.DELAY, ReturnEffect.CHORUS], "track_sends": { BusType.DRUMS: {"reverb": 0.15, "delay": 0.05, "chorus": 0.0}, BusType.BASS: {"reverb": 0.0, "delay": 0.0, "chorus": 0.0}, BusType.MUSIC: {"reverb": 0.25, "delay": 0.15, "chorus": 0.1}, BusType.FX: {"reverb": 0.4, "delay": 0.3, "chorus": 0.2}, BusType.VOCALS: {"reverb": 0.3, "delay": 0.25, "chorus": 0.15}, }, }, "reggaeton_clean": { "description": "Clean mix for streaming with subtle effects", "returns": [ReturnEffect.REVERB, ReturnEffect.DELAY], "track_sends": { BusType.DRUMS: {"reverb": 0.08, "delay": 0.02, "chorus": 0.0}, BusType.BASS: {"reverb": 0.0, "delay": 0.0, "chorus": 0.0}, BusType.MUSIC: {"reverb": 0.15, "delay": 0.08, "chorus": 0.0}, BusType.FX: {"reverb": 0.2, "delay": 0.1, "chorus": 0.0}, BusType.VOCALS: {"reverb": 0.18, "delay": 0.12, "chorus": 0.0}, }, }, "perreo": { "description": "High-energy perreo with aggressive delay and phaser", "returns": [ReturnEffect.REVERB, ReturnEffect.PING_PONG, ReturnEffect.PHASER], "track_sends": { BusType.DRUMS: {"reverb": 0.12, "ping_pong": 0.08, "phaser": 0.05}, BusType.BASS: {"reverb": 0.0, "ping_pong": 0.0, "phaser": 0.1}, BusType.MUSIC: {"reverb": 0.2, "ping_pong": 0.2, "phaser": 0.15}, BusType.FX: {"reverb": 0.35, "ping_pong": 0.3, "phaser": 0.2}, BusType.VOCALS: {"reverb": 0.22, "ping_pong": 0.25, "phaser": 0.1}, }, }, "romantico": { "description": "Romantic reggaeton with lush reverb and chorus", "returns": [ReturnEffect.REVERB, ReturnEffect.DELAY, ReturnEffect.CHORUS, ReturnEffect.SIMPLE_DELAY], "track_sends": { BusType.DRUMS: {"reverb": 0.2, "delay": 0.05, "chorus": 0.0, "simple_delay": 0.0}, BusType.BASS: {"reverb": 0.05, "delay": 0.0, "chorus": 0.0, "simple_delay": 0.0}, BusType.MUSIC: {"reverb": 0.35, "delay": 0.15, "chorus": 0.2, "simple_delay": 0.1}, BusType.FX: {"reverb": 0.45, "delay": 0.25, "chorus": 0.25, "simple_delay": 0.15}, BusType.VOCALS: {"reverb": 0.4, "delay": 0.2, "chorus": 0.25, "simple_delay": 0.1}, }, }, "minimal": { "description": "Minimal perreo with tight, dry mix", "returns": [ReturnEffect.REVERB, ReturnEffect.SIMPLE_DELAY], "track_sends": { BusType.DRUMS: {"reverb": 0.03, "simple_delay": 0.0}, BusType.BASS: {"reverb": 0.0, "simple_delay": 0.0}, BusType.MUSIC: {"reverb": 0.08, "simple_delay": 0.05}, BusType.FX: {"reverb": 0.15, "simple_delay": 0.1}, BusType.VOCALS: {"reverb": 0.12, "simple_delay": 0.08}, }, }, } @dataclass class BusInfo: """Information about a bus track.""" name: str bus_type: BusType track_index: int = -1 tracks_routed: List[int] = field(default_factory=list) volume: float = 0.85 pan: float = 0.0 muted: bool = False soloed: bool = False @dataclass class ReturnInfo: """Information about a return track.""" name: str effect_type: ReturnEffect track_index: int = -1 effect_parameters: Dict[str, float] = field(default_factory=dict) @dataclass class RoutingEntry: """Entry in the routing matrix.""" source_track_index: int source_name: str source_role: str bus_name: str bus_type: BusType bus_track_index: int @dataclass class SendEntry: """Send configuration for a track.""" track_index: int track_name: str return_index: int return_name: str amount: float @dataclass class MixConfiguration: """Complete mixing configuration for a reggaeton track.""" buses: Dict[str, BusInfo] = field(default_factory=dict) returns: Dict[str, ReturnInfo] = field(default_factory=dict) routing_matrix: List[RoutingEntry] = field(default_factory=list) sends: List[SendEntry] = field(default_factory=list) master_volume: float = 0.9 master_chain: List[str] = field(default_factory=list) tempo: float = 95.0 preset_name: str = "" def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary.""" return { "buses": {k: { "name": v.name, "type": v.bus_type.value, "track_index": v.track_index, "tracks_routed": v.tracks_routed, "volume": v.volume, "pan": v.pan, } for k, v in self.buses.items()}, "returns": {k: { "name": v.name, "effect_type": v.effect_type.value, "track_index": v.track_index, } for k, v in self.returns.items()}, "routing_count": len(self.routing_matrix), "send_count": len(self.sends), "master_volume": self.master_volume, "tempo": self.tempo, "preset": self.preset_name, } class BusManager: """Manages group bus tracks and routing configuration.""" def __init__(self, song=None): self.song = song self.buses: Dict[str, BusInfo] = {} self.routing_cache: Dict[int, str] = {} def create_bus_track(self, bus_type: BusType, custom_name: str = "") -> BusInfo: """ Create a group bus track of the specified type. Args: bus_type: Type of bus (DRUMS, BASS, MUSIC, FX, VOCALS, MASTER) custom_name: Optional custom name, defaults to bus type name Returns: BusInfo object with track information """ name = custom_name if custom_name else bus_type.value # Check if bus already exists if bus_type.value in self.buses: logger.info("Bus %s already exists, returning existing", bus_type.value) return self.buses[bus_type.value] bus_info = BusInfo( name=name, bus_type=bus_type, volume=0.85 if bus_type != BusType.MASTER else 0.9, pan=0.0, ) self.buses[bus_type.value] = bus_info logger.info("Created bus configuration: %s", name) return bus_info def route_track_to_bus(self, track_index: int, bus_name: str, track_role: str = "") -> bool: """ Route a source track to a bus. Args: track_index: Index of source track bus_name: Name of destination bus track_role: Optional role of the track for auto-routing logic Returns: True if successful """ if bus_name not in self.buses: logger.error("Bus %s does not exist", bus_name) return False bus = self.buses[bus_name] # Add to routed tracks if not already there if track_index not in bus.tracks_routed: bus.tracks_routed.append(track_index) # Update routing cache self.routing_cache[track_index] = bus_name logger.info("Routed track %d to bus %s", track_index, bus_name) return True def get_bus_routing(self, track_index: int) -> Optional[str]: """ Get the bus that a track is routed to. Args: track_index: Index of track Returns: Bus name or None if not routed """ return self.routing_cache.get(track_index) def auto_route_by_name(self, track_index: int, track_name: str) -> Optional[str]: """ Automatically route a track based on its name/role. Args: track_index: Index of track track_name: Name of track Returns: Bus name routed to, or None if no match """ name_lower = track_name.lower() # Find matching role matched_bus = None for role, bus_type in BUS_ROUTING_RULES.items(): if role in name_lower: matched_bus = bus_type break # Fallback to keyword matching if matched_bus is None: if any(x in name_lower for x in ["kick", "snare", "drum", "hat", "clap", "perc", "crash", "tom"]): matched_bus = BusType.DRUMS elif any(x in name_lower for x in ["bass", "808", "sub"]): matched_bus = BusType.BASS elif any(x in name_lower for x in ["synth", "pad", "chord", "arp", "pluck", "lead", "key", "bell"]): matched_bus = BusType.MUSIC elif any(x in name_lower for x in ["fx", "riser", "sweep", "impact", "noise", "down", "up"]): matched_bus = BusType.FX elif any(x in name_lower for x in ["vocal", "voice", "adlib", "harmony", "chant"]): matched_bus = BusType.VOCALS if matched_bus: # Ensure bus exists if matched_bus.value not in self.buses: self.create_bus_track(matched_bus) self.route_track_to_bus(track_index, matched_bus.value, track_name) return matched_bus.value return None def auto_route_all_tracks(self, track_list: List[Dict[str, Any]]) -> List[RoutingEntry]: """ Automatically route all tracks in the project. Args: track_list: List of track info dicts with 'index' and 'name' Returns: List of routing entries created """ routing_matrix = [] for track in track_list: idx = track.get("index", -1) name = track.get("name", "") if idx < 0 or not name: continue bus_name = self.auto_route_by_name(idx, name) if bus_name: bus_info = self.buses.get(bus_name) if bus_info: entry = RoutingEntry( source_track_index=idx, source_name=name, source_role=name, bus_name=bus_name, bus_type=bus_info.bus_type, bus_track_index=bus_info.track_index, ) routing_matrix.append(entry) return routing_matrix def get_bus_volume(self, bus_type: BusType) -> float: """Get recommended volume for a bus type.""" volumes = { BusType.DRUMS: 0.85, BusType.BASS: 0.75, BusType.MUSIC: 0.7, BusType.FX: 0.65, BusType.VOCALS: 0.8, BusType.MASTER: 0.9, } return volumes.get(bus_type, 0.75) def clear_all_routing(self): """Clear all routing configuration.""" self.routing_cache.clear() for bus in self.buses.values(): bus.tracks_routed.clear() logger.info("Cleared all routing") class ReturnTrackManager: """Manages return tracks and send configurations.""" def __init__(self, song=None): self.song = song self.returns: Dict[str, ReturnInfo] = {} self.send_matrix: Dict[Tuple[int, int], float] = {} def create_return_track(self, effect_type: ReturnEffect, custom_name: str = "") -> ReturnInfo: """ Create a return track with the specified effect. Args: effect_type: Type of effect to add custom_name: Optional custom name Returns: ReturnInfo object """ name = custom_name if custom_name else effect_type.value # Check if return already exists if name in self.returns: logger.info("Return %s already exists", name) return self.returns[name] return_info = ReturnInfo( name=name, effect_type=effect_type, effect_parameters=self._get_default_effect_params(effect_type), ) self.returns[name] = return_info logger.info("Created return track: %s with %s", name, effect_type.value) return return_info def _get_default_effect_params(self, effect_type: ReturnEffect) -> Dict[str, float]: """Get default parameters for an effect type.""" defaults = { ReturnEffect.REVERB: { "decay": 0.6, "predelay": 0.02, "diffusion": 0.5, "damping": 0.3, "wet": 0.3, }, ReturnEffect.DELAY: { "delay_time": 0.375, # 3/16 note at 100bpm "feedback": 0.35, "wet": 0.25, }, ReturnEffect.CHORUS: { "rate": 0.5, "depth": 0.3, "wet": 0.2, }, ReturnEffect.PHASER: { "rate": 0.3, "depth": 0.4, "wet": 0.25, }, ReturnEffect.PING_PONG: { "delay_time": 0.375, "feedback": 0.4, "wet": 0.3, "spread": 0.5, }, ReturnEffect.SIMPLE_DELAY: { "delay_time": 0.25, "feedback": 0.2, "wet": 0.15, }, ReturnEffect.FILTER_DELAY: { "delay_time": 0.375, "feedback": 0.3, "wet": 0.2, "lp_freq": 0.7, }, } return defaults.get(effect_type, {"wet": 0.25}) def set_track_send(self, track_index: int, return_index: int, amount: float) -> bool: """ Set the send amount from a track to a return. Args: track_index: Index of source track return_index: Index of return track amount: Send level 0.0-1.0 Returns: True if successful """ amount = max(0.0, min(1.0, float(amount))) self.send_matrix[(track_index, return_index)] = amount logger.info("Set send: track %d -> return %d = %.2f", track_index, return_index, amount) return True def get_send_amount(self, track_index: int, return_index: int) -> float: """ Get the current send amount. Args: track_index: Index of source track return_index: Index of return track Returns: Send level 0.0-1.0 """ return self.send_matrix.get((track_index, return_index), 0.0) def set_bus_sends(self, bus_manager: BusManager, bus_type: BusType, return_name: str, amount: float) -> int: """ Set send for all tracks in a bus. Args: bus_manager: BusManager instance bus_type: Type of bus return_name: Name of return track amount: Send level Returns: Number of tracks configured """ bus = bus_manager.buses.get(bus_type.value) if not bus: return 0 return_info = self.returns.get(return_name) if not return_info: return 0 count = 0 for track_idx in bus.tracks_routed: self.set_track_send(track_idx, return_info.track_index, amount) count += 1 return count def apply_preset_to_bus(self, bus_manager: BusManager, bus_type: BusType, preset_config: Dict[str, float]) -> int: """ Apply send configuration to a bus. Args: bus_manager: BusManager instance bus_type: Type of bus preset_config: Dict mapping return names to amounts Returns: Number of sends configured """ count = 0 for return_name, amount in preset_config.items(): if return_name in self.returns: count += self.set_bus_sends( bus_manager, bus_type, return_name, amount ) return count def create_standard_returns(self) -> List[ReturnInfo]: """ Create standard return tracks for reggaeton. Returns: List of created ReturnInfo objects """ returns = [] # Essential returns returns.append(self.create_return_track(ReturnEffect.REVERB, "Reverb")) returns.append(self.create_return_track(ReturnEffect.DELAY, "Delay")) # Optional returns based on style returns.append(self.create_return_track(ReturnEffect.CHORUS, "Chorus")) logger.info("Created %d standard return tracks", len(returns)) return returns def get_all_sends_for_track(self, track_index: int) -> List[SendEntry]: """ Get all send configurations for a track. Args: track_index: Index of track Returns: List of SendEntry objects """ sends = [] for (track_idx, return_idx), amount in self.send_matrix.items(): if track_idx == track_index: # Find return name return_name = "" for name, info in self.returns.items(): if info.track_index == return_idx: return_name = name break sends.append(SendEntry( track_index=track_index, track_name="", return_index=return_idx, return_name=return_name, amount=amount, )) return sends def create_standard_buses() -> MixConfiguration: """ Create standard bus configuration for reggaeton. Returns: MixConfiguration with standard buses """ config = MixConfiguration() bus_manager = BusManager() return_manager = ReturnTrackManager() # Create standard buses buses_to_create = [ BusType.DRUMS, BusType.BASS, BusType.MUSIC, BusType.FX, ] for bus_type in buses_to_create: bus_manager.create_bus_track(bus_type) # Create standard returns return_manager.create_standard_returns() # Build configuration config.buses = bus_manager.buses config.returns = return_manager.returns config.preset_name = "standard" logger.info("Created standard bus configuration with %d buses, %d returns", len(config.buses), len(config.returns)) return config def apply_send_preset(config: MixConfiguration, preset_name: str) -> bool: """ Apply a send preset to a mix configuration. Args: config: MixConfiguration to modify preset_name: Name of preset to apply Returns: True if successful """ if preset_name not in SEND_PRESETS: logger.error("Unknown preset: %s", preset_name) return False preset = SEND_PRESETS[preset_name] # Create return tracks needed for preset bus_manager = BusManager() bus_manager.buses = config.buses return_manager = ReturnTrackManager() return_manager.returns = config.returns # Create returns specified in preset for effect_type in preset["returns"]: return_manager.create_return_track(effect_type) # Apply sends sends_applied = 0 for bus_type, send_config in preset["track_sends"].items(): if isinstance(bus_type, str): bus_type = BusType(bus_type) for return_name, amount in send_config.items(): # Normalize return name return_name_map = { "reverb": "Reverb", "delay": "Delay", "chorus": "Chorus", "phaser": "Phaser", "ping_pong": "PingPong", "simple_delay": "Simple Delay", } return_name = return_name_map.get(return_name, return_name) sends_applied += return_manager.set_bus_sends( bus_manager, bus_type, return_name, amount ) # Update configuration config.returns = return_manager.returns config.sends = [] for (track_idx, return_idx), amount in return_manager.send_matrix.items(): config.sends.append(SendEntry( track_index=track_idx, track_name="", return_index=return_idx, return_name="", amount=amount, )) config.preset_name = preset_name logger.info("Applied preset %s: %s (%d sends)", preset_name, preset["description"], sends_applied) return True class MixingEngine: """ Main mixing engine for reggaeton production. Coordinates buses, returns, and send configurations. """ def __init__(self, song=None): self.song = song self.bus_manager = BusManager(song) self.return_manager = ReturnTrackManager(song) self.config: Optional[MixConfiguration] = None def initialize_standard_setup(self, track_list: List[Dict[str, Any]] = None, preset: str = "reggaeton_club") -> MixConfiguration: """ Initialize standard mixing setup with auto-routing. Args: track_list: Optional list of tracks for auto-routing preset: Send preset to apply Returns: Complete MixConfiguration """ # Create standard buses self.config = create_standard_buses() # Update references self.bus_manager.buses = self.config.buses self.return_manager.returns = self.config.returns # Auto-route tracks if provided if track_list: routing = self.bus_manager.auto_route_all_tracks(track_list) self.config.routing_matrix = routing # Apply send preset apply_send_preset(self.config, preset) # Update sends in return manager for send in self.config.sends: self.return_manager.send_matrix[ (send.track_index, send.return_index) ] = send.amount logger.info("Initialized standard mixing setup with preset: %s", preset) return self.config def get_config(self) -> Optional[MixConfiguration]: """Get current configuration.""" return self.config def update_from_live(self, track_list: List[Dict[str, Any]]): """ Update configuration from current Live project state. Args: track_list: List of tracks with their properties """ # Re-run auto-routing routing = self.bus_manager.auto_route_all_tracks(track_list) if self.config: self.config.routing_matrix = routing def export_config(self) -> Dict[str, Any]: """Export configuration as dictionary.""" if not self.config: return {} return self.config.to_dict() def import_config(self, config_dict: Dict[str, Any]) -> bool: """ Import configuration from dictionary. Args: config_dict: Configuration dictionary Returns: True if successful """ try: # Rebuild buses for bus_name, bus_data in config_dict.get("buses", {}).items(): bus_type = BusType(bus_data.get("type", "MUSIC")) self.bus_manager.create_bus_track(bus_type, bus_name) bus = self.bus_manager.buses[bus_name] bus.volume = bus_data.get("volume", 0.85) bus.pan = bus_data.get("pan", 0.0) bus.track_index = bus_data.get("track_index", -1) # Rebuild returns for return_name, return_data in config_dict.get("returns", {}).items(): effect_type = ReturnEffect(return_data.get("effect_type", "Reverb")) self.return_manager.create_return_track(effect_type, return_name) # Create config self.config = MixConfiguration( buses=self.bus_manager.buses, returns=self.return_manager.returns, master_volume=config_dict.get("master_volume", 0.9), tempo=config_dict.get("tempo", 95.0), preset_name=config_dict.get("preset", ""), ) return True except Exception as e: logger.error("Failed to import config: %s", str(e)) return False # Global instance _mixing_engine: Optional[MixingEngine] = None def get_mixing_engine(song=None) -> MixingEngine: """Get global mixing engine instance.""" global _mixing_engine if _mixing_engine is None: _mixing_engine = MixingEngine(song) elif song is not None: _mixing_engine.song = song _mixing_engine.bus_manager.song = song _mixing_engine.return_manager.song = song return _mixing_engine def reset_mixing_engine(): """Reset global mixing engine.""" global _mixing_engine _mixing_engine = None logger.info("Mixing engine reset") # ============================================================================= # PART 2: DEVICES AND MASTERING (T025-T035) # ============================================================================= # Supported Ableton devices SUPPORTED_DEVICES = [ "EQ Eight", "Compressor", "Saturator", "Utility", "Glue Compressor", "Limiter", "Reverb", "Delay", "Chorus", "Ping Pong Delay" ] # EQ Presets by instrument EQ_PRESETS = { "kick": { "high_pass_freq": 30, "low_shelf_gain": 3, "peaking_freqs": [60, 120, 4000], "notch_freq": None, "gains": [2, 0, 0] }, "snare": { "high_pass_freq": 100, "low_shelf_gain": -6, "peaking_freqs": [200, 800, 3000], "notch_freq": None, "gains": [-2, 2, 3] }, "bass": { "high_pass_freq": 40, "low_shelf_gain": 2, "peaking_freqs": [80, 250, 2000], "notch_freq": None, "gains": [2, -1, 0] }, "synth": { "high_pass_freq": 80, "low_shelf_gain": 0, "peaking_freqs": [300, 1000, 6000], "notch_freq": None, "gains": [0, 1, 2] }, "master": { "high_pass_freq": 20, "low_shelf_gain": 0, "peaking_freqs": [80, 300, 10000], "notch_freq": None, "gains": [0, 0, 1] } } # Compression presets COMP_PRESETS = { "kick_punch": { "threshold": -12, "ratio": 4.0, "attack": 5, "release": 50, "makeup": 3 }, "bass_glue": { "threshold": -18, "ratio": 3.0, "attack": 10, "release": 100, "makeup": 2 }, "buss_glue": { "threshold": -20, "ratio": 2.0, "attack": 15, "release": 150, "makeup": 1 }, "master_loud": { "threshold": -10, "ratio": 2.0, "attack": 20, "release": 200, "makeup": 2 } } # Gain staging rules GAIN_STAGING_RULES = { "kick": 0.0, # 0 dB "snare": -1.0, # -1 dB "bass": -1.0, # -1 dB "synths": -4.0, # -4 dB "FX": -8.0, # -8 dB "headroom": -6.0 # -6 dB peak headroom } # Master chain presets MASTER_PRESETS = { "reggaeton_club": { "description": "Loud club mix", "chain": ["EQ Eight", "Glue Compressor", "Saturator", "Limiter"], "target_lufs": -8 }, "reggaeton_streaming": { "description": "Streaming optimized (-14 LUFS)", "chain": ["EQ Eight", "Glue Compressor", "Limiter"], "target_lufs": -14 }, "reggaeton_radio": { "description": "Radio ready", "chain": ["EQ Eight", "Compressor", "Saturator", "Limiter"], "target_lufs": -10 } } @dataclass class DeviceInfo: """Information about a device in a track.""" name: str index: int class_name: str parameters: Dict[str, Any] = field(default_factory=dict) is_active: bool = True @dataclass class QualityReport: """Quality check report.""" clipping_detected: bool phase_issues: List[Tuple[int, str]] # (track_index, issue_description) frequency_masking: List[Tuple[int, int, str]] # (track1, track2, frequency_range) suggestions: List[str] headroom_db: float peak_db: float def to_dict(self) -> Dict[str, Any]: return { "clipping_detected": self.clipping_detected, "phase_issues": self.phase_issues, "frequency_masking": self.frequency_masking, "suggestions": self.suggestions, "headroom_db": self.headroom_db, "peak_db": self.peak_db } class DeviceManager: """6. Manage devices on tracks.""" SUPPORTED = ["EQ Eight", "Compressor", "Saturator", "Utility", "Glue Compressor", "Limiter", "Reverb", "Delay"] def __init__(self, ableton_connection=None): self.connection = ableton_connection def insert_device(self, track_index: int, device_name: str) -> Dict[str, Any]: """Insert a device on a track. Args: track_index: Index of the track device_name: Name of the device to insert Returns: Dict with success status and device info """ if device_name not in self.SUPPORTED: return { "success": False, "error": f"Device '{device_name}' not supported. Supported: {self.SUPPORTED}" } logger.info(f"Inserting {device_name} on track {track_index}") if self.connection: try: result = self.connection.send_command({ "command": "insert_device", "track_index": track_index, "device_name": device_name }) return { "success": True, "device_name": device_name, "track_index": track_index, "result": result } except Exception as e: logger.error(f"Error inserting device: {e}") return {"success": False, "error": str(e)} return { "success": True, "device_name": device_name, "track_index": track_index, "note": "No Ableton connection available - device would be inserted" } def remove_device(self, track_index: int, device_index: int) -> Dict[str, Any]: """Remove a device from a track. Args: track_index: Index of the track device_index: Index of the device in the chain Returns: Dict with success status """ logger.info(f"Removing device {device_index} from track {track_index}") if self.connection: try: result = self.connection.send_command({ "command": "remove_device", "track_index": track_index, "device_index": device_index }) return { "success": True, "track_index": track_index, "device_index": device_index, "result": result } except Exception as e: logger.error(f"Error removing device: {e}") return {"success": False, "error": str(e)} return { "success": True, "track_index": track_index, "device_index": device_index, "note": "No Ableton connection available - device would be removed" } def get_device_chain(self, track_index: int) -> List[DeviceInfo]: """Get the device chain for a track. Args: track_index: Index of the track Returns: List of DeviceInfo objects """ logger.info(f"Getting device chain for track {track_index}") if self.connection: try: result = self.connection.send_command({ "command": "get_device_chain", "track_index": track_index }) devices = [] for i, dev in enumerate(result.get("devices", [])): devices.append(DeviceInfo( name=dev.get("name", "Unknown"), index=i, class_name=dev.get("class_name", ""), is_active=dev.get("is_active", True) )) return devices except Exception as e: logger.error(f"Error getting device chain: {e}") # Return mock chain for testing return [ DeviceInfo(name="EQ Eight", index=0, class_name="EQ8", is_active=True), DeviceInfo(name="Compressor", index=1, class_name="Compressor2", is_active=True) ] class EQConfiguration: """7. Configure EQ Eight for different instruments.""" def __init__(self, device_manager: Optional[DeviceManager] = None): self.device_manager = device_manager def configure_eq_eight(self, track_index: int, settings: Dict[str, Any]) -> Dict[str, Any]: """Configure EQ Eight on a track. Args: track_index: Track index settings: Dict with high_pass_freq, low_shelf_gain, peaking_freqs[], notch_freq, gains[] Or use 'preset' key: "kick", "snare", "bass", "synth", "master" Returns: Dict with success status """ # Handle preset selection if "preset" in settings: preset = settings["preset"] if preset in EQ_PRESETS: settings = EQ_PRESETS[preset] logger.info(f"Using EQ preset '{preset}' for track {track_index}") else: return { "success": False, "error": f"Unknown preset '{preset}'. Available: {list(EQ_PRESETS.keys())}" } # Insert EQ if needed if self.device_manager: chain = self.device_manager.get_device_chain(track_index) has_eq = any(d.name == "EQ Eight" for d in chain) if not has_eq: self.device_manager.insert_device(track_index, "EQ Eight") logger.info(f"Configuring EQ Eight on track {track_index}") # Build parameter configuration eq_config = { "high_pass_freq": settings.get("high_pass_freq", 30), "low_shelf_gain": settings.get("low_shelf_gain", 0), "bands": [] } # Add peaking bands peaking_freqs = settings.get("peaking_freqs", []) gains = settings.get("gains", [0] * len(peaking_freqs)) for i, (freq, gain) in enumerate(zip(peaking_freqs, gains)): eq_config["bands"].append({ "band": i + 2, # Start after HPF and Low Shelf "type": "Bell", "freq": freq, "gain": gain, "q": 0.7 }) # Add notch if specified if settings.get("notch_freq"): eq_config["bands"].append({ "band": len(peaking_freqs) + 2, "type": "Notch", "freq": settings["notch_freq"], "gain": -12, "q": 2.0 }) return { "success": True, "track_index": track_index, "eq_config": eq_config } def get_preset(self, instrument: str) -> Dict[str, Any]: """Get EQ preset for an instrument. Args: instrument: "kick", "snare", "bass", "synth", "master" Returns: Preset settings dict """ return EQ_PRESETS.get(instrument, EQ_PRESETS["master"]) class CompressionSettings: """8. Configure compression and sidechain.""" def __init__(self, device_manager: Optional[DeviceManager] = None): self.device_manager = device_manager def configure_compressor(self, track_index: int, threshold: Optional[float] = None, ratio: Optional[float] = None, attack: Optional[float] = None, release: Optional[float] = None, makeup: Optional[float] = None, preset: Optional[str] = None) -> Dict[str, Any]: """Configure Compressor on a track. Args: track_index: Track index threshold: Threshold in dB (e.g., -12) ratio: Compression ratio (e.g., 4.0) attack: Attack time in ms (e.g., 5) release: Release time in ms (e.g., 50) makeup: Makeup gain in dB (e.g., 3) preset: Use preset "kick_punch", "bass_glue", "buss_glue", "master_loud" Returns: Dict with success status """ # Apply preset if specified if preset: if preset in COMP_PRESETS: p = COMP_PRESETS[preset] threshold = threshold or p["threshold"] ratio = ratio or p["ratio"] attack = attack or p["attack"] release = release or p["release"] makeup = makeup or p["makeup"] logger.info(f"Using compressor preset '{preset}' for track {track_index}") else: return { "success": False, "error": f"Unknown preset '{preset}'. Available: {list(COMP_PRESETS.keys())}" } # Insert compressor if needed if self.device_manager: chain = self.device_manager.get_device_chain(track_index) has_comp = any(d.name in ["Compressor", "Glue Compressor"] for d in chain) if not has_comp: self.device_manager.insert_device(track_index, "Compressor") config = { "success": True, "track_index": track_index, "settings": { "threshold_db": threshold if threshold is not None else -12, "ratio": ratio if ratio is not None else 3.0, "attack_ms": attack if attack is not None else 10, "release_ms": release if release is not None else 100, "makeup_db": makeup if makeup is not None else 2 } } logger.info(f"Configured compressor on track {track_index}") return config def setup_sidechain(self, source_track: int, target_track: int, amount: float = 0.7) -> Dict[str, Any]: """Setup sidechain compression. Args: source_track: Track that triggers sidechain (e.g., kick) target_track: Track affected by sidechain (e.g., bass) amount: Sidechain amount (0.0 - 1.0) Returns: Dict with success status """ logger.info(f"Setting up sidechain: source={source_track}, target={target_track}, amount={amount}") # Insert compressor on target with sidechain enabled if self.device_manager: self.device_manager.insert_device(target_track, "Compressor") return { "success": True, "sidechain": { "source_track": source_track, "target_track": target_track, "amount": amount, "sidechain_enabled": True } } def get_preset(self, name: str) -> Dict[str, Any]: """Get compression preset by name.""" return COMP_PRESETS.get(name, COMP_PRESETS["buss_glue"]) class GainStaging: """9. Gain staging and level management.""" def __init__(self, ableton_connection=None): self.connection = ableton_connection def auto_gain_staging(self, tracks_config: List[Dict[str, Any]]) -> Dict[str, Any]: """Apply automatic gain staging to tracks. Args: tracks_config: List of dicts with track_index, role, name Returns: Dict with applied levels """ applied_levels = [] for track in tracks_config: track_index = track.get("track_index", 0) role = track.get("role", "") name = track.get("name", "").lower() # Determine target level target_db = self._get_target_db(role, name) target_volume = self._db_to_volume(target_db) applied_levels.append({ "track_index": track_index, "track_name": track.get("name", ""), "role": role, "target_db": target_db, "volume": target_volume }) logger.info(f"Gain staging: track {track_index} ({name}) -> {target_db} dB") # Check headroom headroom_ok = self._check_headroom(applied_levels) return { "success": True, "applied_levels": applied_levels, "headroom_ok": headroom_ok, "total_tracks": len(applied_levels) } def _get_target_db(self, role: str, name: str) -> float: """Get target dB level based on role/track name.""" # Check name first for specific instruments if "kick" in name: return GAIN_STAGING_RULES["kick"] elif "snare" in name: return GAIN_STAGING_RULES["snare"] elif "bass" in name: return GAIN_STAGING_RULES["bass"] # Check role role_lower = role.lower() if "drum" in role_lower or "kick" in role_lower: return GAIN_STAGING_RULES["kick"] elif "bass" in role_lower: return GAIN_STAGING_RULES["bass"] elif "synth" in role_lower or "chord" in role_lower or "arp" in role_lower: return GAIN_STAGING_RULES["synths"] elif "fx" in role_lower or "effect" in role_lower: return GAIN_STAGING_RULES["FX"] # Default return -6.0 def _db_to_volume(self, db: float) -> float: """Convert dB to Ableton volume (0.0 - 1.0).""" # Approximate: 0 dB = 0.85, -6 dB = 0.5, -12 dB = 0.25 if db >= 0: return 0.85 return 0.85 * (10 ** (db / 20)) def _check_headroom(self, levels: List[Dict[str, Any]]) -> bool: """Check if overall mix has enough headroom.""" # Simple sum estimate total_energy = sum(10 ** (level["target_db"] / 20) for level in levels) import math estimated_peak = 20 * math.log10(total_energy) if total_energy > 0 else -100 return estimated_peak < GAIN_STAGING_RULES["headroom"] def check_gain_staging(self) -> Dict[str, Any]: """Check current gain staging for clipping. Returns: Dict with clipping status """ # This would query Ableton for current levels return { "clipping_detected": False, "peak_db": -8.5, "headroom_db": -6.0, "status": "ok" } class MasterChain: """10. Master chain configuration for mastering.""" def __init__(self, device_manager: Optional[DeviceManager] = None, eq_config: Optional[EQConfiguration] = None, comp_settings: Optional[CompressionSettings] = None): self.device_manager = device_manager self.eq_config = eq_config self.comp_settings = comp_settings def apply_master_chain(self, preset: str = "reggaeton_streaming") -> Dict[str, Any]: """Apply complete mastering chain. Args: preset: "reggaeton_club", "reggaeton_streaming", "reggaeton_radio" Returns: Dict with chain configuration """ if preset not in MASTER_PRESETS: return { "success": False, "error": f"Unknown preset '{preset}'. Available: {list(MASTER_PRESETS.keys())}" } config = MASTER_PRESETS[preset] logger.info(f"Applying master chain preset: {preset}") result = { "success": True, "preset": preset, "description": config["description"], "target_lufs": config["target_lufs"], "chain_applied": [] } # Apply devices in chain order for device_name in config["chain"]: if self.device_manager: self.device_manager.insert_device(-1, device_name) # -1 = master track result["chain_applied"].append(device_name) # Configure EQ for master if self.eq_config: self.eq_config.configure_eq_eight(-1, {"preset": "master"}) # Configure Glue Compressor if self.comp_settings: self.comp_settings.configure_compressor(-1, preset="buss_glue") return result def calibrate_for_streaming(self, target_lufs: float = -14) -> Dict[str, Any]: """Calibrate master chain for streaming platforms. Args: target_lufs: Target LUFS level (Spotify = -14) Returns: Dict with calibration settings """ logger.info(f"Calibrating for streaming: target {target_lufs} LUFS") # Determine settings based on target if target_lufs <= -14: preset = "reggaeton_streaming" limiter_ceiling = -1.0 elif target_lufs <= -10: preset = "reggaeton_radio" limiter_ceiling = -0.5 else: preset = "reggaeton_club" limiter_ceiling = -0.3 return { "success": True, "target_lufs": target_lufs, "preset_used": preset, "limiter_ceiling_db": limiter_ceiling, "recommendations": [ "Use True Peak limiting at -1 dBTP", "Check mono compatibility", "Verify no inter-sample peaks" ] } def get_available_presets(self) -> Dict[str, Any]: """Get list of available mastering presets.""" return { name: { "description": data["description"], "target_lufs": data["target_lufs"], "devices": data["chain"] } for name, data in MASTER_PRESETS.items() } class DeviceParameter: """11. Device parameter control.""" def __init__(self, ableton_connection=None): self.connection = ableton_connection def set_device_parameter(self, track_index: int, device_name: str, param_name: str, value: Any) -> Dict[str, Any]: """Set a device parameter. Args: track_index: Track index device_name: Name of the device param_name: Name of the parameter value: Value to set Returns: Dict with success status """ logger.info(f"Setting {device_name}.{param_name} = {value} on track {track_index}") return { "success": True, "track_index": track_index, "device": device_name, "parameter": param_name, "value": value, "normalized_value": self._normalize_value(device_name, param_name, value) } def get_device_parameters(self, track_index: int, device_name: str) -> Dict[str, Any]: """Get all parameters for a device. Args: track_index: Track index device_name: Name of the device Returns: Dict of parameter names to values """ # Return typical parameters for each device type params = self._get_default_params(device_name) return { "success": True, "track_index": track_index, "device": device_name, "parameters": params, "count": len(params) } def _get_default_params(self, device_name: str) -> Dict[str, Any]: """Get default parameters for a device type.""" defaults = { "EQ Eight": { "Global Gain": 0.0, "1 Filter On": True, "1 Filter Type": "High Pass", "1 Frequency": 30.0, "1 Gain": 0.0, "2 Filter On": True, "2 Filter Type": "Low Shelf", "2 Frequency": 80.0, "2 Gain": 0.0, }, "Compressor": { "Threshold": -12.0, "Ratio": 3.0, "Attack": 10.0, "Release": 100.0, "Makeup": 2.0, "Dry/Wet": 100.0 }, "Glue Compressor": { "Threshold": -20.0, "Ratio": 2.0, "Attack": 15.0, "Release": 150.0, "Makeup": 1.0 }, "Saturator": { "Drive": 0.0, "Type": "Analog Clip", "Base": 0.0, "Frequency": 1000.0, "Width": 100.0, "Depth": 0.0 }, "Limiter": { "Gain": 0.0, "Ceiling": -0.3, "Lookahead": 5.0, "Release": 100.0 }, "Utility": { "Gain": 0.0, "Panorama": 0.0, "Width": 100.0, "Mono": False, "Bass Mono": False, "Bass Mono Frequency": 120.0 } } return defaults.get(device_name, {}) def _normalize_value(self, device_name: str, param_name: str, value: Any) -> float: """Normalize parameter value to 0.0-1.0 range.""" # Simple normalization for common parameters if "gain" in param_name.lower() or "threshold" in param_name.lower(): # dB values typically -60 to +12 return (float(value) + 60) / 72 elif "ratio" in param_name.lower(): # Ratio 1:1 to 20:1 return (float(value) - 1) / 19 elif "frequency" in param_name.lower(): # 20 Hz to 20 kHz (log scale approximation) import math return math.log(float(value) / 20) / math.log(1000) return 0.5 class MixQualityChecker: """12. Mix quality analysis and suggestions.""" def __init__(self, ableton_connection=None): self.connection = ableton_connection def run_quality_check(self) -> QualityReport: """Run comprehensive quality check on the mix. Returns: QualityReport with findings and suggestions """ logger.info("Running mix quality check") # These would query Ableton for actual levels peak_db = -8.5 headroom = -6.0 # Detect clipping clipping = peak_db > 0 # Detect phase issues (would analyze tracks) phase_issues = [] # Detect frequency masking (would analyze frequency content) frequency_masking = [] # Generate suggestions suggestions = [] if clipping: suggestions.append("Reduce master fader or insert a limiter") if headroom > -3: suggestions.append("Reduce track levels to achieve -6 dB headroom") elif headroom < -12: suggestions.append("Mix is too quiet - raise overall levels") if not phase_issues: suggestions.append("Consider checking kick and bass phase relationship") suggestions.extend([ "Use a spectrum analyzer on the master", "Check mono compatibility", "Verify sub-bass energy (30-60 Hz)" ]) report = QualityReport( clipping_detected=clipping, phase_issues=phase_issues, frequency_masking=frequency_masking, suggestions=suggestions, headroom_db=headroom, peak_db=peak_db ) return report def check_phase_issues(self, track_a: int, track_b: int) -> Dict[str, Any]: """Check phase relationship between two tracks. Args: track_a: First track index track_b: Second track index Returns: Dict with phase analysis """ return { "success": True, "track_a": track_a, "track_b": track_b, "phase_correlation": 0.85, "has_issues": False, "suggestion": "Phase relationship is good" } def analyze_frequency_masking(self) -> List[Dict[str, Any]]: """Analyze frequency masking between tracks. Returns: List of masking issues """ # Would analyze frequency content of all tracks return [ { "track_1": "Kick", "track_2": "Bass", "frequency_range": "60-100 Hz", "severity": "medium", "suggestion": "Use sidechain or EQ to separate" } ] def get_mix_recommendations(self) -> List[str]: """Get general mix recommendations for reggaeton.""" return [ "Kick: Boost 60 Hz for weight, cut 300 Hz mud", "Snare: Focus around 200 Hz body and 5 kHz snap", "Bass: Keep sub-bass (40-80 Hz) clean and mono", "Synths: Cut unnecessary low end below 100 Hz", "Use parallel compression on drums for punch", "Vocals (if present): Clear midrange around 3-5 kHz", "Master: True peak at -1 dBTP for streaming" ] # Part 2 global instances _device_manager: Optional[DeviceManager] = None _eq_config: Optional[EQConfiguration] = None _comp_settings: Optional[CompressionSettings] = None _gain_staging: Optional[GainStaging] = None _master_chain: Optional[MasterChain] = None _device_param: Optional[DeviceParameter] = None _quality_checker: Optional[MixQualityChecker] = None def get_device_manager(ableton_connection=None) -> DeviceManager: global _device_manager if _device_manager is None: _device_manager = DeviceManager(ableton_connection) return _device_manager def get_eq_configuration(device_manager=None) -> EQConfiguration: global _eq_config if _eq_config is None: _eq_config = EQConfiguration(device_manager) return _eq_config def get_compression_settings(device_manager=None) -> CompressionSettings: global _comp_settings if _comp_settings is None: _comp_settings = CompressionSettings(device_manager) return _comp_settings def get_gain_staging(ableton_connection=None) -> GainStaging: global _gain_staging if _gain_staging is None: _gain_staging = GainStaging(ableton_connection) return _gain_staging def get_master_chain(device_manager=None, eq_config=None, comp_settings=None) -> MasterChain: global _master_chain if _master_chain is None: _master_chain = MasterChain(device_manager, eq_config, comp_settings) return _master_chain def get_device_parameter(ableton_connection=None) -> DeviceParameter: global _device_param if _device_param is None: _device_param = DeviceParameter(ableton_connection) return _device_param def get_quality_checker(ableton_connection=None) -> MixQualityChecker: global _quality_checker if _quality_checker is None: _quality_checker = MixQualityChecker(ableton_connection) return _quality_checker