VoiceStudio/notebooks/OmniVoice_Studio_Colab.ipynb

VoiceStudio on Google Colab

Run the full VoiceStudio app — voice cloning, voice design, video dubbing, and TTS in 646 languages — on a free Colab GPU, web UI included.

Before you start: Runtime → Change runtime type → T4 GPU (the notebook also works on CPU, but generation is much slower).

Run the cells top to bottom. Every cell is idempotent — re-running after a hiccup is always safe.

Contents

• Part 1 — Setup & launch (cells 1-7): GPU check, install, backend launch, open the web UI, API smoke test.
• Part 2 — Feature tour (cells 8-20): each major feature as a self-contained API demo with inline playback — multilingual TTS, voice cloning, voice design, voice profiles, transcription, AI-watermark detection, the OpenAI-compatible API, a multi-voice story, a chaptered audiobook, video dubbing, and vocal isolation.
• Part 3 — Troubleshooting.

How this notebook works (and why):

Piece	Approach	Why
Backend install	uv pip install --system .	Mirrors the official Docker image: installs into Colab's Python and keeps Colab's preinstalled CUDA PyTorch when it matches our pinned torch, instead of re-downloading multi-GB wheels into a venv. The pins (deploy/torch-constraints.txt) are passed explicitly so torch, torchaudio and torchvision cannot drift apart (#1357).
Web UI	Built in-notebook with bun (~2 min, once per session)	Official releases ship desktop installers only — there is no prebuilt standalone web bundle to download. The backend serves the built SPA itself from frontend/dist.
Opening the app	Colab's built-in kernel port proxy	No third-party tunnel binaries; the URL is private to your Google session. (A cloudflared public-URL alternative is documented in the launch cell.)

Issues with this notebook are VoiceStudio issues — report them at debpalash/VoiceStudio/issues.

Part 1 — Setup & launch

In [ ]:

# ── 1. GPU check ────────────────────────────────────────────────────────────
# Confirms the runtime has an NVIDIA GPU. Everything still works on CPU, but
# a single sentence can take minutes instead of seconds.
import shutil
import subprocess

if shutil.which("nvidia-smi"):
    print(subprocess.run(["nvidia-smi"], capture_output=True, text=True).stdout)
    print("GPU detected — you're good to go.")
else:
    print("=" * 74)
    print("WARNING: no NVIDIA GPU in this runtime.")
    print("Fix: menu bar -> Runtime -> Change runtime type -> T4 GPU,")
    print("then re-run this notebook from the top.")
    print("(Continuing anyway works, but generation will be very slow on CPU.)")
    print("=" * 74)

In [ ]:

# ── 2. Clone + install (idempotent; first run ~5-8 min) ─────────────────────
import os
import shutil
import subprocess
import sys

REPO_URL = "https://github.com/debpalash/VoiceStudio"
REPO_DIR = "/content/VoiceStudio"
BUN = "/root/.bun/bin/bun"

# Generous network timeouts: Colab -> PyPI is usually fast, but model/wheel
# CDNs occasionally stall and uv's default timeout is short.
os.environ.setdefault("UV_HTTP_TIMEOUT", "300")

def run(cmd, *, cwd=None, what=""):
    """Stream a command's output; stop the cell with an actionable message on failure."""
    shown = cmd if isinstance(cmd, str) else " ".join(cmd)
    print(f"\n$ {shown}")
    rc = subprocess.run(cmd, cwd=cwd, shell=isinstance(cmd, str)).returncode
    if rc != 0:
        raise SystemExit(
            f"\nFAILED: {what or shown} (exit code {rc}).\n"
            "Scroll up for the underlying error. Re-running this cell is safe.\n"
            "Still stuck? Open an issue with the output above:\n"
            "  https://github.com/debpalash/VoiceStudio/issues"
        )

# 2a. Clone the repo (reused if already present)
if os.path.isdir(os.path.join(REPO_DIR, ".git")):
    print(f"Repo already present at {REPO_DIR} — reusing it.")
else:
    run(["git", "clone", "--depth", "1", REPO_URL, REPO_DIR], what="git clone")

# 2b. System packages (ffmpeg: audio/video processing; libsndfile1: soundfile)
run("apt-get -qq update && apt-get -qq install -y ffmpeg libsndfile1", what="apt-get install")

# 2c. bun — builds the web UI (see the intro cell for why we build it here)
if not os.path.exists(BUN):
    run("curl -fsSL https://bun.sh/install | bash", what="bun installer")
os.environ["PATH"] = os.path.dirname(BUN) + os.pathsep + os.environ["PATH"]
run([BUN, "--version"], what="bun version check")

# 2d. Build the frontend (skipped once frontend/dist/index.html exists)
dist_index = os.path.join(REPO_DIR, "frontend", "dist", "index.html")
if os.path.exists(dist_index):
    print("frontend/dist already built — skipping.")
else:
    # The repo is a bun workspace: install at the repo root, build in frontend/.
    run([BUN, "install", "--frozen-lockfile"], cwd=REPO_DIR, what="bun install (workspace)")
    run([BUN, "run", "--cwd", "frontend", "build"], cwd=REPO_DIR, what="frontend build (vite)")
    if not os.path.exists(dist_index):
        raise SystemExit(
            "The frontend build finished but frontend/dist/index.html is missing —\n"
            "scroll up for vite errors, then re-run this cell."
        )

# 2e. Backend dependencies — same command the official Docker image uses.
if not shutil.which("uv"):
    run([sys.executable, "-m", "pip", "install", "-q", "uv"], what="pip install uv")
# --constraint: `uv pip install` ignores pyproject's `[tool.uv]
# constraint-dependencies` (a project-API setting), so without it torch,
# torchaudio and torchvision resolve on their bare lower bounds and Colab's
# preinstalled torchvision can be left behind a newer torch -- which fails at
# import with "operator torchvision::nms does not exist" (#1357).
run(["uv", "pip", "install", "--system", "--no-cache",
     "--constraint", "deploy/torch-constraints.txt", "."],
    cwd=REPO_DIR, what="backend install (uv pip install --system .)")

# 2f. cuDNN 8 side-install for CTranslate2 (WhisperX ASR on GPU). PyTorch 2.8+
# ships cuDNN 9; scripts/setup.py places cuDNN 8 libs where the backend
# preloads them from (<repo>/.venv/.../cudnn8_compat). We create the .venv
# directory only so the script has its expected target — no actual venv is
# used on Colab. Non-fatal: without it, transcription falls back to the
# torch-native Whisper backend automatically.
os.makedirs(os.path.join(REPO_DIR, ".venv"), exist_ok=True)
run([sys.executable, os.path.join(REPO_DIR, "scripts", "setup.py")], what="cuDNN 8 compat setup")

# 2g. Sanity check in a fresh interpreter (this kernel may hold a stale torch).
# Imports the backend's own model stack, not just torch — Colab's system
# Python mixes preinstalled and freshly-resolved wheels, and a torchaudio or
# transformers that can't load together only shows up when the model module is
# imported (#1229). Catching it here beats a 5-minute health timeout in cell 5.
run([sys.executable, "-c",
     # Versions FIRST: if the model-stack import below fails, the cell output
     # still shows what was actually installed, which is the single most
     # useful line for diagnosing a Colab environment (#1229).
     "import torch, torchaudio, uvicorn, fastapi, transformers; "
     "print(f'torch {torch.__version__}, torchaudio {torchaudio.__version__}, "
     "transformers {transformers.__version__}, "
     "CUDA available: {torch.cuda.is_available()}'); "
     "from omnivoice.models.omnivoice import OmniVoice; "
     "from transformers import HiggsAudioV2TokenizerModel; "
     "print('Install OK - backend model stack imports cleanly')"],
    cwd=REPO_DIR, what="import sanity check")

In [ ]:

# ── 3. (Optional) Hugging Face token ────────────────────────────────────────
# Only needed for GATED models — e.g. pyannote speaker diarization, used by
# video dubbing for multi-speaker detection. TTS, voice cloning, and voice
# design all work WITHOUT a token, so feel free to skip this cell.
#
# To use one: accept the model terms at
#   https://huggingface.co/pyannote/speaker-diarization-3.1
# then add a Colab Secret named HF_TOKEN (key icon in the left sidebar),
# toggle "Notebook access" ON, and re-run this cell.
import os

try:
    from google.colab import userdata
    _token = userdata.get("HF_TOKEN")
    os.environ["HF_TOKEN"] = _token
    print("HF_TOKEN loaded from Colab Secrets — gated models (diarization) enabled.")
except Exception:
    print("No HF_TOKEN Colab Secret found — skipping. (Everything except gated "
          "diarization models works without it.)")

In [ ]:

# ── 4. (Optional, recommended) Pre-download the default voice model ─────────
# The default TTS engine fetches k2-fsa/OmniVoice (a few GB) on its very first
# generation. Downloading it up front makes the first request fast instead of
# a several-minute stall. Safe to re-run: already-downloaded files are reused.
from huggingface_hub import snapshot_download

path = snapshot_download("k2-fsa/OmniVoice")
print("Default TTS model cached at:", path)

In [ ]:

# ── 5. Launch the backend ───────────────────────────────────────────────────
# Starts uvicorn on port 3900 and waits for /health. Re-running this cell when
# the backend is already up just reports its status.
import json
import os
import subprocess
import sys
import time
import urllib.request

REPO_DIR = "/content/VoiceStudio"
PORT = 3900
LOG_PATH = "/content/omnivoice_backend.log"
HEALTH_URL = f"http://127.0.0.1:{PORT}/health"

def health():
    try:
        with urllib.request.urlopen(HEALTH_URL, timeout=5) as r:
            return json.load(r)
    except Exception:
        return None

info = health()
if info:
    print(f"Backend already running — {info}")
else:
    # Clear any half-dead process from a previous run of this cell.
    subprocess.run(f"kill -9 $(lsof -t -i:{PORT}) 2>/dev/null || true", shell=True)
    time.sleep(1)

    env = os.environ.copy()
    # Headless-server deployment, same as the official Docker image: relaxes
    # the desktop-only loopback origin gate so the proxied browser session can
    # reach the settings/system routes.
    env["OMNIVOICE_SERVER_MODE"] = "1"
    # Voices, projects, and generated audio live here. Ephemeral! See the
    # troubleshooting cell for persisting it to Google Drive.
    env["OMNIVOICE_DATA_DIR"] = "/content/omnivoice_data"
    env["PYTHONUNBUFFERED"] = "1"

    log = open(LOG_PATH, "ab")
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "main:app",
         "--app-dir", "backend", "--host", "127.0.0.1", "--port", str(PORT)],
        cwd=REPO_DIR, env=env, stdout=log, stderr=subprocess.STDOUT,
    )
    print(f"Backend starting (PID {proc.pid}); log: {LOG_PATH}")

    deadline = time.time() + 300
    while time.time() < deadline:
        if proc.poll() is not None:
            break  # process died — report below
        info = health()
        if info:
            break
        print(".", end="", flush=True)
        time.sleep(3)
    print()

    if info:
        print(f"Backend is up — {info}")
        if "cuda" not in str(info.get("device", "")):
            print("NOTE: device is not CUDA — generation will be slow. "
                  "See cell 1 to enable the GPU runtime.")
    else:
        try:
            with open(LOG_PATH, "r", errors="replace") as f:
                tail = "".join(f.readlines()[-40:])
        except OSError:
            tail = "(no log file found)"
        raise SystemExit(
            "Backend did not become healthy within 5 minutes.\n"
            f"--- last lines of {LOG_PATH} ---\n{tail}\n"
            "--- end of log ---\n"
            "Fix the error above (usually a missing dependency: re-run cell 2),\n"
            "then re-run this cell. Still stuck? Attach the log to an issue:\n"
            "  https://github.com/debpalash/VoiceStudio/issues"
        )

