UmbraAudio/scripts/probe_cross_language_tts.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Does Qwen3-TTS speak Chinese when the voice reference is English?

This is the open prerequisite for any target-language audio lane. The
engine and the server both accept a per-request ``language``, so it is
possible in principle - but "possible in principle" is not a measurement.

The question is answered by transcription rather than by listening. Whisper
is asked to transcribe each generated clip and to say what language it
thinks it heard. If the model is reading the Chinese text with English
phonology, the transcription comes back as English-sounding gibberish or as
the wrong characters; if it is speaking Chinese, the transcription comes back
as the Chinese characters that were sent.

Four clips are generated:

============================  =========================================
clip                          what it answers
============================  =========================================
english-ref-chinese-lang      the case under test: English voice ref,
                              Chinese text, language=chinese
english-ref-english-lang      the control: the same voice ref reading the
                              same Chinese text with English phonology
no-ref-chinese-lang           the model's own Chinese voice, for reference
english-ref-english-text      the known-good case: the voice ref reading
                              English
============================  =========================================

The first two are the comparison that matters: same text, same voice, the
only difference is the language parameter.

Run it with the TTS server up and LM Studio unloaded - on a 16 GB Mac the
two models cannot be resident at the same time.
"""

from __future__ import annotations

import argparse
import json
import sys
import wave
from pathlib import Path

import httpx

CHINESE_TEXT = "凯因达站在舰桥上，看着舷窗外的星云，握紧了手里的扳手。"
CHINESE_TEXT_2 = "我们必须在下一次跃迁之前找到那块燃料。"
ENGLISH_TEXT = "Kainda stood on the bridge, watching the nebula through the viewport."

TTS_URL = "http://127.0.0.1:8881"


def _health(client: httpx.Client) -> dict:
    response = client.get(f"{TTS_URL}/health", timeout=10)
    response.raise_for_status()
    return response.json()


def _load_base(client: httpx.Client) -> None:
    response = client.post(f"{TTS_URL}/load", json={"model_type": "base"}, timeout=1800)
    response.raise_for_status()


def _synthesise(
    client: httpx.Client,
    out_path: Path,
    text: str,
    *,
    language: str,
    ref_audio: Path | None = None,
    ref_text: str | None = None,
    voice_description: str | None = None,
) -> None:
    data: dict[str, str] = {"text": text, "language": language, "sample_rate": "24000"}
    files = None
    if voice_description:
        data["voice_description"] = voice_description
    if ref_audio is not None:
        data["ref_text"] = ref_text or ""
        files = {"ref_audio": (ref_audio.name, ref_audio.read_bytes(), "audio/wav")}
    response = client.post(f"{TTS_URL}/v1/tts", data=data, files=files, timeout=1800)
    response.raise_for_status()
    out_path.write_bytes(response.content)
    print(f"  wrote {out_path} ({out_path.stat().st_size} bytes)", flush=True)


def _duration(path: Path) -> float:
    with wave.open(str(path), "rb") as handle:
        return handle.getnframes() / float(handle.getframerate())


def _transcribe(path: Path) -> dict:
    import whisper

    model = whisper.load_model("base")
    result = model.transcribe(str(path), fp16=False)
    return {
        "text": (result.get("text") or "").strip(),
        "language": result.get("language"),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--ref", type=Path, required=True, help="an English voice reference wav")
    parser.add_argument("--ref-text", default="", help="the transcript of the reference clip")
    parser.add_argument(
        "--out", type=Path, required=True, help="where to write the clips and report"
    )
    args = parser.parse_args()

    args.out.mkdir(parents=True, exist_ok=True)

    with httpx.Client() as client:
        health = _health(client)
        print(f"tts server: {health}", flush=True)
        if not health.get("base_loaded"):
            print("loading the base model (this takes a while)...", flush=True)
            _load_base(client)

        clips = {
            "english-ref-chinese-lang": {
                "text": CHINESE_TEXT,
                "language": "chinese",
                "ref_audio": args.ref,
                "ref_text": args.ref_text,
            },
            "english-ref-english-lang": {
                "text": CHINESE_TEXT,
                "language": "english",
                "ref_audio": args.ref,
                "ref_text": args.ref_text,
            },
            "no-ref-chinese-lang": {
                "text": CHINESE_TEXT,
                "language": "chinese",
                "voice_description": (
                    "A calm, measured male narrator with a deep, radio-announcer voice."
                ),
            },
            "english-ref-english-text": {
                "text": ENGLISH_TEXT,
                "language": "english",
                "ref_audio": args.ref,
                "ref_text": args.ref_text,
            },
        }

        report: dict[str, dict] = {}
        for name, spec in clips.items():
            print(f"generating {name}...", flush=True)
            path = args.out / f"{name}.wav"
            _synthesise(
                client,
                path,
                spec["text"],
                language=spec["language"],
                ref_audio=spec.get("ref_audio"),
                ref_text=spec.get("ref_text"),
                voice_description=spec.get("voice_description"),
            )
            report[name] = {
                "text_sent": spec["text"],
                "language_requested": spec["language"],
                "duration_s": round(_duration(path), 3),
            }

    print("transcribing...", flush=True)
    for name in report:
        path = args.out / f"{name}.wav"
        transcript = _transcribe(path)
        report[name].update(transcript)
        print(f"  {name}: heard {transcript['language']}: {transcript['text']}", flush=True)

    expected = CHINESE_TEXT.replace("，", "").replace("。", "")
    for name in ("english-ref-chinese-lang", "english-ref-english-lang", "no-ref-chinese-lang"):
        heard = report[name]["text"].replace(" ", "").replace(",", "").replace(".", "")
        overlap = len(set(expected) & set(heard))
        report[name]["character_overlap_with_source"] = overlap
        report[name]["source_characters"] = len(set(expected))

    verdict = report["english-ref-chinese-lang"]
    report["verdict"] = {
        "english_voice_reference_speaks_chinese": (
            verdict["language"] == "zh" and verdict["character_overlap_with_source"] >= 8
        ),
        "note": (
            "The English voice reference is kept - the point is whether the same "
            "voice can speak the target language, not whether a Chinese voice "
            "can."
        ),
    }

    report_path = args.out / "probe-report.json"
    report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"\nwrote {report_path}")
    print(json.dumps(report["verdict"], ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())