Explosion_der_Spiele/scripts/tools/generate_voice_dialogs.py

#!/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())