In [ ]:

# ── 6. Open the app ─────────────────────────────────────────────────────────
# Serves localhost:3900 to YOUR browser through Colab's built-in kernel port
# proxy — authenticated to your Google session, no third-party tunnel binary.
# A new browser tab opens with the full VoiceStudio UI (allow pop-ups
# for colab.research.google.com if nothing appears).
#
# Want a PUBLIC URL instead (e.g. to open the app on your phone)? Cloudflare's
# quick tunnel works well — run in a new cell:
#   !wget -q https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64 -O /usr/local/bin/cloudflared
#   !chmod +x /usr/local/bin/cloudflared
#   !nohup cloudflared tunnel --url http://127.0.0.1:3900 --no-autoupdate > /content/cloudflared.log 2>&1 &
#   !sleep 5 && grep -o "https://.*trycloudflare.com" /content/cloudflared.log | head -1
# Anyone with that URL can reach your session — set a share PIN in the app's
# Settings first.
from google.colab import output

output.serve_kernel_port_as_window(3900)
print("If no tab opened, click the link above (and allow pop-ups).")

In [ ]:

# ── 7. Smoke test: generate speech through the API ──────────────────────────
# Proves the whole stack end-to-end without touching the UI: /health, then a
# real TTS generation played inline. POST /generate returns the WAV directly.
import json
import urllib.error
import urllib.parse
import urllib.request

