"""Where the spoken words come from — one interface, two engines. WHY THIS EXISTS. Commentary audio was synthesised by exactly one vendor, reached by exactly one code path, and the coupling cost more than it looked. Three separate problems, one shape: * **Money.** Google's tier auto-selection lands on the most expensive voice that exists, and on 13 August 2026 — one championship day — Text-to-Speech billed **kr 7,480 in Studio voices alone**, against an expected kr 277 for the whole estate. * **Availability.** `TextToSpeechClient()` needs Application Default Credentials at *construction*, so a box without them lost the whole synthesis thread rather than degrading. The office deployment has no ADC and is not getting any. * **Sovereignty.** Every spoken line left the building to be spoken. That is a vendor dependency in the one part of the broadcast that is pure computation. A local engine answers all three at once, so the choice belongs behind an interface rather than an `if` in the middle of the cache loop. WHAT AN ENGINE OWES. Two things and no more: which voice each commentary role speaks in, and how a line becomes audio bytes. Everything else — the queue, the circuit breaker, the digest key, the index — stays in `live_cache.py`, which is engine-agnostic and should remain so. THE ONE-VOICE RULE APPLIES TO BOTH. The broadcast standard is exactly **one male and one female voice per language** — Male Caller for facts, Female Analyst for the story — held stable across a whole programme, because a voice that drifts between renders reads as a different person arriving mid-race. Both engines pin their set deliberately and both let the same `TTS_VOICE__` environment variables override it, so an operator learns one convention rather than two. """ from __future__ import annotations import io import os import wave from pathlib import Path from .synth import LANG_CODE, pick_voices #: Where Piper's `.onnx` voice models live. A directory rather than a package resource because the #: models are ~60 MB each and belong to the deployment, not to the source tree. VOICE_DIR = Path(os.environ.get("VR_PIPER_VOICES", "/srv/vr/voices")) #: The local set, chosen to mirror the cloud one role for role: a male lead, a female analyst, and a #: third distinct voice for rules rulings so a verdict does not arrive in the same voice that just #: described the geometry. `en_GB` throughout, matching the `en-GB` the cloud set uses — a broadcast #: does not change accent because its synthesiser changed. LOCAL_ROLE_VOICES = { "en": { "lead": "en_GB-alan-medium", "tactical": "en_GB-jenny_dioco-medium", "rules": "en_GB-northern_english_male-medium", }, } class GoogleEngine: """Cloud synthesis, unchanged — the SSML path and its plain-text retry both live here.""" suffix = ".mp3" name = "google" def __init__(self) -> None: from google.cloud import texttospeech self._tts = texttospeech self._client = texttospeech.TextToSpeechClient() def voices(self, lang: str) -> dict[str, str]: return pick_voices(self._client, lang) def synthesize(self, *, lang: str, payload: str, is_ssml: bool, spoken: str, voice_name: str, rate: float) -> bytes: tts = self._tts voice = tts.VoiceSelectionParams(language_code=LANG_CODE[lang], name=voice_name) config = tts.AudioConfig(audio_encoding=tts.AudioEncoding.MP3, speaking_rate=rate) inp = tts.SynthesisInput(ssml=payload) if is_ssml else tts.SynthesisInput(text=payload) try: return self._client.synthesize_speech(input=inp, voice=voice, audio_config=config).audio_content except Exception: # A malformed pronunciation hint must not silence a line. Retry the same voice with the # plain text, which is what the SSML was decorating. if not is_ssml: raise return self._client.synthesize_speech( input=tts.SynthesisInput(text=spoken), voice=voice, audio_config=config).audio_content class PiperEngine: """Local neural synthesis on the CPU, no network and no credential. Measured on the producer VM (2 vCPU) with real commentary lines: **~16x realtime**, so a race day's three to four hours of speech is about thirteen minutes of compute. The same models ran at 33.7x on the render host's 24 threads — quote whichever machine will actually do the work. WHAT IS GIVEN UP, stated rather than glossed: Piper takes **plain text**, so the SSML pronunciation layer does not reach it. That layer carries per-language fixes — Swedish `banan` (the course, not the fruit) and the hard `g` in `gippa` — and for English exactly one entry, the broadcast sense of *live* as /laɪv/. So the English loss today is one word, and the Swedish dictionary matters only when Swedish output is switched back on. This is the same degradation the cloud path already performs whenever SSML is rejected, not a new failure mode. """ suffix = ".wav" name = "piper" def __init__(self, *, voice_dir: Path | None = None, on_log=print) -> None: self._dir = voice_dir or VOICE_DIR self._log = on_log self._loaded: dict[str, object] = {} if not self._dir.is_dir(): raise RuntimeError(f"no Piper voice directory at {self._dir} (set $VR_PIPER_VOICES)") def _model(self, voice_name: str) -> Path: return self._dir / f"{voice_name}.onnx" def voices(self, lang: str) -> dict[str, str]: """The role set, with the same environment overrides the cloud engine honours. An override naming a model this box does not have is **ignored with a log line** rather than raising: the commonest way to arrive here is an operator moving a deployment and carrying the cloud's `TTS_VOICE_EN_LEAD=en-GB-Studio-B` with them, and a broadcast that refuses to speak because a voice name is spelled for the other engine is a worse outcome than one that speaks in its curated voice and says so. """ curated = LOCAL_ROLE_VOICES.get(lang) if not curated: raise RuntimeError(f"no local voice set for language {lang!r}") up = lang.upper() chosen: dict[str, str] = {} for role, default in curated.items(): override = os.environ.get(f"TTS_VOICE_{up}_{role.upper()}") if override and self._model(override).exists(): chosen[role] = override else: if override: self._log(f"[live-tts] {voice_env_name(up, role)}={override!r} names no local model; " f"using {default!r}") chosen[role] = default missing = [name for name in chosen.values() if not self._model(name).exists()] if missing: raise RuntimeError(f"missing Piper models in {self._dir}: {', '.join(sorted(set(missing)))}") return {**chosen, "m": chosen["lead"], "f": chosen["tactical"], "r": chosen["rules"]} def _voice(self, voice_name: str): voice = self._loaded.get(voice_name) if voice is None: from piper import PiperVoice # ~60 MB per model, loaded once and held: reloading per line would dominate the cost of # synthesis and undo the reason for running locally. voice = PiperVoice.load(str(self._model(voice_name))) self._loaded[voice_name] = voice return voice def synthesize(self, *, lang: str, payload: str, is_ssml: bool, spoken: str, voice_name: str, rate: float) -> bytes: from piper import SynthesisConfig # `length_scale` is a DURATION multiplier, so it is the reciprocal of a speaking rate: the # broadcast's 1.03 (slightly quick) is 0.97 here. Getting this backwards is silent — the # commentary simply drags — so it is written out rather than inlined. config = SynthesisConfig(length_scale=1.0 / rate if rate else 1.0) buffer = io.BytesIO() with wave.open(buffer, "wb") as out: self._voice(voice_name).synthesize_wav(spoken, out, syn_config=config) return buffer.getvalue() def voice_env_name(lang_upper: str, role: str) -> str: return f"TTS_VOICE_{lang_upper}_{role.upper()}" def make_engine(*, on_log=print): """The deployment's engine, from `$VR_TTS_ENGINE`. Defaults to **google**, so no existing deployment changes behaviour by upgrading. Set `VR_TTS_ENGINE=piper` on a box that has the models and no credential — which is the office deployment exactly. """ choice = (os.environ.get("VR_TTS_ENGINE") or "google").strip().lower() if choice == "piper": return PiperEngine(on_log=on_log) if choice == "google": return GoogleEngine() raise RuntimeError(f"unknown $VR_TTS_ENGINE {choice!r} (expected 'google' or 'piper')")