Blame for speak.py

#!/usr/bin/env python3
"""
speak.py
Generate speech in Brian's voice from text, using Chatterbox (MIT-licensed, runs locally).

Long scripts are split into sentence-sized pieces, each piece is generated in the voice of
the reference clip, and the pieces are joined with natural pauses into one WAV file.
A settings file is saved next to every output so any take can be reproduced exactly.

Examples:
  python speak.py "Hey guys, welcome back to the Reptarium." --ref brian_voice_refs\\ref_01.wav
  python speak.py --file script.txt --ref brian_voice_refs\\ref_01.wav --takes 3
  python speak.py --interactive --ref brian_voice_refs\\ref_01.wav
  python speak.py "Hey guys." --voice voices\\brian.pt     (profile built from all clips)

Every output carries Resemble AI's inaudible Perth watermark, which marks it as AI-generated.
"""

import os
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1")  # Windows: avoid symlink errors
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS", "1")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

import argparse
import json
import random
import re
import sys
import time
import warnings
from datetime import datetime
from pathlib import Path

warnings.filterwarnings("ignore")  # the libraries print a lot of harmless deprecation noise

import numpy as np
import soundfile as sf


# ---------------------------------------------------------------- text splitting

def split_text(text, max_chars=280):
    """Split a script into chunks Chatterbox handles well.

    Returns a list of (chunk, pause_after) where pause_after is "sentence" or "paragraph".
    Chatterbox gets unreliable on very long inputs, so long sentences are split at commas,
    then at word boundaries as a last resort."""
    paragraphs = [p.strip() for p in re.split(r"\n\s*\n", text.strip()) if p.strip()]
    out = []
    for para in paragraphs:
        para = re.sub(r"\s+", " ", para)
        sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", para) if s.strip()]
        pieces = []
        for s in sentences:
            if len(s) <= max_chars:
                pieces.append(s)
                continue
            cur = ""
            for part in re.split(r"(?<=[,;:])\s+", s):
                while len(part) > max_chars:  # no commas to break on: cut at a word boundary
                    cut = part.rfind(" ", 0, max_chars)
                    cut = cut if cut > 0 else max_chars
                    if cur:
                        pieces.append(cur)
                        cur = ""
                    pieces.append(part[:cut].strip())
                    part = part[cut:].strip()
                if cur and len(cur) + 1 + len(part) > max_chars:
                    pieces.append(cur)
                    cur = part
                else:
                    cur = f"{cur} {part}".strip()
            if cur:
                pieces.append(cur)
        # Join very short sentences ("Look at this. Wow.") so they flow naturally.
        merged = []
        for piece in pieces:
            if merged and len(merged[-1]) < 60 and len(merged[-1]) + 1 + len(piece) <= max_chars:
                merged[-1] = f"{merged[-1]} {piece}"
            else:
                merged.append(piece)
        for i, m in enumerate(merged):
            out.append((m, "paragraph" if i == len(merged) - 1 else "sentence"))
    return out


# ---------------------------------------------------------------- model

def pick_device(choice):
    import torch
    if choice != "auto":
        return choice
    if torch.cuda.is_available():
        return "cuda"
    if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available():
        return "mps"
    return "cpu"


def load_model(kind, device):
    print(f"Loading Chatterbox ({kind}) on {device}. First run downloads the model (~2-3 GB)...")
    t = time.time()
    if kind == "turbo":
        from chatterbox.tts_turbo import ChatterboxTurboTTS
        model = ChatterboxTurboTTS.from_pretrained(device=device)
        # Fix: Turbo's loudness step returns 64-bit audio with current numpy, which crashes
        # the reference processing ("expected scalar type Double but found Float").
        orig = model.norm_loudness
        model.norm_loudness = lambda wav, sr: np.asarray(orig(wav, sr), dtype=np.float32)
    else:
        from chatterbox.tts import ChatterboxTTS
        model = ChatterboxTTS.from_pretrained(device=device)
    print(f"Model ready in {time.time() - t:.0f}s.")
    return model


def check_ref(path, kind):
    if not path.exists():
        sys.exit(f"Reference clip not found: {path}")
    info = sf.info(str(path))
    if info.duration < 5.5:
        sys.exit(f"{path.name} is {info.duration:.1f}s. Use a reference clip of 7-15 seconds.")
    if info.duration > 20:
        print(f"Note: {path.name} is {info.duration:.0f}s; only about the first 10-15s is used.")