BASE = "http://127.0.0.1:3900"

with urllib.request.urlopen(f"{BASE}/health", timeout=10) as r:
    print("GET /health ->", json.load(r))

data = urllib.parse.urlencode({
    "text": "VoiceStudio is up and running on Google Colab.",
}).encode()
print("Generating... (first-ever generation loads the model — if you skipped "
      "cell 4 it also downloads it, which can take several minutes)")
try:
    with urllib.request.urlopen(
        urllib.request.Request(f"{BASE}/generate", data=data), timeout=1800
    ) as r:
        wav = r.read()
        print(f"OK — {len(wav)} bytes, duration {r.headers.get('X-Audio-Duration')}s, "
              f"generated in {r.headers.get('X-Gen-Time')}s (seed {r.headers.get('X-Seed')})")
except urllib.error.HTTPError as e:
    detail = e.read().decode("utf-8", errors="replace")
    raise SystemExit(
        f"Generation failed: HTTP {e.code}\n{detail}\n"
        "Check /content/omnivoice_backend.log for the full trace; on a fresh\n"
        "session the usual cause is an interrupted model download — re-run\n"
        "cell 4, then this cell."
    )

OUT = "/content/omnivoice_smoke.wav"
with open(OUT, "wb") as f:
    f.write(wav)

from IPython.display import Audio, display
display(Audio(OUT))

Part 2 — Feature tour

Each cell below demos one major feature straight through the backend's HTTP API — the same API the desktop app and the web UI use. Every cell is self-contained (it regenerates anything it needs), plays its result inline, and states its expected runtime honestly. Requirements: the backend from cell 5 must be running, and run cell 8 (helpers) once first.

