app.py

"""Second Ear — a realtime music production assistant.

Three surfaces over one analysis engine:

  Live       a rolling window of whatever the browser is hearing, metered
             against genre targets, with findings that fire as they happen.
  Bounce     measurement-grade pass on a rendered file, plus the semantic
             layer, the written critique, and an Ableton action plan.
  Bridge     how to wire the plan into a real Live set through MCP.

Every analysis endpoint is also an MCP tool, so an agent that already has
Ableton MCP connected can use this Space as its ears and its own Ableton
connection as its hands.
"""

from __future__ import annotations

import json

import gradio as gr
import numpy as np
import soundfile as sf

from ear import ableton, dsp, knowledge, llm, render, semantic

LIVE_WINDOW_S = 8.0        # what the meters describe
CAPTURE_S = 30.0           # how far back "analyse what just happened" reaches
TEMPO_EVERY = 8            # ticks between tempo/key refresh (they need context)

semantic.SEMANTIC.start()


# --------------------------------------------------------------------------
# shared helpers
# --------------------------------------------------------------------------

def _read(path: str) -> tuple[int, np.ndarray]:
    data, sr = sf.read(path, dtype="float32", always_2d=True)
    return sr, data


def _blank_state() -> dict:
    return {"buf": np.zeros((0, 2), dtype=np.float32), "tick": 0,
            "rhythm": {"bpm": 0.0, "confidence": 0.0, "onset_rate": 0.0},
            "key": {"key": "—", "confidence": 0.0}}


# --------------------------------------------------------------------------
# live loop
# --------------------------------------------------------------------------

def live_tick(chunk, state, genre, source):
    """Fold one streamed chunk into the rolling window and re-meter."""
    state = state or _blank_state()
    if chunk is None:
        return render.idle("waiting for signal…"), render.idle("no findings yet"), state

    sr, data = chunk
    incoming = dsp.to_float_stereo(sr, data)
    if incoming.shape[0] == 0:
        return gr.skip(), gr.skip(), state

    buf = np.concatenate([state["buf"], incoming], axis=0)
    keep = int(CAPTURE_S * dsp.SR)
    if buf.shape[0] > keep:
        buf = buf[-keep:]
    state["buf"] = buf
    state["tick"] += 1

    window = buf[-int(LIVE_WINDOW_S * dsp.SR):]
    if float(np.max(np.abs(window))) < 1e-4:
        return render.idle("signal is silent — check the input device"), gr.skip(), state

    rep = dsp.analyze(dsp.SR, window, fast=True)
    if rep is None:
        return gr.skip(), gr.skip(), state

    # Tempo and key need a longer view than the meter window, and cost more,
    # so they refresh on their own slower clock and are carried between ticks.
    if state["tick"] % TEMPO_EVERY == 1 and buf.shape[0] > 12 * dsp.SR:
        mono = buf[-int(16 * dsp.SR):].mean(axis=1)
        flux, fps = dsp.onset_envelope(mono)
        state["rhythm"] = dsp.tempo_from_onsets(flux, fps)
        freqs, power = dsp.spectrum(mono)
        state["key"] = dsp.key_estimate(freqs, power)
    rep.rhythm, rep.key = state["rhythm"], state["key"]

    diags = knowledge.diagnose(rep, genre, source)
    verdict, tone = knowledge.headline_verdict(rep, diags)
    held = min(buf.shape[0] / dsp.SR, CAPTURE_S)

    return (
        render.meters(rep, genre, verdict, tone,
                      extra=f"{LIVE_WINDOW_S:.0f}s window · {held:.0f}s held"),
        render.cards(diags, limit=4),
        state,
    )


def analyse_capture(state, genre, source, intent, use_llm):
    """Run the full pass on whatever the live loop has been holding."""
    state = state or _blank_state()
    buf = state.get("buf")
    if buf is None or buf.shape[0] < dsp.SR:
        return (render.idle("nothing captured yet — start listening first"),
                "Not enough audio held to analyse.", "{}")
    return _full_pass(dsp.SR, buf, genre, source, intent, use_llm)


# --------------------------------------------------------------------------
# full pass (shared by the bounce tab and the API)
# --------------------------------------------------------------------------