def load_voice_profile(model, kind, path, device):
    """Load a voice profile made by build_voice.py (identity averaged over many clips)."""
    if not path.exists():
        sys.exit(f"Voice profile not found: {path}")
    meta = path.with_suffix(".json")
    if meta.exists():
        made_for = json.loads(meta.read_text(encoding="utf-8")).get("model", kind)
        if made_for != kind:
            sys.exit(f"{path.name} was built for the {made_for} model. "
                     f"Add --model {made_for}, or rebuild it with --model {kind}.")
    if kind == "turbo":
        from chatterbox.tts_turbo import Conditionals
    else:
        from chatterbox.tts import Conditionals
    model.conds = Conditionals.load(str(path), map_location="cpu").to(device)


def prepare_voice(model, kind, ref, exaggeration):
    if kind == "turbo":
        model.prepare_conditionals(str(ref), exaggeration=0.0, norm_loudness=True)
    else:
        model.prepare_conditionals(str(ref), exaggeration=exaggeration)


def generate_piece(model, kind, text, args):
    if kind == "turbo":
        wav = model.generate(text, temperature=args.temperature)
    else:
        wav = model.generate(text, exaggeration=args.exaggeration,
                             cfg_weight=args.cfg, temperature=args.temperature)
    return wav.squeeze().detach().cpu().numpy().astype(np.float32)


def seed_everything(seed):
    import torch
    random.seed(seed)
    np.random.seed(seed % (2 ** 32))
    torch.manual_seed(seed)


# ---------------------------------------------------------------- audio assembly

def trim(wav, sr, top_db=45, keep=0.04):
    """Trim silence at the ends of a piece, leaving a few ms so words aren't clipped."""
    import librosa
    if len(wav) < sr // 10:
        return wav
    _, (start, end) = librosa.effects.trim(wav, top_db=top_db)
    pad = int(sr * keep)
    return wav[max(0, start - pad): min(len(wav), end + pad)]


def assemble(pieces, sr, pause, para_pause):
    parts = []
    for wav, kind in pieces:
        parts.append(wav)
        gap = para_pause if kind == "paragraph" else pause
        parts.append(np.zeros(int(sr * gap), dtype=np.float32))
    audio = np.concatenate(parts[:-1]) if len(parts) > 1 else parts[0]
    peak = float(np.max(np.abs(audio))) if audio.size else 0.0
    if peak > 0:
        audio = audio * (0.89 / peak)  # about -1 dBFS, same level as the dataset clips
    return audio


# ---------------------------------------------------------------- main

def render(model, kind, text, out_path, args, seed):
    chunks = split_text(text, args.max_chars)
    if not chunks:
        print("Nothing to say (empty text).")
        return None
    seed_everything(seed)
    pieces, t0 = [], time.time()
    for i, (chunk, pause_kind) in enumerate(chunks, 1):
        t = time.time()
        print(f"  [{i}/{len(chunks)}] {chunk[:80]}{'...' if len(chunk) > 80 else ''}")
        wav = generate_piece(model, kind, chunk, args)
        pieces.append((trim(wav, model.sr), pause_kind))
        took = time.time() - t
        print(f"      {len(wav) / model.sr:.1f}s of audio in {took:.0f}s")
    audio = assemble(pieces, model.sr, args.pause, args.para_pause)

    out_path.parent.mkdir(parents=True, exist_ok=True)
    sf.write(str(out_path), audio, model.sr, subtype="PCM_16")
    settings = {
        "text": text, "reference": str(args.ref) if args.ref else None,
        "voice_profile": str(args.voice) if args.voice else None, "model": kind, "seed": seed,
        "exaggeration": args.exaggeration, "cfg": args.cfg, "temperature": args.temperature,
        "pause": args.pause, "para_pause": args.para_pause, "max_chars": args.max_chars,
        "created": datetime.now().isoformat(timespec="seconds"),
        "note": "AI-generated voice (Chatterbox, Perth watermark).",
    }
    out_path.with_suffix(".json").write_text(json.dumps(settings, indent=2), encoding="utf-8")
    print(f"Saved {out_path}  ({len(audio) / model.sr:.1f}s, made in {time.time() - t0:.0f}s, "
          f"seed {seed})")
    return out_path


