"""What actually went to air, written down while it is still knowable. The as-run — which race was on screen, from when to when — is the one fact about a broadcast that exists **nowhere else**. `OnAirQueue` decides it, in memory, and forgets it; the race plane records that a race was sailed, not that anybody showed it; and the picture itself is not machine-readable. So a recipient asking "cut me race 1009 out of the day's recording" cannot be answered afterwards unless the producer says so at the time. Sibling of `recording_anchor.py`, and deliberately the same shape: an append-only JSONL file written beside the show. That is the cheapest possible thing to add to a live loop — one line, no network, no database, no schema migration — and if it fails it fails alone. **Two race-boundary rows, and the second one is optional.** `on_air` when a race takes the air; `off_air` when the producer drops back to the waiting picture. A race-to-race hand-over emits only `on_air`, because that is what the producer actually does — `take_next` swaps the race in place without releasing — so `segments()` below closes the previous race at the next one's start rather than expecting a matching close. Getting that wrong is how an as-run acquires gaps that were never off air. The same append-only evidence stream also carries `commentary_selected` when one wording variant crosses the producer seam and `commentary_started` when the programme renderer reports that its audio actually began. The separation is intentional: the microphone scheduler can discard a stale selected line. `segments()` ignores these audit rows when building the race partition. Read it back with `segments()`, which turns the rows into the gap-free partition the metadata contract publishes (`docs/broadcast-metadata.md` §4.2). """ from __future__ import annotations import json import threading import time from collections.abc import Callable from pathlib import Path ON_AIR = "on_air" OFF_AIR = "off_air" COMMENTARY_SELECTED = "commentary_selected" COMMENTARY_STARTED = "commentary_started" # Race hand-overs are appended by the producer loop while audible-start receipts arrive on HTTP # handler threads. One process therefore has several legitimate writers to the same JSONL file. # Serialize the one-line writes so two simultaneous events cannot interleave and corrupt both rows. _APPEND_LOCK = threading.Lock() def on_air_row(*, now_ms: int, instance_id: str, race_name: str | None = None, gun_utc_ms: int | None = None, leaving_instance_id: str | None = None) -> dict: """A race has taken the air. `leaving_instance_id` is the race this one displaced, and only the producer knows it — the queue hands over without either race being told. It is recorded rather than inferred so a reader can tell a hand-over (one race replacing another) from a fresh start after a gap, which look identical once you have only timestamps.""" return {"tUtcMs": int(now_ms), "event": ON_AIR, "instanceId": str(instance_id), "raceName": race_name, "gunUtcMs": None if gun_utc_ms is None else int(gun_utc_ms), "leavingInstanceId": leaving_instance_id} def off_air_row(*, now_ms: int, instance_id: str) -> dict: """Nothing is on air any more — the producer went back to the waiting presentation.""" return {"tUtcMs": int(now_ms), "event": OFF_AIR, "instanceId": str(instance_id)} def commentary_selected_row(*, now_ms: int, instance_id: str | None, race_name: str | None, source_sec: float | int | None, commentary: dict) -> dict | None: """A rotated line crossed the producer seam and became eligible for the programme microphone. This is deliberately distinct from :func:`commentary_started_row`: the renderer may discard a stale colour line while another sentence owns the microphone. Keeping both rows is what makes delivery rate and phrase rotation independently auditable after the show. """ variant = _commentary_variant(commentary) if variant is None: return None return { "tUtcMs": int(now_ms), "event": COMMENTARY_SELECTED, "instanceId": _text(instance_id, 100), "raceName": _text(race_name, 100), "sourceSec": _number(source_sec), **variant, "text": _text(commentary.get("text"), 600), } def commentary_started_row(*, now_ms: int, payload: object) -> dict | None: """Validate one renderer receipt for the instant a rotated line actually began playing. Browser input is untrusted. The server supplies ``tUtcMs`` from its own clock and clamps every descriptive field; the client timestamp is retained only as diagnostic context and is never used as the authoritative as-run time. """ if not isinstance(payload, dict): return None variant = _commentary_variant(payload) renderer_id = _text(payload.get("rendererId"), 16) backend = _text(payload.get("audioBackend"), 16) if variant is None or not renderer_id or backend not in ("neural", "browser"): return None role = _text(payload.get("rendererRole"), 40) return { "tUtcMs": int(now_ms), "event": COMMENTARY_STARTED, "instanceId": _text(payload.get("instanceId"), 100), "raceName": _text(payload.get("raceName"), 100), "sourceSec": _number(payload.get("sourceSec")), **variant, "text": _text(payload.get("text"), 600), "lang": _text(payload.get("lang"), 12), "audioBackend": backend, "rendererId": renderer_id, "rendererRole": role, "clientUtcMs": _integer(payload.get("clientUtcMs"), minimum=0), } def _commentary_variant(value: dict) -> dict | None: category = _text(value.get("commentaryVariantCategory"), 80) index = _integer(value.get("commentaryVariantIndex"), minimum=1, maximum=100) count = _integer(value.get("commentaryVariantCount"), minimum=1, maximum=100) variant_id = _text(value.get("commentaryVariantId"), 64) if not category or index is None or count is None or index > count or not variant_id: return None return {"commentaryVariantCategory": category, "commentaryVariantIndex": index, "commentaryVariantCount": count, "commentaryVariantId": variant_id} def _text(value: object, limit: int) -> str | None: if not isinstance(value, str): return None cleaned = "".join(ch for ch in value if ch.isprintable()).strip()[:limit] return cleaned or None def _integer(value: object, *, minimum: int | None = None, maximum: int | None = None) -> int | None: if isinstance(value, bool): return None try: out = int(value) except (TypeError, ValueError): return None if minimum is not None and out < minimum: return None if maximum is not None and out > maximum: return None return out def _number(value: object) -> float | int | None: if isinstance(value, bool) or value is None: return None try: out = float(value) except (TypeError, ValueError): return None if out != out or out in (float("inf"), float("-inf")): return None return int(out) if out.is_integer() else round(out, 3) def append(path: Path | None, row: dict, *, on_log: Callable[[str], None] = print) -> bool: """Append one row. Returns whether it was written; **never raises**. This is called from the live broadcast loop, so a full disk, a read-only mount or a path that stopped existing must cost a log line and nothing else. `path=None` means as-run logging is not configured, which is a no-op rather than an error — the producer has to run without it.""" if path is None: return False try: with _APPEND_LOCK: path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(row) + "\n") return True except Exception as e: # noqa: BLE001 — the show outranks the log on_log(f"[asrun] could not write {path} ({type(e).__name__}: {e}) — continuing") return False def read_rows(path: Path) -> list[dict]: """Every row written so far. A truncated last line — the run was killed mid-write — is dropped rather than raising, exactly as `recording_anchor.read_samples` does: the file is an append log and its tail is the least valuable part of it.""" if not path.exists(): return [] rows = [] for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def segments(rows: list[dict], *, session_end_ms: int | None = None) -> list[dict]: """The rows as `[{instanceId, raceName, gunUtcMs, fromUtcMs, toUtcMs}]`, in order. `toUtcMs` is EXCLUSIVE and is whichever comes first: an explicit `off_air`, or the next race taking the air. `None` on the last segment means it was still on air when the log ended — which is how "what is playing right now" falls out without a second endpoint. Pass `session_end_ms` to close it at a known end of transmission instead. Rows are sorted by time rather than trusted in file order: the log is appended from one loop so they arrive ordered, but a reader that assumed it would break silently on a concatenated file. An `off_air` for a race that is not the one on air is ignored — it cannot close anything, and treating it as a close would truncate the race that IS on air. """ out: list[dict] = [] for r in sorted(rows, key=lambda x: int(x.get("tUtcMs", 0))): t, event = int(r.get("tUtcMs", 0)), r.get("event") if event == ON_AIR: if out and out[-1]["toUtcMs"] is None: out[-1]["toUtcMs"] = t # a hand-over closes the race it displaced out.append({"instanceId": r.get("instanceId"), "raceName": r.get("raceName"), "gunUtcMs": r.get("gunUtcMs"), "fromUtcMs": t, "toUtcMs": None}) elif event == OFF_AIR: if out and out[-1]["toUtcMs"] is None and out[-1]["instanceId"] == r.get("instanceId"): out[-1]["toUtcMs"] = t if session_end_ms is not None and out and out[-1]["toUtcMs"] is None: out[-1]["toUtcMs"] = int(session_end_ms) return out def default_path(arg: str | None) -> Path | None: """The configured log path, or None when as-run logging is switched off.""" return Path(arg).expanduser() if arg else None def _now_ms() -> int: return int(time.time() * 1000)