"""Ahead-of-air neural TTS cache for trusted live commentary events. The browser can safely use a clip only when its rendered text is an exact match. This cache receives events from the scoring process, which already knows the upcoming pre-start script, and prepares those clips before their air time. A synthesis failure is deliberately non-fatal: the live player keeps its browser-voice fallback and the race never waits on a cloud request. """ from __future__ import annotations import hashlib import json import queue import re import threading import time from .engines import make_engine from .synth import AUDIO_DIR, LANG_CODE, RATE, to_ssml class LiveNeuralTtsCache: """Prepare exact-match neural clips in bounded background workers.""" def __init__(self, *, workers: int = 1, min_request_interval_s: float = 0.20, on_log=print): self._log = on_log self._queue: queue.Queue[tuple[str, str, str, tuple[tuple[str, str, str], ...]] | None] = queue.Queue() self._lock = threading.Lock() self._scheduled: set[tuple[str, str, str]] = set() self._voices: dict[str, dict[str, str]] = {} # A missing ADC login used to make every new commentary line schedule three failing cloud # calls and log all of them. That is not a transient content error; it can overload a live # simulator precisely while the browser is falling back to its local voice. One bounded # circuit breaker keeps the broadcast responsive and retries later without operator action. self._unavailable_until = 0.0 self._unavailable_reason: str | None = None self._unavailable_logged = False # A newly deployed revision joins a race in progress and can discover hundreds of exact # lines at once. Several workers used to burst that backlog into the project-wide TTS # minute quota, taking the already-on-air revision's dynamic commentary down with it. One # worker plus a shared request slot keeps a warming candidate at 300 requests/minute. The # project quota is 500/minute, but pacing at the exact ceiling left no room for one retry or # a no-traffic candidate warming beside production and caused a repeated all-audio 429 loop. # Ordinary live demand still drains much faster than it reaches air. self._min_request_interval_s = max(0.0, float(min_request_interval_s)) self._next_request_at = 0.0 self._index_path = AUDIO_DIR / "index.json" try: self._index = json.loads(self._index_path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): self._index = {} for n in range(max(1, workers)): threading.Thread(target=self._run, daemon=True, name=f"live-neural-tts-{n + 1}").start() def _wait_for_request_slot(self) -> None: """Pace calls against the project quota, shared even if a caller opts into more workers.""" with self._lock: now = time.monotonic() wait_s = max(0.0, self._next_request_at - now) self._next_request_at = max(now, self._next_request_at) + self._min_request_interval_s if wait_s: time.sleep(wait_s) @staticmethod def _role(event: dict) -> str: role = event.get("commentaryRole") if role in {"lead", "tactical", "rules"}: return role return {"m": "lead", "f": "tactical", "r": "rules"}.get(event.get("voice"), "tactical") @staticmethod def _spoken_i18n_options(event: dict) -> list[dict]: """Return the source line plus every approved on-air wording alternative. ``StreamingScorer`` deliberately selects a wording only once the event crosses its settled live seam. The TTS cache sees the event earlier, so it must prepare the whole approved bank rather than only the producer's default wording. Otherwise a rotated line reaches air without a neural clip and the player falls back to a browser voice. """ candidates = [event.get("i18n")] candidates.extend(event.get("_commentaryVariants") or []) options: list[dict] = [] seen: set[tuple[tuple[str, str], ...]] = set() for candidate in candidates: if not isinstance(candidate, dict): continue option = {str(lang): str(text) for lang, text in candidate.items() if str(text or "").strip()} if not option: continue key = tuple(sorted(option.items())) if key not in seen: seen.add(key) options.append(option) return options def submit(self, events: list[dict], teams: list[dict]) -> None: """Queue every known spoken line once; caller remains on the scoring thread. The producer's first i18n value and all of its approved wording alternatives are prepared together. The later live rotation can then select any variant without changing speaker quality or dropping to a browser voice. """ with self._lock: now = time.monotonic() if now < self._unavailable_until: return # The prior pause has elapsed. Allow one fresh cloud attempt and one concise diagnostic # if credentials are still unavailable, rather than suppressing every later outage. if self._unavailable_until: self._unavailable_until = 0.0 self._unavailable_reason = None self._unavailable_logged = False foreign = tuple(sorted((t["name"], t["nameLang"], t.get("ipa") or "") for t in teams if t.get("name") and t.get("nameLang"))) for event in events: role = self._role(event) for option in self._spoken_i18n_options(event): for lang, text in option.items(): text = text.strip() if lang not in LANG_CODE: continue key = (lang, text, role) with self._lock: if text in (self._index.get(lang) or {}) or key in self._scheduled: continue self._scheduled.add(key) self._queue.put((lang, text, role, foreign)) def _run(self) -> None: # Built lazily, INSIDE the guard below, and deliberately not here. Constructing the cloud # client at thread start meant a box without Application Default Credentials lost the whole # synthesis thread to an unhandled `DefaultCredentialsError` — the circuit breaker this class # advertises guarded the synthesis calls but not the client. Measured on the office producer, # which has no ADC and is not getting any. engine = None while True: job = self._queue.get() if job is None: return lang, text, role, foreign = job success = False try: with self._lock: unavailable = time.monotonic() < self._unavailable_until if unavailable: continue if engine is None: engine = make_engine(on_log=self._log) names = self._voices.get(lang) if names is None: names = engine.voices(lang) self._voices[lang] = names spoken = re.sub(r"\s*—\s*", ", ", text) payload, is_ssml = to_ssml(lang, spoken, list(foreign)) voice_name = names[role] digest = hashlib.sha1(f"{voice_name}|{RATE}|{payload}".encode()).hexdigest()[:12] # The engine's own extension, not a fixed `.mp3`: the local engine emits WAV, and the # renderer reads whatever relative path the index names rather than assuming a format. # The digest deliberately does NOT include the engine — the voice name already differs # between engines, and folding it in would invalidate every cached cloud clip and # re-synthesise a championship's commentary at Studio prices to no purpose. rel = f"{lang}/{digest}{engine.suffix}" target = AUDIO_DIR / rel target.parent.mkdir(parents=True, exist_ok=True) if not target.exists(): self._wait_for_request_slot() audio = engine.synthesize(lang=lang, payload=payload, is_ssml=is_ssml, spoken=spoken, voice_name=voice_name, rate=RATE) tmp = target.with_suffix(".tmp") tmp.write_bytes(audio) tmp.replace(target) with self._lock: self._index.setdefault(lang, {})[text] = rel tmp_index = self._index_path.with_suffix(".tmp") tmp_index.write_text(json.dumps(self._index, ensure_ascii=False, indent=0), encoding="utf-8") tmp_index.replace(self._index_path) self._unavailable_until = 0.0 self._unavailable_reason = None self._unavailable_logged = False success = True except Exception as exc: # noqa: BLE001 - live commentary must keep running without cloud TTS with self._lock: self._unavailable_until = time.monotonic() + 120.0 self._unavailable_reason = str(exc) should_log = not self._unavailable_logged self._unavailable_logged = True if should_log: self._log( # Names the engine rather than saying "cloud": on the office deployment there # is none, and a message that misdescribes where synthesis happens sends an # operator to check a credential that was never involved. f"[live-neural-tts] {getattr(engine, 'name', 'tts')} synthesis paused for 120 s; " "browser fallback remains live " f"({self._unavailable_reason})" ) finally: if not success: with self._lock: self._scheduled.discard((lang, text, role)) self._queue.task_done()