ear/ableton.py

"""Diagnosis → executable Ableton Live moves.

This Space is the ears. It has no route into your Live set, and pretending
otherwise would be a lie. What it emits instead is a plan in the exact shape
the Ableton MCP server expects, so an agent with that server connected
locally can execute it verbatim.

Parameter names differ between Live versions and device presets, so every
step that sets a parameter is preceded by a `get_device_parameters` probe and
carries a `match` hint. The executing agent resolves the real name rather
than trusting a hardcoded string.
"""

from __future__ import annotations

import json
from typing import Any

from .knowledge import Diagnosis

# Browser paths as they appear in Live's own device tree. `load_instrument_or_effect`
# wants a browser URI, which the agent resolves with `get_browser_items_at_path`.
DEVICE_PATHS = {
    "EQ Eight": "Audio Effects/EQ Eight",
    "EQ Three": "Audio Effects/EQ Three",
    "Utility": "Audio Effects/Utility",
    "Glue Compressor": "Audio Effects/Glue Compressor",
    "Compressor": "Audio Effects/Compressor",
    "Multiband Dynamics": "Audio Effects/Multiband Dynamics",
    "Limiter": "Audio Effects/Limiter",
    "Saturator": "Audio Effects/Saturator",
    "Drum Buss": "Audio Effects/Drum Buss",
    "Roar": "Audio Effects/Roar",
    "Dynamic Tube": "Audio Effects/Dynamic Tube",
    "Auto Filter": "Audio Effects/Auto Filter",
    "Hybrid Reverb": "Audio Effects/Hybrid Reverb",
    "Spectrum": "Audio Effects/Spectrum",
    "Operator": "Instruments/Operator",
}


def _load(track: Any, device: str, note: str = "") -> dict:
    return {
        "tool": "load_instrument_or_effect",
        "args": {"track_index": track, "uri": DEVICE_PATHS.get(device, device)},
        "device": device,
        "resolve": "call get_browser_tree / get_browser_items_at_path to turn "
                   "this path into a concrete browser URI",
        "skip_if": f"{device} already exists on this track",
        "note": note,
    }


def _probe(track: Any, device: str) -> dict:
    return {
        "tool": "get_device_parameters",
        "args": {"track_index": track, "device_index": "$LAST"},
        "purpose": f"resolve the real parameter names for {device} in this Live version",
    }


def _set(track: Any, match: str, value: Any, why: str) -> dict:
    return {
        "tool": "set_device_parameter",
        "args": {"track_index": track, "device_index": "$LAST",
                 "parameter_name": match, "value": value},
        "match": match,
        "why": why,
    }


# --------------------------------------------------------------------------
# per-diagnosis recipes
# --------------------------------------------------------------------------