Endpoints used, per section: /generate (9-12), /design/describe (11), /profiles (12), /transcribe (13), /watermark/status + /watermark/detect (14), /v1/audio/* (15), /longform/render (16), /audiobook (17), /dub/* + /jobs/* (18-19).

8. Feature-tour helpers

Shared plumbing for the tour: a /generate wrapper that saves WAVs under /content/omnivoice_demos/, inline playback, and loud failures that point at the backend log. Run this once before any feature cell.

In [ ]:

# ── 8. Feature-tour helpers (run once before cells 9-20) ────────────────────
import os

import requests
from IPython.display import Audio, display

BASE = "http://127.0.0.1:3900"
DEMO_DIR = "/content/omnivoice_demos"
LOG_HINT = ("Backend log: /content/omnivoice_backend.log — attach it when "
            "reporting: https://github.com/debpalash/VoiceStudio/issues")
os.makedirs(DEMO_DIR, exist_ok=True)

# Shared sample lines (cells 9/10/12/13 reuse these to stay self-contained).
EN_TEXT = "This sentence was generated locally, on a free cloud graphics card."
ES_TEXT = "La clonación de voz ya no necesita la nube: todo sucede aquí mismo."
BN_TEXT = "আমার কণ্ঠস্বর এখন ছয়শো ছেচল্লিশটি ভাষায় কথা বলতে পারে।"

def check_backend():
    try:
        requests.get(f"{BASE}/health", timeout=5).raise_for_status()
    except Exception:
        raise SystemExit("The backend is not answering — run cell 5 first. " + LOG_HINT)

def api_fail(resp, doing):
    raise SystemExit(f"{doing} failed: HTTP {resp.status_code}\n"
                     f"{resp.text[:2000]}\n{LOG_HINT}")

def speak(text, out_name, files=None, timeout=1800, **form):
    """POST /generate; save the returned WAV under DEMO_DIR; return (path, headers)."""
    check_backend()
    data = {"text": text, **{k: str(v) for k, v in form.items()}}
    r = requests.post(f"{BASE}/generate", data=data, files=files, timeout=timeout)
    if r.status_code != 200:
        api_fail(r, f"/generate ({out_name})")
    path = os.path.join(DEMO_DIR, out_name)
    with open(path, "wb") as f:
        f.write(r.content)
    return path, r.headers

def ensure_wav(out_name, text, **form):
    """Idempotent speak(): reuse the file if an earlier cell already made it."""
    path = os.path.join(DEMO_DIR, out_name)
    if os.path.exists(path) and os.path.getsize(path) > 0:
        return path
    return speak(text, out_name, **form)[0]

def play(path, label=None):
    if label:
        print(label)
    display(Audio(path))

print("Helpers ready. Demo artifacts will land in", DEMO_DIR)

9. Text-to-speech in many languages

Three short generations — English, Spanish, Bengali — through POST /generate, with the per-request seed and generation time from the response headers. No language parameter is passed: the engine autodetects from the text. Expected runtime: ~5-20 s per line on a warm T4 (the very first generation loads the model — minutes if cell 4 was skipped).

In [ ]:

# ── 9. TTS in many languages ────────────────────────────────────────────────
try:
    speak
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

for code, label, text in [("en", "English", EN_TEXT),
                          ("es", "Spanish", ES_TEXT),
                          ("bn", "Bengali", BN_TEXT)]:
    path, h = speak(text, f"tts_{code}.wav")
    print(f"{label}: seed={h.get('X-Seed')}  gen_time={h.get('X-Gen-Time')}s  "
          f"duration={h.get('X-Audio-Duration')}s")
    play(path)

10. Voice cloning (zero-shot)

Clone a voice from a single short reference clip — no training. To stay self-contained, the reference is the English WAV from cell 9 (regenerated if missing): the cloned line should come back in that same voice. A commented variant shows how to clone your own voice from an uploaded 3-10 s clip. Expected runtime: ~10-30 s on a warm T4.

In [ ]:

# ── 10. Voice cloning ───────────────────────────────────────────────────────
try:
    ensure_wav
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

ref_path = ensure_wav("tts_en.wav", EN_TEXT)
ref_text = EN_TEXT  # transcript of the reference clip (skips auto-transcription)

with open(ref_path, "rb") as ref:
    clone_path, h = speak(
        "And this is the very same voice, cloned from three seconds of audio.",
        "clone_demo.wav",
        files={"ref_audio": ("reference.wav", ref, "audio/wav")},
        ref_text=ref_text,
    )
print(f"Cloned in {h.get('X-Gen-Time')}s (seed {h.get('X-Seed')})")
play(ref_path, "Reference voice:")
play(clone_path, "Cloned voice, new sentence:")

# ── Clone YOUR OWN voice instead ──
# from google.colab import files
# uploaded = files.upload()          # pick a clean 3-10 s clip of one speaker
# my_clip = next(iter(uploaded))
# with open(my_clip, "rb") as ref:
#     my_clone, _ = speak("Any text you like, in my voice.", "my_clone.wav",
#                         files={"ref_audio": (my_clip, ref)})
#     # (no ref_text -> the backend transcribes the clip automatically)
# play(my_clone)

11. Voice design

Describe a voice in plain English; POST /design/describe maps the description onto the engine's design-parameter space (gender, age, pitch, accent, style) and returns a validated instruct string, which /generate then uses to synthesize that voice from nothing — no reference audio. The same sentence is rendered with two contrasting designs. Expected runtime: ~10-30 s per voice on a warm T4.

In [ ]:

# ── 11. Voice design ────────────────────────────────────────────────────────
try:
    speak
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

SENTENCE = "Every voice you hear in this notebook was invented by a description."
designs = [
    ("design_a.wav", "a deep, elderly male narrator with a low pitch and a british accent"),
    ("design_b.wav", "a young, energetic female voice with a high pitch"),
]

for out_name, description in designs:
    r = requests.post(f"{BASE}/design/describe", json={"description": description}, timeout=30)
    if r.status_code != 200:
        api_fail(r, "/design/describe")
    d = r.json()
    print(f'"{description}"')
    print(f"  -> instruct: {d['instruct']!r}")
    if d.get("unmatched"):
        print(f"  -> not mappable (ignored): {d['unmatched']}")
    path, h = speak(SENTENCE, out_name, instruct=d["instruct"])
    play(path, f"  gen_time={h.get('X-Gen-Time')}s")

12. Voice profiles — save, list, reuse

Voices become reusable assets: save the cell-9 narrator and the cell-11 designed voice as named profiles (POST /profiles), list them (GET /profiles), then generate by profile_id alone — no reference upload, no instruct string. Cells 16-17 reuse these two profiles. Idempotent: existing profiles with the same names are reused. Expected runtime: seconds, plus one ~10-30 s generation.

In [ ]:

# ── 12. Voice profiles ──────────────────────────────────────────────────────
try:
    ensure_wav
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

check_backend()

def ensure_profile(name, wav_path, ref_text):
    """Create a clone profile from a WAV, or reuse it if this name exists."""
    r = requests.get(f"{BASE}/profiles", timeout=30)
    if r.status_code != 200:
        api_fail(r, "GET /profiles")
    for p in r.json():
        if p.get("name") == name:
            print(f"Profile {name!r} already exists (id={p['id']}) — reusing.")
            return p["id"]
    with open(wav_path, "rb") as ref:
        r = requests.post(
            f"{BASE}/profiles",
            data={"name": name, "ref_text": ref_text, "kind": "clone"},
            files={"ref_audio": (os.path.basename(wav_path), ref, "audio/wav")},
            timeout=120,
        )
    if r.status_code != 200:
        api_fail(r, "POST /profiles")
    created = r.json()
    print(f"Created profile {name!r} (id={created['id']})")
    return created["id"]

NARRATOR_ID = ensure_profile(
    "Colab Narrator", ensure_wav("tts_en.wav", EN_TEXT), EN_TEXT)
GUEST_SENTENCE = "Every voice you hear in this notebook was invented by a description."
GUEST_ID = ensure_profile(
    "Colab Guest",
    ensure_wav("design_b.wav", GUEST_SENTENCE,
               instruct="female, young, high pitch"),
    GUEST_SENTENCE)

profiles = requests.get(f"{BASE}/profiles", timeout=30).json()
print(f"\n{len(profiles)} saved profile(s):")
for p in profiles:
    print(f"  {p['id']}  {p.get('kind', '?'):6s}  {p.get('name')}")

path, h = speak("Generating by profile id — no reference clip attached this time.",
                "profile_demo.wav", profile_id=NARRATOR_ID)
play(path, f"\nVoice of profile {NARRATOR_ID} (gen_time={h.get('X-Gen-Time')}s):")

13. Transcription (speech-to-text)

The round trip: the WAV that TTS produced in cell 9 goes back through POST /transcribe, and the recognized text should match the original sentence. Expected runtime: the first transcription downloads an ASR model (roughly 1-3 GB, a few minutes); afterwards it's seconds.

In [ ]:

# ── 13. Transcription: TTS -> ASR round trip ────────────────────────────────
try:
    ensure_wav
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

wav = ensure_wav("tts_en.wav", EN_TEXT)
print("Transcribing... (first run downloads an ASR model — a few minutes)")
with open(wav, "rb") as f:
    r = requests.post(f"{BASE}/transcribe",
                      files={"audio": ("tts_en.wav", f, "audio/wav")},
                      timeout=1800)
if r.status_code != 200:
    api_fail(r, "POST /transcribe")
res = r.json()
print(f"Engine: {res.get('engine')}  "
      f"(audio {res.get('duration_s')}s, transcribed in {res.get('transcription_time_s')}s)")
print(f"  TTS input : {EN_TEXT}")
print(f"  ASR output: {res.get('text', '').strip()}")

14. AI-watermark detection

Every WAV VoiceStudio generates carries an inaudible AudioSeal watermark by default, so AI-generated audio can be identified later. POST /watermark/detect on a generated clip should report high confidence; the same check on a plain ffmpeg-made tone should not. Expected runtime: seconds (first call loads the small detector model).

In [ ]:

# ── 14. AI watermark: detect generated vs. plain audio ──────────────────────
import subprocess

try:
    ensure_wav
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

status = requests.get(f"{BASE}/watermark/status", timeout=30)
if status.status_code != 200:
    api_fail(status, "GET /watermark/status")
print("Watermark status:", status.json())

def detect(path):
    with open(path, "rb") as f:
        r = requests.post(f"{BASE}/watermark/detect",
                          files={"file": (os.path.basename(path), f, "audio/wav")},
                          timeout=300)
    if r.status_code != 200:
        api_fail(r, "POST /watermark/detect")
    return r.json()

generated = ensure_wav("tts_en.wav", EN_TEXT)
res = detect(generated)
print(f"\nVoiceStudio-generated clip: watermarked={res.get('is_watermarked')} "
      f"confidence={res.get('confidence')}")

plain = os.path.join(DEMO_DIR, "plain_tone.wav")
if not os.path.exists(plain):
    subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                    "-f", "lavfi", "-i", "sine=frequency=440:duration=3", plain],
                   check=True)
res = detect(plain)
print(f"Plain (non-AI) tone:      watermarked={res.get('is_watermarked')} "
      f"confidence={res.get('confidence')}")

15. OpenAI-compatible API

Anything that speaks OpenAI's audio API works against http://127.0.0.1:3900/v1 unchanged — here the official openai Python client does TTS (/v1/audio/speech) and STT (/v1/audio/transcriptions) against the local engine. No key is checked; voice also accepts your profile ids from cell 12. Expected runtime: seconds (warm models).

In [ ]:

# ── 15. OpenAI-compatible API with the official openai client ───────────────
import subprocess
import sys

try:
    ensure_wav
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

check_backend()
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "openai"], check=True)
from openai import OpenAI

client = OpenAI(base_url=f"{BASE}/v1", api_key="none")  # any string — nothing checks it

speech = client.audio.speech.create(
    model="tts-1", voice="alloy", response_format="wav",
    input="This request thinks it is talking to OpenAI. It is not.",
)
oa_path = os.path.join(DEMO_DIR, "openai_compat.wav")
with open(oa_path, "wb") as f:
    f.write(speech.read())
play(oa_path, "POST /v1/audio/speech ->")

with open(oa_path, "rb") as f:
    tr = client.audio.transcriptions.create(model="whisper-1", file=f)
print("POST /v1/audio/transcriptions ->", tr.text.strip())

16. Multi-voice story

Two saved profiles perform a four-line scene through POST /longform/render — the Stories Editor's backend: per-span voices, inter-line pauses, and a single mixed MP3 out, with SSE progress along the way. Requires the profiles from cell 12. Expected runtime: ~1-2 min on a warm T4 (four TTS spans + encode).

In [ ]:

# ── 16. Multi-voice story via /longform/render ──────────────────────────────
import json as _json

try:
    NARRATOR_ID, GUEST_ID
except NameError:
    raise SystemExit("Run cell 12 (voice profiles) first — this cell reuses its two profiles.")

payload = {
    "default_voice": NARRATOR_ID,
    "format": "mp3",
    "chapters": [{
        "title": "A very short scene",
        "spans": [
            {"voice_id": NARRATOR_ID, "text": "Have you noticed we are running inside a notebook?", "pause_ms_after": 300},
            {"voice_id": GUEST_ID, "text": "I have. And nobody uploaded a single voice actor.", "pause_ms_after": 300},
            {"voice_id": NARRATOR_ID, "text": "Two profiles, four lines, one rendered story.", "pause_ms_after": 300},
            {"voice_id": GUEST_ID, "text": "Roll the credits.", "pause_ms_after": 0},
        ],
    }],
}

check_backend()
out_name = None
with requests.post(f"{BASE}/longform/render", json=payload,
                   stream=True, timeout=(10, 1800)) as r:
    if r.status_code != 200:
        api_fail(r, "POST /longform/render")
    for line in r.iter_lines(decode_unicode=True):
        if not line or not line.startswith("data:"):
            continue
        evt = _json.loads(line[len("data:"):].strip())
        etype = evt.get("type")
        if etype == "error":
            raise SystemExit(f"Story render failed: {evt}\n{LOG_HINT}")
        if etype == "done":
            out_name = evt["output"]
            print(f"Done: {evt.get('chapters')} chapter(s), {evt.get('duration_s')}s of audio")
        else:
            print(f"  [{etype}] {evt.get('title') or evt.get('text', '')!s:.60}")

if not out_name:
    raise SystemExit(f"Stream ended without a 'done' event. {LOG_HINT}")
story_path = os.path.join("/content/omnivoice_data/outputs", out_name)
play(story_path, f"Rendered story ({story_path}):")

17. Audiobook (chaptered m4b)

POST /audiobook takes a plain-text script — # Heading starts a chapter, [pause 400ms] inserts silence — and renders a chapterized, tagged .m4b with SSE progress (the same renderer handles resume after interruptions). Narrated by the cell-12 profile. Expected runtime: ~1-2 min on a warm T4 for this two-chapter miniature.

In [ ]:

# ── 17. Audiobook: script -> chaptered m4b ──────────────────────────────────
import base64
import json as _json

from IPython.display import HTML

try:
    NARRATOR_ID
except NameError:
    raise SystemExit("Run cell 12 (voice profiles) first — this cell narrates with its profile.")

SCRIPT = """# Chapter One: The Setup
It began, as most experiments do, with a borrowed graphics card. [pause 400ms] Nothing was installed, and nothing was certain.

# Chapter Two: The Payoff
Two chapters later, the machine could read aloud. [pause 400ms] The end.
"""

payload = {
    "text": SCRIPT,
    "default_voice": NARRATOR_ID,
    "format": "m4b",
    "metadata": {"title": "The Colab Miniature", "author": "VoiceStudio",
                 "narrator": "Colab Narrator"},
}

check_backend()
done = None
with requests.post(f"{BASE}/audiobook", json=payload,
                   stream=True, timeout=(10, 1800)) as r:
    if r.status_code != 200:
        api_fail(r, "POST /audiobook")
    for line in r.iter_lines(decode_unicode=True):
        if not line or not line.startswith("data:"):
            continue
        evt = _json.loads(line[len("data:"):].strip())
        if evt.get("type") == "error":
            raise SystemExit(f"Audiobook render failed: {evt}\n{LOG_HINT}")
        if evt.get("type") == "done":
            done = evt
        else:
            print(f"  [{evt.get('type')}] {evt.get('title') or ''}")

if not done:
    raise SystemExit(f"Stream ended without a 'done' event. {LOG_HINT}")
book_path = os.path.join("/content/omnivoice_data/outputs", done["output"])
print(f"Done: {done.get('chapters')} chapters, {done.get('duration_s')}s -> {book_path}")

# Inline playback: IPython's Audio widget doesn't know .m4b, so embed the
# (small) file as an HTML5 audio tag — Chrome plays AAC-in-MP4 natively.
b64 = base64.b64encode(open(book_path, "rb").read()).decode("ascii")
display(HTML(f'<audio controls src="data:audio/mp4;base64,{b64}"></audio>'))
print("Chapter markers show up in audiobook players (Apple Books, etc.) — "
      "download the file from the Colab file browser to try it.")

18. Video dubbing — heavy, optional

The flagship pipeline, kept honest and miniature: a 6-second synthetic clip (color frame + the cell-9 English narration) is dubbed into Spanish — upload → prep (audio extract + Demucs vocal separation) → transcribe → translate → voice-cloned TTS → mux — all through the same job API the app uses.

Run this cell only if you have 5-15 minutes: the first run downloads the Demucs separation model and (if cell 13 didn't run) an ASR model. Translation here uses the free Google web endpoint via deep-translator (installed in-cell); for a fully offline dub the backend also supports provider="nllb" (a ~2.5 GB one-time model download).

In [ ]:

# ── 18. Video dubbing (mini): English clip -> Spanish dub ───────────────────
import subprocess
import sys
import time

try:
    ensure_wav
except NameError:
    raise SystemExit("Run cell 8 (feature-tour helpers) first.")

from IPython.display import Video

check_backend()
TARGET_LANG, TARGET_LABEL = "es", "Spanish"

# deep-translator backs the default "google" translation provider (free web
# endpoint, no key). Offline alternative: provider="nllb" below (~2.5 GB model).
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "deep-translator"], check=True)

# 18a. Build a tiny source video: a colored frame carrying the EN narration.
src_wav = ensure_wav("tts_en.wav", EN_TEXT)
src_mp4 = os.path.join(DEMO_DIR, "dub_source.mp4")
if not os.path.exists(src_mp4):
    subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                    "-f", "lavfi", "-i", "color=c=steelblue:s=640x360:r=25",
                    "-i", src_wav, "-shortest",
                    "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac",
                    src_mp4], check=True)
print(f"Source clip: {src_mp4} ({os.path.getsize(src_mp4)} bytes)")

def poll_task(task_id, what, timeout_s=1200):
    """Poll the persisted job row until the background task finishes."""
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        r = requests.get(f"{BASE}/jobs/{task_id}", timeout=30)
        status = r.json().get("status") if r.status_code == 200 else "pending"
        if status == "done":
            print(f" {what}: done")
            return
        if status in ("failed", "cancelled"):
            raise SystemExit(f"{what} {status}: {r.text[:1500]}\n{LOG_HINT}")
        print(".", end="", flush=True)
        time.sleep(5)
    raise SystemExit(f"{what} timed out after {timeout_s}s. {LOG_HINT}")

# 18b. Upload -> prep task (audio extract + Demucs separation; model download
#      on first run).
with open(src_mp4, "rb") as f:
    r = requests.post(f"{BASE}/dub/upload",
                      files={"video": ("dub_source.mp4", f, "video/mp4")},
                      data={"input_type": "video"}, timeout=300)
if r.status_code != 202:
    api_fail(r, "POST /dub/upload")
job_id, prep_task = r.json()["job_id"], r.json()["task_id"]
print(f"Dub job {job_id} — preparing (Demucs separation; first run downloads its model)")
poll_task(prep_task, "prep")

# 18c. Transcribe the source audio (synchronous endpoint).
print("Transcribing source audio...")
r = requests.post(f"{BASE}/dub/transcribe/{job_id}", timeout=1800)
if r.status_code != 200:
    api_fail(r, "POST /dub/transcribe")
segments = r.json()["segments"]
for s in segments:
    print(f"  [{s['start']:.2f}-{s['end']:.2f}s] {s['text']}")

# 18d. Translate the segments.
r = requests.post(f"{BASE}/dub/translate", json={
    "job_id": job_id,
    "target_lang": TARGET_LANG,
    "provider": "google",   # or "nllb" for the fully offline translator
    "segments": [{"id": str(s["id"]), "text": s["text"],
                  "start": s["start"], "end": s["end"]} for s in segments],
}, timeout=600)
if r.status_code != 200:
    api_fail(r, "POST /dub/translate")
translated = {t["id"]: t["text"] for t in r.json()["translated"]}
for sid, text in translated.items():
    print(f"  {TARGET_LANG} #{sid}: {text}")

# 18e. Generate the dubbed track — TTS voice-cloned from the source speaker.
r = requests.post(f"{BASE}/dub/generate/{job_id}", json={
    "language": TARGET_LABEL,
    "language_code": TARGET_LANG,
    "segments": [{"start": s["start"], "end": s["end"],
                  "text": translated.get(str(s["id"]), s["text"])}
                 for s in segments],
}, timeout=120)
if r.status_code != 200:
    api_fail(r, "POST /dub/generate")
print(f"Rendering dubbed track (task {r.json()['task_id']})")
poll_task(r.json()["task_id"], "dub generate")

# 18f. Mux and fetch the dubbed video.
r = requests.get(f"{BASE}/dub/download/{job_id}",
                 params={"include_tracks": TARGET_LANG, "default_track": TARGET_LANG},
                 timeout=1800)
if r.status_code != 200:
    api_fail(r, "GET /dub/download")
dub_mp4 = os.path.join(DEMO_DIR, f"dubbed_{TARGET_LANG}.mp4")
with open(dub_mp4, "wb") as f:
    f.write(r.content)
print(f"Dubbed video: {dub_mp4} ({os.path.getsize(dub_mp4)} bytes)")
display(Video(dub_mp4, embed=True, width=480))
DUB_JOB_ID = job_id  # cell 19 (vocal isolation) reuses this job

19. Vocal isolation (stems)

Dub prep already ran Demucs source separation, so the stems exist as job artifacts: GET /dub/export-stems/{job_id} returns the dubbed vocals and the original background as a zip, played here separately. Requires cell 18's job. (There is no job-free standalone isolation endpoint — separation rides the dub pipeline, so this cell does too.) Expected runtime: seconds.

In [ ]:

# ── 19. Vocal isolation: fetch and play the separated stems ─────────────────
import io
import zipfile

try:
    DUB_JOB_ID
except NameError:
    raise SystemExit("Run cell 18 (video dubbing) first — the stems are artifacts of its job.")

r = requests.get(f"{BASE}/dub/export-stems/{DUB_JOB_ID}", timeout=300)
if r.status_code != 200:
    api_fail(r, "GET /dub/export-stems")

stems_dir = os.path.join(DEMO_DIR, "stems")
os.makedirs(stems_dir, exist_ok=True)
with zipfile.ZipFile(io.BytesIO(r.content)) as zf:
    names = zf.namelist()
    zf.extractall(stems_dir)
print("Stems:", names)
for name in names:
    play(os.path.join(stems_dir, name), name)

20. Everything you just generated

A closing inventory: every artifact the tour produced, with paths and sizes. Remember Colab storage is ephemeral — download anything you want to keep (file browser in the left sidebar), or mount Drive as shown in the troubleshooting section.

In [ ]:

# ── 20. Tour summary: artifacts on disk ─────────────────────────────────────
import os

def _ls(root, label):
    print(f"\n{label} ({root})")
    if not os.path.isdir(root):
        print("  (nothing here — the cells that write to it were not run)")
        return
    for dirpath, _dirs, files in sorted(os.walk(root)):
        for name in sorted(files):
            p = os.path.join(dirpath, name)
            print(f"  {os.path.getsize(p):>10,} B  {os.path.relpath(p, root)}")

_ls("/content/omnivoice_demos", "Feature-tour artifacts")
_ls("/content/omnivoice_data/outputs", "Backend outputs (history, stories, audiobooks)")

try:
    r = requests.get(f"{BASE}/profiles", timeout=15)
    if r.status_code == 200:
        print(f"\nSaved voice profiles: "
              f"{', '.join(p['name'] for p in r.json()) or '(none)'}")
except Exception:
    pass  # backend may already be stopped — the file listing above still stands

Part 3 — Troubleshooting

"device": "cpu" in /health, or generation is crawling — the runtime has no GPU. Runtime → Change runtime type → T4 GPU, then Runtime → Run all. Colab's free tier sometimes has no GPUs available; trying again later usually works.

First generation takes minutes — that's the default model (a few GB) downloading and loading. Run cell 4 to front-load the download; after the first generation the model stays warm and requests take seconds.

Session limits (free tier) — Colab disconnects idle sessions and caps total runtime (up to ~12 h, often less). When the VM is recycled, everything under /content is deleted: the repo, models, and your omnivoice_data (voices, projects, generated audio). To keep your data across sessions, mount Google Drive and point the data dir at it before running cell 5:

from google.colab import drive
drive.mount('/content/drive')
import os
os.environ['OMNIVOICE_DATA_DIR'] = '/content/drive/MyDrive/omnivoice_data'

VRAM — a T4 has 16 GB: plenty for TTS, cloning, and voice design. Full dubbing pipelines (separation + transcription + diarization + TTS) run closer to the limit; the app evicts idle engines automatically, but very long videos may need shorter chunks.

UI tab is blank or errors out — re-run cell 5 (it verifies /health without restarting a healthy backend), then cell 6. Pop-ups must be allowed for colab.research.google.com.

Feature-tour cells fail with NameError / "Run cell 8 first" — the tour's helper functions live in cell 8; run it once after every kernel restart.

Backend logs — /content/omnivoice_backend.log. The launch, smoke-test, and feature cells print the relevant tail or point here on failure; attach the full file when reporting an issue.

Reporting issues — this notebook is maintained in the main repo: debpalash/VoiceStudio/issues.