Spaces:
Running on Zero
Running on Zero
| """SemBridge — unified speech & singing generation (TTS · SVS · mixed) on ZeroGPU.""" | |
| import json | |
| import os | |
| import tempfile | |
| import time | |
| from pathlib import Path | |
| os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True") | |
| import spaces # MUST come before any CUDA-touching import | |
| from sembridge import SemBridge | |
| from sembridge.infer import _write_pcm16 | |
| import gradio as gr | |
| MODEL_ID = "ASLP-lab/SemBridge" | |
| EXAMPLES_DIR = Path(__file__).parent / "examples" | |
| MAX_PATCHES = 500 # 40 s hard cap of the released checkpoint (2 × 25 Hz frames per patch) | |
| print("Loading SemBridge checkpoint (≈4.3 GB)…", flush=True) | |
| _t0 = time.perf_counter() | |
| model = SemBridge.from_pretrained(MODEL_ID, device="cuda", dtype="bfloat16") | |
| print(f"SemBridge loaded in {time.perf_counter() - _t0:.1f}s", flush=True) | |
| # --------------------------------------------------------------- catalog | |
| # (demo id, request json, task) — packaged straight from the official repo. | |
| CATALOG = [ | |
| ("tts_zh_r001", "tts/zh_r001.json", "tts"), | |
| ("tts_zh_r002", "tts/zh_r002.json", "tts"), | |
| ("tts_en_r001", "tts/en_r001.json", "tts"), | |
| ("tts_en_r002", "tts/en_r002.json", "tts"), | |
| ("svs_zh_r001", "svs/zh_r001.json", "svs"), | |
| ("svs_zh_r002", "svs/zh_r002.json", "svs"), | |
| ("svs_en_r001", "svs/en_r001.json", "svs"), | |
| ("svs_en_r002", "svs/en_r002.json", "svs"), | |
| ("unified_zh_novel_wukong", "unified/zh_novel_wukong.json", "unified"), | |
| ("unified_zh_robust_caihong", "unified/zh_robust_caihong.json", "unified"), | |
| ] | |
| def _read_json(relative: str) -> dict: | |
| record = json.loads((EXAMPLES_DIR / relative).read_text(encoding="utf-8")) | |
| if not isinstance(record, dict): | |
| raise ValueError("request JSON must contain one object") | |
| return record | |
| def _request_prompt_paths(relative: str) -> dict: | |
| """Resolve prompt_audio / prompt_text / generation from a packaged request. | |
| `prompt_audio` in the packaged JSON is relative to the *request file's* | |
| directory (mirroring sembridge.requests.load_request), i.e. "../audio/x.wav" | |
| means examples/audio/x.wav. | |
| """ | |
| record = _read_json(relative) | |
| prompt_audio = record.get("prompt_audio") | |
| if prompt_audio is None: | |
| raise ValueError(f"{relative} has no prompt_audio") | |
| request_dir = (EXAMPLES_DIR / relative).parent | |
| prompt_path = Path(prompt_audio) | |
| if not prompt_path.is_absolute(): | |
| prompt_path = request_dir / prompt_path | |
| if not prompt_path.is_file(): | |
| raise FileNotFoundError(prompt_path) | |
| out = { | |
| "prompt_audio": str(prompt_path), | |
| "prompt_text": record.get("prompt_text", ""), | |
| "generation": record.get("generation") or {}, | |
| } | |
| return out | |
| VOICE_CHOICES = [ # (label, request path) for the reference-voice dropdown | |
| ("ZH speech 1 — Seed-TTS", "tts/zh_r001.json"), | |
| ("ZH speech 2 — Seed-TTS", "tts/zh_r002.json"), | |
| ("EN speech 1 — Common Voice", "tts/en_r001.json"), | |
| ("EN speech 2 — Common Voice", "tts/en_r002.json"), | |
| ("ZH singer 1 — GTSinger", "svs/zh_r001.json"), | |
| ("ZH singer 2 — GTSinger", "svs/zh_r002.json"), | |
| ("EN singer 1 — GTSinger", "svs/en_r001.json"), | |
| ("EN singer 2 — Opencpop", "svs/en_r002.json"), | |
| ("ZH host (spoken ref) — 悟空", "unified/zh_novel_wukong.json"), | |
| ("ZH ballad host (sung ref) — 彩虹", "unified/zh_robust_caihong.json"), | |
| ] | |
| VOICE_BY_LABEL = dict(VOICE_CHOICES) | |
| DEFAULT_TTS_VOICE = VOICE_CHOICES[0][0] | |
| DEFAULT_SVS_VOICE = "ZH singer 1 — GTSinger" | |
| # ------------------------------------------------------------ helpers | |
| def _clean_number(value, fallback): | |
| """Gradio Number inputs give None for blank; fall back to the model default.""" | |
| if value is None or (isinstance(value, float) and value == 0) or value == "": | |
| return fallback | |
| return value | |
| def _generation_kwargs(request_generation: dict, flow_steps, cfg_value, temperature, seed): | |
| """Request `generation` block first, explicit UI values on top; None = no override.""" | |
| legacy = { | |
| k: v for k, v in (request_generation or {}).items() | |
| if k in {"max_new_patches", "min_new_patches", "flow_steps", "cfg_value", | |
| "temperature", "seed"} and v is not None | |
| } | |
| ui = { | |
| "flow_steps": flow_steps, "cfg_value": cfg_value, | |
| "temperature": temperature, "seed": seed, | |
| } | |
| legacy.update({k: v for k, v in ui.items() if v is not None}) | |
| return legacy | |
| def _parse_score(score_text: str) -> list[dict]: | |
| """Parse 'lyric|pitch|seconds' lines into sembridge note dicts.""" | |
| notes = [] | |
| for raw in (score_text or "").splitlines(): | |
| line = raw.strip() | |
| if not line: | |
| continue | |
| parts = [p.strip() for p in line.split("|")] | |
| if len(parts) != 3: | |
| raise gr.Error(f"Each note line is 'lyric|pitch|seconds' — got: {line!r}") | |
| lyric, pitch, duration = parts | |
| try: | |
| pitch_i = int(pitch) | |
| duration_f = float(duration) | |
| except ValueError: | |
| raise gr.Error(f"Pitch must be an integer, duration a float — got: {line!r}") | |
| if not 0 <= pitch_i <= 128: | |
| raise gr.Error(f"MIDI pitch {pitch_i} is outside [0, 128].") | |
| if duration_f <= 0: | |
| raise gr.Error(f"Duration must be positive — got: {line!r}") | |
| notes.append({"text": lyric, "pitch": pitch_i, "duration": duration_f}) | |
| if not notes: | |
| raise gr.Error("Enter at least one note: one 'lyric|pitch|seconds' per line.") | |
| return notes | |
| def _score_to_ui(notes: list[dict]) -> str: | |
| """Render packaged SVS notes as editable UI score text.""" | |
| return "\n".join( | |
| f"{n.get('text', '')}|{n.get('pitch', 0)}|{n.get('duration', 0)}" for n in notes | |
| ) | |
| def _estimate_duration(task: str, *args, **kwargs) -> int: | |
| """Callable @spaces.GPU duration: reserve what this request needs, not a blanket worst case. | |
| Receives the same inputs as the decorated function (positional layout of | |
| the handler signature when ZeroGPU passes args positionally, keywords as | |
| a fallback — both are accepted). Measured on ZeroGPU (RTX 6000 Ada): | |
| ~6.5 ms per patch×flow-step plus a small cold-start allowance and codec | |
| decode time. | |
| """ | |
| text = kwargs.get("text") or kwargs.get("score_text") or (args[0] if args else "") or "" | |
| flow_steps = kwargs.get("flow_steps") | |
| if flow_steps is None and len(args) >= 5: | |
| flow_steps = args[4] | |
| steps = int(flow_steps) if flow_steps else 10 | |
| steps = max(steps, 2) | |
| if task == "svs" and isinstance(text, str) and "|" in text: | |
| # Score text: sum note durations directly. | |
| total_seconds = 0.0 | |
| for line in text.splitlines(): | |
| if line.strip(): | |
| try: | |
| total_seconds += float(line.split("|")[2]) | |
| except (IndexError, ValueError): | |
| total_seconds += 0.3 | |
| seconds = total_seconds | |
| else: | |
| # Speech: characters ≈ syllables ≈ 155 ms each (CJK 1 char/syllable). | |
| seconds = (len(text) if isinstance(text, str) else 0) * 0.155 | |
| patches = min(MAX_PATCHES, max(2, int(seconds * 12.5))) | |
| total = 10.0 + patches * steps * 0.008 + seconds * 0.3 | |
| return min(240, int(total * 1.3)) | |
| def _tts_duration(*args, **kwargs) -> int: | |
| return _estimate_duration("tts", *args, **kwargs) | |
| def _svs_duration(*args, **kwargs) -> int: | |
| return _estimate_duration("svs", *args, **kwargs) | |
| def _save_and_report(waveform, patches, started, status_box: gr.Textbox | None = None): | |
| out = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) | |
| _write_pcm16(Path(out.name), waveform, model.codec.sample_rate) | |
| seconds = waveform.size(-1) / model.codec.sample_rate | |
| elapsed = time.perf_counter() - started | |
| return out.name, ( | |
| f"{seconds:.1f}s audio · {patches} patches · generated in {elapsed:.1f}s" | |
| ) | |
| # ---------------------------------------------------------------- GPU fns | |
| def tts_generate(text: str, voice: str, ref_audio, ref_text: str, | |
| flow_steps=None, cfg_value=None, temperature=None, seed=None, | |
| progress=gr.Progress()): | |
| """Text-to-speech in the reference voice. | |
| Args: | |
| text: Sentence to speak (English or Chinese). | |
| voice: Packaged reference voice preset; empty string means custom upload. | |
| ref_audio: Optional uploaded reference recording (must be given with ref_text). | |
| ref_text: Transcript of the reference recording. | |
| flow_steps: Flow-matching steps; blank = request/model default (10). | |
| cfg_value: Classifier-free guidance; blank = 2.0. | |
| temperature: Sampling temperature; blank = 1.0. | |
| seed: RNG seed; blank = 42. | |
| """ | |
| started = time.perf_counter() | |
| if not (text or "").strip(): | |
| raise gr.Error("Enter some text to speak.") | |
| request_gen = {} | |
| if voice and voice in VOICE_BY_LABEL: | |
| preset = _request_prompt_paths(VOICE_BY_LABEL[voice]) | |
| prompt_audio, prompt_text = preset["prompt_audio"], preset["prompt_text"] | |
| request_gen = preset["generation"] | |
| elif ref_audio: | |
| if not (ref_text or "").strip(): | |
| raise gr.Error("An uploaded voice needs its transcript.") | |
| prompt_audio, prompt_text = ref_audio, ref_text.strip() | |
| else: | |
| raise gr.Error("Pick a packaged voice or upload a reference recording.") | |
| kwargs = _generation_kwargs(request_gen, flow_steps, cfg_value, temperature, seed) | |
| waveform, patches = model.generate( | |
| task="tts", text=text, prompt_audio=prompt_audio, prompt_text=prompt_text, | |
| prompt_task="speech", **kwargs, | |
| ) | |
| return _save_and_report(waveform, patches.size(1), started) | |
| def svs_generate(score_text: str, voice: str, ref_audio, ref_text: str, | |
| flow_steps=None, cfg_value=None, temperature=None, seed=None, | |
| progress=gr.Progress()): | |
| """Note-conditioned singing from a score string. | |
| Args: | |
| score_text: One note per line, "lyric|pitch|duration_seconds". "<SP>" = rest. | |
| voice: Packaged sung reference preset; empty string means custom upload. | |
| ref_audio: Optional uploaded sung reference recording. | |
| ref_text: Score-formatted transcript of the sung reference. | |
| flow_steps: Flow-matching steps; blank = 20 for singing. | |
| cfg_value: Classifier-free guidance; blank = 2.0. | |
| temperature: Sampling temperature; blank = 1.0. | |
| seed: RNG seed; blank = 42. | |
| """ | |
| started = time.perf_counter() | |
| notes = _parse_score(score_text) | |
| request_gen = {} | |
| if voice and voice in VOICE_BY_LABEL: | |
| preset = _request_prompt_paths(VOICE_BY_LABEL[voice]) | |
| prompt_audio, prompt_text = preset["prompt_audio"], preset["prompt_text"] | |
| request_gen = preset["generation"] | |
| elif ref_audio: | |
| if not (ref_text or "").strip(): | |
| raise gr.Error("An uploaded sung reference needs its score-formatted transcript.") | |
| prompt_audio, prompt_text = ref_audio, ref_text.strip() | |
| else: | |
| raise gr.Error("Pick a packaged voice or upload a sung reference.") | |
| kwargs = _generation_kwargs(request_gen, flow_steps, cfg_value, temperature, seed) | |
| waveform, patches = model.generate( | |
| task="svs", notes=notes, prompt_audio=prompt_audio, prompt_text=prompt_text, | |
| prompt_task="singing", **kwargs, | |
| ) | |
| return _save_and_report(waveform, patches.size(1), started) | |
| def _unified_duration(case_id: str, *args, **kwargs) -> int: | |
| """Duration for the mixed speech+song cases: longest packaged sequence is | |
| 15 s of audio at flow 20 ≈ 26 s measured; reserve ~1.4× for cold starts.""" | |
| return 45 | |
| def unified_generate(case_id: str, flow_steps=None, cfg_value=None, | |
| temperature=None, seed=None, progress=gr.Progress()): | |
| """Run one packaged speech→singing→speech sequence end to end. | |
| Args: | |
| case_id: Packaged case id: unified_zh_novel_wukong or unified_zh_robust_caihong. | |
| flow_steps: Optional flow-matching steps override. | |
| cfg_value: Optional guidance override. | |
| temperature: Optional temperature override. | |
| seed: Optional seed override. | |
| """ | |
| started = time.perf_counter() | |
| if case_id not in {"unified_zh_novel_wukong", "unified_zh_robust_caihong"}: | |
| raise gr.Error("Case must be unified_zh_novel_wukong or unified_zh_robust_caihong.") | |
| record = _read_json(f"unified/{case_id.split('_', 1)[1]}.json") | |
| preset = _request_prompt_paths(f"unified/{case_id.split('_', 1)[1]}.json") | |
| kwargs = _generation_kwargs( | |
| preset["generation"], flow_steps, cfg_value, temperature, seed | |
| ) | |
| waveform, patches = model.generate( | |
| task="unified", segments=record["segments"], | |
| prompt_audio=preset["prompt_audio"], prompt_text=preset["prompt_text"], | |
| prompt_task=record.get("prompt_task", "speech"), **kwargs, | |
| ) | |
| return _save_and_report(waveform, patches.size(1), started) | |
| # --------------------------------------------------------------------- UI | |
| CSS = """ | |
| #col-container { max-width: 1100px; margin: 0 auto; } | |
| .dark .gradio-container { color: var(--body-text-color); } | |
| """ | |
| CUSTOM = "⬆️ Custom upload" | |
| def _voice_row(request: str) -> str: | |
| return request | |
| def on_voice_change(voice: str): | |
| """Preview the packaged recording; prefill its transcript and generation settings.""" | |
| if voice not in VOICE_BY_LABEL: | |
| return [gr.update(value=None, visible=True, interactive=True), | |
| gr.update(value=""), None, None, None, None] | |
| preset = _request_prompt_paths(VOICE_BY_LABEL[voice]) | |
| gen = preset["generation"] | |
| return [ | |
| gr.update(value=preset["prompt_audio"], visible=True, interactive=False), | |
| gr.update(value=preset["prompt_text"]), | |
| gen.get("flow_steps"), | |
| gen.get("cfg_value"), | |
| gen.get("temperature"), | |
| gen.get("seed"), | |
| ] | |
| def tts_examples() -> list[list]: | |
| rows = [] | |
| for label, request in VOICE_CHOICES[:4]: | |
| record = _read_json(request) | |
| rows.append([record.get("text", ""), label]) | |
| return rows | |
| def svs_examples() -> list[list]: | |
| rows = [] | |
| for label, request in VOICE_CHOICES[4:8]: | |
| record = _read_json(request) | |
| rows.append([_score_to_ui(record.get("notes", [])), label]) | |
| return rows | |
| with gr.Blocks(title="SemBridge — Speech & Singing") as demo: | |
| gr.HTML( | |
| "<div style='text-align:center'>" | |
| "<h1>🎤 SemBridge</h1>" | |
| "<p style='margin:4px 0'><b>Semantic Token Anchoring for Continuous-Latent " | |
| "Autoregressive Speech Generation</b> — speech, singing, one voice.</p>" | |
| "<p style='margin:4px 0'>Model: <a href='https://huggingface.co/ASLP-lab/SemBridge'>" | |
| "ASLP-lab/SemBridge</a> · <a href='https://github.com/ASLP-lab/SemBridge'>Code</a> · " | |
| "<a href='https://arxiv.org/abs/2608.07462'>Paper</a> · " | |
| "<a href='https://tiamojames.github.io/SemBridge_Demo/'>Demo page</a></p>" | |
| "</div>" | |
| ) | |
| gr.Markdown( | |
| "🔊 The model generates from a **reference voice** — choose one of the ten " | |
| "packaged speakers/singers (exact references from the authors' demo, press ▶ on " | |
| "the preview to listen) or upload ~4–10 s of clean audio with its transcript." | |
| ) | |
| with gr.Column(elem_id="col-container"): | |
| with gr.Tabs(): | |
| with gr.Tab("🗣️ Text to Speech"): | |
| tts_text = gr.Textbox( | |
| label="Text to speak (English or Chinese)", | |
| placeholder="Reading between the lines requires understanding.", | |
| lines=2, autofocus=True, | |
| ) | |
| tts_run = gr.Button("Generate speech", variant="primary") | |
| tts_out = gr.Audio(label="Generated speech", type="filepath") | |
| tts_status = gr.Textbox(label="Info", show_label=False, interactive=False) | |
| with gr.Accordion("Reference voice (required)", open=True): | |
| tts_voice = gr.Dropdown(list(VOICE_BY_LABEL) + [CUSTOM], | |
| value=DEFAULT_TTS_VOICE, label="Voice") | |
| tts_ref_audio = gr.Audio( | |
| label="Reference recording (preview; upload when custom)", | |
| type="filepath", sources=["upload"], | |
| ) | |
| tts_ref_text = gr.Textbox(label="Reference transcript", lines=2, | |
| placeholder="Upload + transcript when set to Custom upload") | |
| with gr.Accordion("Generation settings", open=False): | |
| with gr.Row(): | |
| tts_flow = gr.Number(label="Flow steps (default 10)", precision=0) | |
| tts_cfg = gr.Number(label="CFG (default 2.0)") | |
| tts_temp = gr.Number(label="Temperature (default 1.0)") | |
| tts_seed = gr.Number(label="Seed (default 42)", precision=0) | |
| gr.Examples( | |
| examples=tts_examples(), | |
| inputs=[tts_text, tts_voice], | |
| outputs=[tts_out, tts_status], | |
| fn=tts_generate, cache_examples=True, cache_mode="lazy", | |
| label="Official demo sentences (Seed-TTS references)", | |
| ) | |
| with gr.Tab("🎵 Singing (SVS)"): | |
| svs_score = gr.Textbox( | |
| label="Score — one note per line: lyric | MIDI pitch | duration (s). " | |
| "'<SP>' = rest.", | |
| placeholder="like|58|0.30\nthis|61|0.36\n<SP>|50|0.25", | |
| lines=8, autofocus=True, | |
| ) | |
| svs_run = gr.Button("Generate singing", variant="primary") | |
| svs_out = gr.Audio(label="Generated singing", type="filepath") | |
| svs_status = gr.Textbox(label="Info", show_label=False, interactive=False) | |
| with gr.Accordion("Sung reference voice (required)", open=True): | |
| svs_voice = gr.Dropdown(list(VOICE_BY_LABEL) + [CUSTOM], | |
| value=DEFAULT_SVS_VOICE, label="Voice") | |
| svs_ref_audio = gr.Audio( | |
| label="Sung reference (preview; upload when custom)", | |
| type="filepath", sources=["upload"], | |
| ) | |
| svs_ref_text = gr.Textbox(label="Score-formatted reference transcript", | |
| lines=4) | |
| with gr.Accordion("Generation settings", open=False): | |
| with gr.Row(): | |
| svs_flow = gr.Number(label="Flow steps (default 20)", precision=0) | |
| svs_cfg = gr.Number(label="CFG (default 2.0)") | |
| svs_temp = gr.Number(label="Temperature (default 1.0)") | |
| svs_seed = gr.Number(label="Seed (default 42)", precision=0) | |
| gr.Examples( | |
| examples=svs_examples(), | |
| inputs=[svs_score, svs_voice], | |
| outputs=[svs_out, svs_status], | |
| fn=svs_generate, cache_examples=True, cache_mode="lazy", | |
| label="Official demo scores (GTSinger / Opencpop)", | |
| ) | |
| with gr.Tab("🎭 Mixed speech + song"): | |
| gr.Markdown( | |
| "The unified model speaks, sings, then speaks again — in one " | |
| "sequence, one voice. These are the two official packaged cases." | |
| ) | |
| case_id = gr.Radio( | |
| ["unified_zh_novel_wukong", "unified_zh_robust_caihong"], | |
| value="unified_zh_novel_wukong", | |
| label="Case (speech → singing → speech)", | |
| ) | |
| uni_run = gr.Button("Generate mixed sequence", variant="primary") | |
| uni_out = gr.Audio(label="Generated mixed audio", type="filepath") | |
| uni_status = gr.Textbox(label="Info", show_label=False, interactive=False) | |
| with gr.Accordion("Generation overrides (optional)", open=False): | |
| with gr.Row(): | |
| uni_flow = gr.Number(label="Flow steps", precision=0) | |
| uni_cfg = gr.Number(label="CFG") | |
| uni_temp = gr.Number(label="Temperature") | |
| uni_seed = gr.Number(label="Seed", precision=0) | |
| gr.Markdown( | |
| "📝 Output is mono 44.1 kHz WAV, capped at 500 patches (~40 s). Singing " | |
| "durations are quantized to 25 ms frames; the score format mirrors the " | |
| "authors' JSON (lyric / MIDI pitch / seconds). Packaged references are " | |
| "Seed-TTS / GTSinger / Opencpop samples from the official demo and keep " | |
| "their original terms." | |
| ) | |
| tts_voice.change(on_voice_change, tts_voice, | |
| [tts_ref_audio, tts_ref_text, tts_flow, tts_cfg, | |
| tts_temp, tts_seed]) | |
| svs_voice.change(on_voice_change, svs_voice, | |
| [svs_ref_audio, svs_ref_text, svs_flow, svs_cfg, | |
| svs_temp, svs_seed]) | |
| tts_run.click(tts_generate, | |
| inputs=[tts_text, tts_voice, tts_ref_audio, tts_ref_text, | |
| tts_flow, tts_cfg, tts_temp, tts_seed], | |
| outputs=[tts_out, tts_status], api_name="tts") | |
| svs_run.click(svs_generate, | |
| inputs=[svs_score, svs_voice, svs_ref_audio, svs_ref_text, | |
| svs_flow, svs_cfg, svs_temp, svs_seed], | |
| outputs=[svs_out, svs_status], api_name="svs") | |
| uni_run.click(unified_generate, | |
| inputs=[case_id, uni_flow, uni_cfg, uni_temp, uni_seed], | |
| outputs=[uni_out, uni_status], api_name="unified") | |
| demo.launch(mcp_server=True, theme=gr.themes.Citrus(), css=CSS) |