def _recipe(d: Diagnosis, track: Any) -> list[dict] | None:
    if d.id in ("clipping", "no_headroom"):
        return [
            _load(track, "Limiter", "master chain, last in the signal path"),
            _probe(track, "Limiter"),
            _set(track, "Ceiling", -1.0, "true peak target of -1.0 dBTP survives lossy encoding"),
            _set(track, "Release", 200.0, "slow enough not to pump on sustained low end"),
        ]

    if d.id == "too_loud":
        return [
            _probe(track, "Limiter"),
            _set(track, "Gain", round(-d.amount, 1),
                 f"back off {d.amount:.1f} dB of limiter drive"),
            _load(track, "Saturator", "recover density without more ceiling"),
            _probe(track, "Saturator"),
            _set(track, "Drive", 3.0, "harmonic weight instead of gain"),
            _set(track, "Dry/Wet", 35.0, "parallel amount"),
        ]

    if d.id == "over_limited":
        return [
            _probe(track, "Limiter"),
            _set(track, "Gain", -2.5, "give the transients back 2.5 dB"),
            _load(track, "Glue Compressor", "parallel glue on the drum bus, not the master"),
            _probe(track, "Glue Compressor"),
            _set(track, "Ratio", 4.0, "parallel crush setting"),
            _set(track, "Attack", 30.0, "let the stick through before gain reduction"),
            _set(track, "Release", 0.4, "follows the groove"),
            _set(track, "Dry/Wet", 30.0, "parallel, keeps the original transient"),
        ]

    if d.id in ("mud", "hollow"):
        gain = -abs(d.amount) if d.id == "mud" else 2.0
        return [
            _load(track, "EQ Eight", f"surgical work at {d.freq:.0f} Hz"),
            _probe(track, "EQ Eight"),
            _set(track, "2 Filter Type A", "Bell", "band 2 as a bell"),
            _set(track, "2 Frequency A", round(d.freq, 1), f"centre on the problem at {d.freq:.0f} Hz"),
            _set(track, "2 Gain A", round(gain, 1), d.headline),
            _set(track, "2 Resonance A", 1.4 if d.id == "mud" else 0.7,
                 "narrow to cut, wide to add"),
        ]

    if d.id == "harsh":
        return [
            _load(track, "Multiband Dynamics", "dynamic control at 3 kHz, not a static cut"),
            _probe(track, "Multiband Dynamics"),
            _set(track, "Low Crossover", 1500.0, "isolate 1.5–4k as the mid band"),
            _set(track, "High Crossover", 4000.0, "isolate 1.5–4k as the mid band"),
            _set(track, "Above Threshold (Mid)", -18.0, "engage only on peaks"),
            _set(track, "Above Ratio (Mid)", 3.0, f"{d.amount:.1f} dB of movement at the top"),
        ]

    if d.id in ("dull", "brittle"):
        gain = 2.0 if d.id == "dull" else -2.0
        return [
            _load(track, "EQ Eight", "high shelf"),
            _probe(track, "EQ Eight"),
            _set(track, "8 Filter Type A", "High Shelf", "band 8 as a shelf"),
            _set(track, "8 Frequency A", round(d.freq, 1), f"shelf from {d.freq/1000:.0f} kHz up"),
            _set(track, "8 Gain A", gain, d.headline),
        ]

    if d.id in ("stereo_sub", "out_of_phase"):
        return [
            _load(track, "Utility", "before the limiter, after everything else"),
            _probe(track, "Utility"),
            _set(track, "Bass Mono", 1, "collapse the low end to mono"),
            _set(track, "Bass Mono Frequency", 120.0, "everything under 120 Hz centred"),
        ]

    if d.id == "sub_heavy":
        return [
            _load(track, "EQ Eight", "control the shelf under 60 Hz"),
            _probe(track, "EQ Eight"),
            _set(track, "1 Filter Type A", "High Pass 48", "steep high-pass"),
            _set(track, "1 Frequency A", 29.0, "kill infrasonic energy the system cannot use"),
            _set(track, "2 Filter Type A", "Low Shelf", "tame the remaining sub"),
            _set(track, "2 Frequency A", 60.0, "shelf at the sub/bass boundary"),
            _set(track, "2 Gain A", round(-abs(d.amount), 1), d.headline),
        ]

    if d.id == "no_sub":
        return [
            {"tool": "create_midi_track", "args": {"index": -1},
             "why": "dedicated sub track — layering it on the bass channel means "
                    "the sub inherits the bass processing, which is what killed it"},
            _load("$NEW", "Operator", "single sine partial"),
            _probe("$NEW", "Operator"),
            _set("$NEW", "Oscillator A Coarse", 1, "fundamental only"),
            _load("$NEW", "Utility", "force mono"),
            _set("$NEW", "Width", 0.0, "sub is mono, always"),
        ]

    if d.id == "over_wide":
        return [
            _load(track, "Utility", "pull the image back in"),
            _probe(track, "Utility"),
            _set(track, "Width", 110.0, "down from whatever the widener is doing"),
            _set(track, "Bass Mono", 1, "and keep the low end centred regardless"),
        ]

    if d.id == "narrow":
        return [
            _load(track, "Utility", "apply to reverb returns and pads, NOT bass or kick"),
            _probe(track, "Utility"),
            _set(track, "Width", 130.0, "widen only the elements that can afford it"),
        ]

    if d.id == "flat_drums":
        return [
            _load(track, "Drum Buss", "transient shaping plus drive in one device"),
            _probe(track, "Drum Buss"),
            _set(track, "Transients", 35.0, "bring the attack back"),
            _set(track, "Drive", 15.0, "density without the limiter"),
            _set(track, "Crunch", 10.0, "top-end bite"),
        ]

    if d.id == "sibilance":
        return [
            _load(track, "Multiband Dynamics", "de-ess band"),
            _probe(track, "Multiband Dynamics"),
            _set(track, "High Crossover", 6500.0, "isolate the sibilant band"),
            _set(track, "Above Threshold (High)", -22.0, "catch only the esses"),
            _set(track, "Above Ratio (High)", 4.0, "3 dB of range on peaks"),
        ]

    if d.id == "loud_range":
        return [
            _load(track, "Glue Compressor", "slow master glue"),
            _probe(track, "Glue Compressor"),
            _set(track, "Ratio", 2.0, "gentle"),
            _set(track, "Attack", 30.0, "let transients through"),
            _set(track, "Release", 0.6, "follows sections, not hits"),
            _set(track, "Threshold", -14.0, "aim for 1–2 dB of gain reduction"),
        ]

    return None


