570 lines
20 KiB
Python
570 lines
20 KiB
Python
"""
|
|
Parallel Compression System for AbletonMCP_AI
|
|
|
|
Implements New York-style parallel compression for professional mixing:
|
|
- Create parallel compression chains with wet/dry blending
|
|
- Duplicate tracks with heavy compression while preserving original
|
|
- Blend signals for punch and clarity
|
|
- Presets for drums, vocals, and buses
|
|
|
|
Agente 8: Parallel Compression System
|
|
"""
|
|
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("ParallelCompression")
|
|
|
|
|
|
class CompressionPreset(Enum):
|
|
"""Standard parallel compression presets."""
|
|
DRUM_PARALLEL = "drum_parallel"
|
|
VOCAL_PARALLEL = "vocal_parallel"
|
|
BUS_PARALLEL = "bus_parallel"
|
|
|
|
|
|
@dataclass
|
|
class ParallelChainSettings:
|
|
"""Settings for a parallel compression chain."""
|
|
ratio: float = 4.0
|
|
threshold: float = -20.0
|
|
attack: float = 10.0
|
|
release: float = 100.0
|
|
makeup_gain: float = 0.0
|
|
dry_wet: float = 0.5 # Blend ratio: 0.0 = dry only, 1.0 = wet only
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"ratio": self.ratio,
|
|
"threshold": self.threshold,
|
|
"attack": self.attack,
|
|
"release": self.release,
|
|
"makeup_gain": self.makeup_gain,
|
|
"dry_wet": self.dry_wet,
|
|
}
|
|
|
|
|
|
# Preset configurations for different use cases
|
|
PARALLEL_PRESETS = {
|
|
CompressionPreset.DRUM_PARALLEL: ParallelChainSettings(
|
|
ratio=8.0,
|
|
threshold=-16.0,
|
|
attack=2.0, # Fast attack for drums
|
|
release=30.0, # Fast release
|
|
makeup_gain=6.0,
|
|
dry_wet=0.35, # 35% compressed signal
|
|
),
|
|
CompressionPreset.VOCAL_PARALLEL: ParallelChainSettings(
|
|
ratio=4.0,
|
|
threshold=-18.0,
|
|
attack=8.0, # Medium attack
|
|
release=80.0, # Medium release
|
|
makeup_gain=4.0,
|
|
dry_wet=0.45, # 45% compressed signal
|
|
),
|
|
CompressionPreset.BUS_PARALLEL: ParallelChainSettings(
|
|
ratio=2.0,
|
|
threshold=-20.0,
|
|
attack=15.0, # Slow attack for bus glue
|
|
release=150.0, # Slow release
|
|
makeup_gain=2.0,
|
|
dry_wet=0.25, # 25% compressed signal (subtle)
|
|
),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ParallelChain:
|
|
"""Represents a complete parallel compression chain."""
|
|
name: str
|
|
original_track_index: int
|
|
compressed_track_index: int
|
|
settings: ParallelChainSettings
|
|
preset_used: Optional[str] = None
|
|
active: bool = True
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"original_track": self.original_track_index,
|
|
"compressed_track": self.compressed_track_index,
|
|
"settings": self.settings.to_dict(),
|
|
"preset": self.preset_used,
|
|
"active": self.active,
|
|
}
|
|
|
|
|
|
class ParallelCompression:
|
|
"""
|
|
Professional parallel compression system for Ableton Live.
|
|
|
|
Implements New York-style parallel compression where:
|
|
1. Original track remains uncompressed (dry)
|
|
2. Duplicate track gets heavy compression (wet)
|
|
3. Both are blended for punch and clarity
|
|
"""
|
|
|
|
def __init__(self, ableton_conn=None):
|
|
"""
|
|
Initialize parallel compression system.
|
|
|
|
Args:
|
|
ableton_conn: Ableton Live connection (self from __init__.py)
|
|
"""
|
|
self.conn = ableton_conn
|
|
self._song = ableton_conn._song if hasattr(ableton_conn, '_song') else None
|
|
self._chains: Dict[str, ParallelChain] = {}
|
|
|
|
def create_parallel_chain(self, track_index: int,
|
|
ratio: float = 4.0,
|
|
threshold: float = -20.0,
|
|
makeup_gain: float = 0.0,
|
|
name: str = "") -> Dict[str, Any]:
|
|
"""
|
|
Create a parallel compression chain on a track.
|
|
|
|
This creates:
|
|
1. Keeps original track as "dry" signal
|
|
2. Creates duplicate track with heavy compression as "wet" signal
|
|
3. Blends both via volume levels
|
|
|
|
Args:
|
|
track_index: Index of the original track
|
|
ratio: Compression ratio (e.g., 4.0 for 4:1)
|
|
threshold: Threshold in dB (e.g., -20.0)
|
|
makeup_gain: Makeup gain in dB
|
|
name: Optional custom name for the chain
|
|
|
|
Returns:
|
|
Dict with chain creation status and details
|
|
"""
|
|
if self._song is None:
|
|
return {"error": "No song connection available"}
|
|
|
|
try:
|
|
orig_idx = int(track_index)
|
|
if orig_idx < 0 or orig_idx >= len(self._song.tracks):
|
|
return {"error": "Track index %d out of range" % orig_idx}
|
|
|
|
original_track = self._song.tracks[orig_idx]
|
|
orig_name = str(original_track.name)
|
|
|
|
# Create settings object
|
|
settings = ParallelChainSettings(
|
|
ratio=float(ratio),
|
|
threshold=float(threshold),
|
|
makeup_gain=float(makeup_gain),
|
|
dry_wet=0.5, # Default 50/50 blend
|
|
)
|
|
|
|
# Create duplicate track for compressed version
|
|
dup_result = self.duplicate_track_with_compression(orig_idx, settings.to_dict())
|
|
|
|
if not dup_result.get("success"):
|
|
return {
|
|
"error": "Failed to create compressed track: %s" % dup_result.get("error", "Unknown")
|
|
}
|
|
|
|
compressed_idx = dup_result.get("compressed_track_index", -1)
|
|
|
|
# Set blend levels
|
|
blend_result = self.blend_wet_dry(compressed_idx, orig_idx, 50.0)
|
|
|
|
# Generate chain name
|
|
chain_name = name if name else "Parallel_%s" % orig_name
|
|
|
|
# Store chain info
|
|
chain = ParallelChain(
|
|
name=chain_name,
|
|
original_track_index=orig_idx,
|
|
compressed_track_index=compressed_idx,
|
|
settings=settings,
|
|
preset_used=None,
|
|
)
|
|
self._chains[chain_name] = chain
|
|
|
|
return {
|
|
"success": True,
|
|
"chain_name": chain_name,
|
|
"original_track": orig_idx,
|
|
"original_name": orig_name,
|
|
"compressed_track": compressed_idx,
|
|
"compressed_name": dup_result.get("compressed_name", ""),
|
|
"settings": settings.to_dict(),
|
|
"blend_applied": blend_result.get("success", False),
|
|
"note": "Parallel compression chain created. Adjust track volumes to taste."
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error creating parallel chain: %s" % str(e))
|
|
return {"error": str(e)}
|
|
|
|
def duplicate_track_with_compression(self, original_track: int,
|
|
settings: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Duplicate a track and apply heavy compression to the duplicate.
|
|
|
|
Args:
|
|
original_track: Index of the track to duplicate
|
|
settings: Compression settings dict with ratio, threshold, attack, release, makeup_gain
|
|
|
|
Returns:
|
|
Dict with duplication status and compressed track info
|
|
"""
|
|
if self._song is None:
|
|
return {"error": "No song connection available", "success": False}
|
|
|
|
try:
|
|
orig_idx = int(original_track)
|
|
if orig_idx < 0 or orig_idx >= len(self._song.tracks):
|
|
return {"error": "Track index out of range", "success": False}
|
|
|
|
original = self._song.tracks[orig_idx]
|
|
orig_name = str(original.name)
|
|
|
|
# Determine track type
|
|
is_midi = getattr(original, "has_midi_input", False)
|
|
|
|
# Create new track of same type
|
|
if is_midi:
|
|
self._song.create_midi_track(-1)
|
|
else:
|
|
self._song.create_audio_track(-1)
|
|
|
|
compressed_idx = len(self._song.tracks) - 1
|
|
compressed = self._song.tracks[compressed_idx]
|
|
compressed_name = "%s (Comp)" % orig_name
|
|
compressed.name = compressed_name
|
|
|
|
# Copy volume/pan from original
|
|
try:
|
|
compressed.mixer_device.volume.value = original.mixer_device.volume.value
|
|
compressed.mixer_device.panning.value = original.mixer_device.panning.value
|
|
except Exception as e:
|
|
logger.warning("Could not copy mixer settings: %s" % str(e))
|
|
|
|
# Insert compressor and configure
|
|
comp_result = self._insert_and_configure_compressor(
|
|
compressed_idx,
|
|
ratio=settings.get("ratio", 4.0),
|
|
threshold=settings.get("threshold", -20.0),
|
|
attack=settings.get("attack", 10.0),
|
|
release=settings.get("release", 100.0),
|
|
makeup=settings.get("makeup_gain", 0.0)
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"original_track_index": orig_idx,
|
|
"compressed_track_index": compressed_idx,
|
|
"original_name": orig_name,
|
|
"compressed_name": compressed_name,
|
|
"is_midi": is_midi,
|
|
"compressor_configured": comp_result.get("configured", False),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error duplicating track: %s" % str(e))
|
|
return {"error": str(e), "success": False}
|
|
|
|
def blend_wet_dry(self, wet_track: int, dry_track: int,
|
|
mix_percent: float) -> Dict[str, Any]:
|
|
"""
|
|
Blend wet (compressed) and dry (original) tracks.
|
|
|
|
Args:
|
|
wet_track: Index of the compressed track
|
|
dry_track: Index of the original track
|
|
mix_percent: Blend percentage 0-100 (0 = dry only, 100 = wet only)
|
|
|
|
Returns:
|
|
Dict with blend status
|
|
"""
|
|
if self._song is None:
|
|
return {"error": "No song connection available", "success": False}
|
|
|
|
try:
|
|
wet_idx = int(wet_track)
|
|
dry_idx = int(dry_track)
|
|
mix = float(mix_percent) / 100.0 # Convert to 0.0-1.0
|
|
|
|
# Clamp mix
|
|
mix = max(0.0, min(1.0, mix))
|
|
|
|
# Calculate volumes
|
|
# When mix is 0.5 (50%), both tracks at full volume
|
|
# When mix is 0.0 (0%), dry at full, wet at 0
|
|
# When mix is 1.0 (100%), dry at 0, wet at full
|
|
dry_volume = 1.0 - (mix * 0.5) # At 50%, dry = 0.75
|
|
wet_volume = 0.5 + (mix * 0.5) # At 50%, wet = 0.75
|
|
|
|
# Apply volumes
|
|
dry_track_obj = self._song.tracks[dry_idx]
|
|
wet_track_obj = self._song.tracks[wet_idx]
|
|
|
|
dry_track_obj.mixer_device.volume.value = dry_volume
|
|
wet_track_obj.mixer_device.volume.value = wet_volume
|
|
|
|
return {
|
|
"success": True,
|
|
"dry_track": dry_idx,
|
|
"wet_track": wet_idx,
|
|
"mix_percent": mix_percent,
|
|
"dry_volume": dry_volume,
|
|
"wet_volume": wet_volume,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error blending tracks: %s" % str(e))
|
|
return {"error": str(e), "success": False}
|
|
|
|
def apply_preset(self, track_index: int, preset: CompressionPreset,
|
|
name: str = "") -> Dict[str, Any]:
|
|
"""
|
|
Apply a preset parallel compression chain.
|
|
|
|
Args:
|
|
track_index: Index of the track
|
|
preset: CompressionPreset enum value
|
|
name: Optional custom chain name
|
|
|
|
Returns:
|
|
Dict with application status
|
|
"""
|
|
if preset not in PARALLEL_PRESETS:
|
|
return {
|
|
"error": "Unknown preset: %s. Available: %s" % (
|
|
preset, [p.value for p in PARALLEL_PRESETS.keys()]
|
|
)
|
|
}
|
|
|
|
settings = PARALLEL_PRESETS[preset]
|
|
|
|
# Create the chain with preset settings
|
|
result = self.create_parallel_chain(
|
|
track_index=track_index,
|
|
ratio=settings.ratio,
|
|
threshold=settings.threshold,
|
|
makeup_gain=settings.makeup_gain,
|
|
name=name if name else preset.value
|
|
)
|
|
|
|
if result.get("success"):
|
|
result["preset_used"] = preset.value
|
|
result["preset_settings"] = settings.to_dict()
|
|
|
|
# Adjust blend based on preset
|
|
if "compressed_track" in result and "original_track" in result:
|
|
blend_pct = settings.dry_wet * 100.0
|
|
self.blend_wet_dry(
|
|
result["compressed_track"],
|
|
result["original_track"],
|
|
blend_pct
|
|
)
|
|
result["blend_percent"] = blend_pct
|
|
|
|
return result
|
|
|
|
def get_preset_settings(self, preset_name: str) -> Optional[ParallelChainSettings]:
|
|
"""
|
|
Get settings for a named preset.
|
|
|
|
Args:
|
|
preset_name: Name of preset (drum_parallel, vocal_parallel, bus_parallel)
|
|
|
|
Returns:
|
|
ParallelChainSettings or None
|
|
"""
|
|
try:
|
|
preset = CompressionPreset(preset_name)
|
|
return PARALLEL_PRESETS.get(preset)
|
|
except ValueError:
|
|
return None
|
|
|
|
def list_presets(self) -> List[Dict[str, Any]]:
|
|
"""List all available presets with descriptions."""
|
|
presets = []
|
|
for preset, settings in PARALLEL_PRESETS.items():
|
|
presets.append({
|
|
"name": preset.value,
|
|
"description": self._get_preset_description(preset),
|
|
"settings": settings.to_dict(),
|
|
})
|
|
return presets
|
|
|
|
def _get_preset_description(self, preset: CompressionPreset) -> str:
|
|
"""Get human-readable description for a preset."""
|
|
descriptions = {
|
|
CompressionPreset.DRUM_PARALLEL:
|
|
"Aggressive 8:1 ratio with fast attack/release for drum punch and impact",
|
|
CompressionPreset.VOCAL_PARALLEL:
|
|
"Smooth 4:1 ratio with medium timing for vocal presence and control",
|
|
CompressionPreset.BUS_PARALLEL:
|
|
"Gentle 2:1 ratio with slow timing for bus glue and cohesion",
|
|
}
|
|
return descriptions.get(preset, "Custom preset")
|
|
|
|
def _insert_and_configure_compressor(self, track_index: int,
|
|
ratio: float,
|
|
threshold: float,
|
|
attack: float,
|
|
release: float,
|
|
makeup: float) -> Dict[str, Any]:
|
|
"""
|
|
Insert and configure a compressor on a track.
|
|
|
|
Args:
|
|
track_index: Track index
|
|
ratio: Compression ratio
|
|
threshold: Threshold in dB
|
|
attack: Attack time in ms
|
|
release: Release time in ms
|
|
makeup: Makeup gain in dB
|
|
|
|
Returns:
|
|
Dict with configuration status
|
|
"""
|
|
try:
|
|
track = self._song.tracks[int(track_index)]
|
|
|
|
# Try to find existing compressor
|
|
compressor = None
|
|
for d in track.devices:
|
|
name = str(d.name).lower()
|
|
if "compressor" in name and "glue" not in name:
|
|
compressor = d
|
|
break
|
|
|
|
configured = False
|
|
if compressor and hasattr(compressor, "parameters"):
|
|
for param in compressor.parameters:
|
|
param_name = str(param.name).lower()
|
|
try:
|
|
if "ratio" in param_name:
|
|
param.value = float(ratio)
|
|
configured = True
|
|
elif "threshold" in param_name:
|
|
param.value = float(threshold)
|
|
elif "attack" in param_name:
|
|
param.value = float(attack)
|
|
elif "release" in param_name:
|
|
param.value = float(release)
|
|
elif "makeup" in param_name or "gain" in param_name:
|
|
param.value = float(makeup)
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"configured": configured,
|
|
"device_found": compressor is not None,
|
|
"device_name": str(compressor.name) if compressor else None,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error configuring compressor: %s" % str(e))
|
|
return {"configured": False, "error": str(e)}
|
|
|
|
def get_chain(self, name: str) -> Optional[ParallelChain]:
|
|
"""Get a parallel chain by name."""
|
|
return self._chains.get(name)
|
|
|
|
def list_chains(self) -> List[Dict[str, Any]]:
|
|
"""List all active parallel chains."""
|
|
return [chain.to_dict() for chain in self._chains.values()]
|
|
|
|
def remove_chain(self, name: str) -> Dict[str, Any]:
|
|
"""
|
|
Remove a parallel chain.
|
|
|
|
Note: This only removes the chain from tracking, not the actual tracks.
|
|
|
|
Args:
|
|
name: Name of the chain to remove
|
|
|
|
Returns:
|
|
Dict with removal status
|
|
"""
|
|
if name in self._chains:
|
|
chain = self._chains.pop(name)
|
|
return {
|
|
"removed": True,
|
|
"chain": chain.to_dict(),
|
|
"note": "Chain removed from tracking. Tracks remain in project."
|
|
}
|
|
return {"removed": False, "error": "Chain '%s' not found" % name}
|
|
|
|
|
|
# Module-level convenience functions
|
|
|
|
def create_parallel_compression(ableton_conn, track_index: int,
|
|
ratio: float = 4.0,
|
|
threshold: float = -20.0,
|
|
makeup_gain: float = 0.0,
|
|
name: str = "") -> Dict[str, Any]:
|
|
"""
|
|
Create a parallel compression chain (module-level convenience function).
|
|
|
|
Args:
|
|
ableton_conn: Ableton Live connection
|
|
track_index: Track index
|
|
ratio: Compression ratio
|
|
threshold: Threshold in dB
|
|
makeup_gain: Makeup gain in dB
|
|
name: Optional chain name
|
|
|
|
Returns:
|
|
Dict with creation status
|
|
"""
|
|
comp = ParallelCompression(ableton_conn)
|
|
return comp.create_parallel_chain(track_index, ratio, threshold, makeup_gain, name)
|
|
|
|
|
|
def apply_preset(ableton_conn, track_index: int, preset_name: str,
|
|
name: str = "") -> Dict[str, Any]:
|
|
"""
|
|
Apply a preset parallel compression (module-level convenience function).
|
|
|
|
Args:
|
|
ableton_conn: Ableton Live connection
|
|
track_index: Track index
|
|
preset_name: Preset name (drum_parallel, vocal_parallel, bus_parallel)
|
|
name: Optional custom chain name
|
|
|
|
Returns:
|
|
Dict with application status
|
|
"""
|
|
try:
|
|
preset = CompressionPreset(preset_name)
|
|
comp = ParallelCompression(ableton_conn)
|
|
return comp.apply_preset(track_index, preset, name)
|
|
except ValueError:
|
|
return {
|
|
"error": "Unknown preset '%s'. Available: %s" % (
|
|
preset_name, [p.value for p in CompressionPreset]
|
|
)
|
|
}
|
|
|
|
|
|
def list_presets() -> List[Dict[str, Any]]:
|
|
"""List all available parallel compression presets."""
|
|
comp = ParallelCompression(None)
|
|
return comp.list_presets()
|
|
|
|
|
|
# Global instance for caching
|
|
_parallel_compression_instance: Optional[ParallelCompression] = None
|
|
|
|
|
|
def get_parallel_compression(ableton_conn=None) -> ParallelCompression:
|
|
"""Get or create the global ParallelCompression instance."""
|
|
global _parallel_compression_instance
|
|
if _parallel_compression_instance is None:
|
|
_parallel_compression_instance = ParallelCompression(ableton_conn)
|
|
elif ableton_conn is not None:
|
|
_parallel_compression_instance.conn = ableton_conn
|
|
_parallel_compression_instance._song = ableton_conn._song if hasattr(ableton_conn, '_song') else None
|
|
return _parallel_compression_instance
|