def main():
    p = argparse.ArgumentParser(description="Generate speech in a cloned voice (Chatterbox).")
    p.add_argument("text", nargs="?", help="Text to speak (or use --file / --interactive)")
    p.add_argument("--file", type=Path, help="Text file with the script (blank line = paragraph)")
    p.add_argument("--interactive", action="store_true",
                   help="Keep the model loaded and type lines one after another")
    p.add_argument("--ref", type=Path, help="Reference clip of Brian (7-15 s WAV)")
    p.add_argument("--voice", type=Path,
                   help="Voice profile from build_voice.py (built from all clips), instead of --ref")
    p.add_argument("--out", type=Path, help="Output WAV (default: output\\brian_<time>.wav)")
    p.add_argument("--model", choices=["standard", "turbo"], default="standard",
                   help="standard: best control and quality. turbo: faster, fewer controls")
    p.add_argument("--exaggeration", type=float, default=0.5,
                   help="Emotion/energy, 0.25 calm to 1.0 very animated (standard model only)")
    p.add_argument("--cfg", type=float, default=0.5,
                   help="Pacing: lower (0.3) = slower, more deliberate; higher = faster (standard only)")
    p.add_argument("--temperature", type=float, default=0.8,
                   help="Variation between takes, 0.5 steady to 1.0 lively")
    p.add_argument("--seed", type=int, help="Repeat an exact take (seed is in each .json file)")
    p.add_argument("--takes", type=int, default=1, help="Make several versions to choose from")
    p.add_argument("--pause", type=float, default=0.25, help="Seconds between sentences")
    p.add_argument("--para-pause", type=float, default=0.7, help="Seconds between paragraphs")
    p.add_argument("--max-chars", type=int, default=280, help="Longest piece sent to the model")
    p.add_argument("--device", default="auto", help="auto, cpu, cuda")
    args = p.parse_args()

    if not (args.text or args.file or args.interactive):
        p.error("give some text, --file script.txt, or --interactive")
    if bool(args.ref) == bool(args.voice):
        p.error("use either --ref clip.wav or --voice profile.pt (one of them)")
    text = args.file.read_text(encoding="utf-8") if args.file else args.text
    if args.ref:
        check_ref(args.ref, args.model)
    else:
        if not args.voice.exists():
            sys.exit(f"Voice profile not found: {args.voice}")
        meta = args.voice.with_suffix(".json")
        if meta.exists():
            made_for = json.loads(meta.read_text(encoding="utf-8")).get("model", args.model)
            if made_for != args.model:
                sys.exit(f"{args.voice.name} was built for the {made_for} model. "
                         f"Add --model {made_for}, or rebuild it with --model {args.model}.")

    device = pick_device(args.device)
    if device == "cpu":
        print("Running on CPU: expect roughly 3-8 seconds of processing per second of speech.")
    model = load_model(args.model, device)
    if args.voice:
        print(f"Loading voice profile {args.voice.name}...")
        load_voice_profile(model, args.model, args.voice, device)
    else:
        print(f"Learning the voice from {args.ref.name}...")
        prepare_voice(model, args.model, args.ref, args.exaggeration)

    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    base_seed = args.seed if args.seed is not None else random.randint(0, 2 ** 31 - 1)

    if args.interactive:
        print("\nType a line and press Enter to generate it. Empty line to quit.")
        n = 0
        while True:
            try:
                line = input("\n> ").strip()
            except (EOFError, KeyboardInterrupt):
                break
            if not line:
                break
            n += 1
            render(model, args.model, line, Path("output") / f"line_{stamp}_{n:03d}.wav",
                   args, base_seed + n)
        return

    out = args.out or Path("output") / f"brian_{stamp}.wav"
    for k in range(args.takes):
        path = out if args.takes == 1 else out.with_name(f"{out.stem}_take{k + 1}{out.suffix}")
        if args.takes > 1:
            print(f"\nTake {k + 1} of {args.takes}")
        render(model, args.model, text, path, args, base_seed + k)


if __name__ == "__main__":
    main()