# --------------------------------------------------------------------------
# plan assembly
# --------------------------------------------------------------------------

def build_plan(
    diags: list[Diagnosis],
    *,
    track_index: Any = "$MASTER",
    genre: str = "",
    bpm: float = 0.0,
    max_steps: int = 4,
) -> dict:
    """Turn the top diagnoses into an Ableton MCP call sequence."""
    steps = []
    for d in diags:
        if d.severity == "good":
            continue
        calls = _recipe(d, track_index)
        if not calls:
            continue
        steps.append({
            "step": len(steps) + 1,
            "severity": d.severity,
            "intent": d.headline,
            "evidence": d.evidence,
            "expected_result": d.move,
            "calls": calls,
        })
        if len(steps) >= max_steps:
            break

    preflight = [
        {"tool": "get_session_info", "args": {},
         "purpose": "confirm the set is reachable and read the track count"},
        {"tool": "get_track_info", "args": {"track_index": track_index},
         "purpose": "read the existing device chain so steps can be skipped "
                    "when the device is already there"},
    ]
    if bpm > 0:
        preflight.append({
            "tool": "set_tempo", "args": {"tempo": round(bpm)},
            "purpose": f"detected {bpm:.1f} BPM in the incoming audio",
            "confirm_first": True,
        })

    return {
        "target": {"track_index": track_index, "genre": genre},
        "contract": (
            "Execute against a locally connected Ableton MCP server. "
            "$MASTER = the master track index, $LAST = the device index returned "
            "by the preceding load call, $NEW = the track index returned by the "
            "preceding create call. Run every get_device_parameters probe and "
            "match parameter names by substring before calling "
            "set_device_parameter — names differ across Live versions."
        ),
        "preflight": preflight,
        "steps": steps,
        "rollback": "Every step is additive. To undo, delete_device on the "
                    "devices added by this plan, newest first.",
    }


def plan_to_markdown(plan: dict) -> str:
    if not plan.get("steps"):
        return "No corrective moves needed — nothing measurable is wrong."

    lines = ["**Preflight**", ""]
    for c in plan["preflight"]:
        lines.append(f"- `{c['tool']}({json.dumps(c['args'])})` — {c['purpose']}")
    lines.append("")

    for step in plan["steps"]:
        badge = {"critical": "🔴", "warn": "🟠", "note": "🔵"}.get(step["severity"], "·")
        lines.append(f"**{badge} Step {step['step']} — {step['intent']}**")
        lines.append("")
        lines.append(f"> {step['evidence']}")
        lines.append("")
        for c in step["calls"]:
            args = json.dumps(c["args"], ensure_ascii=False)
            reason = c.get("why") or c.get("purpose") or c.get("note") or ""
            lines.append(f"- `{c['tool']}({args})`" + (f" — {reason}" if reason else ""))
        lines.append("")
        lines.append(f"*What you should hear:* {step['expected_result']}")
        lines.append("")

    return "\n".join(lines)