"""audio-brief MVP v1 — Gradio UI.
Run: python app.py
Drop a file → 40-60 s later you get a one-page brief: a client-facing
paragraph, the structured data, and three copy-paste outputs for the next
workflow step (SA3 variation, SA3 match-style, Ableton clip plan).
"""
from __future__ import annotations
import json
import secrets
import sys
import traceback
import urllib.parse
from pathlib import Path
import gradio as gr
### crate_picker rationale
# Gradio v6's gr.Radio strict-validates the incoming value against the
# component's `choices` attribute on the server side. We use the picker
# as a dynamic chip strip over the on-disk crate — `choices` is updated
# via `gr.update(choices=…)` from handlers — but in v6 those updates
# only flow to the client UI, not to the server-side validator, so
# legitimate tile ids get rejected as "not in choices".
#
# gr.Dropdown(allow_custom_value=True) skips that strict check and lets
# any string through to our handlers (which already look up the tile id
# against the crate dir themselves). We keep the strip look via CSS
# (.dc-crate-strip in theme.py) — the visual is unchanged.
import outputs
import share
import waveform
import wallet
import sa3
import crate
from narrative import (
write_brief, write_mix_chain, write_audio_only,
write_sa3_prompt_blended, write_sa3_prompt, write_sa3_prompt_lens,
write_sa3_prompt_upload_lens,
)
import morpho
from pipeline import BPM_PRIORS, Analysis, analyze, to_json
from theme import (
THEME, CUSTOM_CSS,
brand_html, pollen_pill_html, metric_grid_html,
session_spend_pill_html, cost_pip_html,
palette_toggle_html,
)
GENRE_CHOICES = [
"default", "dnb", "jungle", "trap", "hip-hop", "house",
"deep-house", "techno", "ambient", "downtempo", "pop", "rock",
]
# Model dropdowns are populated at startup from Pollinations' /v1/models
# catalog (see models.py), filtered by modality. A/B get text→text models;
# C gets audio-input multimodals only. Falls back to a curated list if the
# catalog fetch fails.
import models as _models
TEXT_MODELS = _models.text_models()
AUDIO_MODEL_CHOICES = _models.audio_model_choices() # [(label, value), ...]
DEFAULT_MODEL_A = "claude" if "claude" in TEXT_MODELS else (TEXT_MODELS[0] if TEXT_MODELS else "openai")
DEFAULT_MODEL_B = "openai-large" if "openai-large" in TEXT_MODELS else (
"openai" if "openai" in TEXT_MODELS else (TEXT_MODELS[1] if len(TEXT_MODELS) > 1 else "openai")
)
DEFAULT_MODEL_C = "gemini" if "gemini" in [v for _, v in AUDIO_MODEL_CHOICES] else "openai-audio"
def _short_caption(a: Analysis) -> str:
parts = []
if a.bpm:
parts.append(f"{int(round(a.bpm))} BPM")
if a.key and a.key_mode:
parts.append(f"{a.key} {a.key_mode}")
if a.tags_mood:
parts.append(a.tags_mood[0]["label"])
if a.tags_genre:
parts.append(a.tags_genre[0]["label"])
if a.duration_s:
parts.append(f"{int(round(a.duration_s))}s")
return " · ".join(parts) if parts else "(empty)"
def _metrics_html(a: Analysis | None) -> str:
"""6-up metric tile grid for the Analysis tab.
Renders the measured numbers the design highlights: BPM, KEY, LUFS,
TRUE PK, LRA, LENGTH. Empty placeholders when a stage hasn't filled
the value yet."""
if a is None:
return metric_grid_html([
("BPM", "—"), ("KEY", "—"), ("LUFS", "—"),
("TRUE PK", "—"), ("LRA", "—"), ("LENGTH", "—"),
])
bpm = "—" if a.bpm is None else f"{a.bpm:.0f}"
if a.key and a.key_mode:
key = f"{a.key} {a.key_mode[:3]}"
elif a.key:
key = a.key
else:
key = "—"
# LUFS-I and true peak are always negative dB; render with a real minus sign.
lufs = "—" if a.lufs_i is None else f"−{abs(a.lufs_i):.1f}"
peak = "—" if a.true_peak_db is None else f"−{abs(a.true_peak_db):.1f}"
lra = "—" if a.lufs_lra is None else f"{a.lufs_lra:.1f}"
if a.duration_s:
secs = int(round(a.duration_s))
length = f"{secs // 60}:{secs % 60:02d}"
else:
length = "—"
return metric_grid_html([
("BPM", bpm), ("KEY", key), ("LUFS", lufs),
("TRUE PK", peak), ("LRA", lra), ("LENGTH", length),
])
def _data_table(a: Analysis) -> list[list[str]]:
rows = [
["duration", f"{a.duration_s} s"],
["bpm", str(a.bpm)],
["bpm_prior", str(a.bpm_prior)],
["key", f"{a.key} {a.key_mode}".strip()],
["key_correlation", str(a.key_correlation)],
["lufs_i", str(a.lufs_i)],
["lufs_lra", str(a.lufs_lra)],
["true_peak_db", str(a.true_peak_db)],
["voiceover_present", str(a.voiceover_present)],
["sections", str(len(a.sections))],
["downbeats", str(len(a.downbeats))],
["stems", ", ".join(sorted(a.stems)) or "(none)"],
["bass_midi", a.bass_midi_path or "(none)"],
]
return rows
def _sections_table(a: Analysis) -> list[list[str]]:
return [
[s["label"], str(s["start"]), str(s["end"]), str(s["length"])]
for s in a.sections
]
def _tags_table(a: Analysis) -> list[list[str]]:
out: list[list[str]] = []
for kind, items in (
("genre", a.tags_genre),
("mood", a.tags_mood),
("instrument", a.tags_instrument),
):
for t in items[:5]:
out.append([kind, t.get("label", "?"), f"{t.get('score', 0):.3f}"])
return out
def _format_errors(a: Analysis) -> str:
if not a.errors:
return "no errors"
return "\n".join(f"• {e['stage']}: {e['error']}" for e in a.errors)
def _format_timings(a: Analysis) -> str:
if not a.timings:
return "no timings"
total = sum(a.timings.values())
rows = [f"{k:>14} {v:>6.2f} s" for k, v in a.timings.items()]
rows.append(f"{'total':>14} {total:>6.2f} s")
return "\n".join(rows)
_PITCH_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
def _midi_note_name(p: int) -> str:
"""MIDI note number → 'A1' / 'G#2' style label."""
octave = (p // 12) - 1
return f"{_PITCH_NAMES[p % 12]}{octave}"
def _bass_midi_summary(a: Analysis) -> dict | None:
"""Compact one-line description of the basic-pitch bass-stem MIDI for
the brief LLM. Without this, basic-pitch runs every analysis but its
output is silently dropped from the brief surface — only the deterministic
Ableton clip-plan output ever sees it. With this, the brief can say
'bass walks G1 → C2 at ~1.4 notes/sec' instead of generic flannel.
Returns None if no notes were extracted (no bass stem, MIDI parse
failed, etc.) — keeps the payload tidy."""
notes = a.bass_notes or []
if not notes:
return None
pitches = [int(n["pitch"]) for n in notes if "pitch" in n]
if not pitches:
return None
durations = [
float(n["end"]) - float(n["start"])
for n in notes
if "start" in n and "end" in n and float(n["end"]) > float(n["start"])
]
span_s = max(float(n.get("end", 0.0)) for n in notes) or 1e-6
return {
"note_count": len(notes),
"pitch_low": _midi_note_name(min(pitches)),
"pitch_high": _midi_note_name(max(pitches)),
"pitch_low_midi": int(min(pitches)),
"pitch_high_midi": int(max(pitches)),
"notes_per_sec": round(len(notes) / span_s, 2),
"mean_note_dur_s": round(sum(durations) / len(durations), 3) if durations else None,
}
def _brief_payload(a: Analysis) -> dict:
"""Numeric + categorical analysis data fed to the brief LLM. Keep
this minimal but include EVERY measurement we actually paid for —
silently dropping demucs/basic-pitch output leaves quality on the
table (demucs alone is ~40s of the analysis wall-clock)."""
return {
"bpm": a.bpm,
"key": f"{a.key} {a.key_mode}" if a.key else None,
"duration_s": a.duration_s,
"sections": a.sections,
"lufs_i": a.lufs_i,
"lufs_lra": a.lufs_lra,
"voiceover_present": a.voiceover_present,
"top_genre": a.top_genre(),
"top_mood": a.top_mood(),
"top_instrument": a.top_instrument(),
# Top-3 measured tags (essentia). These carry the vibe the brief
# paragraph used to infer — the SA3 prompt-writer reads them
# directly now, so the brief-writing LLM call can be skipped.
# Empty lists where tagging was skipped (fast mode / no essentia).
"genre_tags": [t["label"] for t in a.tags_genre[:3]],
"mood_tags": [t["label"] for t in a.tags_mood[:3]],
"instrument_tags": [t["label"] for t in a.tags_instrument[:3]],
"stems_found": sorted(a.stems),
# Per-stem RMS / peak / spectral-centroid — demucs already ran;
# without this the brief LLM gets 'yes there are stems' and nothing
# else. With it, the LLM can characterise the mix (heavy bass,
# bright vocals, hi-hat-forward) without guessing.
"stem_stats": a.stem_stats or None,
# basic-pitch ran on the bass stem; surface a compact summary so
# the brief can describe the bassline character.
"bass_midi_summary": _bass_midi_summary(a),
}
def _chain_payload(a: Analysis) -> dict:
return {
**_brief_payload(a),
"true_peak_db": a.true_peak_db,
# stem_stats is already in _brief_payload now (it always should
# have been); keep this for backwards compatibility with bundles
# that look for it explicitly here.
"stem_stats": a.stem_stats,
}
# ── Ingest: video containers + YouTube links ─────────────────────────────
# librosa/soundfile can decode wav/mp3/flac/ogg/aiff natively, but NOT
# video containers (mp4/mov/webm/mkv) or the aac/m4a/opus family. For
# those we shell out to ffmpeg once and hand the pipeline a plain
# 44.1 kHz stereo WAV. Extracted files land in CRATE_DIR/ingest — the
# crate dir is whitelisted via gr.set_static_paths, so the anchor
# player's /gradio_api/file= URL can serve them.
_EXTRACT_EXTS = {
".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi", # video containers
".m4a", ".aac", ".opus", ".wma", # audio soundfile can't read
}
_INGEST_DIR = crate.CRATE_DIR / "ingest"
def _prepare_source_audio(audio_path: str) -> str:
"""Return a path the analysis pipeline can decode.
Pass-through for formats librosa handles natively; ffmpeg-extract to
WAV for video containers and aac-family audio. Raises RuntimeError
with a user-facing message when ffmpeg is missing or the file has no
audio stream (e.g. a screen recording with the mic muted).
"""
import shutil
import subprocess
src = Path(audio_path)
if src.suffix.lower() not in _EXTRACT_EXTS:
return audio_path
if not shutil.which("ffmpeg"):
raise RuntimeError(
f"'{src.suffix}' needs ffmpeg to extract the audio track, and "
"ffmpeg isn't installed on this server. Convert to WAV/MP3 "
"and re-upload."
)
_INGEST_DIR.mkdir(parents=True, exist_ok=True)
# Keep the original stem so the crate tile / anchor player show a
# recognisable name; secrets suffix avoids collisions between users
# uploading files with the same name.
safe_stem = "".join(c if c.isalnum() or c in "-_ " else "_" for c in src.stem)[:60]
out = _INGEST_DIR / f"{safe_stem}-{secrets.token_hex(4)}.wav"
proc = subprocess.run(
["ffmpeg", "-y", "-i", str(src), "-vn",
"-ar", "44100", "-ac", "2", "-acodec", "pcm_s16le", str(out)],
capture_output=True, text=True, timeout=300,
)
if proc.returncode != 0 or not out.exists() or out.stat().st_size < 1024:
tail = (proc.stderr or "").strip().splitlines()[-1:] or ["unknown ffmpeg error"]
raise RuntimeError(
f"couldn't extract audio from '{src.name}' — {tail[0]}. "
"Does the file actually contain an audio track?"
)
print(f"[ingest] extracted {src.name} → {out.name} "
f"({out.stat().st_size // 1024} KB)", flush=True)
return str(out)
_YT_HOSTS = {"youtube.com", "www.youtube.com", "m.youtube.com",
"music.youtube.com", "youtu.be"}
_YT_MAX_SECONDS = 15 * 60
def fetch_youtube_audio(url: str):
"""Download a YouTube video's audio track as WAV into the ingest dir.
Returns (gr.update for audio_in, status markdown). Never raises —
all failure modes come back as a status message, because on HF
Spaces YouTube frequently blocks datacenter IPs ('Sign in to
confirm you're not a bot') and that shouldn't traceback the UI.
"""
url = (url or "").strip()
if not url:
return gr.update(), "⚠️ paste a YouTube link first"
try:
host = urllib.parse.urlparse(url).netloc.lower()
except ValueError:
host = ""
if host not in _YT_HOSTS:
return gr.update(), f"⚠️ not a YouTube link (host: `{host or '?'}`)"
try:
import yt_dlp
except ImportError:
return gr.update(), ("⚠️ `yt-dlp` isn't installed on this server — "
"`pip install yt-dlp` and restart.")
import shutil
if not shutil.which("ffmpeg"):
return gr.update(), ("⚠️ ffmpeg isn't installed — needed to convert "
"the YouTube audio to WAV.")
_INGEST_DIR.mkdir(parents=True, exist_ok=True)
ydl_opts = {
"format": "bestaudio/best",
"outtmpl": str(_INGEST_DIR / "yt-%(id)s.%(ext)s"),
"noplaylist": True,
"quiet": True,
"noprogress": True,
"no_warnings": True,
"postprocessors": [
{"key": "FFmpegExtractAudio", "preferredcodec": "wav"},
],
}
try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(url, download=False)
duration = int(info.get("duration") or 0)
if duration > _YT_MAX_SECONDS:
return gr.update(), (
f"⚠️ that video is {duration // 60} min — too long to "
f"analyse (limit {_YT_MAX_SECONDS // 60} min). demucs "
"on a track that size would take a very long time."
)
info = ydl.extract_info(url, download=True)
except Exception as e: # yt-dlp raises many exception types
msg = str(e)
if "Sign in to confirm" in msg or "bot" in msg.lower():
return gr.update(), (
"⚠️ YouTube blocked this server's IP (common on hosted "
"Spaces). Try again later, or download the audio locally "
"and upload the file instead."
)
return gr.update(), f"⚠️ download failed: {msg[:300]}"
out = _INGEST_DIR / f"yt-{info['id']}.wav"
if not out.exists():
return gr.update(), "⚠️ download finished but no WAV landed — check server logs"
title = info.get("title") or info["id"]
print(f"[ingest] youtube {info['id']} → {out.name} "
f"({out.stat().st_size // 1024} KB)", flush=True)
return (gr.update(value=str(out)),
f"✅ fetched **{title}** ({duration or '?'}s) — now hit **Analyze audio**")
def run_brief(audio_path: str | None, bpm_mode: str, bpm_prior_choice: str, bpm_prior_num: float, model_choice: str, api_key: str = "", original_prompt: str = ""):
"""Streaming generator — yields partial outputs as each stage finishes so
the UI fills in progressively instead of all-at-once after a long wait.
`original_prompt`: the source SA3 prompt that produced this audio (when
the tile came from a Generate-then-Use-for-Analysis flow). When set,
the derived SA3 variation prompt is built via the LLM `write_sa3_prompt_blended`
call which fuses the user's intent vocabulary with the measured arc —
so variants stay in the user's stylistic lane instead of drifting toward
whatever the LLM thought it heard. Empty string for uploaded files.
Output tuple (16 slots):
paragraph, caption, anchor_player, wave, section_seek,
data_tbl, sections_tbl, tags_tbl,
errors, timings, sa3_var, sa3_match, clip_plan, mix_chain,
raw_json, state
"""
# Waveform Image + section-seek strip are hidden by default; revealed
# only once we have a real rendered PNG (stages 2+). On error or
# pre-analysis we keep them hidden so empty bordered boxes don't sit.
wave_hide = gr.update(value=None, visible=False)
seek_hide = gr.update(value="", visible=False)
player_hide = gr.update(value="", visible=False)
if not audio_path:
empty = "(drop a file first)"
yield (empty, empty, player_hide, wave_hide, seek_hide, _metrics_html(None), [], [], [], empty, empty, empty, empty, empty, empty, "{}", None)
return
# Stage 1 — announce; clear stale state.
yield ("_analyzing audio…_", "running pipeline", player_hide, wave_hide, seek_hide, _metrics_html(None), [], [], [],
"", "", "(waiting on analysis)", "(waiting on analysis)",
"(waiting on analysis)", "(waiting on analysis)", "{}", None)
prior: float | str
if bpm_mode and bpm_mode.startswith("Manual") and bpm_prior_num and bpm_prior_num > 0:
prior = float(bpm_prior_num)
else:
prior = bpm_prior_choice or "default"
try:
# Video containers / aac-family uploads: extract the audio track
# to WAV first — librosa can't decode them directly. No-op for
# wav/mp3/flac/ogg. audio_path is rebound so every downstream
# consumer (anchor player, crate tile, share bundle) uses the WAV.
audio_path = _prepare_source_audio(audio_path)
# run_tags=True → measured genre/mood/instrument via essentia when
# it's available (Linux/Spaces). Degrades to a skipped stage on
# platforms without the wheels, so this is safe to leave on.
a = analyze(audio_path, bpm_prior=prior, run_tags=True, run_embedding=False)
except Exception as e:
traceback.print_exc()
err = f"pipeline failed: {type(e).__name__}: {e}"
yield (err, err, player_hide, wave_hide, seek_hide, _metrics_html(None), [], [], [], err, err, err, err, err, err, "{}", None)
return
# Stage 2 — analysis done; render every deterministic output. LLM still pending.
caption = _short_caption(a)
wave_png = waveform.render(a)
# Reveal the anchor player + waveform + section-seek strip now that
# we have the data. Anchor player uses Gradio's `/gradio_api/file=...`
# proxy to serve from the crate dir (whitelisted via set_static_paths).
player_show = gr.update(value=_anchor_player_html(audio_path), visible=True)
wave_show = gr.update(value=wave_png, visible=True)
seek_show = gr.update(value=_section_seek_html(a), visible=True)
sa3_var_partial = outputs.sa3_variation_prompt(a, "")
sa3_match_partial = outputs.sa3_match_style_prompt(a, "")
clip_plan = outputs.ableton_clip_plan(a)
raw = to_json(a)
state = {"analysis": a, "model": model_choice, "audio_path": audio_path,
"brief_payload": _brief_payload(a), "chain_payload": _chain_payload(a)}
yield (f"_generating narrative via **{model_choice}**…_", caption,
player_show, wave_show, seek_show, _metrics_html(a),
_data_table(a), _sections_table(a), _tags_table(a),
_format_errors(a), _format_timings(a),
sa3_var_partial, sa3_match_partial, clip_plan,
f"_generating mix chain via **{model_choice}**…_", raw, state)
# Stage 3 — fire both LLM calls in parallel, yield whichever finishes first.
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
# Short-circuit when there's no wallet: the LLM calls would just 401
# and surface raw LLMError text in the brief / mix-chain panels, which
# looks like the whole analysis broke. Render a friendly inline nudge
# instead, and skip the round-trips. Measurement-only insights are
# still complete — section table, BPM, key, waveform, derived prompt
# (basic compression) are all already yielded above.
if not wallet.get_key(session_key=api_key):
wallet_nudge = ("_📌 Pollinations not connected — connect the pollen pill (top right) "
"to write the brief + mix chain. Measurement and the derived "
"SA3 prompt are ready below._")
yield (wallet_nudge, caption,
player_show, wave_show, seek_show, _metrics_html(a),
_data_table(a), _sections_table(a), _tags_table(a),
_format_errors(a), _format_timings(a),
sa3_var_partial, sa3_match_partial, clip_plan,
wallet_nudge, raw, state)
return
paragraph = "(narrative pending)"
llm_chain = ""
sa3_var_final = sa3_var_partial
sa3_match_final = sa3_match_partial
mix_chain_final = f"_generating mix chain via **{model_choice}**…_"
with ThreadPoolExecutor(max_workers=2) as pool:
fut_brief = pool.submit(write_brief, _brief_payload(a), model=model_choice, api_key=api_key)
fut_chain = pool.submit(write_mix_chain, _chain_payload(a), model=model_choice, api_key=api_key)
pending = {fut_brief, fut_chain}
while pending:
done, pending = wait(pending, return_when=FIRST_COMPLETED)
for fut in done:
try:
res = fut.result()
# write_brief / write_mix_chain now return (text, resolved_model)
text = res[0] if isinstance(res, tuple) else res
except Exception as e:
text = f"(failed — {type(e).__name__}: {e})"
if fut is fut_brief:
paragraph = text
safe = text if not text.startswith("(") else ""
sa3_match_final = outputs.sa3_match_style_prompt(a, safe)
# Derived prompt — fan out THREE lens variants in
# parallel (MATCH/LOOSE/FREE). Default-selected is
# LOOSE — the middle ground that keeps feel but doesn't
# inherit the anchor's exact section timings (which is
# what makes variants come back sparse when the anchor
# has a long quiet intro). All three results land in
# state["sa3_lenses"] so the chip-row in the UI can
# swap between them instantly without re-calling.
sa3_var_final = outputs.sa3_variation_prompt(a, safe)
state["sa3_lenses"] = {"match": "", "loose": "", "free": ""}
if safe:
try:
if original_prompt:
# Run all 3 lenses concurrently — one wall-time cost.
with ThreadPoolExecutor(max_workers=3) as lens_pool:
lens_futs = {
lens_pool.submit(
write_sa3_prompt_lens,
original_prompt=original_prompt,
brief_payload=_brief_payload(a),
brief_text=safe,
lens=lens_name,
model=model_choice,
api_key=api_key,
): lens_name
for lens_name in ("match", "loose", "free")
}
for lf, lens_name in lens_futs.items():
try:
lp, _ = lf.result()
if lp and not lp.startswith("("):
state["sa3_lenses"][lens_name] = lp
except Exception as e:
print(f"[lens {lens_name}] {type(e).__name__}: {e}",
file=sys.stderr, flush=True)
# Default-display: LOOSE (middle ground).
if state["sa3_lenses"]["loose"]:
sa3_var_final = state["sa3_lenses"]["loose"]
elif state["sa3_lenses"]["match"]:
sa3_var_final = state["sa3_lenses"]["match"]
else:
# Uploaded file — no original prompt to blend
# against. Previously only one compression
# was run + stored as 'loose', so clicking
# MATCH or FREE chips no-op'd. Fan out all
# three lenses concurrently against the
# upload-only system prompts in narrative.py.
with ThreadPoolExecutor(max_workers=3) as upload_lens_pool:
upload_lens_futs = {
upload_lens_pool.submit(
write_sa3_prompt_upload_lens,
brief_payload=_brief_payload(a),
brief_text=safe,
lens=lens_name,
model=model_choice,
api_key=api_key,
): lens_name
for lens_name in ("match", "loose", "free")
}
for lf, lens_name in upload_lens_futs.items():
try:
lp, _ = lf.result()
if lp and not lp.startswith("("):
state["sa3_lenses"][lens_name] = lp
except Exception as e:
print(f"[upload-lens {lens_name}] "
f"{type(e).__name__}: {e}",
file=sys.stderr, flush=True)
# Default-display: LOOSE (middle ground).
if state["sa3_lenses"]["loose"]:
sa3_var_final = state["sa3_lenses"]["loose"]
elif state["sa3_lenses"]["match"]:
sa3_var_final = state["sa3_lenses"]["match"]
elif state["sa3_lenses"]["free"]:
sa3_var_final = state["sa3_lenses"]["free"]
except Exception as e:
print(f"[run_brief] sa3-prompt LLM failed: {type(e).__name__}: {e}",
file=sys.stderr, flush=True)
else:
llm_chain = text
mix_chain_final = outputs.mix_chain_text(a, llm_chain)
yield (paragraph, caption,
player_show, wave_show, seek_show, _metrics_html(a),
_data_table(a), _sections_table(a), _tags_table(a),
_format_errors(a), _format_timings(a),
sa3_var_final, sa3_match_final, clip_plan,
mix_chain_final, raw, state)
def load_brief_from_bundle(file_path: str | None, session_id: str | None = ""):
"""Import a .abv1 bundle and populate the ANALYSE view directly.
The bundle already contains the analysis JSON + brief text + mix chain
+ derived SA3 lenses. We deserialize and render every UI surface the
same way run_brief would — without a second LLM round-trip and
without forcing the user to click Analyze. This is the import-as-
load semantics Codex called out: the card says 'Import shared
analysis', so it should load the analysis.
Returns the same 17-slot tuple shape as run_brief so we can wire
this handler into the same set of output components, plus three
trailing slots for the import-card status + crate refresh.
"""
# Hidden / empty placeholders for the failure path — same shape as
# run_brief's no-audio yield so Gradio's output binding matches.
wave_hide = gr.update(value=None, visible=False)
seek_hide = gr.update(value="", visible=False)
player_hide = gr.update(value="", visible=False)
empty_run_outputs = (
"", "", player_hide, wave_hide, seek_hide, _metrics_html(None),
[], [], [], "", "",
"", "", "", "", "{}", None,
)
if not file_path:
return (*empty_run_outputs, "", gr.update(), _crate_header_html(session_id=session_id))
bundle = share.import_bundle(file_path)
if not bundle:
return (*empty_run_outputs,
"_❌ Not a valid .abv1 bundle._",
gr.update(), _crate_header_html(session_id=session_id))
payload = bundle.get("analysis", {}) or {}
analysis_dict = payload.get("analysis", {}) or {}
anchor_path = bundle["anchor_path"]
# Reconstruct Analysis dataclass. Field-by-field setattr so a bundle
# written by a future-version Space with extra fields doesn't crash.
try:
a = Analysis(source_path=anchor_path)
for k, v in analysis_dict.items():
if hasattr(a, k) and k != "source_path":
setattr(a, k, v)
except Exception:
return (*empty_run_outputs,
"_❌ Bundle analysis JSON is malformed._",
gr.update(), _crate_header_html(session_id=session_id))
# Add anchor + variants to crate so the user can regenerate / replay.
src_prompt = payload.get("source_prompt", "")
model_used = payload.get("model", "shared-import")
try:
anchor_tile = crate.add_tile(
audio_path=anchor_path,
source_prompt=src_prompt,
model=model_used,
duration_s=float(a.duration_s or 0.0),
session_id=session_id,
)
for vp in bundle.get("variant_paths", []) or []:
try:
crate.add_tile(
audio_path=vp,
source_prompt=src_prompt,
parent_id=anchor_tile.id,
model=model_used,
session_id=session_id,
)
except Exception:
pass
except Exception:
pass
# Build the same last_run state shape run_brief produces.
state = {
"analysis": a,
"model": model_used,
"audio_path": anchor_path,
"source_prompt": src_prompt,
"sa3_lenses": payload.get("sa3_lenses", {}) or {},
"brief_payload": _brief_payload(a),
"chain_payload": _chain_payload(a),
}
# Render every deterministic output from Analysis (same helpers
# run_brief calls in stage 2).
caption = _short_caption(a)
paragraph = payload.get("brief") or "_imported analysis (brief text not bundled)_"
mix_chain_md = payload.get("mix_chain") or "_imported analysis (mix chain not bundled)_"
try:
wave_png = waveform.render(a)
wave_out = gr.update(value=wave_png, visible=True)
except Exception:
wave_out = wave_hide
try:
seek_out = gr.update(value=_section_seek_html(a), visible=True)
except Exception:
seek_out = seek_hide
player_out = gr.update(value=_anchor_player_html(anchor_path), visible=True)
# Derived SA3 prompt — prefer the lens cached in the bundle; fall
# back to the local outputs helper if the bundle predates the lens
# feature. brief text needs to be non-empty for the helper to work.
safe_brief = paragraph if not paragraph.startswith("(") and not paragraph.startswith("_") else ""
lenses = state["sa3_lenses"]
sa3_var_out = (lenses.get("loose") or lenses.get("match")
or outputs.sa3_variation_prompt(a, safe_brief))
sa3_match_out = outputs.sa3_match_style_prompt(a, safe_brief)
clip_plan_out = outputs.ableton_clip_plan(a)
raw_out = to_json(a)
status = (f"📥 Shared analysis loaded · `[{anchor_tile.id}]` **{anchor_tile.label}** "
f"({len(bundle.get('variant_paths') or [])} bundled variants).")
return (
paragraph, caption, player_out, wave_out, seek_out,
_metrics_html(a),
_data_table(a), _sections_table(a), _tags_table(a),
_format_errors(a), _format_timings(a),
sa3_var_out, sa3_match_out, clip_plan_out, mix_chain_md,
raw_out, state,
# 3 trailing: import-card status, crate radio, crate header.
status,
gr.update(choices=crate.tile_choices(session_id=session_id), value=anchor_tile.id),
_crate_header_html(session_id=session_id),
)
def _status_badge(state: str, t_s: float | None, ok_count: int, fail_count: int) -> str:
"""One-line status: ⏳ running / ✓ done / ⚠ partial / ✗ failed."""
if state == "running":
return "⏳ _running…_"
if state == "done":
if fail_count == 0:
return f"✅ **done in {t_s:.1f}s**"
if ok_count == 0:
return f"❌ **both calls failed** · {t_s:.1f}s"
return f"⚠️ **partial** · {ok_count}/{ok_count + fail_count} ok · {t_s:.1f}s"
return ""
def _compare_header(label: str, model: str, badge: str, resolved: str | None = None) -> str:
# When Pollinations resolves an alias (`gemini` → `gemini-3.5-flash`),
# show the resolved name in parens so the user sees what actually ran.
if resolved and resolved != model:
model_str = f"`{model}` → `{resolved}`"
else:
model_str = f"`{model}`"
return f"### {label} · {model_str}\n\n{badge}"
def _fail_cell(label: str, reason: str) -> str:
"""Distinct visual for a failed cell — blockquote + ❌ icon."""
return f"> ❌ **{label} failed**\n>\n> `{reason}`"
def export_scorecard(compare_state: dict | None):
# Guard: compare_state is populated on the FIRST yield (before any LLM
# call returns), so without the done check a click during a running
# Compare would export a card full of "_running…_" cells.
if not compare_state or not compare_state.get("done"):
gr.Warning("Run side-by-side hasn't finished yet — wait for all three columns to show ✅ before exporting.")
return None
return share.render_scorecard_png(compare_state["analysis"], compare_state["columns"])
def export_report(compare_state: dict | None):
if not compare_state or not compare_state.get("done"):
gr.Warning("Run side-by-side hasn't finished yet — wait for all three columns to show ✅ before exporting.")
return None
return share.render_full_report_md(compare_state["analysis"], compare_state["columns"])
def _columns_from_local(local: dict) -> list[dict]:
"""Flatten the streaming `local` dict into the column structure that
share.render_scorecard_png / render_full_report_md expects."""
out = []
for col in ("a", "b", "c"):
s = local[col]
out.append({
"col": s["label"],
"model": s["model"],
"brief": s.get("brief", ""),
"chain": s.get("raw_chain", ""),
"mode": "audio-only" if col == "c" else "measured",
"elapsed_s": s.get("t"),
})
return out
def run_compare(state: dict | None, model_a: str, model_b: str, model_c: str, api_key: str = ""):
"""Three-column comparison.
Column A: measured analysis + LLM brief+chain (model_a)
Column B: measured analysis + LLM brief+chain (model_b)
Column C: AUDIO-ONLY — model_c receives the raw audio file with NO
measurements and has to guess BPM/key/loudness by ear. This is
the demo of what the audio-brief wedge buys.
Outputs (11 slots):
a_header, a_brief,
b_header, b_brief,
c_header, c_brief,
comparison_table,
a_timing, b_timing, c_timing,
compare_state (gr.State with analysis + columns for export)
"""
if not state or not state.get("brief_payload"):
msg = "_Run **analyze** first — Compare reuses the most recent analysis._"
yield (
_compare_header("A · measured", model_a, "_no analysis yet_"), msg,
_compare_header("B · measured", model_b, "_no analysis yet_"), msg,
_compare_header("C · audio-only", model_c, "_no analysis yet_"), msg,
"_Run analyze, then click **Run side-by-side**._",
"", "", "",
None,
)
return
from narrative import LLMError
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
import time
a = state["analysis"]
brief_payload = state["brief_payload"]
chain_payload = state["chain_payload"]
audio_path = state.get("audio_path")
t0 = time.perf_counter()
local: dict[str, dict] = {
"a": {"model": model_a, "resolved": None, "label": "A · measured", "brief": "_running…_",
"chain": "_running…_", "raw_chain": "", "brief_ok": None, "chain_ok": None, "t": None},
"b": {"model": model_b, "resolved": None, "label": "B · measured", "brief": "_running…_",
"chain": "_running…_", "raw_chain": "", "brief_ok": None, "chain_ok": None, "t": None},
"c": {"model": model_c, "resolved": None, "label": "C · audio-only", "brief": "_running… (sending raw audio)_",
"chain": "_running…_", "raw_chain": "", "brief_ok": None, "chain_ok": None, "t": None},
}
def status_for(side: str) -> str:
s = local[side]
if s["brief_ok"] is None or s["chain_ok"] is None:
return _status_badge("running", None, 0, 0)
ok = int(bool(s["brief_ok"])) + int(bool(s["chain_ok"]))
fail = 2 - ok
return _status_badge("done", s["t"], ok, fail)
def _briefs_for_table() -> dict:
return {
col: (local[col]["brief"] if local[col]["brief_ok"] else None)
for col in ("a", "b", "c")
}
def _chains_for_table() -> dict:
# Pass the raw LLM chain (not the formatted one with the measured
# header prepended) so the parser can split on `## ` headings.
return {
col: (local[col].get("raw_chain") if local[col]["chain_ok"] else None)
for col in ("a", "b", "c")
}
def _labels_for_table() -> dict:
return {
"a": f"A · {local['a']['model']}",
"b": f"B · {local['b']['model']}",
"c": f"C · {local['c']['model']} (audio-only)",
}
def render(*, done: bool = False):
# The `done` flag gates the export buttons — until the final yield,
# compare_state.done is False and clicking Download… returns nothing
# rather than exporting a partial scorecard with "_running…_" cells.
return (
_compare_header(local["a"]["label"], local["a"]["model"], status_for("a"), local["a"].get("resolved")),
local["a"]["brief"],
_compare_header(local["b"]["label"], local["b"]["model"], status_for("b"), local["b"].get("resolved")),
local["b"]["brief"],
_compare_header(local["c"]["label"], local["c"]["model"], status_for("c"), local["c"].get("resolved")),
local["c"]["brief"],
outputs.compare_table_markdown(_briefs_for_table(), _chains_for_table(), _labels_for_table()),
"", "", "",
{"analysis": a, "columns": _columns_from_local(local), "done": done},
)
yield render()
def _audio_only_call():
"""Wrap write_audio_only so the pool sees a single dict-returning task."""
if not audio_path:
raise RuntimeError("no audio_path in state — re-run analyze")
return write_audio_only(audio_path, model=model_c, no_cache=True, api_key=api_key)
with ThreadPoolExecutor(max_workers=5) as pool:
futs: dict = {
pool.submit(write_brief, brief_payload, model=model_a, no_cache=True, api_key=api_key): ("a", "brief"),
pool.submit(write_mix_chain, chain_payload, model=model_a, no_cache=True, api_key=api_key): ("a", "chain"),
pool.submit(write_brief, brief_payload, model=model_b, no_cache=True, api_key=api_key): ("b", "brief"),
pool.submit(write_mix_chain, chain_payload, model=model_b, no_cache=True, api_key=api_key): ("b", "chain"),
pool.submit(_audio_only_call): ("c", "both"), # single call → both brief+chain
}
pending = set(futs)
while pending:
done, pending = wait(pending, return_when=FIRST_COMPLETED)
for fut in done:
side, kind = futs[fut]
try:
res = fut.result()
ok = True
except LLMError as e:
res = _fail_cell(kind if kind != "both" else "audio-only", str(e))
ok = False
except Exception as e: # noqa: BLE001
res = _fail_cell(kind if kind != "both" else "audio-only", f"{type(e).__name__}: {e}")
ok = False
if kind == "both":
# Audio-only returns {"brief", "chain", "model_resolved"} on success.
if ok and isinstance(res, dict):
local[side]["brief"] = res.get("brief", "_(no brief)_")
raw = res.get("chain", "")
local[side]["raw_chain"] = raw
local[side]["chain"] = outputs.normalize_chain_markdown(raw)
local[side]["brief_ok"] = True
local[side]["chain_ok"] = True
local[side]["resolved"] = res.get("model_resolved")
else:
local[side]["brief"] = res
local[side]["chain"] = "_(audio-only call failed — see brief cell)_"
local[side]["raw_chain"] = ""
local[side]["brief_ok"] = False
local[side]["chain_ok"] = False
else:
# write_brief / write_mix_chain now return (text, resolved_model).
if ok and isinstance(res, tuple):
text, resolved = res
if not local[side].get("resolved"):
local[side]["resolved"] = resolved
res = text
if kind == "chain" and ok:
local[side]["raw_chain"] = res
res = outputs.mix_chain_text(a, res)
elif kind == "chain":
local[side]["raw_chain"] = ""
local[side][kind] = res
local[side][f"{kind}_ok"] = ok
# Stamp final time when both calls (or the single audio-only call)
# for this side have landed.
if (local[side]["brief_ok"] is not None
and local[side]["chain_ok"] is not None
and local[side]["t"] is None):
local[side]["t"] = time.perf_counter() - t0
yield render(done=not pending)
def _wallet_status_md() -> str:
key = wallet.get_key()
if not key:
return "**Pollinations wallet:** not connected — narrative will fall back to (or fail). Click *Connect* to authorise abv1 with your Pollinations balance."
info = wallet.load_wallet().get("user", {}) or {}
who = info.get("preferred_username") or info.get("name") or "connected"
masked = key[:6] + "…" + key[-4:] if len(key) > 12 else "•••"
return f"**Pollinations wallet:** connected as **{who}** (`{masked}`)."
def _extract_balance(info: dict) -> float | None:
"""Try every common path Pollinations might surface a balance on.
The userinfo response shape isn't documented publicly, so we
probe top-level keys first, then a couple of nested containers
we've seen on similar wallet APIs. Returns None if nothing
numeric is found — caller renders a graceful fallback then."""
if not isinstance(info, dict):
return None
# Top-level keys we've already tried + common variants.
candidates = [
info.get("balance"),
info.get("pollen"),
info.get("credits"),
info.get("pollen_balance"),
info.get("remaining_pollen"),
info.get("available"),
]
# Nested under .account / .wallet / .usage if Pollinations groups them.
for nest_key in ("account", "wallet", "usage", "user"):
sub = info.get(nest_key)
if isinstance(sub, dict):
candidates.extend([
sub.get("balance"),
sub.get("pollen"),
sub.get("credits"),
sub.get("remaining"),
sub.get("available"),
])
for c in candidates:
if c is None:
continue
try:
return float(c)
except (TypeError, ValueError):
continue
return None
def _topbar_html(session_key: str = "", session_spend: float = 0.0) -> str:
"""Top bar HTML: brand wordmark left, pollen pill (clickable wallet) right.
The pill IS the wallet button:
- disconnected → click triggers OAuth redirect via inline JS
- connected → click triggers the hidden #wallet-disconnect-trigger
Pollen balance: Pollinations userinfo doesn't always include it; we
show the balance if present, otherwise a connected dot.
`session_key` is the per-visitor key from gr.State (HF Spaces mode).
Desktop falls back to env vars and the on-disk wallet.
`session_spend` renders a small `0.84 ◆ session` pill left of the
wallet pill when > 0 (per v2 design — cost transparency)."""
key = wallet.get_key(session_key=session_key)
if not key:
return (
'
'
)
# On HF Spaces we don't have userinfo cached; on desktop, wallet file
# may have userinfo + account from the connect flow. Prefer the
# account dict (whole-wallet total) over the userinfo dict (per-key
# scope) — matches what the Pollinations consent screen shows.
w = wallet.load_wallet()
account = w.get("account", {}) or {}
info = w.get("user", {}) or {}
balance = _extract_balance(account) or _extract_balance(info)
return (
'
'
)
def save_key_from_fragment(api_key: str):
"""Page-load handler: if Pollinations redirected back to us with
#api_key=sk_… in the URL fragment, JS strips it and passes the key
here. We persist it to the wallet file and refresh the status pill.
Matches the abv1 OAuth pattern (standalone/public/index.html:4959)."""
import time, json as _json, sys as _sys
if api_key and api_key.startswith("sk_"):
try:
info = wallet.userinfo(api_key)
except Exception as e:
print(f"[wallet] userinfo failed: {type(e).__name__}: {e}",
file=_sys.stderr, flush=True)
info = {}
# Log only the response SHAPE — keys + value types — never raw
# values. The Pollinations userinfo response may include account
# identifiers (email, user id, plan name) we don't want winding
# up in HF's stderr aggregate. Knowing which key holds the
# balance is enough to narrow _extract_balance.
try:
shape = {k: type(v).__name__ for k, v in info.items()} if isinstance(info, dict) else type(info).__name__
print(f"[wallet] userinfo shape: {shape}",
file=_sys.stderr, flush=True)
except Exception:
pass
try:
account = wallet.account_info(api_key)
except Exception as e:
print(f"[wallet] account_info failed: {type(e).__name__}: {e}",
file=_sys.stderr, flush=True)
account = {}
# save_wallet is a no-op on HF Spaces; persists to disk on desktop.
wallet.save_wallet({
"api_key": api_key,
"user": info,
"account": account,
"scope": "generate account:usage",
"saved_at": int(time.time()),
})
# Render the connected pill regardless of where the key is stored.
# session_spend resets to 0 here — a fresh connect starts the
# session-spend counter from zero.
return _topbar_html(session_key=api_key, session_spend=0.0), api_key, 0.0
return _topbar_html(), "", 0.0
def disconnect_wallet():
wallet.clear_wallet()
# Disconnect also clears the session-spend pill — it's per-visit.
return _topbar_html(), "", 0.0
# Pollinations SA3 cost: flat per call regardless of duration (verified
# 2026-06-23, see sa3.py). Each /audio/{text} hit on stable-audio-3-medium
# is 0.04 pollen. Used to tick the session-spend pill and render cost pips.
SA3_COST_PER_CALL = 0.04
def tick_session_spend(curr: float, delta: float, api_key: str = ""):
"""Increment the session-spend counter and refresh the top bar so the
`0.84 ◆ session` pill updates after a gen lands.
Returns (topbar_html, new_spend) — both used as outputs in a `.then(...)`
chain after each Pollinations-billing handler."""
try:
new = float(curr or 0.0) + float(delta or 0.0)
except (TypeError, ValueError):
new = 0.0
return _topbar_html(session_key=api_key, session_spend=new), new
# ── MVP2 · Generate-tab handlers ────────────────────────────────────────────
def _tile_meta_md(tile: crate.Tile | None, session_id: str | None = "") -> str:
if not tile:
return "_no tile selected_"
parent_line = ""
if tile.parent_id:
parent_tile = crate.get_tile(tile.parent_id, session_id=session_id)
if parent_tile:
parent_line = f"\n- **descended from**: `[{tile.parent_id}]` {parent_tile.label}"
return (
f"**`[{tile.id}]` {tile.label}**\n\n"
f"- model: `{tile.model}`\n"
f"- duration: {tile.duration_s:.1f}s\n"
f"- prompt: _{tile.source_prompt}_"
f"{parent_line}"
)
_LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1", "[::1]"}
def _is_loopback_url(url: str) -> bool:
"""True if `url` points at a loopback host (localhost, 127.0.0.1, [::1]).
Hardening: the local-gen flow fetches whatever URL the user pastes
into the textbox. Without this check, a saved-in-localStorage URL
pointing at any internal/private endpoint would be POSTed to from
the browser. Restricting to loopback keeps the feature scoped to
'a server on this machine' as documented."""
if not url:
return False
try:
from urllib.parse import urlparse
p = urlparse(url.strip())
except Exception:
return False
if p.scheme not in ("http", "https"):
return False
return (p.hostname or "").lower() in _LOOPBACK_HOSTS
def generate_sa3(prompt: str, model: str, duration: float,
api_key: str = "", local_gen_url: str = "", local_gen_b64: str = "",
local_gen_error: str = "",
session_id: str | None = "",
duration_custom: float = 0):
"""Click handler for the Generate button.
Two paths:
- `model == "local-server"`: a JS prelude on the click already fetched
the user's localhost gen server and base64-encoded the audio bytes
in `local_gen_b64`. We just decode and write to the crate. No
Pollinations call.
- else: existing Pollinations SA3 flow via sa3.generate(api_key=...).
"""
import sys as _sys, traceback as _tb, base64 as _b64
print(f"[gen] click: model={model} dur={duration} prompt={prompt[:80]!r}",
file=_sys.stderr, flush=True)
# Hidden placeholders (no value, not visible) — used in every failure
# path so the latest-gen + selected-tile + meta stay collapsed when
# nothing valid landed. Keeps the layout clean on error (v2 spec).
# gen_audio slot was dropped (see comment near crate_preview definition);
# the return tuple shape went 6 → 5: (status, crate_picker, crate_preview,
# crate_meta, crate_header).
_audio_hide = gr.update(value=None, visible=False)
_meta_hide = gr.update(value="", visible=False)
_sess_dir = crate.crate_dir(session_id)
if not (prompt or "").strip():
return ("_❌ Enter a prompt first._",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
# Custom chip → sentinel 0 in the Radio slot; the real seconds ride
# in duration_custom (a gr.Number, so no choices validation). The
# radio value can't be rewritten client-side — Gradio v6 validates
# it against `choices` on preprocess.
duration = int(duration or 0)
if duration == 0:
try:
duration = int(duration_custom or 0)
except (TypeError, ValueError):
duration = 0
duration = max(1, min(sa3.ABSOLUTE_MAX_S, duration)) if duration else 10
# ── Local server path ────────────────────────────────────────────────
if model == "local-server":
url_raw = (local_gen_url or "http://localhost:7864").rstrip("/")
if not _is_loopback_url(url_raw):
# Defense-in-depth — the JS prelude refuses to fetch a
# non-loopback URL, so if we got here with a populated b64,
# the prelude was bypassed. Reject the bytes too.
return ("_❌ Local gen URL must be a loopback address (localhost / 127.0.0.1 / [::1]). The local-server path only fetches from this machine._",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
if local_gen_error:
# Browser-side error — TypeError (CORS / mixed-content /
# PNA), HTTP 4xx/5xx, empty body, etc. The JS prelude
# surfaces the actual reason via the local_gen_error slot
# so we can show it instead of the misleading generic.
return (f"_❌ Local gen browser fetch failed: {local_gen_error}_",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
if not local_gen_b64:
return (f"_❌ Local gen server didn't return audio. Is it running at `{url_raw}/generate`?_",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
try:
raw = _b64.b64decode(local_gen_b64)
except Exception as e:
return (f"_❌ Local audio decode failed: {e}_",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
# Guess extension — wav vs mp3 by magic bytes; default .wav for raw PCM containers.
ext = "wav" if raw[:4] == b"RIFF" else ("mp3" if raw[:3] == b"ID3" or (raw and raw[0] == 0xFF) else "wav")
out_path = _sess_dir / f"local-{crate.new_id()}.{ext}"
out_path.write_bytes(raw)
tile = crate.add_tile(
audio_path=str(out_path),
source_prompt=prompt,
model="local-server",
duration_s=float(duration),
session_id=session_id,
)
new_audio = _sess_dir / f"{tile.id}.{ext}"
try:
out_path.rename(new_audio)
tile.audio_path = str(new_audio)
tile.save()
except Exception:
pass
status = (f"✅ Local gen `[{tile.id}]` **{tile.label}** "
f"({len(raw)/1024:.0f} KB) — model `local-server`")
return (status,
gr.update(choices=crate.tile_choices(session_id=session_id), value=tile.id),
gr.update(value=tile.audio_path, visible=True),
gr.update(value=_tile_meta_md(tile, session_id=session_id), visible=True),
_crate_header_html(session_id=session_id))
# ── Pollinations path (default) ──────────────────────────────────────
if not wallet.get_key(session_key=api_key):
return ("_❌ Click the pollen pill (top right) to connect a Pollinations wallet first._",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
out_path = _sess_dir / f"{crate.new_id()}.mp3"
print(f"[gen] calling sa3.generate → {out_path}", file=_sys.stderr, flush=True)
try:
info = sa3.generate(prompt, model=model, duration=duration, out_path=out_path,
api_key=api_key)
except sa3.SA3Error as e:
print(f"[gen] SA3Error: {e}", file=_sys.stderr, flush=True)
return (f"_❌ {e}_",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
except Exception as e:
print(f"[gen] {type(e).__name__}: {e}", file=_sys.stderr, flush=True)
_tb.print_exc(file=_sys.stderr)
return (f"_❌ unexpected {type(e).__name__}: {e}_",
gr.update(choices=crate.tile_choices(session_id=session_id)),
_audio_hide, _meta_hide, _crate_header_html(session_id=session_id))
print(f"[gen] sa3 ok: {info}", file=_sys.stderr, flush=True)
tile = crate.add_tile(
audio_path=info["path"],
source_prompt=prompt,
model=model,
duration_s=float(duration),
session_id=session_id,
)
# Move the audio file to a name that matches the tile id for tidiness.
new_audio = _sess_dir / f"{tile.id}.mp3"
try:
Path(info["path"]).rename(new_audio)
tile.audio_path = str(new_audio)
tile.save()
except Exception:
pass
status = (f"✅ Generated `[{tile.id}]` **{tile.label}** "
f"({info['bytes']/1024:.0f} KB, {info['wall_s']:.1f}s wall, model `{info['model']}`)")
choices = crate.tile_choices(session_id=session_id)
return (status,
gr.update(choices=choices, value=tile.id),
gr.update(value=tile.audio_path, visible=True),
gr.update(value=_tile_meta_md(tile, session_id=session_id), visible=True),
_crate_header_html(session_id=session_id))
def _read_var_path(update_or_value) -> str | None:
"""Audio updates can come as either raw filepaths (legacy) or
gr.update dicts. Extract the underlying path string either way."""
if update_or_value is None:
return None
if isinstance(update_or_value, dict):
return update_or_value.get("value") or None
if isinstance(update_or_value, str):
return update_or_value or None
return None
def commit_regen_to_history(history, lens, prompt_text,
v1, v2, v3, v4, v5, matches):
"""Post-regen handler — appends the just-completed batch to history,
rebuilds the tab-strip choices, auto-selects the new tab.
Var paths come in as the current values of var1..5 (already populated
by the streaming yields of regenerate_variants). `matches` is the
BPM/Harm/Rhyt breakdown list snapshotted by regenerate_variants into
current_matches_state — persisting it here means tab replay restores
the mint-fill bars instead of resetting to the placeholder state."""
paths = [_read_var_path(v) for v in (v1, v2, v3, v4, v5)]
if not any(paths):
# No successful gens — don't pollute history.
return history, gr.update(), gr.update()
safe_matches = list(matches or [None] * 5)
# Pad/truncate defensively in case the state shape drifts.
safe_matches = (safe_matches + [None] * 5)[:5]
new_history, new_label = _append_regen_history(
history, lens or "loose", prompt_text or "",
paths, safe_matches,
)
choices = [(e["label"], i) for i, e in enumerate(new_history)]
return (
new_history,
gr.update(choices=choices, value=len(new_history) - 1, visible=True),
gr.update(value=_variants_prompt_card_html(prompt_text), visible=True),
)
def replay_regen_from_history(history, idx):
"""Tab click handler — replays a saved regen batch into the variants
row + prompt card. Returns 11 updates: 5 audios, 5 titles, prompt card."""
if not isinstance(history, list) or idx is None:
return [gr.update()] * 11
try:
idx = int(idx)
entry = history[idx]
except (ValueError, IndexError, TypeError):
return [gr.update()] * 11
audio_updates = []
title_updates = []
for i in range(5):
p = (entry["paths"][i] if i < len(entry["paths"]) else None)
m = (entry["matches"][i] if i < len(entry["matches"]) else None)
if p:
audio_updates.append(gr.update(value=p, visible=True))
else:
audio_updates.append(gr.update(value=None, visible=True))
title_updates.append(gr.update(value=_variant_title_html(i, m)))
prompt_card = gr.update(
value=_variants_prompt_card_html(entry.get("prompt", "")),
visible=True,
)
return [*audio_updates, *title_updates, prompt_card]
def _append_regen_history(history: list | None, lens: str, prompt_text: str,
var_paths: list, var_matches: list) -> tuple[list, str]:
"""Append a completed regen run to the history list. Returns
(new_history, new_label_for_just-added_entry).
Label format: `{LENS}·{N}` where N is the nth time this lens has been
used in the current session. Stable across re-renders because it's
computed at append time and stored on the entry."""
history = list(history or [])
same_lens_count = sum(1 for e in history if e.get("lens") == lens)
label = f"{lens.upper()}·{same_lens_count + 1}"
entry = {
"lens": lens,
"label": label,
"prompt": prompt_text or "",
"paths": list(var_paths or [None] * 5),
"matches": list(var_matches or [None] * 5),
}
history.append(entry)
return history, label
def regenerate_variants(prompt_text: str, last_run_state, duration: int = 15,
api_key: str = "", n_variants: int = 5,
session_id: str | None = "",
gen_model: str = "",
local_gen_url: str = "",
local_regen_b64_json: str = "",
local_regen_error: str = "",
duration_custom: float = 0):
"""Spawn `n_variants` SA3 variants from the (possibly edited) derived prompt.
Two paths, selected by `gen_model`:
- `local-server`: the JS prelude on regen_btn already fetched N WAVs
from the user's loopback bridge and base64-encoded them into a JSON
array in `local_regen_b64_json`. We decode + write tiles. No wallet,
no session-spend tick.
- else: existing Pollinations SA3 flow (requires wallet key).
Streams output: each variant fills its audio slot as it finishes. Slots
beyond `n_variants` are explicitly cleared so the previous batch's audio
doesn't linger in the unused cards.
Variants are added to the crate with parent_id set to the source tile
when one is known (i.e. the user came from Generate → Use for analysis).
`api_key` is the session-scoped Pollinations token from gr.State.
Output tuple: var1..5, var1_title..5_title, status_md, prompt_card,
current_matches_state
"""
# Clamp to the 1–8 chip range. Anything outside that is a caller bug.
try:
n_variants = int(n_variants or 5)
except (TypeError, ValueError):
n_variants = 5
# UI only has 5 audio slots; cap defensively.
n_variants = max(1, min(5, n_variants))
import sys as _sys, traceback as _tb
# Audio outputs use gr.update wrappers so the file-proxy properly
# registers each path as a servable static asset. Yielding raw
# filepath strings in a streaming generator was leaving v1..v5 with
# paths the browser couldn't fetch (Audio rendered the slot but no
# file URL was wired up).
paths: list[str | None] = [None, None, None, None, None]
matches: list[dict | None] = [None, None, None, None, None]
def _audio_slot(idx: int):
"""gr.update for the Audio component at slot idx. Slots beyond
n_variants are hidden so a Preview (1) doesn't leave four blank
cards in the row; a Trio (3) hides the last two. The label is
empty because the visible title is now the gr.HTML above it."""
if idx >= n_variants:
return gr.update(value=None, visible=False)
if paths[idx]:
return gr.update(value=paths[idx], visible=True)
return gr.update(value=None, visible=True)
def _title_slot(idx: int):
"""gr.update for the gr.HTML title above slot idx. Matches the
audio slot's visibility so the title disappears with its card."""
if idx >= n_variants:
return gr.update(value=_variant_title_html(idx, None), visible=False)
return gr.update(value=_variant_title_html(idx, matches[idx]), visible=True)
def _emit(status: str, prompt_card: str | None = None):
# Output order matches the regen_btn.click outputs binding:
# (var1..5, var1_title..5_title, regen_status, variants_prompt_used,
# current_matches_state)
# The matches snapshot at the end lets commit_regen_to_history
# persist per-axis bars into the regen tab — otherwise tab auto-
# select after streaming would replay the entry with None matches
# and wipe the mint fill the user just saw.
if prompt_card is None:
# Default — render the current prompt_text as the card body.
prompt_card = _variants_prompt_card_html(prompt_text)
return (
*(_audio_slot(i) for i in range(5)),
*(_title_slot(i) for i in range(5)),
status,
gr.update(value=prompt_card, visible=True),
list(matches),
)
if not (prompt_text or "").strip():
yield _emit("_❌ Enter a prompt to regenerate from._")
return
# Source-of-truth for path selection: the GEN-tab dropdown. If the user
# chose local-server there, they shouldn't be blocked by Pollinations
# state on Analyse-tab Regenerate. The JS prelude on regen_btn will
# have already done the N browser fetches by the time we're here.
use_local = (gen_model == "local-server")
if not use_local:
if not wallet.get_key(session_key=api_key):
yield _emit("_❌ Connect a Pollinations wallet first (click the pollen pill, top right)._")
return
else:
# Local path — surface JS-side errors before doing any decode work.
if local_regen_error:
yield _emit(f"_❌ Local regen browser fetch failed: {local_regen_error}_")
return
if not local_regen_b64_json:
yield _emit("_❌ Local regen returned no audio. Is the bridge running at the URL in the Local server panel?_")
return
# Try to thread parent lineage + anchor measurements from last_run.
# parent_id sets the crate-tile lineage; anchor_bpm/anchor_key/anchor_path
# power the per-variant match % readout (BPM + chroma + onset-rate
# comparison). last_run_state is populated by run_brief on
# analysis-complete.
parent_id: str | None = None
anchor_bpm: float | None = None
anchor_feats: dict | None = None
_sess_dir = crate.crate_dir(session_id)
if isinstance(last_run_state, dict):
src_path = last_run_state.get("audio_path", "")
for t in crate.list_tiles(session_id=session_id):
if t.audio_path == src_path:
parent_id = t.id
break
an = last_run_state.get("analysis")
if an is not None:
anchor_bpm = getattr(an, "bpm", None)
# Compute the anchor's chroma + onset fingerprint ONCE per batch
# so each of the 5 variant scorings reuses it (saves ~1s × 5).
if src_path:
anchor_feats = _anchor_features(src_path)
# Coerce duration — Radio choices yield int, but Gradio may pass a
# str on cold-start race conditions; default to 15 if anything's off.
# The Custom chip carries the sentinel 0 in the Radio slot with the
# real seconds in duration_custom (a gr.Number — no choices
# validation). Floor is 1 so short reference clips can be matched
# exactly (a 4s ident → 4s variants).
try:
duration = int(duration or 0)
except (TypeError, ValueError):
duration = 0
if duration == 0:
try:
duration = int(duration_custom or 0)
except (TypeError, ValueError):
duration = 0
duration = max(1, min(sa3.ABSOLUTE_MAX_S, duration)) if duration else 15
# ── Local-server path ────────────────────────────────────────────────
if use_local:
import json as _json, base64 as _b64
try:
b64_arr = _json.loads(local_regen_b64_json)
except Exception as e:
yield _emit(f"_❌ Could not parse local regen payload: {e}_")
return
if not isinstance(b64_arr, list) or not b64_arr:
yield _emit("_❌ Local regen returned an empty payload._")
return
# JS may have fetched fewer than n_variants if a mid-batch failure
# was caught; honour the actual array length.
n_variants = min(n_variants, len(b64_arr))
for i in range(n_variants):
yield _emit(f"_writing local variant {i+1}/{n_variants}…_")
try:
raw = _b64.b64decode(b64_arr[i])
except Exception as e:
print(f"[regen v{i+1}] local decode failed: {e}",
file=_sys.stderr, flush=True)
yield _emit(f"_❌ variant {i+1}/{n_variants}: decode failed_")
continue
ext = "wav" if raw[:4] == b"RIFF" else (
"mp3" if (raw[:3] == b"ID3" or (raw and raw[0] == 0xFF)) else "wav")
out_path = _sess_dir / f"local-{crate.new_id()}.{ext}"
out_path.write_bytes(raw)
tile = crate.add_tile(
audio_path=str(out_path),
source_prompt=prompt_text,
parent_id=parent_id,
model="local-server",
duration_s=float(duration),
session_id=session_id,
)
new_audio = _sess_dir / f"{tile.id}.{ext}"
try:
out_path.rename(new_audio)
tile.audio_path = str(new_audio)
tile.save()
except Exception:
pass
paths[i] = tile.audio_path
matches[i] = _quick_match(tile.audio_path, anchor_bpm, anchor_feats)
print(f"[regen v{i+1}] local tile={tile.id} bytes={len(raw)} "
f"match={matches[i]} path={tile.audio_path}",
file=_sys.stderr, flush=True)
parent_note = f" (descended from `[{parent_id}]`)" if parent_id else ""
yield _emit(f"✅ {n_variants} local variant{'s' if n_variants != 1 else ''} ready — added to the crate{parent_note}.")
return
model = sa3.DEFAULT_MODEL
for i in range(n_variants):
yield _emit(f"_generating variant {i+1}/{n_variants} via `{model}` ({duration}s)…_")
out_path = _sess_dir / f"{crate.new_id()}.mp3"
try:
info = sa3.generate(prompt_text, model=model, duration=duration,
out_path=out_path, api_key=api_key)
except sa3.SA3Error as e:
print(f"[regen v{i+1}] SA3Error: {e}", file=_sys.stderr, flush=True)
yield _emit(f"_❌ variant {i+1}/5: {e}_")
continue
except Exception as e:
print(f"[regen v{i+1}] {type(e).__name__}: {e}", file=_sys.stderr, flush=True)
_tb.print_exc(file=_sys.stderr)
yield _emit(f"_❌ variant {i+1}/5: unexpected {type(e).__name__}_")
continue
tile = crate.add_tile(
audio_path=info["path"],
source_prompt=prompt_text,
parent_id=parent_id,
model=model,
duration_s=float(duration),
session_id=session_id,
)
# Rename file to match tile id, same as generate_sa3.
new_audio = _sess_dir / f"{tile.id}.mp3"
try:
Path(info["path"]).rename(new_audio)
tile.audio_path = str(new_audio)
tile.save()
except Exception:
pass
paths[i] = tile.audio_path
# Quick match — librosa BPM + chroma + onset-rate similarity vs
# cached anchor fingerprint. ~1.5s per variant. Stores the dict
# so the slot label can render per-axis bars instead of a scalar %.
matches[i] = _quick_match(tile.audio_path, anchor_bpm, anchor_feats)
print(f"[regen v{i+1}] tile={tile.id} bytes={info.get('bytes')} "
f"match={matches[i]} path={tile.audio_path}",
file=_sys.stderr, flush=True)
parent_note = ""
if parent_id:
parent_note = f" (descended from `[{parent_id}]`)"
yield _emit(f"✅ 5 variants ready — added to the crate{parent_note}.")
def _crate_header_html(session_id: str | None = "") -> str:
"""Header above the crate chip strip: 'CRATE · N takes', or NOTHING
when empty. The previous dashed empty-state card duplicated what the
Generate button two lines above already invites — a designer-level
cleanup pass removed it. When the user has tiles, the header is a
tight one-line label. When empty, the entire crate row is silent."""
tiles = crate.list_tiles(session_id=session_id)
n = len(tiles)
if n == 0:
return ""
return (
'
'
'
CRATE
'
'
'
f'{n} take{"s" if n != 1 else ""}'
'
'
)
def refresh_crate(session_id: str | None = ""):
"""Re-scan the crate dir and refresh the chip strip + header."""
return (gr.update(choices=crate.tile_choices(session_id=session_id)),
_crate_header_html(session_id=session_id))
def select_tile(tile_id: str | None, session_id: str | None = ""):
"""Selecting a tile from the dropdown previews it + shows metadata +
reveals the Use-for-analysis / delete action row. Returns gr.update
wrappers so the preview/meta/actions show only when a tile is actually
selected (otherwise stay hidden — designer cleanup pass: no dead controls)."""
if not tile_id:
return (gr.update(value=None, visible=False),
gr.update(value="", visible=False),
gr.update(visible=False))
tile = crate.get_tile(tile_id, session_id=session_id)
if not tile:
return (gr.update(value=None, visible=False),
gr.update(value="_tile not found (was it deleted elsewhere?)_", visible=True),
gr.update(visible=False))
return (gr.update(value=tile.audio_path, visible=True),
gr.update(value=_tile_meta_md(tile, session_id=session_id), visible=True),
gr.update(visible=True))
def delete_tile(tile_id: str | None, session_id: str | None = ""):
if not tile_id:
return (gr.update(choices=crate.tile_choices(session_id=session_id)),
gr.update(value=None, visible=False),
gr.update(value="", visible=False),
_crate_header_html(session_id=session_id),
gr.update(visible=False))
crate.delete_tile(tile_id, session_id=session_id)
return (gr.update(choices=crate.tile_choices(session_id=session_id), value=None),
gr.update(value=None, visible=False),
gr.update(value="", visible=False),
_crate_header_html(session_id=session_id),
gr.update(visible=False))
def _anchor_features(anchor_path: str) -> dict | None:
"""Pre-compute chroma + onset-rate fingerprint of the anchor so we don't
re-decode + re-FFT it once per variant. Returns dict or None on failure.
Called ONCE per regen batch; cached results reused 5× in _quick_match."""
try:
import librosa
import numpy as np
y, sr = librosa.load(anchor_path, sr=22050, mono=True, duration=15.0)
chroma = np.mean(librosa.feature.chroma_cqt(y=y, sr=sr), axis=1)
norm = float(np.linalg.norm(chroma))
dur = max(len(y) / sr, 0.1)
onsets_per_s = len(librosa.onset.onset_detect(y=y, sr=sr)) / dur
return {
"chroma": chroma,
"chroma_norm": norm,
"onsets_per_s": onsets_per_s,
}
except Exception as e:
print(f"[_anchor_features] {type(e).__name__}: {e}",
file=sys.stderr, flush=True)
return None
def _quick_match(variant_path: str, anchor_bpm: float | None,
anchor_feats: dict | None) -> dict | None:
"""Per-axis similarity scores (0-100 each) for a regen variant vs anchor.
Returns `{"bpm": int, "harm": int, "rhyt": int}` or None on failure.
Why three axes (not one scalar): the previous match % was a single
weighted sum that compressed too much information and didn't track
what the listener actually heard. Two variants in the same key at the
same tempo could sound nothing alike (different timbres, different
instrumentation density) yet show 100%. Surfacing the axes lets the
user see WHICH dimension is matching/diverging — and feed that intuition
back into the prompt.
- bpm: tempo proximity (tolerates half/double-time confusion)
- harm: chroma-vector cosine similarity (12-pitch harmonic profile)
- rhyt: onset-rate similarity (rhythm density: busy vs sparse)
`anchor_feats` is the cached output of `_anchor_features` so we don't
reload the anchor audio 5x per regen batch."""
if not (anchor_bpm and anchor_feats):
return None
try:
import librosa
import numpy as np
y_v, sr = librosa.load(variant_path, sr=22050, mono=True, duration=15.0)
# Honest BPM — neutral prior so detector isn't biased to anchor BPM.
tempo_v, _ = librosa.beat.beat_track(y=y_v, sr=sr, start_bpm=120, tightness=100)
v_bpm = float(np.asarray(tempo_v).item())
ratios = [
abs(v_bpm - anchor_bpm),
abs(v_bpm * 2 - anchor_bpm),
abs(v_bpm / 2 - anchor_bpm),
]
best_bpm_err = min(ratios) / max(float(anchor_bpm), 1.0)
bpm = int(round(max(0.0, 100.0 - best_bpm_err * 500.0)))
chroma_v = np.mean(librosa.feature.chroma_cqt(y=y_v, sr=sr), axis=1)
denom = float(np.linalg.norm(chroma_v) * anchor_feats["chroma_norm"])
cos = float(np.dot(chroma_v, anchor_feats["chroma"]) / denom) if denom > 0 else 0.0
harm = int(round(max(0.0, min(100.0, (cos - 0.4) / 0.6 * 100.0))))
v_dur = max(len(y_v) / sr, 0.1)
v_onsets = len(librosa.onset.onset_detect(y=y_v, sr=sr)) / v_dur
denom = max(anchor_feats["onsets_per_s"], 0.5)
density_err = abs(v_onsets - anchor_feats["onsets_per_s"]) / denom
rhyt = int(round(max(0.0, 100.0 - density_err * 200.0)))
return {"bpm": bpm, "harm": harm, "rhyt": rhyt}
except Exception as e:
print(f"[_quick_match] {type(e).__name__}: {e}",
file=sys.stderr, flush=True)
return None
def _variants_prompt_card_html(prompt_text: str) -> str:
"""Compact card shown above the ANCHOR + 5 VARIANTS title, echoing the
derived prompt that produced the current batch. Lets the user see at a
glance which phrasing led to which 5 gens (especially useful when
iterating across MATCH / LOOSE / FREE lenses)."""
import html as _html
text = _html.escape((prompt_text or "").strip()) or "(empty)"
return (
'
'
'
PROMPT USED
'
f'
{text}
'
'
'
)
def _bars(v: int) -> str:
"""Map a 0–100 score to a 4-block bar HTML string. Quartiles at 12/38/63/88.
Returns inline HTML with a `.fill` span on the filled portion so the
measured axis (mint) reads as colour-coded fill against the muted
`▯` track behind it — that's the metering colour the variant titles
rely on for at-a-glance match strength."""
if v >= 88: n = 4
elif v >= 63: n = 3
elif v >= 38: n = 2
elif v >= 12: n = 1
else: n = 0
return f'{"▮" * n}{"▯" * (4 - n)}'
def _variant_title_html(idx: int, breakdown: dict | None) -> str:
"""Render a per-variant title bar above the Audio component.
Layout: BPM/Harm/Rhyt bars on ONE horizontal line; v-name on the right.
Compact (~30px tall) so the variant cards stay tight. Pre-regen state
shows faded placeholder bars."""
name = f"v{idx + 1}"
if not breakdown:
bars = 'BPM ▯▯▯▯ · Harm ▯▯▯▯ · Rhyt ▯▯▯▯'
cls = 'bars placeholder'
else:
bars = (
f'BPM {_bars(breakdown["bpm"])} · '
f'Harm {_bars(breakdown["harm"])} · '
f'Rhyt {_bars(breakdown["rhyt"])}'
)
cls = 'bars'
return (
'