384 lines
14 KiB
Python
384 lines
14 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Render the sample chapter in every language Qwen3-TTS supports.
|
|
|
|
Reads the translation JSONs produced by ``build_sample_translations.py`` and
|
|
renders each utterance with the speaker's own voice, so the reviewer can hear
|
|
whether an identity survives the language change.
|
|
|
|
Output: ``<project>/Lang_test/<LANG>/uttNN_<Speaker>.wav`` plus a stitched
|
|
per-language file, which is what you actually listen to.
|
|
|
|
Qwen3-TTS auto-detects language from the text - ``/v1/tts`` takes no language
|
|
argument - so there is nothing to set per clip.
|
|
|
|
Also writes ``_ALL_LANGUAGES_full_sample.wav`` - every language back to back,
|
|
in the operator's list order with English first - for listening to the whole
|
|
set in one pass.
|
|
|
|
Voice is supplied two ways:
|
|
|
|
``--voice-mode ref`` (default) sends ``ref_audio`` + ``ref_text`` and lets the
|
|
model build its own prompt. ``--voice-mode prompt`` uses the cached ICL prompt,
|
|
which is more faithful but is **model-specific**: a prompt built by the 1.7B
|
|
model has a ref_code whose dimensions the 0.6B model rejects, with
|
|
"Expected size 1024 but got size 2048". Cached prompts are therefore only safe
|
|
when rendering with the model that built them.
|
|
|
|
/v1/tts returns raw WAV bytes, not a JSON envelope.
|
|
|
|
Usage:
|
|
python scripts/render_lang_test.py --project DealHunterShip_4.0e_whole-to-audio
|
|
python scripts/render_lang_test.py --languages EN FR --timeout 600
|
|
python scripts/render_lang_test.py --dry-run
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from umbraaudio.config import load_config # noqa: E402
|
|
|
|
TTS_URL = "http://127.0.0.1:8881"
|
|
|
|
# Qwen3-TTS language names, from the model's own codec_language_id keys.
|
|
LANG_CODES = {
|
|
"EN": "english", "ZH": "chinese", "JA": "japanese", "KO": "korean",
|
|
"DE": "german", "ES": "spanish", "FR": "french", "RU": "russian",
|
|
"IT": "italian", "PT": "portuguese",
|
|
}
|
|
|
|
# Chapter 1 of Deal Hunter is the voice-introduction sample.
|
|
CHAPTER_ID = 1
|
|
TRANSLATION_DIRNAME = "Lang_test"
|
|
|
|
# The reference WAVs were generated from this text, so it is the correct
|
|
# ref_text for the clone path. A wrong transcript degrades the clone.
|
|
from umbraaudio.voices.qwen_tts import VOICE_REFERENCE_TEXT as REF_TEXT # noqa: E402
|
|
|
|
|
|
def slugify(name: str) -> str:
|
|
return name.strip().lower().replace(" ", "_")
|
|
|
|
|
|
def build_translation_paths(config) -> Path:
|
|
"""Where build_sample_translations.py put them."""
|
|
return config.project_dir / "samples" / "translations"
|
|
|
|
|
|
def speaker_refs(data_dir: Path) -> dict[str, Path]:
|
|
voices = data_dir / "voices"
|
|
return {p.name[: -len("_ref.wav")]: p for p in voices.glob("*_ref.wav")}
|
|
|
|
|
|
# How the voice page's accent is phrased when handed to the model. The
|
|
# language parameter sets phonology; this sets accent character, which is how
|
|
# Deal Hunter stays Australian while speaking French.
|
|
ACCENT_INSTRUCT = {
|
|
"australian": "with a strong Australian accent",
|
|
"canadian": "with a Canadian accent",
|
|
"american": "with a neutral American accent",
|
|
"british": "with a British accent",
|
|
"indian": "with an Indian accent",
|
|
"nigerian": "with a Nigerian accent",
|
|
"kenyan": "with a Kenyan accent",
|
|
"hong_kong": "with a Hong Kong accent",
|
|
}
|
|
|
|
|
|
def accent_for(display: str, data_dir: Path) -> str:
|
|
"""The speaker page's accent, as an instruction phrase.
|
|
|
|
The narrator is excluded on purpose: its accent is the translation language
|
|
(see generate_narrator_refs.py), and restating "American" for the German
|
|
narrator is precisely the bug this replaces.
|
|
"""
|
|
import json as _json
|
|
|
|
if display.lower() == "narrator":
|
|
return ""
|
|
speakers = _json.loads((data_dir / "speakers.json").read_text())["speakers"]
|
|
by_lower = {s.get("primary_name", "").lower(): s for s in speakers}
|
|
rec = by_lower.get(display.lower())
|
|
if not rec:
|
|
for s in speakers:
|
|
aliases = {a.lower() for a in (s.get("aliases") or [])}
|
|
if display.lower() in aliases:
|
|
rec = s
|
|
break
|
|
accent = (rec or {}).get("accent") or ""
|
|
return ACCENT_INSTRUCT.get(accent.lower(), "")
|
|
|
|
|
|
def build_resolver(data_dir: Path, refs: dict[str, Path]) -> dict[str, str]:
|
|
"""Map the display name in a translation to the real voice slug.
|
|
|
|
The translations say "Kainda"; the speaker record and the reference file say
|
|
"Kainda Walters Foster Murphy". Resolving through the alias map rather than
|
|
by string-munging the display name is what keeps the two in step when a
|
|
speaker is renamed - the same lesson as the speaker-review audit, where a
|
|
stale name silently produced no voice at all.
|
|
"""
|
|
from umbraaudio.analysis.attribution_qa import build_alias_map
|
|
|
|
alias_map = build_alias_map(data_dir / "speakers.json")
|
|
by_lower = {k.lower(): v for k, v in alias_map.items()}
|
|
out: dict[str, str] = {}
|
|
for display in {u["speaker"] for u in _ALL_SPEAKERS}:
|
|
canonical = by_lower.get(display.lower())
|
|
slug = slugify(canonical or display)
|
|
if slug in refs:
|
|
out[display] = slug
|
|
return out
|
|
|
|
|
|
_ALL_SPEAKERS: list[dict[str, str]] = []
|
|
|
|
|
|
def post_tts(
|
|
url: str,
|
|
text: str,
|
|
*,
|
|
prompt_slug: str | None,
|
|
ref_audio: Path | None,
|
|
ref_text: str | None,
|
|
timeout: int,
|
|
language: str = "english",
|
|
voice_description: str = "",
|
|
) -> bytes:
|
|
"""POST multipart by hand; the boundary must match byte for byte."""
|
|
boundary = f"----umbra{uuid.uuid4().hex}"
|
|
parts: list[bytes] = []
|
|
|
|
def field(name: str, value: str) -> None:
|
|
parts.append(
|
|
f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n'
|
|
f"{value}\r\n".encode()
|
|
)
|
|
|
|
def file_field(name: str, path: Path) -> None:
|
|
parts.append(
|
|
f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"; '
|
|
f'filename="{path.name}"\r\nContent-Type: audio/wav\r\n\r\n'.encode()
|
|
)
|
|
parts.append(path.read_bytes())
|
|
parts.append(b"\r\n")
|
|
|
|
field("text", text)
|
|
# Must be sent: the server used to hardcode language="english" in every
|
|
# branch, so German and French were synthesised with English phonology.
|
|
field("language", language)
|
|
if voice_description:
|
|
field("voice_description", voice_description)
|
|
if prompt_slug:
|
|
field("prompt_slug", prompt_slug)
|
|
if ref_audio:
|
|
file_field("ref_audio", ref_audio)
|
|
if ref_text:
|
|
field("ref_text", ref_text)
|
|
parts.append(f"--{boundary}--\r\n".encode())
|
|
body = b"".join(parts)
|
|
|
|
req = urllib.request.Request(
|
|
f"{url}/v1/tts",
|
|
data=body,
|
|
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
|
method="POST",
|
|
)
|
|
# Raw WAV, not JSON. An earlier version of this script parsed
|
|
# {"audio_base64": ...} and would have failed on every single clip.
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
body = resp.read()
|
|
if not body.startswith(b"RIFF"):
|
|
raise RuntimeError(
|
|
f"expected WAV, got {body[:80]!r} "
|
|
f"(HTTP {getattr(resp, 'status', '?')})"
|
|
)
|
|
return body
|
|
|
|
|
|
def stitch(parts: list[Path], out: Path, gap_ms: int = 400) -> None:
|
|
"""Concatenate with a short gap so speakers are audibly separate."""
|
|
import subprocess
|
|
|
|
if not parts:
|
|
return
|
|
listing = out.parent / f".{out.stem}_concat.txt"
|
|
listing.write_text(
|
|
"".join(f"file '{p.resolve()}'\n" for p in parts)
|
|
)
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(listing),
|
|
"-af", f"apad=pad_dur={gap_ms/1000}", "-ar", "24000", "-ac", "1",
|
|
str(out),
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
listing.unlink(missing_ok=True)
|
|
|
|
|
|
# English first, then the operator's stated order for the languages rendered.
|
|
COMBINED_ORDER = ["EN", "ZH", "JA", "KO", "DE", "ES", "FR", "RU", "IT", "PT"]
|
|
|
|
|
|
def build_combined(out_root: Path) -> Path | None:
|
|
"""Concatenate every per-language chapter into one listenable file."""
|
|
import subprocess
|
|
|
|
lines = []
|
|
for lang in COMBINED_ORDER:
|
|
f = out_root / lang / f"_full_sample_{lang}.wav"
|
|
if f.exists():
|
|
lines.append(f"file '{f.resolve()}'")
|
|
if not lines:
|
|
return None
|
|
listing = out_root / "._combined.txt"
|
|
listing.write_text("\n".join(lines) + "\n")
|
|
out = out_root / "_ALL_LANGUAGES_full_sample.wav"
|
|
subprocess.run(
|
|
["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(listing),
|
|
"-c", "copy", str(out)],
|
|
check=True, capture_output=True,
|
|
)
|
|
listing.unlink(missing_ok=True)
|
|
return out
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--project")
|
|
ap.add_argument("--languages", help="comma-separated suffixes, default all found")
|
|
ap.add_argument("--timeout", type=int, default=900)
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
ap.add_argument(
|
|
"--voice-mode", choices=("ref", "prompt"), default="ref",
|
|
help="ref: ref_audio+ref_text (model-portable). prompt: cached ICL "
|
|
"prompt (more faithful, but only valid for the model that built it)",
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
config = load_config(args.project)
|
|
data_dir = config.data_dir
|
|
tdir = build_translation_paths(config)
|
|
if not tdir.exists():
|
|
print(f" no translations at {tdir} - run build_sample_translations.py first")
|
|
return 1
|
|
|
|
refs = speaker_refs(data_dir)
|
|
prompts = {p.stem for p in (data_dir / "voice_prompts").glob("*.pt")}
|
|
global _ALL_SPEAKERS
|
|
_ALL_SPEAKERS = [
|
|
u for f in sorted(tdir.glob("deal_hunter_sample_ch1_*.json"))
|
|
for u in json.loads(f.read_text(encoding="utf-8"))["utterances"]
|
|
]
|
|
resolved = build_resolver(data_dir, refs)
|
|
for name in sorted({u["speaker"] for u in _ALL_SPEAKERS}):
|
|
if name not in resolved:
|
|
print(f" WARNING: no reference audio resolves for {name!r}")
|
|
langs = sorted(tdir.glob("deal_hunter_sample_ch1_*.json"))
|
|
if args.languages:
|
|
want = {s.strip().upper() for s in args.languages.split(",")}
|
|
langs = [p for p in langs if p.stem.rsplit("_", 1)[-1].upper() in want]
|
|
if not langs:
|
|
print(" no language files matched")
|
|
return 1
|
|
|
|
out_root = config.project_dir / TRANSLATION_DIRNAME
|
|
total = sum(len(json.loads(p.read_text(encoding="utf-8"))["utterances"]) for p in langs)
|
|
print(f" {len(langs)} languages, {total} clips -> {out_root}")
|
|
missing = []
|
|
for p in langs:
|
|
for u in json.loads(p.read_text(encoding="utf-8"))["utterances"]:
|
|
slug = resolved.get(u["speaker"], "")
|
|
if slug not in refs:
|
|
missing.append((p.stem.rsplit("_", 1)[-1], u["speaker"]))
|
|
|
|
if args.dry_run:
|
|
for lang in langs:
|
|
doc = json.loads(lang.read_text(encoding="utf-8"))
|
|
print(f"\n {doc['language_name']} ({doc['language_code']})")
|
|
for u in doc["utterances"]:
|
|
slug = resolved.get(u["speaker"], "")
|
|
mode = "ICL prompt" if slug in prompts else (
|
|
"ref_audio" if slug in refs else "NO REFERENCE"
|
|
)
|
|
print(f" {u['index']:2d} {u['speaker']:32s} {mode}")
|
|
if missing:
|
|
print(f"\n speakers with no reference audio: {sorted(set(missing))}")
|
|
return 0
|
|
|
|
out_root.mkdir(parents=True, exist_ok=True)
|
|
ok = failed = 0
|
|
for lang_file in langs:
|
|
doc = json.loads(lang_file.read_text(encoding="utf-8"))
|
|
suffix = lang_file.stem.rsplit("_", 1)[-1]
|
|
out_dir = out_root / suffix
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
made: list[Path] = []
|
|
for u in doc["utterances"]:
|
|
slug = slugify(u["speaker"])
|
|
dest = out_dir / f"utt{u['index']:02d}_{u['speaker'].replace(' ', '_')}.wav"
|
|
if slug not in refs:
|
|
print(f" SKIP {suffix} utt{u['index']} - no reference for {u['speaker']}")
|
|
failed += 1
|
|
continue
|
|
t0 = time.time()
|
|
try:
|
|
use_prompt = args.voice_mode == "prompt" and slug in prompts
|
|
# The narrator's accent is the translation language, so it uses
|
|
# a reference designed for that language. Every other character
|
|
# keeps its own reference and carries its page accent through.
|
|
# The narrator uses the speaker page's own reference in EVERY
|
|
# language. Generating a reference per language invents a
|
|
# different narrator each time - measured 0.77-0.96 band-energy
|
|
# similarity to the real one - so the voice changed with the
|
|
# language. The language parameter supplies phonology; the one
|
|
# reference supplies identity. That is the same source the
|
|
# vmatrix was built from, at x_vector resolution instead of the
|
|
# 8k matrix, which is the only form this machine can run.
|
|
ref = refs[slug]
|
|
instruct = accent_for(u["speaker"], data_dir)
|
|
audio = post_tts(
|
|
TTS_URL, u["text"],
|
|
prompt_slug=slug if use_prompt else None,
|
|
ref_audio=None if use_prompt else ref,
|
|
ref_text=None if use_prompt else REF_TEXT,
|
|
timeout=args.timeout,
|
|
language=LANG_CODES.get(suffix, "english"),
|
|
voice_description=instruct,
|
|
)
|
|
dest.write_bytes(audio)
|
|
made.append(dest)
|
|
ok += 1
|
|
print(f" {suffix} utt{u['index']:02d} {u['speaker']:26s} "
|
|
f"{len(audio)/1024:6.0f} KB {time.time()-t0:5.1f}s")
|
|
except (urllib.error.URLError, TimeoutError, RuntimeError) as e:
|
|
failed += 1
|
|
print(f" {suffix} utt{u['index']:02d} {u['speaker']:26s} FAILED: {e}")
|
|
try:
|
|
stitch(made, out_dir / f"_full_sample_{suffix}.wav")
|
|
print(f" {suffix}: stitched {len(made)} clips -> _full_sample_{suffix}.wav")
|
|
except Exception as e: # stitching is a convenience, not the deliverable
|
|
print(f" {suffix}: stitch failed: {e}")
|
|
|
|
combined = build_combined(out_root)
|
|
print(f"\n done: {ok} rendered, {failed} failed")
|
|
if combined:
|
|
print(f" combined: {combined}")
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|