#!/usr/bin/env python3 """Generate high-quality Piper TTS for dialog_* / lang_voice_* from locales. Uses large/high Piper models from data/voice_cast.json. Until real voice takes exist, run --watch to auto-rebuild when scripts change. Usage: python3 scripts/tools/generate_voice_dialogs.py --force python3 scripts/tools/generate_voice_dialogs.py --watch python3 scripts/tools/generate_voice_dialogs.py --lang de --force """ from __future__ import annotations import argparse import hashlib import json import os import re import subprocess import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parents[2] CAST_PATH = ROOT / "data" / "voice_cast.json" LOCALES = ROOT / "data" / "locales" OUT_ROOT = ROOT / "assets" / "audio" / "voice" MANIFEST = OUT_ROOT / "_manifest.json" PIPER = os.path.expanduser("~/.local/bin/piper") DEFAULT_VOICES = Path(os.path.expanduser("~/.local/share/piper/voices")) STAGE_DIR = re.compile(r"\*[^*]+\*") EMOJI_ETC = re.compile(r"[⬜]+") def clean_tts(text: str) -> str: text = STAGE_DIR.sub(" ", text) text = EMOJI_ETC.sub(" ", text) text = text.replace("…", "...") text = text.replace("—", " - ") text = text.replace("–", " - ") text = re.sub(r"\s+", " ", text).strip() return text def load_cast() -> dict: return json.loads(CAST_PATH.read_text(encoding="utf-8")) def voices_dir(cast: dict) -> Path: return Path(os.path.expanduser(cast.get("voices_dir", str(DEFAULT_VOICES)))) def resolve_model(cast: dict, model_key: str) -> Path: models = cast["models"] if model_key not in models: raise KeyError(f"unknown model key {model_key}") path = voices_dir(cast) / models[model_key] return path def npc_params(cast: dict, npc_id: str, lang: str) -> dict: npc = cast["npcs"][npc_id] # New format: per-lang dict; legacy: flat speaker ints entry = npc.get(lang, npc) if isinstance(entry, dict) and ("model" in entry or "speaker" in entry): params = dict(entry) else: defaults = cast.get("lang_defaults", {}) params = { "model": defaults.get(lang, "de_high"), "speaker": int(entry) if not isinstance(entry, dict) else 0, "length_scale": float(npc.get("length_scale", 1.0)), } params.setdefault("speaker", 0) params.setdefault("length_scale", 1.0) params.setdefault("noise_scale", 0.667) params.setdefault("noise_w", 0.8) params.setdefault("sentence_silence", 0.35) return params def fingerprint(text: str, params: dict, model_path: Path) -> str: payload = { "text": text, "model": model_path.name, "speaker": int(params.get("speaker", 0)), "length_scale": float(params.get("length_scale", 1.0)), "noise_scale": float(params.get("noise_scale", 0.667)), "noise_w": float(params.get("noise_w", 0.8)), "sentence_silence": float(params.get("sentence_silence", 0.35)), } raw = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16] def synthesize(model: Path, text: str, out_wav: Path, params: dict) -> None: out_wav.parent.mkdir(parents=True, exist_ok=True) tmp = out_wav.with_suffix(".tmp.wav") cmd = [ PIPER, "-m", str(model), "-s", str(int(params.get("speaker", 0))), "--length_scale", str(float(params.get("length_scale", 1.0))), "--noise_scale", str(float(params.get("noise_scale", 0.667))), "--noise_w", str(float(params.get("noise_w", 0.8))), "--sentence_silence", str(float(params.get("sentence_silence", 0.35))), "-f", str(tmp), ] proc = subprocess.run(cmd, input=text.encode("utf-8"), capture_output=True) if proc.returncode != 0 or not tmp.exists() or tmp.stat().st_size < 44: err = proc.stderr.decode("utf-8", errors="replace") if tmp.exists(): tmp.unlink(missing_ok=True) raise RuntimeError(f"piper failed for {out_wav.name}: {err[-500:]}") tmp.replace(out_wav) def load_manifest() -> dict: if MANIFEST.exists(): try: return json.loads(MANIFEST.read_text(encoding="utf-8")) except json.JSONDecodeError: return {} return {} def save_manifest(data: dict) -> None: OUT_ROOT.mkdir(parents=True, exist_ok=True) MANIFEST.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def generate(force: bool = False, langs: list[str] | None = None) -> tuple[int, int]: if not Path(PIPER).exists(): raise SystemExit(f"piper not found at {PIPER}") cast = load_cast() lines = cast["lines"] all_langs = langs or ["de", "en", "by"] locale_data = {} for lang in all_langs: locale_data[lang] = json.loads((LOCALES / f"{lang}.json").read_text(encoding="utf-8")) manifest = {} if force else load_manifest() done = 0 skipped = 0 for lang in all_langs: print(f"== {lang}", flush=True) for key, npc_id in lines.items(): text_raw = locale_data[lang].get(key) if not text_raw: print(f" skip missing {key}") skipped += 1 continue tts = clean_tts(str(text_raw)) if not tts: skipped += 1 continue params = npc_params(cast, npc_id, lang) model_path = resolve_model(cast, str(params["model"])) if not model_path.exists(): raise SystemExit(f"MISSING model {model_path}") fp = fingerprint(tts, params, model_path) out = OUT_ROOT / lang / f"{key}.wav" mkey = f"{lang}/{key}.wav" if ( not force and out.exists() and out.stat().st_size > 1000 and manifest.get(mkey) == fp ): skipped += 1 continue print(f" {key} npc={npc_id} model={model_path.name} spk={params.get('speaker')}") synthesize(model_path, tts, out, params) manifest[mkey] = fp done += 1 save_manifest(manifest) print(f"generated={done} skipped={skipped} out={OUT_ROOT}", flush=True) return done, skipped def watch_sources_mtime() -> float: mtimes = [CAST_PATH.stat().st_mtime] for p in LOCALES.glob("*.json"): mtimes.append(p.stat().st_mtime) return max(mtimes) def watch_loop(interval: float = 2.0) -> int: print(f"watching {LOCALES} + {CAST_PATH} (interval={interval}s) — Ctrl+C to stop", flush=True) print("placeholder TTS only; replace WAVs later with real takes (same paths).", flush=True) last = 0.0 # initial sync generate(force=False) last = watch_sources_mtime() try: while True: time.sleep(interval) now = watch_sources_mtime() if now > last: print("\n[watch] locale/cast changed — regenerating…") generate(force=False) last = watch_sources_mtime() except KeyboardInterrupt: print("\n[watch] stopped") return 0 def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--force", action="store_true", help="Rebuild all WAVs") ap.add_argument("--lang", action="append", help="Only these langs (de/en/by)") ap.add_argument("--watch", action="store_true", help="Auto-regenerate when locales/cast change") ap.add_argument("--interval", type=float, default=2.0, help="Watch poll seconds") args = ap.parse_args() if args.watch: return watch_loop(args.interval) generate(force=args.force, langs=args.lang) return 0 if __name__ == "__main__": raise SystemExit(main())