def _full_pass(sr, data, genre, source, intent, use_llm):
    rep = dsp.analyze(sr, data)
    if rep is None:
        return render.idle("clip too short"), "Clip too short to analyse.", "{}"

    diags = knowledge.diagnose(rep, genre, source)
    verdict, tone = knowledge.headline_verdict(rep, diags)

    mono = dsp.to_float_stereo(sr, data).mean(axis=1)
    tags = semantic.SEMANTIC.describe(mono)
    tag_line = semantic.tags_line(tags)

    plan = ableton.build_plan(diags, genre=genre, bpm=rep.rhythm.get("bpm", 0.0))

    parts = [render.meters(rep, genre, verdict, tone,
                           extra=f"{rep.duration:.1f}s · {genre}")]
    if tag_line:
        parts.append(
            f'{render.STYLE}<div class="se-wrap" style="margin-top:10px">'
            f'<div class="se-num-k">sounds like</div>'
            f'<div style="font-size:13px;margin-top:5px">{tag_line}</div></div>'
        )
    parts.append(f'<div style="margin-top:10px">{render.cards(diags)}</div>')

    written = [f"### {verdict}", ""]
    if tag_line:
        written.append(f"*Sounds like: {tag_line}*\n")
    if use_llm:
        note = llm.critique(rep.to_dict(), [d.to_dict() for d in diags],
                            tags, genre, source, intent)
        written += [note, "", "---", ""] if note else [f"*{llm.available()[1]}*", ""]
    written += ["## Ableton action plan", "", ableton.plan_to_markdown(plan)]

    export = {"report": rep.to_dict(),
              "findings": [d.to_dict() for d in diags],
              "sounds_like": {g: [t for t, _ in v] for g, v in tags.items()},
              "ableton_plan": plan}

    return "".join(parts), "\n".join(written), json.dumps(export, indent=2)


def analyse_file(audio_path: str, genre: str = "Dubstep / Riddim",
                 source: str = "Full mix / master", intent: str = "",
                 use_llm: bool = True):
    """Analyse a rendered audio file and return a full production report.

    Args:
        audio_path: path to the audio file to listen to (wav, mp3, flac, aiff).
        genre: which target window to judge against, e.g. "Dubstep / Riddim".
        source: what the audio is — "Full mix / master", "Drum bus", "Bass / 808",
            "Lead / synth", "Vocal" or "Pad / atmosphere".
        intent: optional free text describing what you were going for.
        use_llm: include the written engineer's critique (needs HF_TOKEN on the Space).
    """
    if not audio_path:
        return render.idle("load a file first"), "No audio supplied.", "{}"
    sr, data = _read(audio_path)
    return _full_pass(sr, data, genre, source, intent, use_llm)


def measure(audio_path: str) -> str:
    """Measure an audio file and return the raw metrics as JSON.

    Loudness (LUFS-I/S, LRA), true peak, crest factor, seven-band balance,
    band ratios, stereo correlation and width, tempo and key. No opinions.

    Args:
        audio_path: path to the audio file to measure.
    """
    if not audio_path:
        return json.dumps({"error": "no audio supplied"})
    sr, data = _read(audio_path)
    rep = dsp.analyze(sr, data)
    if rep is None:
        return json.dumps({"error": "clip too short"})
    return json.dumps(rep.to_dict(), indent=2)


def health() -> str:
    """Report which optional layers are live: the CLAP semantic ear and the
    written critique. Returns JSON."""
    return json.dumps({
        "semantic": semantic.SEMANTIC.status(),
        "semantic_ready": semantic.SEMANTIC.ready,
        "critique": llm.available()[1],
    }, indent=2)


def _status_line() -> str:
    return (f"<sub>Semantic ear: {semantic.SEMANTIC.status()} · "
            f"written critique: {llm.available()[1]}</sub>")


