diff --git a/.gitignore b/.gitignore index 1a5cae0..b3c7b95 100644 --- a/.gitignore +++ b/.gitignore @@ -37,3 +37,6 @@ media-files/video/*.mp4 # Small public-domain demo evidence clip for the auk voice-editing demo !auk/*.wav + +# VibeVoice 4-speaker spike test clip - committed for remote review +!vibevoice-spike/*.mp3 diff --git a/README.md b/README.md index db266ab..c1b61ac 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ Audio processing demos: WhisperX, Demucs, and TTS technology proofs-of-concept. - `tts/` — text-to-speech service demos (voice cloning via MOSS-TTS) - `auk/` — voice *editing* demos (emotion/pitch/speed/content editing on existing recordings, as opposed to `tts/`'s clone-and-generate approach) +- `webui/` — interactive pipeline studio chaining demucs -> AuK -> MOSS-TTS with previews Each subdirectory has its own README with setup and run instructions. diff --git a/mastering/master.py b/mastering/master.py new file mode 100644 index 0000000..8707786 --- /dev/null +++ b/mastering/master.py @@ -0,0 +1,82 @@ +""" +Mastering chain for generated TTS/voice clips: fixes the "tinny" spectral +imbalance common to neural vocoder output (excess 3-6kHz presence energy, +starved low end relative to a naturally-recorded voice - see the project's +spectral comparison: raw recordings skew toward 200-1000Hz, generated clips +dump ~35% of their energy into 3-6kHz) and adds the density/space that +reads as "produced" rather than dry/synthetic. + +Chain order matters: EQ before compression (so the compressor reacts to +the corrected tonal balance, not the raw tinny one), reverb after +compression (so the room tail doesn't get squashed), limiter last. +""" + +import argparse + +import numpy as np +import soundfile as sf +from pedalboard import ( + Compressor, + HighpassFilter, + HighShelfFilter, + Limiter, + LowShelfFilter, + PeakFilter, + Pedalboard, + Reverb, +) + + +def build_chain( + warmth_db: float, + presence_cut_db: float, + air_db: float, + comp_ratio: float, + reverb_wet: float, +) -> Pedalboard: + return Pedalboard([ + HighpassFilter(cutoff_frequency_hz=60), + LowShelfFilter(cutoff_frequency_hz=150, gain_db=warmth_db, q=0.7), + PeakFilter(cutoff_frequency_hz=3800, gain_db=presence_cut_db, q=1.0), + HighShelfFilter(cutoff_frequency_hz=6500, gain_db=air_db, q=0.7), + Compressor(threshold_db=-20, ratio=comp_ratio, attack_ms=12, release_ms=120), + Reverb(room_size=0.08, damping=0.85, wet_level=reverb_wet, dry_level=1.0 - reverb_wet), + Limiter(threshold_db=-1.0, release_ms=100), + ]) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Apply a studio mastering chain to a voice clip") + parser.add_argument("--input", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--warmth-db", type=float, default=5.5, help="Low-shelf boost @150Hz") + parser.add_argument("--presence-cut-db", type=float, default=-5.0, help="Peak cut @3.8kHz (the tinny band)") + parser.add_argument("--air-db", type=float, default=-4.5, help="High-shelf @6.5kHz, for softness") + parser.add_argument("--comp-ratio", type=float, default=2.5) + parser.add_argument("--reverb-wet", type=float, default=0.025) + args = parser.parse_args() + + audio, sr = sf.read(args.input, always_2d=True) + audio = audio.T.astype(np.float32) # pedalboard wants (channels, samples) + + board = build_chain( + args.warmth_db, args.presence_cut_db, args.air_db, args.comp_ratio, args.reverb_wet + ) + processed = board(audio, sr) + + # pedalboard's Limiter is attack/release-based, not a true brick-wall + # limiter - fast transients can still punch through its threshold (seen + # in practice: in-memory peaks hit 0dBFS despite threshold_db=-1.0, + # producing audible clipping distortion on playback). Hard safety + # ceiling here guarantees the written file never clips regardless. + target_peak = 10 ** (-1.0 / 20) + peak = np.max(np.abs(processed)) + if peak > target_peak: + processed = processed * (target_peak / peak) + + sf.write(args.output, processed.T, sr) + print(f"[INFO] Saved {args.output}") + + +if __name__ == "__main__": + main() diff --git a/sfx-stable-audio-3/generate.py b/sfx-stable-audio-3/generate.py new file mode 100644 index 0000000..c133d1f --- /dev/null +++ b/sfx-stable-audio-3/generate.py @@ -0,0 +1,32 @@ +""" +Stable Audio 3 Small-SFX CLI: generate a sound effect/ambience clip from a +text description. Runs on CPU by design - this checkpoint is the +"lightweight, no GPU required" model in Stable Audio 3's lineup, so it +never contends with ComfyUI/other models for VRAM (unlike generate_sfx.py's +MOSS-SoundEffect, which needs the 3090 and ComfyUI's memory freed first). +""" + +import argparse + +import torchaudio +from stable_audio_3 import StableAudioModel + + +def main() -> None: + parser = argparse.ArgumentParser(description="Stable Audio 3 Small-SFX text-to-sound-effect") + parser.add_argument("--text", required=True, help="Description of the sound effect/ambience") + parser.add_argument("--output", required=True, help="Output wav path") + parser.add_argument("--duration", type=float, default=10.0, help="Clip duration in seconds (max 120)") + parser.add_argument("--seed", type=int, default=-1, help="Random seed, or -1 for random") + args = parser.parse_args() + + model = StableAudioModel.from_pretrained("small-sfx", device="cpu") + sample_rate = model.model.sample_rate + + audio = model.generate(prompt=args.text, duration=args.duration, seed=args.seed) + torchaudio.save(args.output, audio[0].cpu(), sample_rate) + print(f"[INFO] Saved {args.output}") + + +if __name__ == "__main__": + main() diff --git a/tts/generate.py b/tts/generate.py new file mode 100644 index 0000000..b261b00 --- /dev/null +++ b/tts/generate.py @@ -0,0 +1,109 @@ +""" +Generic MOSS-TTS zero-shot cloning CLI: clone the voice in --reference and +say --text in it. Used directly, and as the final stage of webui/app.py. +""" + +import argparse +import importlib.util + +import soundfile as sf +import torch +from transformers import AutoModel, AutoProcessor + +PRETRAINED = "OpenMOSS-Team/MOSS-TTS-Local-Transformer-v1.5" + + +def resolve_attn_implementation(device: str, dtype: torch.dtype) -> str: + if ( + device == "cuda" + and importlib.util.find_spec("flash_attn") is not None + and dtype in {torch.float16, torch.bfloat16} + ): + major, _ = torch.cuda.get_device_capability() + if major >= 8: + return "flash_attention_2" + return "sdpa" if device == "cuda" else "eager" + + +def main() -> None: + parser = argparse.ArgumentParser(description="MOSS-TTS zero-shot voice cloning") + parser.add_argument("--text", required=True, help="Text to synthesize") + parser.add_argument("--reference", required=True, help="Reference audio path") + parser.add_argument("--output", required=True, help="Output wav path") + parser.add_argument("--language", default="English", help="Language tag") + parser.add_argument( + "--instruction", default=None, + help='Style/delivery instruction (e.g. "Change the emotion to excited."). ' + "Undocumented MOSS-TTS field for this checkpoint - confirmed to work for " + "emotion in testing, passed through as-is.", + ) + parser.add_argument( + "--seed", type=int, default=None, + help="Random seed for reproducibility. Unset = nondeterministic.", + ) + parser.add_argument("--audio_temperature", type=float, default=1.7) + parser.add_argument("--audio_top_p", type=float, default=0.8) + parser.add_argument("--audio_top_k", type=int, default=25) + parser.add_argument("--audio_repetition_penalty", type=float, default=1.0) + parser.add_argument("--max_new_tokens", type=int, default=4096) + args = parser.parse_args() + + device = "cuda" if torch.cuda.is_available() else "cpu" + dtype = torch.bfloat16 if device == "cuda" else torch.float32 + + if args.seed is not None: + torch.manual_seed(args.seed) + if device == "cuda": + torch.cuda.manual_seed_all(args.seed) + + torch.backends.cuda.enable_cudnn_sdp(False) + torch.backends.cuda.enable_flash_sdp(True) + torch.backends.cuda.enable_mem_efficient_sdp(True) + torch.backends.cuda.enable_math_sdp(True) + + attn_implementation = resolve_attn_implementation(device, dtype) + print(f"[INFO] attn_implementation={attn_implementation}") + + processor = AutoProcessor.from_pretrained(PRETRAINED, trust_remote_code=True) + processor.audio_tokenizer = processor.audio_tokenizer.to(device) + + model = AutoModel.from_pretrained( + PRETRAINED, + trust_remote_code=True, + attn_implementation=attn_implementation, + dtype=dtype, + ).to(device) + model.eval() + + conversation = [ + processor.build_user_message( + text=args.text, + language=args.language, + reference=[args.reference], + instruction=args.instruction, + ) + ] + + with torch.no_grad(): + batch = processor([conversation], mode="generation") + outputs = model.generate( + input_ids=batch["input_ids"].to(device), + attention_mask=batch["attention_mask"].to(device), + max_new_tokens=args.max_new_tokens, + do_sample=True, + audio_temperature=args.audio_temperature, + audio_top_p=args.audio_top_p, + audio_top_k=args.audio_top_k, + audio_repetition_penalty=args.audio_repetition_penalty, + ) + message = processor.decode(outputs)[0] + audio = message.audio_codes_list[0] + data = audio.detach().cpu().to(torch.float32).numpy() + if data.ndim == 2: + data = data.T + sf.write(args.output, data, processor.model_config.sampling_rate) + print(f"[INFO] Saved {args.output}") + + +if __name__ == "__main__": + main() diff --git a/tts/generate_sfx.py b/tts/generate_sfx.py new file mode 100644 index 0000000..9a3a387 --- /dev/null +++ b/tts/generate_sfx.py @@ -0,0 +1,99 @@ +""" +MOSS-SoundEffect CLI: generate an ambient/sound-effect clip from a text +description. Sibling of generate.py (MOSS-TTS voice cloning) - same +backend setup and model-loading pattern, different checkpoint and a +text-only build_user_message call (no reference audio, no cloning). +""" + +import argparse +import importlib.util +import os + +# This checkpoint's weights alone use ~22.3GB of this box's ~23.56GB usable +# VRAM, leaving generate()'s own small allocations (a vocab-sized mask +# tensor, etc.) no room without this - confirmed via OOM otherwise. Must be +# set before torch's CUDA allocator initializes, so it's set here rather +# than relying on the caller's shell environment. +os.environ.setdefault("PYTORCH_ALLOC_CONF", "expandable_segments:True") +os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True") + +import soundfile as sf +import torch +from transformers import AutoModel, AutoProcessor + +PRETRAINED = "OpenMOSS-Team/MOSS-SoundEffect" + + +def resolve_attn_implementation(device: str, dtype: torch.dtype) -> str: + if ( + device == "cuda" + and importlib.util.find_spec("flash_attn") is not None + and dtype in {torch.float16, torch.bfloat16} + ): + major, _ = torch.cuda.get_device_capability() + if major >= 8: + return "flash_attention_2" + return "sdpa" if device == "cuda" else "eager" + + +def main() -> None: + parser = argparse.ArgumentParser(description="MOSS-SoundEffect text-to-sound-effect") + parser.add_argument("--text", required=True, help="Description of the sound effect/ambience") + parser.add_argument("--output", required=True, help="Output wav path") + parser.add_argument( + "--seed", type=int, default=None, + help="Random seed for reproducibility. Unset = nondeterministic.", + ) + parser.add_argument("--max_new_tokens", type=int, default=4096) + args = parser.parse_args() + + device = "cuda" if torch.cuda.is_available() else "cpu" + dtype = torch.bfloat16 if device == "cuda" else torch.float32 + + if args.seed is not None: + torch.manual_seed(args.seed) + if device == "cuda": + torch.cuda.manual_seed_all(args.seed) + + torch.backends.cuda.enable_cudnn_sdp(False) + torch.backends.cuda.enable_flash_sdp(True) + torch.backends.cuda.enable_mem_efficient_sdp(True) + torch.backends.cuda.enable_math_sdp(True) + + attn_implementation = resolve_attn_implementation(device, dtype) + print(f"[INFO] attn_implementation={attn_implementation}") + + processor = AutoProcessor.from_pretrained(PRETRAINED, trust_remote_code=True) + processor.audio_tokenizer = processor.audio_tokenizer.to(device) + + model = AutoModel.from_pretrained( + PRETRAINED, + trust_remote_code=True, + attn_implementation=attn_implementation, + dtype=dtype, + ).to(device) + model.eval() + + conversation = [processor.build_user_message(ambient_sound=args.text)] + + with torch.no_grad(): + batch = processor([conversation], mode="generation") + # This checkpoint's generate() doesn't accept the TTS model's sampling + # kwargs (do_sample/audio_temperature/etc raise TypeError) - matches + # the official README example, which only passes these three. + outputs = model.generate( + input_ids=batch["input_ids"].to(device), + attention_mask=batch["attention_mask"].to(device), + max_new_tokens=args.max_new_tokens, + ) + message = processor.decode(outputs)[0] + audio = message.audio_codes_list[0] + data = audio.detach().cpu().to(torch.float32).numpy() + if data.ndim == 2: + data = data.T + sf.write(args.output, data, processor.model_config.sampling_rate) + print(f"[INFO] Saved {args.output}") + + +if __name__ == "__main__": + main() diff --git a/vibevoice-spike/README.md b/vibevoice-spike/README.md new file mode 100644 index 0000000..d139418 --- /dev/null +++ b/vibevoice-spike/README.md @@ -0,0 +1,52 @@ +# VibeVoice (community fork) 4-person podcast spike + +Standalone experiment, not integrated into the webui or the production +pipeline. Tests whether VibeVoice (https://github.com/vibevoice-community/VibeVoice) +can generate a 4-speaker podcast scene in one pass, cloning our four +established character voices (Kevin, Marge, Terry, Priya) from their +existing approved `reference_audio` in the-service-station repo. + +- `scripts/trailer_promo_4speaker.txt` - the test script: the real + "sausage rolls" trailer scene from `the-service-station/scripts/trailer-promo.md`, + converted from our `[Character|tags]` format to VibeVoices required + `Speaker N: ...` plain-text format (tags dropped - VibeVoice has no + per-line emotion/timing control). +- `trailer_promo_4speaker_vibevoice-1.5b.mp3` - the generated result + (VibeVoice-1.5B, seed 42). 47s of audio generated in 14.5s (RTF 0.31x). + +Setup lives outside this repo (not committed): `.venv/` here is a fresh +isolated venv (`pip install -e /root/projects/VibeVoice`, pinned +`transformers==4.51.3`), pointed at the already-downloaded +`VibeVoice-1.5B` weights under `/root/projects/ComfyUI/models/tts/VibeVoice/`. + +## Findings (2026-09-30) + +Whisper-transcript check against the script: mostly faithful, not +word-perfect. Two small unscripted insertions ("I am here!" before Priyas +line, "For me," before Marges reply) and one substitution ("He will." -> +"You will."). No repetition/collapse like the AuK chain-corruption bug - +just minor ad-libbing, not catastrophic. + +No audible AI disclaimer or obvious background music in this run - both +are documented risks for this model family (see +`the-service-station/research/vibevoice-2026-08-21.md` and +`vibevoice-community-fork-2026-08-21.md`), so treat this one clean run as +encouraging, not conclusive. + +Steves call after listening: pursue 7B (weights already downloaded, same +venv/script, swap `--model_path`) or not worth it given the limitations +the prior research already flagged. + +## 7B result (2026-09-30, same script/seed/voices) + +`trailer_promo_4speaker_vibevoice-7b.mp3` - VibeVoice-7B, same seed (42), +same script, same reference voices. 34.8s of audio generated in 22.2s +(RTF 0.64x - still faster than real-time, just less headroom than 1.5B). + +Whisper-transcript check: cleaner than 1.5B - neither of 1.5Bs unscripted +insertions ("I am here!", "For me,") appeared, and "He will." came through +exactly right this time (1.5B had drifted it to "You will."). One real +deviation: Marges defensive line "I get one fact wrong per episode." +came out as a question, "Can I get one fact wrong per episode?" - changes +the delivery/meaning, not just wording. Still no audible disclaimer and +nothing that reads as injected background music. diff --git a/vibevoice-spike/scripts/trailer_promo_4speaker.txt b/vibevoice-spike/scripts/trailer_promo_4speaker.txt new file mode 100644 index 0000000..7f627c5 --- /dev/null +++ b/vibevoice-spike/scripts/trailer_promo_4speaker.txt @@ -0,0 +1,9 @@ +Speaker 1: The Service Station. Three facts, one car park, and absolutely no reason to be in a hurry. +Speaker 2: Except for me, obviously. I'm always in a hurry. +Speaker 3: I had a Sinclair Spectrum in 1984, and I will find a way to bring it up. +Speaker 1: He will. +Speaker 4: I still don't understand why the sausage rolls are relevant, but here we are. +Speaker 2: They're always relevant, Priya. +Speaker 1: Every week: one underdog, one argument, and one fact Marge gets wrong. +Speaker 2: I get one fact wrong per episode. Tops. +Speaker 3: It's usually about football. diff --git a/vibevoice-spike/trailer_promo_4speaker_vibevoice-1.5b.mp3 b/vibevoice-spike/trailer_promo_4speaker_vibevoice-1.5b.mp3 new file mode 100644 index 0000000..1a34033 Binary files /dev/null and b/vibevoice-spike/trailer_promo_4speaker_vibevoice-1.5b.mp3 differ diff --git a/vibevoice-spike/trailer_promo_4speaker_vibevoice-7b.mp3 b/vibevoice-spike/trailer_promo_4speaker_vibevoice-7b.mp3 new file mode 100644 index 0000000..34c43a4 Binary files /dev/null and b/vibevoice-spike/trailer_promo_4speaker_vibevoice-7b.mp3 differ diff --git a/webui/README.md b/webui/README.md new file mode 100644 index 0000000..3adf302 --- /dev/null +++ b/webui/README.md @@ -0,0 +1,56 @@ +# webui + +A pipeline studio: strip a voice with **demucs**, edit it with **AuK** (emotion, pitch, speed, +free-form instructions — with preview and undo/reset), then generate new speech in that voice +with **MOSS-TTS**. + +## Why this exists + +Manually chaining demucs → AuK → MOSS-TTS across three separate venvs (via SSH one-off scripts) +worked but made it slow to experiment — especially with AuK's pitch/emotion controls, where you +really want to preview a result and back out if it's wrong (see the pitch-overshoot findings in +`auk/README.md`) rather than committing to a specific edit blind. + +## Architecture + +A single lightweight Gradio app (`app.py`) that **shells out** to each tool's own isolated venv +via subprocess, rather than merging their dependencies into one environment — MOSS-TTS pins +`transformers==5.0.0`, AuK needs `4.57.6`; they can't coexist in one venv. This app's own venv +only needs `gradio` itself. + +Each stage's output becomes the next stage's input by filesystem path; a `history` list backs +the undo/reset buttons in the voice editor. + +## Setup + +```bash +python3 -m venv .venv && . .venv/bin/activate +pip install gradio +``` + +Assumes `demucs/`, `auk/`, and `tts/` (siblings in this repo) already have their own working +venvs and, for `auk/`, downloaded model weights — see their READMEs. + +## Running + +```bash +python3 app.py +``` + +Serves on `0.0.0.0:7860`. The app automatically calls ComfyUI's `/free` endpoint (if its queue +is empty) before each heavy step, to reduce GPU contention on a shared box — see the DNS/routing +and GPU-contention notes earlier in this project's history for why that matters here. + +## Known limitations (inherited from the underlying tools) + +- **AuK's real usable clip length is ~25-30s**, not several minutes — feeding it a long recording + will OOM even with `--cpu_offload` (see `auk/README.md`). This app doesn't chunk long inputs + automatically; keep reference clips short. +- **AuK's pitch-editing instruction-following is unreliable** — even "1 semitone" has measured + out at ~4+ semitones of actual shift in testing. The UI surfaces a warning; preview before + trusting a pitch edit. +- Each stage is a blocking subprocess call (model load + inference each time) — the UI will + appear to hang for anywhere from ~10s to a couple of minutes per click, with no progress bar + yet. +- No queueing/locking across browser tabs or concurrent users — this is a single-GPU box, so two + simultaneous generations will contend for VRAM. diff --git a/webui/app.py b/webui/app.py new file mode 100644 index 0000000..28ce5c2 --- /dev/null +++ b/webui/app.py @@ -0,0 +1,1106 @@ +""" +Pipeline studio: Demucs (strip) -> AuK (edit voice, with preview/undo) -> +MOSS-TTS (generate speech in the edited voice). Plus a Characters tab for +browsing the-service-station's cast and creating new characters/emotion +candidates without SSH + manual YAML editing. + +Orchestrates across the three existing isolated venvs (demucs/, auk/, tts/) +via subprocess, rather than merging their conflicting dependencies +(MOSS-TTS needs transformers 5.0.0, AuK needs 4.57.6) into one environment. + +Run with: webui/.venv/bin/python webui/app.py + +2026-09-15 refresh, after building the-service-station's cast surfaced real +gaps in this UI: +- Speed editing was a silent no-op (AuK needs --gen_seconds or it just + preserves the input's duration - see vendor/AuK/docs/COOKBOOK.md 3.2). +- AuK's Emotion task has a fixed 8-word vocabulary; anything else needs the + Timbre task's free-form template instead. +- A chained AuK edit can leak fragments of its own instruction text into + the spoken content. This went undetected through a full listen-and- + approve cycle on a real character voice because nothing had ever + transcribed the *reference* audio, only generated lines. Every AuK edit + and MOSS generation here now gets transcribed automatically so that's + visible immediately instead of discovered later by ear. +""" + +import glob +import json +import os +import re +import shutil +import subprocess +import tempfile +import time +from datetime import date +from pathlib import Path + +import gradio as gr +import yaml + +REPO_ROOT = Path(__file__).resolve().parent.parent +DEMUCS_DIR = REPO_ROOT / "demucs" +AUK_DIR = REPO_ROOT / "auk" +TTS_DIR = REPO_ROOT / "tts" + +DEMUCS_PYTHON = DEMUCS_DIR / ".venv" / "bin" / "python3" +AUK_INFER = AUK_DIR / ".venv" / "bin" / "auk-infer" +AUK_CKPT = AUK_DIR / "ckpts" / "AuK" / "auk_base.safetensors" +AUK_FLASH_CKPT = AUK_DIR / "ckpts" / "AuK-Flash" / "auk_flash.safetensors" +AUK_QWEN_PATH = AUK_DIR / "ckpts" / "Qwen2.5-Omni-3B" +TTS_PYTHON = TTS_DIR / ".venv" / "bin" / "python3" +TTS_GENERATE = TTS_DIR / "generate.py" +TTS_GENERATE_SFX = TTS_DIR / "generate_sfx.py" + +# Stable Audio 3 Small-SFX - a separate isolated venv/repo, not the tts/ one. +# Runs on CPU by design (the model's own "lightweight, no GPU" tier), so it +# never needs ComfyUI's VRAM freed first, unlike MOSS-SoundEffect below. +SFX_SA3_DIR = REPO_ROOT / "sfx-stable-audio-3" +SFX_SA3_PYTHON = SFX_SA3_DIR / ".venv" / "bin" / "python3" +SFX_SA3_GENERATE = SFX_SA3_DIR / "generate.py" + +SFX_MODEL_SA3 = "Stable Audio 3 (Small-SFX)" +SFX_MODEL_MOSS = "MOSS-SoundEffect" + +# Mastering - own isolated venv (pedalboard), fixes the "tinny" spectral +# imbalance generated TTS output has (see mastering/master.py's docstring +# for the measured before/after). Defaults below match what Steve approved +# after an A/B listen on real character clips. +MASTERING_DIR = REPO_ROOT / "mastering" +MASTERING_PYTHON = MASTERING_DIR / ".venv" / "bin" / "python3" +MASTERING_SCRIPT = MASTERING_DIR / "master.py" +MASTERING_DEFAULTS = { + "warmth_db": 5.5, + "presence_cut_db": -5.0, + "air_db": -4.5, + "comp_ratio": 2.5, + "reverb_wet": 0.025, +} + +MODEL_BASE = "Base (full quality)" +MODEL_FLASH = "Flash (fast draft, 4-step)" + +# The-service-station checkout on this same box - where character cards +# live. Override with the env var if running this UI somewhere else. +SERVICE_STATION_DIR = Path(os.environ.get("SERVICE_STATION_REPO", "/root/projects/the-service-station")) +SS_VOICES_DIR = SERVICE_STATION_DIR / "voices" + +# Saved sound effects just live on disk here - no metadata/db, per Steve's +# "just on disk FOR NOW is fine". +SFX_LIBRARY_DIR = SERVICE_STATION_DIR / "sfx_library" + +WORKDIR = Path(tempfile.mkdtemp(prefix="audio-demo-webui-")) + +# AuK's Emotion task has exactly this vocabulary (vendor/AuK/docs/COOKBOOK.md +# 4.1) - anything else needs the Timbre task's free-form template instead +# (see auk_emotion_instruction below). Kept as the dropdown's options since +# these are the only words guaranteed to hit the dedicated task. +EMOTIONS = ["happy", "angry", "sad", "fearful", "surprised", "disgusted", "calm", "excited"] +PITCH_OPTIONS = ["-3", "-2", "-1", "+1", "+2", "+3"] +SPEED_OPTIONS = ["0.5", "0.75", "1.25", "1.5", "2.0"] +VOLUME_OPTIONS = ["+5", "+10", "+15", "-5", "-10", "-15"] +LANGUAGES = [ + "English", "Spanish", "French", "German", "Portuguese", "Italian", + "Chinese", "Japanese", "Korean", "Russian", +] + +COMFYUI_BASE = "http://localhost:8188" + + +def auk_emotion_instruction(word: str) -> str: + """Same logic as the-service-station/workflows/generate_episode_lines.py's + auk_emotion_instruction() - duplicated rather than shared because this + repo and the-service-station use separate venvs/dependency sets, but + kept in sync deliberately. AuK's Emotion task only understands the 8 + words in EMOTIONS; anything else goes through Timbre's free-form + description instead, which has no fixed vocabulary.""" + if word in EMOTIONS: + return f"Change the emotion to {word}." + return f'Keep the spoken content unchanged and change the timbre to: "the same voice, sounding {word}".' + + +def _comfyui_queue_empty() -> bool: + try: + import urllib.error + import urllib.request + with urllib.request.urlopen(f"{COMFYUI_BASE}/queue", timeout=3) as resp: + data = json.loads(resp.read()) + return not data.get("queue_running") and not data.get("queue_pending") + except Exception: + return True # ComfyUI not reachable -> nothing to worry about + + +def _free_comfyui_memory() -> None: + import urllib.error + import urllib.request + if not _comfyui_queue_empty(): + return + try: + req = urllib.request.Request( + f"{COMFYUI_BASE}/free", + data=json.dumps({"unload_models": True, "free_memory": True}).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + urllib.request.urlopen(req, timeout=5) + except (urllib.error.URLError, OSError): + pass + + +def _run(cmd: list[str], cwd: Path, env: dict | None = None) -> None: + result = subprocess.run( + cmd, cwd=cwd, env=env, capture_output=True, text=True, timeout=1800 + ) + if result.returncode != 0: + tail = "\n".join(result.stderr.strip().splitlines()[-25:]) + raise gr.Error(f"Command failed ({cmd[0]}):\n{tail}") + + +def _ensure_wav(path: str) -> str: + """AuK's loader (librosa/soundfile) can't read compressed containers like + .m4a/.mp3 - only demucs's loader handles those. Normalize once, here, at + the point a reference is set, so everything downstream is always wav.""" + if Path(path).suffix.lower() == ".wav": + return path + out_path = WORKDIR / f"converted_{int(time.time() * 1000)}.wav" + result = subprocess.run( + ["ffmpeg", "-nostdin", "-y", "-i", path, "-ar", "44100", "-ac", "1", str(out_path)], + capture_output=True, text=True, timeout=120, + ) + if result.returncode != 0: + tail = "\n".join(result.stderr.strip().splitlines()[-15:]) + raise gr.Error(f"Could not convert '{path}' to wav:\n{tail}") + return str(out_path) + + +def _duration(path: str) -> float: + r = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", path], + capture_output=True, text=True, + ) + return float(r.stdout.strip()) + + +def _ld_library_path_for_venv(venv_dir: Path) -> str: + matches = list(venv_dir.glob(".venv/lib/python*/site-packages")) + if not matches: + return "" + site_packages = matches[0] + nvidia_dir = site_packages / "nvidia" + if not nvidia_dir.is_dir(): + return "" + lib_dirs = sorted(p for p in nvidia_dir.glob("*/lib") if p.is_dir()) + return ":".join(str(p) for p in lib_dirs) + + +TTS_LD_LIBRARY_PATH = _ld_library_path_for_venv(TTS_DIR) + + +# --------------------------------------------------------------------------- +# Whisper content verification - added 2026-09-15 after a chained AuK edit +# leaked fragments of its own instruction text into a character's spoken +# reference, undetected through a full listen-and-approve round because +# nobody had transcribed the reference itself. Loaded once, lazily, since +# most sessions of this UI won't need it immediately. +# --------------------------------------------------------------------------- +_whisper_model = None + + +def _whisper_transcribe(path: str) -> str: + global _whisper_model + try: + if _whisper_model is None: + from faster_whisper import WhisperModel + # "small" mis-heard casual contractions as formal phrasing on + # short/quiet clips (e.g. "it's gonna" -> "it is going to be") - + # "medium" is noticeably more faithful to actual speech patterns + # at the cost of a few more seconds per transcription on CPU, + # worth it since this check exists specifically to catch content + # problems - a transcript that itself distorts casual speech + # defeats that purpose. + _whisper_model = WhisperModel("medium", device="cpu", compute_type="int8") + segments, _ = _whisper_model.transcribe(path, language="en", beam_size=5) + return " ".join(s.text.strip() for s in segments) + except Exception as e: + return f"(transcription failed: {e})" + + +# --------------------------------------------------------------------------- +# Stage 1: input / demucs stripping +# --------------------------------------------------------------------------- + + +def strip_vocals(input_audio: str): + if not input_audio: + raise gr.Error("Upload or record some audio first.") + _free_comfyui_memory() + out_dir = WORKDIR / f"demucs_{int(time.time())}" + out_dir.mkdir(parents=True, exist_ok=True) + _run( + [ + str(DEMUCS_PYTHON), "-m", "demucs", + "--two-stems=vocals", "-n", "htdemucs", + "-o", str(out_dir), input_audio, + ], + cwd=DEMUCS_DIR, + ) + stem_name = Path(input_audio).stem + vocals = out_dir / "htdemucs" / stem_name / "vocals.wav" + no_vocals = out_dir / "htdemucs" / stem_name / "no_vocals.wav" + return str(vocals), str(no_vocals) + + +# --------------------------------------------------------------------------- +# Stage 2: AuK voice editing, with history for undo/reset +# +# history_state is a list of {"path": str, "instruction": str|None} dicts, +# one entry per actual AuK call (not per button click - a click can queue +# several edits, e.g. enhance+emotion, and each becomes its own history +# entry so "Save as character" can derive a sensibly-scoped +# generation_recipe and Undo is granular). +# --------------------------------------------------------------------------- + + +def use_as_reference(audio_path: str, history: list): + if not audio_path: + raise gr.Error("No audio to use yet.") + audio_path = _ensure_wav(audio_path) + history = [{"path": audio_path, "instruction": None}] + transcript = _whisper_transcribe(audio_path) + return audio_path, history, _history_label(history), transcript + + +def _history_label(history: list) -> str: + n = len(history) - 1 + if n <= 0: + return "No edits applied yet." if history else "No reference set yet." + steps = "\n".join(f"{i}. {h['instruction']}" for i, h in enumerate(history[1:], 1)) + return f"{n} edit(s) applied on top of the original reference:\n{steps}" + + +ENHANCE_INSTRUCTIONS = { + "none": None, + "General (denoise + dereverb)": ( + "Preserve all speakers, remove noise and reverberation, and output " + "clean speech of the same length." + ), + "Telephone/call recording": ( + "Repair the telephone effect and restore natural, clear speech." + ), +} + + +def _auk_model_args(model_choice: str) -> list[str]: + """AuK-Flash is the same architecture distilled to 4 fixed diffusion + steps with CFG=0 (vendor README) - same --qwen_path, different --ckpt + plus --nfe/--cfg. VRAM footprint is essentially identical to Base + (measured in the README), so no extra offload concerns switching.""" + if model_choice == MODEL_FLASH: + if not AUK_FLASH_CKPT.exists(): + raise gr.Error( + f"AuK-Flash isn't downloaded yet ({AUK_FLASH_CKPT} not found) - " + f"run `hf download tencent/AuK-Flash --local-dir ./ckpts/AuK-Flash` " + f"from the auk/ directory first, or switch back to Base." + ) + return ["--ckpt", str(AUK_FLASH_CKPT), "--qwen_path", str(AUK_QWEN_PATH), "--nfe", "4", "--cfg", "0"] + return ["--ckpt", str(AUK_CKPT), "--qwen_path", str(AUK_QWEN_PATH)] + + +def apply_edit( + current_ref: str, + history: list, + model_choice: str, + enhance: str, + timbre_description: str, + emotion: str, + other_emotion: str, + pitch: str, + speed: str, + deaccent: bool, + volume: str, + whisper_direction: str, + nonverbal_instruction: str, + nonverbal_gen_seconds: float, + custom_instruction: str, +): + if not current_ref: + raise gr.Error("Set a voice reference first (upload, strip with Demucs, or use a previous edit).") + + instructions = [] + # Enhancement goes first when picked - clean the recording up before any + # character/accent editing, not after (garbage in, garbage timbre-edited). + if enhance and ENHANCE_INSTRUCTIONS.get(enhance): + instructions.append(ENHANCE_INSTRUCTIONS[enhance]) + if timbre_description and timbre_description.strip(): + # This is AuK's Timbre task - the one actually used to build every + # character's base voice in this project (accent, age, register, + # personality all described in one free-form sentence). It has no + # fixed vocabulary, unlike Emotion below. + instructions.append( + f'Keep the spoken content unchanged and change the timbre to: "{timbre_description.strip()}".' + ) + if emotion and emotion != "none": + instructions.append(auk_emotion_instruction(emotion)) + if other_emotion and other_emotion.strip(): + # Free-text emotion word, not one of AuK's 8 - always goes through + # the Timbre task's free-form template (see auk_emotion_instruction). + instructions.append(auk_emotion_instruction(other_emotion.strip())) + if pitch and pitch != "none": + direction = "Raise" if pitch.startswith("+") else "Lower" + instructions.append(f"{direction} the pitch by {pitch.lstrip('+-')} semitone(s).") + if speed and speed != "none": + speed_gen_seconds = _duration(current_ref) / float(speed) + instructions.append((f"Adjust the speech speed to {speed}x.", speed_gen_seconds)) + if deaccent: + instructions.append("Remove the regional accent while preserving the speaker's voice and content.") + if volume and volume != "none": + direction = "Increase" if volume.startswith("+") else "Decrease" + instructions.append(f"{direction} the volume by {volume.lstrip('+-')} dB.") + if whisper_direction and whisper_direction != "none": + if whisper_direction == "To whisper": + instructions.append("Convert this speech into a soft whisper while preserving the speaker and content.") + else: + instructions.append("Convert this whispered speech into a normal speaking voice while preserving the speaker and content.") + if nonverbal_instruction and nonverbal_instruction.strip(): + # Add/remove changes duration unpredictably (vendor cookbook 4.4 sets + # --gen_seconds for both directions, not just "add") - no way to + # auto-estimate this one like Speed, the user has to know roughly + # what they want. Defaults to the current reference's own duration + # if left blank. + target_seconds = float(nonverbal_gen_seconds) if nonverbal_gen_seconds else _duration(current_ref) + instructions.append((nonverbal_instruction.strip(), target_seconds)) + if custom_instruction and custom_instruction.strip(): + instructions.append(custom_instruction.strip()) + + if not instructions: + raise gr.Error("Pick at least one edit.") + + _free_comfyui_memory() + ref = current_ref + out_dir = WORKDIR / f"auk_{int(time.time())}" + out_dir.mkdir(parents=True, exist_ok=True) + last_transcript = "" + for i, instruction in enumerate(instructions): + gen_seconds = None + if isinstance(instruction, tuple): + instruction, gen_seconds = instruction + out_path = out_dir / f"edit_{i}.wav" + cmd = ( + [str(AUK_INFER)] + _auk_model_args(model_choice) + + ["--audio", ref, "--instruction", instruction, "--output", str(out_path), "--cpu_offload"] + ) + if gen_seconds is not None: + cmd += ["--gen_seconds", str(round(gen_seconds, 2))] + _run(cmd, cwd=AUK_DIR) + ref = str(out_path) + last_transcript = _whisper_transcribe(ref) + history = history + [{"path": ref, "instruction": instruction}] + + return ref, history, _history_label(history), last_transcript + + +def undo_edit(history: list): + if len(history) <= 1: + raise gr.Error("Nothing to undo.") + history = history[:-1] + transcript = _whisper_transcribe(history[-1]["path"]) + return history[-1]["path"], history, _history_label(history), transcript + + +def reset_to_original(history: list): + if not history: + raise gr.Error("No reference set yet.") + history = [history[0]] + transcript = _whisper_transcribe(history[0]["path"]) + return history[0]["path"], history, _history_label(history), transcript + + +# --------------------------------------------------------------------------- +# Stage 3: MOSS-TTS generation +# --------------------------------------------------------------------------- + + +def apply_mastering( + audio_path: str, + warmth_db: float = MASTERING_DEFAULTS["warmth_db"], + presence_cut_db: float = MASTERING_DEFAULTS["presence_cut_db"], + air_db: float = MASTERING_DEFAULTS["air_db"], + comp_ratio: float = MASTERING_DEFAULTS["comp_ratio"], + reverb_wet: float = MASTERING_DEFAULTS["reverb_wet"], +) -> str: + """Run mastering/master.py's pedalboard chain (warmth EQ, presence dip, + compression, a touch of reverb, hard-ceiling limiter) on an existing + clip. Shared by the standalone Mastering tab and Pipeline Studio's + optional auto-apply-after-generation toggle.""" + if not audio_path: + raise gr.Error("No audio to master yet.") + out_path = WORKDIR / f"mastered_{int(time.time())}.wav" + _run( + [ + str(MASTERING_PYTHON), str(MASTERING_SCRIPT), + "--input", audio_path, + "--output", str(out_path), + "--warmth-db", str(warmth_db), + "--presence-cut-db", str(presence_cut_db), + "--air-db", str(air_db), + "--comp-ratio", str(comp_ratio), + "--reverb-wet", str(reverb_wet), + ], + cwd=MASTERING_DIR, + ) + return str(out_path) + + +def generate_speech(reference: str, text: str, language: str, master_after: bool): + if not reference: + raise gr.Error("Set a voice reference first (upload, strip, and/or edit it above).") + if not text or not text.strip(): + raise gr.Error("Enter some text to synthesize.") + + _free_comfyui_memory() + out_path = WORKDIR / f"tts_{int(time.time())}.wav" + env = None + if TTS_LD_LIBRARY_PATH: + env = os.environ.copy() + env["LD_LIBRARY_PATH"] = TTS_LD_LIBRARY_PATH + + _run( + [ + str(TTS_PYTHON), str(TTS_GENERATE), + "--text", text, + "--reference", reference, + "--language", language, + "--output", str(out_path), + ], + cwd=TTS_DIR, + env=env, + ) + final_path = apply_mastering(str(out_path)) if master_after else str(out_path) + transcript = _whisper_transcribe(final_path) + return final_path, transcript + + +def generate_sound_effect(description: str, model_choice: str, duration: float): + """Text-only ambience/SFX generation, no reference audio or character + voice involved (different generation primitive from the AuK/MOSS-TTS + voice-cloning flow above, so it's a standalone section rather than + wired into that pipeline's history/reference state). + Two model backends: Stable Audio 3 Small-SFX (own venv, CPU-only, no + ComfyUI VRAM contention, `duration` matters) and MOSS-SoundEffect (tts/ + venv, needs the GPU + ComfyUI's memory freed first, length is implicit + in its own generation rather than a parameter). + No whisper verification here - there's no speech content to check.""" + if not description or not description.strip(): + raise gr.Error("Enter a description of the sound effect/ambience.") + + out_path = WORKDIR / f"sfx_{int(time.time())}.wav" + + if model_choice == SFX_MODEL_SA3: + _run( + [ + str(SFX_SA3_PYTHON), str(SFX_SA3_GENERATE), + "--text", description.strip(), + "--output", str(out_path), + "--duration", str(duration), + ], + cwd=SFX_SA3_DIR, + ) + else: + _free_comfyui_memory() + env = None + if TTS_LD_LIBRARY_PATH: + env = os.environ.copy() + env["LD_LIBRARY_PATH"] = TTS_LD_LIBRARY_PATH + _run( + [str(TTS_PYTHON), str(TTS_GENERATE_SFX), "--text", description.strip(), "--output", str(out_path)], + cwd=TTS_DIR, + env=env, + ) + return str(out_path) + + +def save_sound_effect_to_library(audio_path: str, description: str) -> str: + """Copy the current clip into SFX_LIBRARY_DIR so it survives past + WORKDIR (a tempdir, wiped on restart). On-disk only, no metadata/db - + per Steve's "just on disk FOR NOW is fine".""" + if not audio_path: + raise gr.Error("Generate a sound effect first.") + SFX_LIBRARY_DIR.mkdir(parents=True, exist_ok=True) + slug = _slugify(description or "") or "sfx" + dest = SFX_LIBRARY_DIR / f"{slug}_{int(time.time())}.wav" + shutil.copy2(audio_path, dest) + return f"Saved to {dest}" + + +# --------------------------------------------------------------------------- +# Characters (the-service-station cast) +# --------------------------------------------------------------------------- + + +def _slugify(text: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") + + +def read_character_cards() -> dict: + """name -> parsed card dict (frontmatter + prose), read-only. Never + round-trips these through yaml.dump - only used for display and for + resolving a base reference_audio, never for rewriting an existing file + (that would silently strip hand-authored inline comments).""" + cards = {} + if not SS_VOICES_DIR.is_dir(): + return cards + for card_path in sorted(SS_VOICES_DIR.glob("*/CHARACTER.md")): + try: + text = card_path.read_text(encoding="utf-8") + parts = text.split("---", 2) + if len(parts) < 3: + continue + frontmatter = yaml.safe_load(parts[1]) or {} + frontmatter["_prose"] = parts[2].strip() + frontmatter["_dir"] = card_path.parent + cards[frontmatter.get("name", card_path.parent.name)] = frontmatter + except Exception as e: + cards[card_path.parent.name] = {"_error": str(e), "_dir": card_path.parent} + return cards + + +def character_names() -> list[str]: + return sorted(read_character_cards().keys()) + + +def load_character(name: str): + if not name: + return "Pick a character.", None + cards = read_character_cards() + card = cards.get(name) + if not card: + return f"No card found for '{name}'.", None + if "_error" in card: + return f"Could not parse {name}'s card: {card['_error']}", None + + ref_audio = card.get("reference_audio") + ref_path = str(SERVICE_STATION_DIR / ref_audio) if ref_audio else None + + lib = card.get("emotion_library") or {} + lib_lines = [f"- **{emo}**: {entry.get('status', '?')}" for emo, entry in lib.items()] or ["(none approved yet)"] + + cache_dir = card["_dir"] / "emotion-cache" + cached = sorted(p.stem for p in cache_dir.glob("*.wav")) if cache_dir.is_dir() else [] + cache_lines = [f"- {e} (cached, not yet approved)" for e in cached] or ["(none pending)"] + + info = ( + f"**{name}** — {card.get('role', '?')}\n\n" + f"Status: `{card.get('status', '?')}` | Engine: `{card.get('engine', '?')}` | " + f"Default emotion (documentation only): `{card.get('default_emotion', '?')}`\n\n" + f"### Approved emotion_library\n" + "\n".join(lib_lines) + "\n\n" + f"### Pending (emotion-cache, not approved)\n" + "\n".join(cache_lines) + "\n\n" + f"---\n{card.get('_prose', '')}" + ) + return info, ref_path + + +def list_character_voices(name: str) -> dict[str, str]: + """label -> reference audio path, for every voice available for a + character: the base voice, every approved emotion_library entry, and + every pending (not yet approved) emotion-cache candidate. Used both by + the clone-as-character tool below and could back a similar picker + anywhere else a character's variants need listing.""" + cards = read_character_cards() + card = cards.get(name) + if not card or "_error" in card: + return {} + voices = {} + ref_audio = card.get("reference_audio") + if ref_audio: + voices["Base (natural voice)"] = str(SERVICE_STATION_DIR / ref_audio) + for emo, entry in (card.get("emotion_library") or {}).items(): + ref = entry.get("reference_audio") + if ref: + voices[f"{emo} (approved)"] = str(SERVICE_STATION_DIR / ref) + cache_dir = card["_dir"] / "emotion-cache" + if cache_dir.is_dir(): + for p in sorted(cache_dir.glob("*.wav")): + voices[f"{p.stem} (cached, not approved)"] = str(p) + return voices + + +def refresh_voice_choices(name: str): + voices = list_character_voices(name) + labels = list(voices.keys()) + return gr.update(choices=labels, value=labels[0] if labels else None) + + +def clone_as_character(character_name: str, voice_label: str, text: str, language: str): + if not character_name: + raise gr.Error("Pick a character first.") + if not voice_label: + raise gr.Error("Pick a voice (base or one of its emotions).") + if not text or not text.strip(): + raise gr.Error("Enter some text for them to say.") + + voices = list_character_voices(character_name) + reference = voices.get(voice_label) + if not reference: + raise gr.Error(f"Could not resolve '{voice_label}' for {character_name} - try Refresh.") + + _free_comfyui_memory() + env = None + if TTS_LD_LIBRARY_PATH: + env = os.environ.copy() + env["LD_LIBRARY_PATH"] = TTS_LD_LIBRARY_PATH + out_path = WORKDIR / f"clone_{int(time.time())}.wav" + _run( + [str(TTS_PYTHON), str(TTS_GENERATE), "--text", text.strip(), + "--reference", reference, "--language", language, "--output", str(out_path)], + cwd=TTS_DIR, env=env, + ) + transcript = _whisper_transcribe(str(out_path)) + return str(out_path), transcript + + +def create_character( + name: str, role: str, prose: str, language: str, + current_ref: str, history: list, +): + if not name or not name.strip(): + raise gr.Error("Give the character a name.") + if not current_ref or not history: + raise gr.Error("Set up a reference and edit it above first - there's nothing to save yet.") + + slug = _slugify(name) + char_dir = SS_VOICES_DIR / slug / "webui-created" + if (SS_VOICES_DIR / slug / "CHARACTER.md").exists(): + raise gr.Error(f"voices/{slug}/CHARACTER.md already exists - this UI won't overwrite an existing character.") + char_dir.mkdir(parents=True, exist_ok=True) + + source_dest = char_dir / f"{slug}_source.wav" + ref_dest = char_dir / f"{slug}_ref_v1.wav" + subprocess.run(["cp", history[0]["path"], str(source_dest)], check=True) + subprocess.run(["cp", current_ref, str(ref_dest)], check=True) + + recipe = [ + {"tool": "auk", "instruction": h["instruction"]} + for h in history[1:] if h["instruction"] + ] + + frontmatter = { + "name": name.strip(), + "role": role.strip() or "Standing guest", + "engine": "moss-tts", + "model": "OpenMOSS-Team/MOSS-TTS-Local-Transformer-v1.5", + "reference_audio": str(ref_dest.relative_to(SERVICE_STATION_DIR)), + "source_clip": str(source_dest.relative_to(SERVICE_STATION_DIR)), + "source_clip_note": "Created via the webui pipeline studio - see generation_recipe for the exact edit chain.", + "generation_recipe": recipe, + "seed": None, + "emotion_library": {}, + "language": language, + "status": "draft", + "last_updated": date.today().isoformat(), + "provenance": "webui", + "appearances": [], + } + frontmatter_yaml = yaml.dump(frontmatter, sort_keys=False, allow_unicode=True) + body = f"---\n{frontmatter_yaml}---\n\n# {name.strip()}\n\n{prose.strip()}\n" + (SS_VOICES_DIR / slug / "CHARACTER.md").write_text(body, encoding="utf-8") + + return ( + f"Created voices/{slug}/CHARACTER.md with status: draft. " + f"Not committed - review, listen, and commit it yourself (or ask an agent to) " + f"once you're happy, same as every other voice in this project." + ) + + +def generate_emotion_candidate(character_name: str, emotion_word: str, model_choice: str): + if not character_name: + raise gr.Error("Pick a character first.") + if not emotion_word or not emotion_word.strip(): + raise gr.Error("Enter an emotion word.") + emotion_word = emotion_word.strip() + + cards = read_character_cards() + card = cards.get(character_name) + if not card or "_error" in card: + raise gr.Error(f"Could not load {character_name}'s card.") + + base_reference = str(SERVICE_STATION_DIR / card["reference_audio"]) + cache_dir = card["_dir"] / "emotion-cache" + cache_dir.mkdir(parents=True, exist_ok=True) + cache_path = cache_dir / f"{_slugify(emotion_word)}.wav" + + _free_comfyui_memory() + instruction = auk_emotion_instruction(emotion_word) + cmd = ( + [str(AUK_INFER)] + _auk_model_args(model_choice) + + ["--audio", base_reference, "--instruction", instruction, + "--output", str(cache_path), "--cpu_offload"] + ) + _run(cmd, cwd=AUK_DIR) + + ref_transcript = _whisper_transcribe(str(cache_path)) + + # Preview clone so the candidate can actually be judged, not just the + # edited reference in isolation. + env = None + if TTS_LD_LIBRARY_PATH: + env = os.environ.copy() + env["LD_LIBRARY_PATH"] = TTS_LD_LIBRARY_PATH + preview_text = f"Testing the {emotion_word} emotion for {character_name}." + preview_path = WORKDIR / f"emotion_preview_{int(time.time())}.wav" + _run( + [str(TTS_PYTHON), str(TTS_GENERATE), "--text", preview_text, + "--reference", str(cache_path), "--language", "English", "--output", str(preview_path)], + cwd=TTS_DIR, env=env, + ) + preview_transcript = _whisper_transcribe(str(preview_path)) + + model_note = ( + " Generated with **Flash** - if this direction is right, regenerate the same " + "word with **Base** before promoting it, since Flash trades fidelity for speed " + "and this is a candidate you might lock in permanently." + if model_choice == MODEL_FLASH else "" + ) + status = ( + f"Saved to `{cache_path.relative_to(SERVICE_STATION_DIR)}` (cached, **not approved**).{model_note}\n\n" + f"Reference transcript: {ref_transcript}\n\n" + f"Preview clone transcript: {preview_transcript}\n\n" + f"If this sounds right, promote it manually per `.claude/skills/produce-episode/" + f"SKILL.md`'s \"Adding an emotion to a character's library\" - this UI won't do " + f"that step for you." + ) + return str(cache_path), str(preview_path), status + + +# --------------------------------------------------------------------------- +# UI +# --------------------------------------------------------------------------- + +with gr.Blocks(title="Audio-demo pipeline studio") as demo: + with gr.Tabs(): + with gr.Tab("Pipeline Studio"): + gr.Markdown( + "# Pipeline studio\n" + "Strip a voice with **Demucs**, edit it with **AuK**, then generate new " + "speech in that voice with **MOSS-TTS**. Each stage shells out to its own " + "isolated venv, so this box's GPU is the real constraint — heavy edits " + "(long clips) can OOM AuK; keep reference clips short (under ~25s)." + ) + + reference_state = gr.State(None) + history_state = gr.State([]) + + with gr.Row(): + with gr.Column(): + gr.Markdown("## 1. Input") + input_audio = gr.Audio(label="Upload or record", type="filepath") + strip_btn = gr.Button("Strip vocals with Demucs") + with gr.Row(): + vocals_out = gr.Audio(label="Vocals (isolated)", type="filepath") + no_vocals_out = gr.Audio(label="Background (no vocals)", type="filepath") + with gr.Row(): + use_vocals_btn = gr.Button("Use vocals as reference") + use_no_vocals_btn = gr.Button("Use background as reference") + use_original_btn = gr.Button("Use uploaded audio as-is") + + with gr.Column(): + gr.Markdown("## 2. Voice editor (AuK)") + current_ref_audio = gr.Audio(label="Current reference", type="filepath", interactive=False) + ref_transcript_box = gr.Textbox( + label="Whisper transcript of current reference (editable - fix any mis-hearing before using it)", + info="Whisper can mishear casual speech as more formal phrasing on short/quiet clips (e.g. \"it's gonna\" -> \"it is going to be\") - correct it here before sending it anywhere.", + ) + use_ref_transcript_btn = gr.Button("Use this transcript as 'Text to speak' →") + history_label = gr.Markdown("No reference set yet.") + model_dd = gr.Radio( + [MODEL_BASE, MODEL_FLASH], value=MODEL_BASE, label="AuK model", + info="Draft with Flash (fast, 4-step, lower fidelity), confirm with Base before committing a voice or emotion you're actually going to keep.", + ) + enhance_dd = gr.Dropdown( + list(ENHANCE_INSTRUCTIONS.keys()), value="none", + label="Enhance (denoise/dereverb/restore) - runs first, before character edits", + ) + timbre_tb = gr.Textbox( + label="Timbre / voice description (accent, age, register, personality)", + placeholder=( + 'e.g. "British male, mid-30s, dry and understated, slight Essex ' + 'accent, deadpan delivery" - this is what actually built every ' + "character's base voice in this project. Free-form, no fixed " + "vocabulary (unlike Emotion below)." + ), + lines=2, + ) + emotion_dd = gr.Dropdown( + ["none"] + EMOTIONS, value="none", + label="Emotion (AuK's fixed vocabulary - anything else, use the field below)", + ) + other_emotion_tb = gr.Textbox( + label="Other emotion word (not in AuK's list above)", + placeholder='e.g. "bemused" - a single word for a momentary reaction, wrapped ' + "as a Timbre instruction automatically (different from the " + "voice-description field above, which is for the character's " + "lasting accent/age/personality)", + ) + pitch_dd = gr.Dropdown(["none"] + PITCH_OPTIONS, value="none", label="Pitch (semitones)") + speed_dd = gr.Dropdown(["none"] + SPEED_OPTIONS, value="none", label="Speed") + gr.Markdown( + "*Pitch editing has been unreliable in testing — even \"1 semitone\" " + "has overshot to ~4+ semitones on some voices. Prefer a free-form " + "custom instruction like \"a little lower in register\" over the Pitch " + "dropdown for character work.*\n\n" + "*Speed now sets AuK's required `--gen_seconds` automatically. On " + "references longer than ~15s this can stall (a 27s target hung this " + "box for 20+ minutes swapping to disk with 0% GPU use, while a 12s " + "source completed in seconds) - trim your reference shorter first if " + "using Speed.*" + ) + with gr.Accordion("More AuK tasks (de-accent, volume, whisper, nonverbal)", open=False): + deaccent_cb = gr.Checkbox(label="De-accent (remove regional accent, keep voice)") + volume_dd = gr.Dropdown( + ["none"] + VOLUME_OPTIONS, value="none", label="Volume (dB)", + ) + whisper_dd = gr.Dropdown( + ["none", "To whisper", "From whisper"], value="none", + label="Whisper conversion", + ) + nonverbal_tb = gr.Textbox( + label="Nonverbal editing (advanced)", + placeholder='e.g. "Add a cough before \'We tested\'" or "Remove all laughs from the audio."', + ) + nonverbal_seconds_nb = gr.Number( + label="Target duration for the nonverbal edit above (seconds)", + value=None, + info="AuK needs an explicit target length for add/remove nonverbal edits - it can't be auto-computed like Speed. Defaults to the current reference's own duration; increase it a little for \"add\", decrease it a little for \"remove\".", + ) + custom_instruction = gr.Textbox( + label="Custom instruction (advanced, optional)", + placeholder='e.g. "Lower the pitch very slightly."', + ) + apply_btn = gr.Button("Apply edit(s)", variant="primary") + with gr.Row(): + undo_btn = gr.Button("Undo last edit") + reset_btn = gr.Button("Reset to original") + + with gr.Column(): + gr.Markdown("## 3. Generate (MOSS-TTS)") + gen_text = gr.Textbox(label="Text to speak", lines=6) + gen_language = gr.Dropdown(LANGUAGES, value="English", label="Language") + gen_master_cb = gr.Checkbox( + label="Apply mastering after generation (warmth/de-tin/light reverb - see Mastering tab)", + value=True, + ) + generate_btn = gr.Button("Generate speech", variant="primary") + generated_audio = gr.Audio(label="Generated speech", type="filepath") + gen_transcript_box = gr.Textbox(label="Whisper transcript of generated speech") + + gr.Markdown("## 4. Save as a the-service-station character") + new_char_name = gr.Textbox(label="Name") + new_char_role = gr.Textbox(label="Role", placeholder="e.g. Standing guest - weather") + new_char_prose = gr.Textbox(label="Character brief (prose)", lines=4) + save_char_btn = gr.Button("Save as new character (status: draft)") + save_char_status = gr.Markdown() + + strip_btn.click(strip_vocals, inputs=[input_audio], outputs=[vocals_out, no_vocals_out]) + use_vocals_btn.click( + use_as_reference, inputs=[vocals_out, history_state], + outputs=[current_ref_audio, history_state, history_label, ref_transcript_box], + ) + use_no_vocals_btn.click( + use_as_reference, inputs=[no_vocals_out, history_state], + outputs=[current_ref_audio, history_state, history_label, ref_transcript_box], + ) + use_original_btn.click( + use_as_reference, inputs=[input_audio, history_state], + outputs=[current_ref_audio, history_state, history_label, ref_transcript_box], + ) + apply_btn.click( + apply_edit, + inputs=[current_ref_audio, history_state, model_dd, enhance_dd, timbre_tb, emotion_dd, + other_emotion_tb, pitch_dd, speed_dd, deaccent_cb, volume_dd, whisper_dd, + nonverbal_tb, nonverbal_seconds_nb, custom_instruction], + outputs=[current_ref_audio, history_state, history_label, ref_transcript_box], + ) + undo_btn.click( + undo_edit, inputs=[history_state], + outputs=[current_ref_audio, history_state, history_label, ref_transcript_box], + ) + reset_btn.click( + reset_to_original, inputs=[history_state], + outputs=[current_ref_audio, history_state, history_label, ref_transcript_box], + ) + generate_btn.click( + generate_speech, inputs=[current_ref_audio, gen_text, gen_language, gen_master_cb], + outputs=[generated_audio, gen_transcript_box], + ) + use_ref_transcript_btn.click(lambda t: t, inputs=[ref_transcript_box], outputs=[gen_text]) + save_char_btn.click( + create_character, + inputs=[new_char_name, new_char_role, new_char_prose, gen_language, current_ref_audio, history_state], + outputs=[save_char_status], + ) + with gr.Tab("Sound Effects"): + gr.Markdown( + "# Sound effects\n" + "Text-only ambience/SFX generation - a separate model from the " + "voice pipeline in Pipeline Studio, no reference audio or " + "character voice involved." + ) + sfx_model = gr.Radio( + choices=[SFX_MODEL_SA3, SFX_MODEL_MOSS], + value=SFX_MODEL_SA3, + label="Model", + ) + sfx_description = gr.Textbox( + label="Description", + placeholder=( + 'e.g. "A door creaking open slowly, then closing with a soft thud." ' + 'or "A busy outdoor market with distant chatter and footsteps."' + ), + lines=2, + ) + sfx_duration = gr.Slider( + minimum=1, maximum=60, value=10, step=1, + label="Duration (seconds) - Stable Audio 3 only", + ) + sfx_btn = gr.Button("Generate sound effect", variant="primary") + sfx_audio = gr.Audio(label="Generated sound effect", type="filepath") + with gr.Row(): + sfx_save_btn = gr.Button("Save to library") + sfx_save_status = gr.Markdown() + sfx_btn.click( + generate_sound_effect, + inputs=[sfx_description, sfx_model, sfx_duration], + outputs=[sfx_audio], + ) + sfx_save_btn.click( + save_sound_effect_to_library, + inputs=[sfx_audio, sfx_description], + outputs=[sfx_save_status], + ) + + with gr.Tab("Mastering"): + gr.Markdown( + "# Mastering\n" + "Fixes the \"tinny\" spectral imbalance common to generated " + "TTS output (too much 3-6kHz presence, not enough low end) " + "and adds a bit of warmth/density/space. Works on any clip - " + "Pipeline Studio's Generate step can also apply this " + "automatically (see its checkbox)." + ) + with gr.Row(): + with gr.Column(): + master_input = gr.Audio(label="Input audio", type="filepath") + master_warmth = gr.Slider( + minimum=0, maximum=10, step=0.5, value=MASTERING_DEFAULTS["warmth_db"], + label="Warmth (low-shelf boost @150Hz, dB)", + ) + master_presence = gr.Slider( + minimum=-10, maximum=0, step=0.5, value=MASTERING_DEFAULTS["presence_cut_db"], + label="De-tin (presence cut @3.8kHz, dB)", + ) + master_air = gr.Slider( + minimum=-10, maximum=2, step=0.5, value=MASTERING_DEFAULTS["air_db"], + label="Softness (high-shelf @6.5kHz, dB)", + ) + master_comp = gr.Slider( + minimum=1, maximum=6, step=0.5, value=MASTERING_DEFAULTS["comp_ratio"], + label="Compression ratio", + ) + master_reverb = gr.Slider( + minimum=0, maximum=0.2, step=0.005, value=MASTERING_DEFAULTS["reverb_wet"], + label="Reverb wet level", + ) + master_btn = gr.Button("Apply mastering", variant="primary") + with gr.Column(): + master_output = gr.Audio(label="Mastered output", type="filepath") + master_btn.click( + apply_mastering, + inputs=[master_input, master_warmth, master_presence, master_air, master_comp, master_reverb], + outputs=[master_output], + ) + + with gr.Tab("Characters"): + gr.Markdown( + f"# The Service Station cast\n" + f"Reading from `{SERVICE_STATION_DIR}`. Approving a character or an emotion " + f"is a manual step (editing the card) - this tab creates drafts and " + f"candidates for you to listen to and promote yourself, it never marks " + f"anything `approved` on its own." + ) + with gr.Row(): + char_dd = gr.Dropdown(label="Character", choices=character_names()) + refresh_btn = gr.Button("Refresh") + char_info = gr.Markdown() + char_audio = gr.Audio(label="Reference audio", type="filepath", interactive=False) + + gr.Markdown( + "## Clone text as this character\n" + "Say anything in the selected character's voice - the base voice, an " + "approved emotion, or a pending (not yet approved) candidate from the " + "cache. Useful both for \"how would they say this?\" and for testing an " + "emotion candidate with real dialogue instead of the fixed test line below." + ) + clone_voice_dd = gr.Dropdown(label="Voice", choices=list(list_character_voices(None).keys())) + clone_text_tb = gr.Textbox(label="Text to speak", lines=3, placeholder="Type anything.") + clone_language_dd = gr.Dropdown(LANGUAGES, value="English", label="Language") + clone_btn = gr.Button("Clone", variant="primary") + clone_audio = gr.Audio(label="Cloned speech", type="filepath", interactive=False) + clone_transcript_box = gr.Textbox(label="Whisper transcript") + + gr.Markdown("## Generate a new emotion candidate") + with gr.Row(): + emotion_char_dd = gr.Dropdown(label="Character", choices=character_names()) + emotion_word_tb = gr.Textbox(label="Emotion word", placeholder="e.g. surprised, bemused, wistful") + emotion_model_dd = gr.Radio( + [MODEL_BASE, MODEL_FLASH], value=MODEL_FLASH, label="AuK model", + info="Flash defaults here since this is exploring candidates - regenerate the same word with Base before promoting one.", + ) + gen_emotion_btn = gr.Button("Generate candidate", variant="primary") + emotion_ref_audio = gr.Audio(label="Edited reference (candidate)", type="filepath", interactive=False) + emotion_preview_audio = gr.Audio(label="Preview clone", type="filepath", interactive=False) + emotion_status = gr.Markdown() + + refresh_btn.click( + lambda: (gr.update(choices=character_names()), gr.update(choices=character_names())), + outputs=[char_dd, emotion_char_dd], + ) + char_dd.change(load_character, inputs=[char_dd], outputs=[char_info, char_audio]) + char_dd.change(refresh_voice_choices, inputs=[char_dd], outputs=[clone_voice_dd]) + clone_btn.click( + clone_as_character, + inputs=[char_dd, clone_voice_dd, clone_text_tb, clone_language_dd], + outputs=[clone_audio, clone_transcript_box], + ) + gen_emotion_btn.click( + generate_emotion_candidate, + inputs=[emotion_char_dd, emotion_word_tb, emotion_model_dd], + outputs=[emotion_ref_audio, emotion_preview_audio, emotion_status], + ).then( + # So a freshly-generated candidate is immediately pickable in + # "Clone text as this character" without reselecting the + # character to force a refresh. Keyed off char_dd (the + # browser above), not emotion_char_dd - the clone tool + # always uses whichever character char_dd is showing, and + # those two dropdowns can legitimately point at different + # characters at once. + refresh_voice_choices, inputs=[char_dd], outputs=[clone_voice_dd], + ) + # Dropdown choices are otherwise fixed at server-start time, so a + # character created earlier in the same running session (or by + # someone else) wouldn't show up without this - refresh + # automatically on every page load, in addition to the manual + # Refresh button above for mid-session use without reloading. + demo.load( + lambda: (gr.update(choices=character_names()), gr.update(choices=character_names())), + outputs=[char_dd, emotion_char_dd], + ) + +if __name__ == "__main__": + # Gradio refuses to serve a file back to the browser unless it's under + # the app's own directory, the system temp dir, or explicitly allowed - + # every character's reference audio lives in the-service-station repo, + # a different path entirely, so every single Characters-tab selection + # errored without this (found via a real user report + the server log, + # not by inspection - the error only surfaces server-side, not in the + # browser's generic error toast). + demo.launch(server_name="0.0.0.0", server_port=7860, allowed_paths=[str(SERVICE_STATION_DIR)])