Spaces:
Sleeping
Sleeping
| """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, | |
| ) | |