def ableton_plan(audio_path: str, genre: str = "Dubstep / Riddim",
                 source: str = "Full mix / master",
                 track_index: str = "$MASTER") -> str:
    """Return an Ableton Live action plan for an audio file, as MCP call JSON.

    The plan is a sequence of Ableton MCP tool calls (load_instrument_or_effect,
    get_device_parameters, set_device_parameter, …) that an agent with a local
    Ableton MCP server connected can execute directly against a running set.

    Args:
        audio_path: path to the audio file to analyse.
        genre: target window to judge against.
        source: what the audio is (master, drum bus, bass, vocal, …).
        track_index: which Live track the plan targets. "$MASTER" for the master.
    """
    if not audio_path:
        return json.dumps({"error": "no audio supplied"})
    sr, data = _read(audio_path)
    rep = dsp.analyze(sr, data)
    if rep is None:
        return json.dumps({"error": "clip too short"})
    diags = knowledge.diagnose(rep, genre, source)
    plan = ableton.build_plan(diags, track_index=track_index, genre=genre,
                              bpm=rep.rhythm.get("bpm", 0.0))
    return json.dumps(plan, indent=2)


# --------------------------------------------------------------------------
# UI
# --------------------------------------------------------------------------

CSS = """
#col-container{max-width:1180px;margin:0 auto;}
.dark .gradio-container{color:var(--body-text-color);}
#se-title h1{font-size:30px;letter-spacing:-0.02em;margin-bottom:2px;}
"""

GENRE_CHOICES = list(knowledge.GENRES.keys())

BRIDGE = """
## Wiring it into Live

This Space listens and decides. It does not touch your set — nothing hosted on
someone else's machine can, and a tool that pretended otherwise would be lying
to you. The split is deliberate:

| | |
|---|---|
| **Second Ear** (this Space) | ears + judgement — measures, diagnoses, writes the plan |
| **Ableton MCP** (your machine) | hands — executes the plan against the live set |

### 1. Point your agent at both

Every endpoint here is exposed as an MCP tool. Add this Space alongside your
existing Ableton MCP server:

```json
{
  "mcpServers": {
    "second-ear": {
      "command": "npx",
      "args": ["mcp-remote", "https://apolithosstudios-second-ear.hf.space/gradio_api/mcp/sse"]
    },
    "ableton": { "command": "...your existing Ableton MCP entry..." }
  }
}
```

Tools you get: `analyse_file`, `measure`, `ableton_plan`.

### 2. Ask for the loop

> "Bounce the drop, run it through second-ear as Dubstep / Riddim, then execute
> the plan on my master."

The agent calls `ableton_plan`, gets back Ableton MCP calls, and runs them.
Every parameter write is preceded by a `get_device_parameters` probe, because
Live's parameter names move between versions — the plan resolves names at
execution time instead of guessing.

### 3. Feeding it live audio

The Live tab listens to whatever the browser's input device is. To point it at
your master bus instead of the room:

1. Install a loopback driver — **BlackHole** (free) or **Loopback**.
2. In Live, set the output (or a dedicated send) to that device.
3. Pick it as the input when the browser asks for microphone permission.

One honest caveat: browsers apply echo cancellation, noise suppression and auto
gain to captured audio by default. That is fine for *direction* — balance
drifting, sub running hot, the drop losing punch — and it is not fine for
absolute numbers. **For measurement-grade LUFS and true peak, bounce a file and
use the Bounce tab.** The Live tab is the ear on your shoulder; the Bounce tab
is the meter.
"""


