Add Gradio app

app.py

@@ -0,0 +1,873 @@
+"""audio.cpp on Hugging Face ZeroGPU.
+
+A Gradio front end for https://github.com/0xShug0/audio.cpp: 100+ TTS, voice
+cloning, ASR, diarization, separation, voice conversion and music models,
+served by the native audio.cpp CUDA runtime. GPU work runs inside a ZeroGPU
+call; when no GPU is available the same binary runs on CPU.
+"""
+from __future__ import annotations
+
+import spaces  # import first so ZeroGPU can patch the process
+
+import html
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import traceback
+import urllib.parse
+import urllib.request
+import uuid
+import zipfile
+from pathlib import Path
+
+import gradio as gr
+
+import audiocpp_runtime as ac
+
+APP_DIR = Path(__file__).resolve().parent
+ALLOW_PYTHON = os.environ.get("ALLOW_PYTHON", "1").lower() not in {"0", "false", "no", "off"}
+MAX_BATCH = int(os.environ.get("MAX_BATCH", "24"))
+MAX_GPU_SECONDS = int(os.environ.get("MAX_GPU_SECONDS", "240"))
+MAX_DOWNLOAD_BYTES = int(os.environ.get("MAX_DOWNLOAD_BYTES", str(300 * 1024 * 1024)))
+SAMPLE_URL = "https://raw.githubusercontent.com/0xShug0/audio.cpp/main/assets/resources/{name}"
+SAMPLE_DIR = ac.HOME / "samples"
+
+
+def sample_file(name: str) -> str:
+    """Fetch a small sample clip from the audio.cpp repo (cached)."""
+    SAMPLE_DIR.mkdir(parents=True, exist_ok=True)
+    dest = SAMPLE_DIR / name
+    if not dest.exists():
+        urllib.request.urlretrieve(SAMPLE_URL.format(name=name), dest)
+    return str(dest)
+
+# --------------------------------------------------------------------------- boot
+SETUP_ERROR = ""
+try:
+    ac.ensure_runtime()
+    CATALOG = ac.catalog()
+except Exception as exc:  # keep the UI alive so the error is visible
+    SETUP_ERROR = f"{type(exc).__name__}: {exc}"
+    traceback.print_exc()
+    CATALOG = {}
+
+try:
+    MODEL_PARAMS = json.loads((APP_DIR / "model_params.json").read_text(encoding="utf-8"))
+except Exception:
+    MODEL_PARAMS = {}
+
+TASK_ORDER = ["tts", "clone", "design", "asr", "align", "diar", "vad", "sep", "vc", "svc",
+              "s2s", "music", "sfx", "edit", "midi", "turn"]
+TASKS = [t for t in TASK_ORDER if any(t in f["tasks"] for f in CATALOG.values())]
+TASK_CHOICES = [(ac.TASK_LABELS.get(t, t), t) for t in TASKS]
+
+TEXT_TASKS = {"tts", "clone", "design", "align", "music", "sfx", "edit", "s2s"}
+AUDIO_TASKS = {"asr", "align", "vad", "diar", "sep", "vc", "svc", "s2s", "edit", "midi", "turn"}
+VOICE_TASKS = {"tts", "clone", "design", "vc", "svc"}
+INSTRUCT_TASKS = {"tts", "clone", "design", "s2s", "edit"}
+MUSIC_TASKS = {"music", "sfx", "edit"}
+TEXT_LABELS = {
+    "tts": "Text to speak", "clone": "Text to speak", "design": "Text to speak",
+    "align": "Transcript to align", "music": "Music prompt / caption", "sfx": "Sound prompt",
+    "edit": "Edit instruction / prompt", "s2s": "Optional text prompt",
+}
+
+FAMILY_PREFS = {
+    "tts": ["supertonic", "kokoro_tts", "pocket_tts", "omnivoice", "qwen3_tts"],
+    "clone": ["pocket_tts", "omnivoice", "chatterbox", "qwen3_tts", "index_tts2"],
+    "design": ["omnivoice", "qwen3_tts", "maya1"],
+    "asr": ["qwen3_asr", "parakeet_tdt", "nemotron_asr", "sense_asr", "fun_asr_nano"],
+    "align": ["qwen3_forced_aligner"],
+    "diar": ["moss_transcribe_diarize", "nemotron_3_diar", "sortformer_diar"],
+    "vad": ["silero_vad", "pulsevad"],
+    "sep": ["htdemucs", "mel_band_roformer", "bs_roformer"],
+    "vc": ["seed_vc", "chatterbox", "meanvc2"],
+    "music": ["ace_step", "stable_audio", "heartmula"],
+    "sfx": ["stable_audio", "midashenglm_gen"],
+}
+
+
+def families_for(task: str) -> list[tuple[str, str]]:
+    fams = ac.families_for_task(task) if CATALOG else []
+    prefs = [f for f in FAMILY_PREFS.get(task, []) if f in fams]
+    ordered = prefs + [f for f in fams if f not in prefs]
+    return [(f"{CATALOG[f]['display_name']} ({f})", f) for f in ordered]
+
+
+def packages_for(family: str) -> list[tuple[str, str]]:
+    if family not in CATALOG:
+        return []
+    rows = []
+    for p in CATALOG[family]["packages"]:
+        mark = " [downloaded]" if ac.is_installed(p["id"]) else ""
+        fmt = p.get("format", "")
+        label = f"{p.get('display_name') or p['id']} [{fmt}]{mark}"
+        rows.append((label, p["id"]))
+    return rows
+
+
+def _human(n: int | None) -> str:
+    if not n:
+        return "unknown size"
+    for unit in ["B", "KB", "MB", "GB", "TB"]:
+        if n < 1024:
+            return f"{n:.1f} {unit}"
+        n /= 1024
+    return f"{n:.1f} PB"
+
+
+def family_info(family: str, pkg: str | None) -> str:
+    if family not in CATALOG:
+        return ""
+    f = CATALOG[family]
+    langs = ", ".join(f["languages"][:30]) + (" ..." if len(f["languages"]) > 30 else "")
+    docs = " · ".join(f"[{Path(d).name}](https://github.com/0xShug0/audio.cpp/blob/main/{d})" for d in f["docs"][:4])
+    lines = [f"**{f['display_name']}** · tasks: `{', '.join(f['tasks'])}`" + (f" · status: {f['status']}" if f['status'] else ""),
+             f["description"]]
+    if langs:
+        lines.append(f"Languages: {langs}")
+    if f["voices"]:
+        lines.append(f"Built-in voices: {len(f['voices'])}")
+    if pkg:
+        state = "downloaded" if ac.is_installed(pkg) else "downloads on first run"
+        lines.append(f"Package `{pkg}`: {state}")
+    if docs:
+        lines.append(f"Docs: {docs}")
+    return "\n\n".join(lines)
+
+
+def param_defaults(family: str) -> tuple[str, str, str]:
+    """(request options JSON, session options JSON, markdown reference table)."""
+    req, sess = {}, {}
+    for p in MODEL_PARAMS.get(family, []) if isinstance(MODEL_PARAMS.get(family), list) else []:
+        default = p.get("default")
+        if default in (None, ""):
+            continue
+        (sess if p.get("scope") == "session" else req)[p["name"]] = default
+    rows = ["| option | type | default | description |", "|---|---|---|---|"]
+    for o in CATALOG.get(family, {}).get("request_options", [])[:60]:
+        desc = str(o.get("description", "")).replace("|", "/")
+        rng = ""
+        if "min" in o or "max" in o:
+            rng = f" ({o.get('min', '')}..{o.get('max', '')})"
+        vals = f" values: {', '.join(map(str, o['values']))}" if o.get("values") else ""
+        rows.append(f"| `{o['name']}` | {o.get('type', '')}{rng} | {o.get('default', '')} | {desc}{vals} |")
+    table = "\n".join(rows) if len(rows) > 2 else "_This model's spec declares no request options._"
+    return (json.dumps(req, indent=2, ensure_ascii=False) if req else "{}",
+            json.dumps(sess, indent=2, ensure_ascii=False) if sess else "{}",
+            table)
+
+
+# --------------------------------------------------------------------------- text input helpers
+def extract_text(path: str) -> str:
+    p = Path(path)
+    ext = p.suffix.lower()
+    if ext == ".pdf":
+        from pypdf import PdfReader
+        return "\n\n".join((page.extract_text() or "") for page in PdfReader(str(p)).pages)
+    if ext == ".docx":
+        import docx
+        return "\n".join(par.text for par in docx.Document(str(p)).paragraphs)
+    if ext == ".epub":
+        parts = []
+        with zipfile.ZipFile(p) as z:
+            for name in sorted(z.namelist()):
+                if name.lower().endswith((".xhtml", ".html", ".htm")):
+                    parts.append(strip_html(z.read(name).decode("utf-8", "replace")))
+        return "\n\n".join(parts)
+    raw = p.read_bytes().decode("utf-8", "replace")
+    if ext in {".html", ".htm", ".xhtml", ".xml"}:
+        return strip_html(raw)
+    if ext in {".srt", ".vtt"}:
+        return strip_subtitles(raw)
+    if ext == ".json":
+        try:
+            data = json.loads(raw)
+            if isinstance(data, str):
+                return data
+            if isinstance(data, dict):
+                for key in ("text", "input", "content", "transcript"):
+                    if isinstance(data.get(key), str):
+                        return data[key]
+        except Exception:
+            pass
+    return raw
+
+
+def strip_html(raw: str) -> str:
+    raw = re.sub(r"(?is)<(script|style|head).*?</\1>", " ", raw)
+    raw = re.sub(r"(?i)<br\s*/?>|</p>|</div>|</h\d>|</li>", "\n", raw)
+    raw = re.sub(r"<[^>]+>", " ", raw)
+    raw = html.unescape(raw)
+    return re.sub(r"[ \t]+", " ", re.sub(r"\n\s*\n+", "\n\n", raw)).strip()
+
+
+def strip_subtitles(raw: str) -> str:
+    out = []
+    for line in raw.splitlines():
+        s = line.strip()
+        if not s or s.isdigit() or "-->" in s or s.upper().startswith("WEBVTT"):
+            continue
+        out.append(re.sub(r"<[^>]+>", "", s))
+    return " ".join(out)
+
+
+def fetch_url(url: str, dest_dir: Path) -> Path:
+    url = url.strip()
+    if not re.match(r"^https?://", url):
+        raise gr.Error("URLs must start with http:// or https://")
+    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (audio.cpp Space)"})
+    with urllib.request.urlopen(req, timeout=60) as resp:
+        ctype = resp.headers.get("Content-Type", "")
+        name = Path(urllib.parse.urlparse(resp.geturl()).path).name or "download"
+        if "." not in name:
+            guess = {"text/html": ".html", "application/pdf": ".pdf", "audio/mpeg": ".mp3",
+                     "audio/wav": ".wav", "text/plain": ".txt"}.get(ctype.split(";")[0].strip(), "")
+            name += guess
+        dest = dest_dir / f"{uuid.uuid4().hex[:6]}-{name[-80:]}"
+        size = 0
+        with open(dest, "wb") as out:
+            while chunk := resp.read(1 << 20):
+                size += len(chunk)
+                if size > MAX_DOWNLOAD_BYTES:
+                    raise gr.Error(f"Download is larger than {_human(MAX_DOWNLOAD_BYTES)}.")
+                out.write(chunk)
+    return dest
+
+
+def load_text(files, url: str, current: str, mode: str) -> str:
+    chunks = []
+    tmp = Path(tempfile.mkdtemp(prefix="text-"))
+    for f in files or []:
+        path = f if isinstance(f, str) else getattr(f, "name", str(f))
+        chunks.append(extract_text(path).strip())
+    if url and url.strip():
+        chunks.append(extract_text(str(fetch_url(url, tmp))).strip())
+    if not chunks:
+        raise gr.Error("Add a text file or a URL first.")
+    loaded = "\n\n".join(c for c in chunks if c)
+    if mode == "Append" and current and current.strip():
+        return current.rstrip() + "\n\n" + loaded
+    return loaded
+
+
+def split_text(text: str, mode: str) -> list[str]:
+    text = text.strip()
+    if mode == "One clip per paragraph":
+        parts = [p.strip() for p in re.split(r"\n\s*\n", text)]
+    elif mode == "One clip per line":
+        parts = [p.strip() for p in text.splitlines()]
+    else:
+        parts = [text]
+    return [p for p in parts if p]
+
+
+# --------------------------------------------------------------------------- GPU entry points
+def _gpu_seconds(job: dict, seconds: int) -> int:
+    return int(max(20, min(int(seconds or 90), MAX_GPU_SECONDS)))
+
+
+@spaces.GPU(duration=_gpu_seconds)
+def gpu_execute(job: dict, seconds: int) -> dict:
+    return ac.execute(job, prefer_gpu=True)
+
+
+def _python_seconds(code_path: str, seconds: int) -> int:
+    return int(max(20, min(int(seconds or 90), MAX_GPU_SECONDS)))
+
+
+@spaces.GPU(duration=_python_seconds)
+def gpu_python(code_path: str, seconds: int) -> dict:
+    return run_python_file(code_path, seconds)
+
+
+def _looks_like_gpu_unavailable(err: Exception) -> bool:
+    msg = str(err).lower()
+    return any(k in msg for k in ("quota", "zerogpu", "no gpu", "gpu task aborted", "unlogged",
+                                  "gpu is not available", "too many requests"))
+
+
+# --------------------------------------------------------------------------- main run
+def _parse_json(text: str, what: str) -> dict:
+    text = (text or "").strip()
+    if not text:
+        return {}
+    try:
+        val = json.loads(text)
+    except json.JSONDecodeError as err:
+        raise gr.Error(f"{what} is not valid JSON: {err}")
+    if not isinstance(val, dict):
+        raise gr.Error(f"{what} must be a JSON object.")
+    return val
+
+
+def _collect_audio(mic, files, url, work: Path) -> list[str]:
+    paths: list[str] = []
+    if mic:
+        paths.append(mic)
+    for f in files or []:
+        paths.append(f if isinstance(f, str) else getattr(f, "name", str(f)))
+    if url and url.strip():
+        for u in re.split(r"[\s,]+", url.strip()):
+            if u:
+                paths.append(str(fetch_url(u, work)))
+    return paths
+
+
+def _fmt_time(sample: int | float | None, rate: int | None) -> str:
+    if sample is None or not rate:
+        return "?"
+    s = float(sample) / float(rate)
+    return f"{int(s // 60):02d}:{s % 60:06.3f}"
+
+
+def _srt_time(sec: float) -> str:
+    ms = int(round(sec * 1000))
+    h, ms = divmod(ms, 3600000)
+    m, ms = divmod(ms, 60000)
+    s, ms = divmod(ms, 1000)
+    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
+
+
+def describe_result(res: dict, label: str, out_dir: Path) -> tuple[str, str | None]:
+    """Human-readable text for one result, plus an SRT file when timings exist."""
+    lines = [f"### {label}"] if label else []
+    rate = res.get("sample_rate")
+    if res.get("text"):
+        lines.append(res["text"])
+    if res.get("language"):
+        lines.append(f"_language: {res['language']}_")
+    cues = []
+    for key, title in (("speaker_turns", "Speaker turns"), ("segments", "Segments")):
+        items = res.get(key) or []
+        if items:
+            lines.append(f"\n**{title}**")
+            for it in items[:2000]:
+                who = f" speaker {it['speaker_id']}:" if "speaker_id" in it else ""
+                txt = f" {it['text']}" if it.get("text") else ""
+                conf = f" ({it['confidence']:.2f})" if isinstance(it.get("confidence"), (int, float)) and not txt else ""
+                lines.append(f"[{_fmt_time(it.get('start_sample'), rate)} - {_fmt_time(it.get('end_sample'), rate)}]{who}{txt}{conf}")
+                if rate and it.get("text"):
+                    cues.append((it["start_sample"] / rate, it["end_sample"] / rate,
+                                 (f"[{it['speaker_id']}] " if "speaker_id" in it else "") + it["text"]))
+    words = res.get("words") or []
+    if words:
+        shown = " ".join(f"{w.get('word')}@{_fmt_time(w.get('start_sample'), rate)}" for w in words[:400])
+        lines.append(f"\n**Words** ({len(words)}): {shown}")
+        if not cues and rate:
+            group: list[dict] = []
+            for w in words:
+                group.append(w)
+                if len(group) >= 10 or str(w.get("word", "")).endswith((".", "?", "!")):
+                    cues.append((group[0]["start_sample"] / rate, group[-1]["end_sample"] / rate,
+                                 " ".join(str(g.get("word", "")) for g in group)))
+                    group = []
+            if group:
+                cues.append((group[0]["start_sample"] / rate, group[-1]["end_sample"] / rate,
+                             " ".join(str(g.get("word", "")) for g in group)))
+    if res.get("custom_schema_output"):
+        lines.append("```json\n" + json.dumps(res["custom_schema_output"], indent=1, ensure_ascii=False)[:20000] + "\n```")
+    srt = None
+    if cues:
+        srt_path = out_dir / f"{re.sub(r'[^A-Za-z0-9_-]+', '_', label or 'transcript')}.srt"
+        srt_path.write_text("\n".join(f"{i}\n{_srt_time(a)} --> {_srt_time(b)}\n{t}\n"
+                                      for i, (a, b, t) in enumerate(cues, 1)), encoding="utf-8")
+        srt = str(srt_path)
+    return "\n".join(lines), srt
+
+
+def concat_wavs(paths: list[str], out: Path) -> str | None:
+    if len(paths) < 2:
+        return None
+    lst = out.with_suffix(".txt")
+    lst.write_text("".join(f"file '{p}'\n" for p in paths))
+    res = subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "concat", "-safe", "0",
+                          "-i", str(lst), "-c:a", "pcm_s16le", str(out)], capture_output=True)
+    return str(out) if res.returncode == 0 else None
+
+
+def run_task(task, family, pkg, device, gpu_seconds, text, split_mode, audio_mic, audio_files, audio_url,
+             voice_id, voice_ref, ref_text, instruct, lyrics, language, seed, duration, speaking_rate,
+             extra_json, load_json, session_json, raw_json, progress=gr.Progress()):
+    if SETUP_ERROR:
+        raise gr.Error("audio.cpp failed to start: " + SETUP_ERROR)
+    if not family or family not in CATALOG:
+        raise gr.Error("Pick a model family.")
+    pkg = pkg or ac.default_package(family)
+    work = ac.WORK_ROOT / f"{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}"
+    inputs = work / "inputs"
+    inputs.mkdir(parents=True, exist_ok=True)
+    status = []
+
+    # ---- base request
+    options = _parse_json(extra_json, "Model options")
+    base: dict = {}
+    if language and str(language).strip():
+        base["language"] = str(language).strip()
+    if task in VOICE_TASKS and voice_id and str(voice_id).strip():
+        base["voice_id"] = str(voice_id).strip()
+    if task in VOICE_TASKS and voice_ref:
+        progress(0.02, desc="Preparing reference voice")
+        base["voice_ref"] = ac.to_wav(voice_ref, inputs, mono=True, max_seconds=60)
+    if ref_text and ref_text.strip() and task in VOICE_TASKS:
+        base["reference_text"] = ref_text.strip()
+    if instruct and instruct.strip() and task in INSTRUCT_TASKS:
+        base["instruct"] = instruct.strip()
+    if task in MUSIC_TASKS:
+        if lyrics and lyrics.strip():
+            base["lyrics"] = lyrics.strip()
+        if duration and float(duration) > 0:
+            base["duration_seconds"] = float(duration)
+    if seed is not None and int(seed) >= 0:
+        base["seed"] = int(seed)
+    if speaking_rate and abs(float(speaking_rate) - 1.0) > 1e-6 and task in {"tts", "clone", "design"}:
+        base["speaking_rate"] = float(speaking_rate)
+    if options:
+        base["options"] = options
+    base.update(_parse_json(raw_json, "Raw request JSON"))
+
+    if task == "clone" and "voice_ref" not in base:
+        raise gr.Error("Voice cloning needs a reference voice recording.")
+    if task in {"vc", "svc"} and "voice_ref" not in base:
+        raise gr.Error("Voice conversion needs a target voice recording.")
+
+    # ---- requests
+    requests: list[dict] = []
+    rate, mono = ac.input_format(task, family)
+    audio_inputs = _collect_audio(audio_mic, audio_files, audio_url, inputs) if task in AUDIO_TASKS else []
+    if task in AUDIO_TASKS and not audio_inputs and task not in {"s2s", "edit"} and "audio" not in base:
+        raise gr.Error("Add input audio (upload, record, file, or URL).")
+    texts = split_text(text or "", split_mode) if task in TEXT_TASKS else []
+    if task in {"tts", "clone", "design", "music", "sfx", "align"} and not texts and "text" not in base:
+        raise gr.Error("Enter some text (or load it from a file / URL).")
+
+    if audio_inputs:
+        for i, src in enumerate(audio_inputs[:MAX_BATCH]):
+            progress(0.05 + 0.1 * i / max(1, len(audio_inputs)), desc=f"Converting {Path(src).name}")
+            r = dict(base)
+            r["audio"] = ac.to_wav(src, inputs, rate=rate, mono=mono)
+            if texts:
+                r["text"] = "\n\n".join(texts)
+            r["id"] = re.sub(r"[^A-Za-z0-9_-]+", "_", Path(src).stem)[:40] + f"-{i + 1}"
+            requests.append(r)
+        if len(audio_inputs) > MAX_BATCH:
+            status.append(f"Only the first {MAX_BATCH} audio inputs were processed.")
+    elif texts:
+        for i, t in enumerate(texts[:MAX_BATCH]):
+            requests.append({**base, "text": t, "id": f"clip{i + 1:02d}"})
+        if len(texts) > MAX_BATCH:
+            status.append(f"Only the first {MAX_BATCH} text chunks were processed.")
+    else:
+        requests.append({**base, "id": "out01"})
+
+    # ---- model download (CPU side, before the GPU is attached)
+    if not ac.is_installed(pkg):
+        progress(0.15, desc=f"Downloading {pkg}")
+    model_path = ac.install(pkg, progress=progress)
+    sess = {**ac.session_options_for(pkg), **_parse_json(session_json, "Session options")}
+    job = {"family": family, "task": task, "model_path": str(model_path), "requests": requests,
+           "load_options": _parse_json(load_json, "Load options"), "session_options": sess,
+           "out_dir": str(work), "timeout": max(30, int(gpu_seconds) - 5)}
+
+    # ---- run
+    progress(0.6, desc="Running on ZeroGPU" if device == "gpu" else "Running on CPU")
+    t0 = time.time()
+    try:
+        if device == "gpu":
+            try:
+                out = gpu_execute(job, int(gpu_seconds))
+            except Exception as err:
+                if not _looks_like_gpu_unavailable(err):
+                    raise
+                status.append(f"ZeroGPU unavailable ({str(err)[:160]}). Fell back to CPU.")
+                job["timeout"] = 1800
+                out = ac.execute(job, prefer_gpu=False)
+        else:
+            job["timeout"] = 1800
+            out = ac.execute(job, prefer_gpu=False)
+    except gr.Error:
+        raise
+    except Exception as err:
+        raise gr.Error(str(err)[-3000:], duration=None)
+
+    # ---- outputs
+    files, audios, texts_out = [], [], []
+    for req, res in zip(requests, out["results"]):
+        files.extend(res.get("files", []))
+        if res.get("audio_file"):
+            audios.append(res["audio_file"])
+        desc, srt = describe_result(res, req["id"] if len(requests) > 1 else "", work)
+        if desc.strip():
+            texts_out.append(desc)
+        if srt:
+            files.append(srt)
+    combined = concat_wavs(audios, work / "combined.wav") if task in {"tts", "clone", "design"} else None
+    if combined:
+        files.insert(0, combined)
+    main_audio = combined or (audios[0] if audios else next((f for f in files if f.endswith(".wav")), None))
+    transcript = "\n\n".join(texts_out)
+    if transcript:
+        tpath = work / "result.md"
+        tpath.write_text(transcript, encoding="utf-8")
+        files.append(str(tpath))
+    status.insert(0, f"Done on **{out['backend'].upper()}** in {time.time() - t0:.1f}s "
+                     f"({len(requests)} request{'s' if len(requests) != 1 else ''}, package `{pkg}`).")
+    details = {"backend": out["backend"], "elapsed_s": out["elapsed_s"], "package": pkg,
+               "model_path": str(model_path), "requests": requests, "results": out["results"]}
+    return (main_audio, files or None, transcript or "", details, "\n\n".join(status), out.get("log", ""),
+            gr.update(choices=packages_for(family), value=pkg))
+
+
+# --------------------------------------------------------------------------- UI callbacks
+def on_task(task):
+    fams = families_for(task)
+    fam = fams[0][1] if fams else None
+    is_text, is_audio = task in TEXT_TASKS, task in AUDIO_TASKS
+    return (
+        gr.update(choices=fams, value=fam),
+        gr.update(visible=is_text, label=TEXT_LABELS.get(task, "Text")),
+        gr.update(visible=is_text),
+        gr.update(visible=is_audio),
+        gr.update(visible=task in VOICE_TASKS),
+        gr.update(visible=task in INSTRUCT_TASKS,
+                  label="Voice description" if task == "design" else "Instruction / style (optional)"),
+        gr.update(visible=task in MUSIC_TASKS),
+        gr.update(visible=task in {"tts", "clone", "design"}),
+    )
+
+
+def on_family(family):
+    if family not in CATALOG:
+        return gr.update(choices=[], value=None), "", gr.update(), "{}", "{}", ""
+    pkg = ac.default_package(family)
+    f = CATALOG[family]
+    voices = f["voices"]
+    table = param_defaults(family)[2]
+    return (gr.update(choices=packages_for(family), value=pkg),
+            family_info(family, pkg),
+            gr.update(choices=voices, value=f["default_voice"] if f["default_voice"] in voices else None),
+            "{}", "{}", table)
+
+
+def on_package(family, pkg):
+    return family_info(family, pkg)
+
+
+def check_size(pkg):
+    if not pkg:
+        return "Pick a package."
+    size = ac.package_size(pkg)
+    return f"`{pkg}`: {_human(size)} ({'downloaded' if ac.is_installed(pkg) else 'not downloaded yet'})"
+
+
+# --------------------------------------------------------------------------- models tab
+def catalog_rows():
+    rows = []
+    for f in sorted(CATALOG.values(), key=lambda x: x["family"]):
+        inst = [p["id"] for p in f["packages"] if ac.is_installed(p["id"])]
+        rows.append([f["family"], f["display_name"], ", ".join(f["tasks"]), len(f["packages"]),
+                     ", ".join(inst), (f["description"] or "")[:160]])
+    return rows
+
+
+def all_package_choices():
+    return [(f"{p['id']} ({f['family']})", p["id"]) for f in CATALOG.values() for p in f["packages"]]
+
+
+def disk_report():
+    used = 0
+    for p in ac.MODELS_ROOT.rglob("*"):
+        if p.is_file():
+            used += p.stat().st_size
+    free = shutil.disk_usage(ac.MODELS_ROOT).free if ac.MODELS_ROOT.exists() else 0
+    return f"Models folder `{ac.MODELS_ROOT}` uses {_human(used)}, {_human(free)} free."
+
+
+def models_download(pkg, progress=gr.Progress()):
+    if not pkg:
+        raise gr.Error("Pick a package.")
+    ac.install(pkg, progress=progress)
+    return f"Downloaded `{pkg}`.\n\n{disk_report()}", catalog_rows()
+
+
+def models_delete(pkg):
+    if not pkg:
+        raise gr.Error("Pick a package.")
+    ac.uninstall(pkg)
+    return f"Removed `{pkg}`.\n\n{disk_report()}", catalog_rows()
+
+
+# --------------------------------------------------------------------------- python tab
+PY_EXAMPLE = '''# Helper module: audiocpp_runtime (imported as ac). Files you write to the
+# current directory show up below (audio is previewed).
+import audiocpp_runtime as ac
+
+print("tasks:", sorted({t for r in ac.list_packages() for t in r["tasks"]}))
+print("GPU visible:", ac.gpu_available())
+
+result = ac.run(
+    family="supertonic", task="tts",
+    request={"text": "Hello from audio.cpp, running inside a Hugging Face Space."},
+    out_dir=".",
+)
+print(result["backend"], result["results"][0]["files"])
+'''
+
+SECRET_HINTS = ("TOKEN", "SECRET", "PASSWORD", "API_KEY", "ACCESS_KEY", "PRIVATE")
+
+
+def run_python_file(code_path: str, seconds: int) -> dict:
+    workdir = Path(code_path).parent
+    env = {k: v for k, v in os.environ.items() if not any(h in k.upper() for h in SECRET_HINTS)}
+    env["PYTHONPATH"] = str(APP_DIR) + (":" + env["PYTHONPATH"] if env.get("PYTHONPATH") else "")
+    env["MPLBACKEND"] = "Agg"
+    t0 = time.time()
+    try:
+        proc = subprocess.run([sys.executable, "-u", code_path], cwd=workdir, env=env,
+                              capture_output=True, text=True, timeout=max(10, int(seconds) - 3))
+        out, code = proc.stdout + ("\n" + proc.stderr if proc.stderr else ""), proc.returncode
+    except subprocess.TimeoutExpired as err:
+        out = (err.stdout or "") + "\n" + (err.stderr or "") if isinstance(err.stdout, str) else ""
+        out += f"\n[timed out after {seconds}s]"
+        code = -1
+    files = sorted(str(p) for p in workdir.rglob("*") if p.is_file() and p.name != "main.py"
+                   and not p.name.startswith("server-"))
+    return {"output": out[-50000:], "returncode": code, "files": files[:50], "seconds": round(time.time() - t0, 1)}
+
+
+def run_python(code, on_gpu, seconds):
+    if not ALLOW_PYTHON:
+        raise gr.Error("The Python runner is disabled on this Space (ALLOW_PYTHON=0).")
+    if not code or not code.strip():
+        raise gr.Error("Write some code first.")
+    workdir = ac.WORK_ROOT / f"py-{uuid.uuid4().hex[:8]}"
+    workdir.mkdir(parents=True, exist_ok=True)
+    code_path = workdir / "main.py"
+    code_path.write_text(code, encoding="utf-8")
+    if on_gpu:
+        try:
+            res = gpu_python(str(code_path), int(seconds))
+        except Exception as err:
+            if not _looks_like_gpu_unavailable(err):
+                raise gr.Error(str(err)[-2000:])
+            res = run_python_file(str(code_path), int(seconds))
+            res["output"] = f"[ZeroGPU unavailable, ran on CPU: {str(err)[:160]}]\n" + res["output"]
+    else:
+        res = run_python_file(str(code_path), int(seconds))
+    audio = next((f for f in res["files"] if f.lower().endswith((".wav", ".mp3", ".flac", ".ogg"))), None)
+    header = f"exit code {res['returncode']} · {res['seconds']}s\n" + "-" * 40 + "\n"
+    return header + res["output"], res["files"] or None, audio
+
+
+# --------------------------------------------------------------------------- layout
+CSS = """
+#title {margin-bottom: 0}
+.small-note {font-size: 0.85em; opacity: 0.8}
+"""
+
+TEXT_FILE_TYPES = [".txt", ".md", ".markdown", ".srt", ".vtt", ".json", ".csv", ".html", ".htm",
+                   ".xml", ".pdf", ".docx", ".epub", ".rtf", ".log"]
+AUDIO_FILE_TYPES = ["audio", "video", ".wav", ".mp3", ".flac", ".ogg", ".opus", ".m4a", ".aac", ".webm",
+                    ".mp4", ".mkv", ".mov"]
+
+
+def build_ui() -> gr.Blocks:
+    first_task = TASKS[0] if TASKS else "tts"
+    first_fams = families_for(first_task)
+    first_fam = first_fams[0][1] if first_fams else None
+    first_table = param_defaults(first_fam)[2] if first_fam else ""
+    first_voices = CATALOG.get(first_fam, {}).get("voices", []) if first_fam else []
+
+    with gr.Blocks(title="audio.cpp · ZeroGPU") as demo:
+        gr.Markdown(
+            "# audio.cpp on ZeroGPU\n"
+            "Run [audio.cpp](https://github.com/0xShug0/audio.cpp) (a native C++/ggml engine for 100+ audio models) "
+            "in the browser: TTS, voice cloning and design, ASR, alignment, diarization, VAD, stem separation, "
+            "voice conversion, restoration, and music or sound generation. Models download on first use from "
+            "[audio-cpp/audio.cpp-gguf](https://huggingface.co/audio-cpp/audio.cpp-gguf).",
+            elem_id="title")
+        if SETUP_ERROR:
+            gr.Markdown(f"**Startup error:** `{SETUP_ERROR}`")
+
+        with gr.Tabs():
+            # ------------------------------------------------------------- run tab
+            with gr.Tab("Run"):
+                with gr.Row():
+                    task = gr.Dropdown(TASK_CHOICES, value=first_task, label="Task", scale=2)
+                    family = gr.Dropdown(first_fams, value=first_fam, label="Model family", scale=3)
+                    pkg = gr.Dropdown(packages_for(first_fam) if first_fam else [],
+                                      value=ac.default_package(first_fam) if first_fam else None,
+                                      label="Package / precision", scale=3)
+                with gr.Row():
+                    device = gr.Radio([("ZeroGPU (CPU fallback)", "gpu"), ("CPU only", "cpu")], value="gpu",
+                                      label="Device", scale=2)
+                    gpu_seconds = gr.Slider(30, MAX_GPU_SECONDS, value=120, step=10,
+                                            label="GPU time budget (seconds, includes model load)", scale=3)
+                    size_btn = gr.Button("Check download size", scale=1)
+                info = gr.Markdown(family_info(first_fam, ac.default_package(first_fam)) if first_fam else "")
+
+                with gr.Row():
+                    with gr.Column(scale=5):
+                        text = gr.Textbox(label=TEXT_LABELS.get(first_task, "Text"), lines=7,
+                                          placeholder="Type or paste text here, or load it from files / a URL below.",
+                                          visible=first_task in TEXT_TASKS)
+                        with gr.Accordion("Load text from files or a URL", open=False,
+                                          visible=first_task in TEXT_TASKS) as text_loader:
+                            text_files = gr.File(label="Text files (txt, md, srt, vtt, pdf, docx, epub, html, json)",
+                                                 file_count="multiple", file_types=TEXT_FILE_TYPES)
+                            text_url = gr.Textbox(label="Or a web page / text URL", placeholder="https://...")
+                            with gr.Row():
+                                load_mode = gr.Radio(["Replace", "Append"], value="Replace", label="Load mode")
+                                load_btn = gr.Button("Load into text box")
+                            split_mode = gr.Radio(["Whole text", "One clip per paragraph", "One clip per line"],
+                                                  value="Whole text", label="Batch mode")
+
+                        with gr.Group(visible=first_task in AUDIO_TASKS) as audio_group:
+                            audio_mic = gr.Audio(sources=["upload", "microphone"], type="filepath",
+                                                 label="Input audio (upload or record)")
+                            audio_files = gr.File(label="More audio / video files (batch, any format ffmpeg reads)",
+                                                  file_count="multiple", file_types=AUDIO_FILE_TYPES)
+                            audio_url = gr.Textbox(label="Or audio / video URLs (space or comma separated)",
+                                                   placeholder="https://.../speech.mp3")
+                            sample_audio_btn = gr.Button("Use a sample speech clip", size="sm")
+
+                        with gr.Group(visible=first_task in VOICE_TASKS) as voice_group:
+                            voice_id = gr.Dropdown(first_voices, value=None, allow_custom_value=True,
+                                                   label="Built-in voice id (optional)")
+                            voice_ref = gr.Audio(sources=["upload", "microphone"], type="filepath",
+                                                 label="Reference / target voice (cloning and conversion)")
+                            ref_text = gr.Textbox(label="Reference transcript (helps some cloning models)", lines=2)
+                            sample_voice_btn = gr.Button("Use a sample reference voice", size="sm")
+
+                        instruct = gr.Textbox(label="Instruction / style (optional)", lines=2,
+                                              visible=first_task in INSTRUCT_TASKS)
+                        with gr.Group(visible=first_task in MUSIC_TASKS) as music_group:
+                            lyrics = gr.Textbox(label="Lyrics (optional)", lines=5)
+                            duration = gr.Number(label="Duration in seconds (0 = model default)", value=30)
+
+                        with gr.Row():
+                            language = gr.Textbox(label="Language (optional, e.g. en, zh, auto)", scale=2)
+                            seed = gr.Number(label="Seed (-1 = random)", value=-1, precision=0, scale=1)
+                            speaking_rate = gr.Slider(0.5, 2.0, value=1.0, step=0.05, label="Speaking rate",
+                                                      visible=first_task in {"tts", "clone", "design"}, scale=2)
+
+                        with gr.Accordion("Advanced model options", open=False):
+                            gr.Markdown("Values here are sent as per-request options (overriding the fields above). "
+                                        "The table lists every option the model's spec declares.",
+                                        elem_classes="small-note")
+                            extra_json = gr.Code("{}", language="json", label="Request options (JSON)", lines=6)
+                            defaults_btn = gr.Button("Insert this model's suggested defaults", size="sm")
+                            session_json = gr.Code("{}", language="json", label="Session options (JSON)", lines=3)
+                            load_json = gr.Code("{}", language="json", label="Load options (JSON)", lines=3)
+                            raw_json = gr.Code("{}", language="json",
+                                               label="Raw request fields merged last (JSON, e.g. {\"emotion\": \"happy\"})",
+                                               lines=3)
+                            options_table = gr.Markdown(first_table)
+
+                        run_btn = gr.Button("Run", variant="primary", size="lg")
+
+                    with gr.Column(scale=4):
+                        out_audio = gr.Audio(label="Output audio", type="filepath", interactive=False)
+                        out_text = gr.Markdown(label="Text output")
+                        out_files = gr.File(label="All output files (wav, stems, srt, midi, ...)", file_count="multiple",
+                                            interactive=False)
+                        out_status = gr.Markdown()
+                        with gr.Accordion("Details (JSON)", open=False):
+                            out_json = gr.JSON()
+                        with gr.Accordion("Server log", open=False):
+                            out_log = gr.Textbox(lines=12, max_lines=30, show_label=False)
+
+                task.change(on_task, task, [family, text, text_loader, audio_group, voice_group, instruct,
+                                            music_group, speaking_rate])
+                family.change(on_family, family, [pkg, info, voice_id, extra_json, session_json, options_table])
+                defaults_btn.click(lambda f: param_defaults(f)[:2] if f else ("{}", "{}"), family,
+                                   [extra_json, session_json])
+                pkg.change(on_package, [family, pkg], info)
+                size_btn.click(check_size, pkg, info)
+                sample_audio_btn.click(lambda: sample_file("sample_16k.wav"), None, audio_mic)
+                sample_voice_btn.click(lambda: sample_file("b.wav"), None, voice_ref)
+                load_btn.click(load_text, [text_files, text_url, text, load_mode], text)
+                run_btn.click(
+                    run_task,
+                    [task, family, pkg, device, gpu_seconds, text, split_mode, audio_mic, audio_files, audio_url,
+                     voice_id, voice_ref, ref_text, instruct, lyrics, language, seed, duration, speaking_rate,
+                     extra_json, load_json, session_json, raw_json],
+                    [out_audio, out_files, out_text, out_json, out_status, out_log, pkg],
+                    concurrency_limit=2,
+                )
+
+            # ------------------------------------------------------------- models tab
+            with gr.Tab("Models"):
+                gr.Markdown("Browse every model family in the audio.cpp catalog and pre-download packages so the "
+                            "first run starts faster. Downloads are lost when the Space restarts unless it has "
+                            "persistent storage.")
+                with gr.Row():
+                    m_pkg = gr.Dropdown(all_package_choices(), label="Package", scale=4, filterable=True)
+                    m_size = gr.Button("Size", scale=1)
+                    m_dl = gr.Button("Download", variant="primary", scale=1)
+                    m_rm = gr.Button("Delete", variant="stop", scale=1)
+                m_status = gr.Markdown(disk_report() if CATALOG else "")
+                m_table = gr.Dataframe(catalog_rows(), headers=["family", "name", "tasks", "packages", "downloaded",
+                                                                "description"],
+                                       interactive=False, wrap=True)
+                m_size.click(check_size, m_pkg, m_status)
+                m_dl.click(models_download, m_pkg, [m_status, m_table])
+                m_rm.click(models_delete, m_pkg, [m_status, m_table])
+
+            # ------------------------------------------------------------- python tab
+            with gr.Tab("Python"):
+                if ALLOW_PYTHON:
+                    gr.Markdown("Run Python with the `audiocpp_runtime` helpers (`ac.run`, `ac.install`, "
+                                "`ac.list_packages`, `ac.execute`, `ac.to_wav`). Code runs in a separate process with "
+                                "a time limit and no Space secrets. Anything written to the working directory is "
+                                "returned below.")
+                    py_code = gr.Code(PY_EXAMPLE, language="python", lines=18, label="Code")
+                    with gr.Row():
+                        py_gpu = gr.Checkbox(value=True, label="Run inside a ZeroGPU call")
+                        py_secs = gr.Slider(20, MAX_GPU_SECONDS, value=120, step=10, label="Time limit (seconds)")
+                        py_btn = gr.Button("Run Python", variant="primary")
+                    py_out = gr.Textbox(lines=16, max_lines=40, label="Output")
+                    py_audio = gr.Audio(label="First audio file", type="filepath", interactive=False)
+                    py_files = gr.File(label="Files", file_count="multiple", interactive=False)
+                    py_btn.click(run_python, [py_code, py_gpu, py_secs], [py_out, py_files, py_audio],
+                                 concurrency_limit=2)
+                else:
+                    gr.Markdown("The Python runner is disabled on this Space.")
+
+            # ------------------------------------------------------------- about tab
+            with gr.Tab("About"):
+                gr.Markdown(f"""
+**How it works.** The Space downloads the official audio.cpp `{ac.AUDIOCPP_VERSION}` CUDA 12.8 Linux build
+(the same prebuilt used by the project's Colab notebook) plus NVIDIA's CUDA 12 runtime wheels. Each run
+starts `audiocpp_server` with a one-model config inside a ZeroGPU call, sends the request to `/v1/tasks/run`,
+then shuts it down. If ZeroGPU is unavailable (no quota, or the Space is on CPU hardware), the same binary
+runs on CPU through a small `libcuda` stub, so everything still works, only slower.
+
+**Tips**
+- The GPU budget includes model loading. Small models (Supertonic, Kokoro, PocketTTS, Qwen3-ASR 0.6B,
+  HTDemucs) finish in a few seconds; 4B to 8B models may need 120 s or more.
+- Use **Batch mode** to turn each paragraph or line into its own clip; TTS clips are also joined into
+  `combined.wav`. ASR, separation and the other audio tasks accept many files at once.
+- Some families expose extra controls; open **Advanced model options** for the full option table.
+- Packages marked `[safetensors]` may need a conversion step that this Space does not run. Prefer GGUF.
+
+**Credits.** audio.cpp by [0xShug0](https://github.com/0xShug0) and contributors (Apache-2.0). Model weights
+keep their original licenses; see the project's
+[model license list](https://github.com/0xShug0/audio.cpp/blob/main/docs/model_licenses.md).
+Per-model option defaults come from the project's WebUI config.
+""")
+    return demo
+
+
+demo = build_ui()
+
+if __name__ == "__main__":
+    demo.queue(default_concurrency_limit=4, max_size=64).launch(
+        theme=gr.themes.Soft(), css=CSS, ssr_mode=False, show_error=True,
+        allowed_paths=[str(ac.WORK_ROOT), str(SAMPLE_DIR)],
+    )