with gr.Blocks(title="Second Ear") as demo:
    with gr.Column(elem_id="col-container"):
        gr.Markdown(
            "# Second Ear\n"
            "A realtime production assistant that listens to what you're making, "
            "tells you what's wrong in engineer's language, and hands your agent "
            "an Ableton plan to fix it.",
            elem_id="se-title",
        )

        with gr.Row():
            genre = gr.Dropdown(GENRE_CHOICES, value=GENRE_CHOICES[0],
                                label="Target sound", scale=2)
            source = gr.Dropdown(knowledge.SOURCES, value=knowledge.SOURCES[0],
                                 label="Listening to", scale=2)

        with gr.Tabs():
            # ---------------------------------------------------------- live
            with gr.Tab("Live"):
                gr.Markdown(
                    "Route your master through a loopback device, start the input, "
                    "and leave it running. The meters show a rolling 8-second "
                    "window against the target you picked; findings fire as they "
                    "happen. The ghost block behind each bar is where that band "
                    "should sit for this genre."
                )
                live_in = gr.Audio(sources=["microphone"], streaming=True,
                                   type="numpy", label="Studio input")
                live_meters = gr.HTML(render.idle("waiting for signal…"))
                live_notes = gr.HTML(render.idle("no findings yet"))

                with gr.Row():
                    capture_btn = gr.Button("Analyse the last 30 seconds",
                                            variant="primary", scale=2)
                    live_llm = gr.Checkbox(value=True, label="Written critique",
                                           scale=1)
                live_intent = gr.Textbox(
                    label="What were you going for? (optional)",
                    placeholder="heavier drop, needs to hit on a club rig",
                    lines=1,
                )
                cap_report = gr.HTML()
                cap_text = gr.Markdown()
                with gr.Accordion("Raw export (JSON)", open=False):
                    cap_json = gr.Code(language="json")

                state = gr.State(_blank_state())
                live_in.stream(
                    live_tick,
                    inputs=[live_in, state, genre, source],
                    outputs=[live_meters, live_notes, state],
                    stream_every=0.5,
                    show_progress="hidden",
                    concurrency_limit=None,
                )
                capture_btn.click(
                    analyse_capture,
                    inputs=[state, genre, source, live_intent, live_llm],
                    outputs=[cap_report, cap_text, cap_json],
                    api_name=False,   # gr.State can't cross the MCP boundary
                )

            # -------------------------------------------------------- bounce
            with gr.Tab("Bounce"):
                gr.Markdown(
                    "Measurement-grade pass on a rendered file — real LUFS, real "
                    "true peak, the semantic layer, the written critique, and the "
                    "Ableton plan."
                )
                with gr.Row():
                    file_in = gr.Audio(sources=["upload", "microphone"],
                                       type="filepath", label="Bounce")
                    with gr.Column():
                        file_intent = gr.Textbox(
                            label="What were you going for? (optional)",
                            placeholder="dark and heavy, has to survive a club system",
                            lines=2,
                        )
                        file_llm = gr.Checkbox(value=True, label="Written critique")
                        run_btn = gr.Button("Listen", variant="primary")

                file_report = gr.HTML()
                file_text = gr.Markdown()
                with gr.Accordion("Raw export (JSON)", open=False):
                    file_json = gr.Code(language="json")

                run_btn.click(
                    analyse_file,
                    inputs=[file_in, genre, source, file_intent, file_llm],
                    outputs=[file_report, file_text, file_json],
                    api_name="analyse_file",
                )

                with gr.Accordion("Numbers only (no opinions)", open=False):
                    meas_btn = gr.Button("Measure")
                    meas_out = gr.Code(language="json", label="Metrics")
                    meas_btn.click(measure, inputs=[file_in], outputs=meas_out,
                                   api_name="measure")

            # -------------------------------------------------------- bridge
            with gr.Tab("Ableton bridge"):
                gr.Markdown(BRIDGE)
                with gr.Row():
                    plan_audio = gr.Audio(type="filepath", label="Bounce")
                    with gr.Column():
                        plan_track = gr.Textbox("$MASTER", label="Target track index")
                        plan_btn = gr.Button("Build the plan", variant="primary")
                plan_out = gr.Code(language="json", label="Ableton MCP calls")
                plan_btn.click(
                    ableton_plan,
                    inputs=[plan_audio, genre, source, plan_track],
                    outputs=plan_out,
                    api_name="ableton_plan",
                )

        # Rendered per page load, not at import — the CLAP ear finishes warming
        # up well after the Blocks tree is built.
        status = gr.Markdown(_status_line())
        demo.load(_status_line, outputs=status, api_name=False)
        gr.Button("health", visible=False).click(
            health, outputs=gr.Textbox(visible=False), api_name="health")


if __name__ == "__main__":
    # Gradio 6 moved theme and css off the Blocks constructor onto launch().
    demo.queue(default_concurrency_limit=4).launch(
        theme=gr.themes.Citrus(),
        css=CSS,
        mcp_server=True,
    )