""" Live feed server — §6 seam #3: engine → renderer (SSE, stdlib-only) The demo serves the feed as static JSON consumed by core/web/feed.js `FileFeed`. Production pushes the same payload live; the renderer switches to it purely via config (core/web/config.js `feed:'live'` → app.js lazy-loads live/web/live_feed.js `LiveFeed`). The renderer is otherwise unchanged — both sources honour the one BroadcastFeed contract. Transport = **Server-Sent Events** over the Python stdlib threading HTTP server (no extra deps): broadcasting is one-way (server → many renderers), which is exactly SSE's shape, and the browser's `EventSource` gives auto-reconnect for free. This server also serves the static broadcast files (from the v3/broadcast root) so the page and the `/events` stream share one origin — no CORS, one command. Wire messages (SSE `event:` types; `data:` is JSON — mirror the file feed's shape, sent as deltas): init : {meta, course, teams, params, media, video} # static race header (also the backlog head) delta : {sec, standings, positions, events, ais, program} # one tick, from StreamingScorer.push_frame result: {winner, winnerName, winnerSec, ...} # once the race finishes `program` (when the driver computes it — `program_state()`) is the explicit presentation phase (countdown / racing / result), so a client never infers liveness from standings alone. Late joiners (e.g. OBS reconnecting) get the full backlog — the latest `init` plus every `delta`/`result` so far — before live deltas resume, so a fresh client renders the whole race-so-far. Sub-phase: B-B (live feed transport). Drive it with live/serve_live.py (replays the demo race through StreamingScorer); the bounded-LLM editor (B-C) edits each delta's newly-finalised colour beats first. """ from __future__ import annotations import hmac import json import os import queue import sys import threading from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs _BROADCAST_ROOT = Path(__file__).resolve().parent.parent # v3/broadcast (static web root) _EVENTS_PATH = "/events" #: How long a stream may stay silent before a keepalive comment goes out. Comfortably under the #: renderer's staleness threshold (12 s, `live_feed.js`), so a quiet-but-healthy producer never trips #: the NO SIGNAL badge, and comfortably under any proxy's idle timeout. _KEEPALIVE_S = 5.0 #: The scene's ion-token module. Served from `VR_CESIUM_ION_TOKEN` when no file is on disk, so the #: photoreal world is a property of the DEPLOYMENT rather than of whoever built the image. _CESIUM_CONFIG_PATH = "/core/web/cesium-config.js" _CESIUM_CONFIG_FILE = _BROADCAST_ROOT / "core" / "web" / "cesium-config.js" def _sse(event: str, data: dict) -> bytes: """Encode one Server-Sent Event frame.""" return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode() def last_start_signal(to_gun_s: float, pre_start_s: int = 180, signals: list | None = None, gun_utc_ms: int | None = None) -> str | None: """Which start-sequence signal has most recently gone, at `to_gun_s` seconds before the gun. A commentator cannot say what moment the race is in without this, and until now nothing could: the warning, preparatory and one-minute signals are sounded by the hooter and beeped by the renderer, but nothing ever published WHICH of them had passed. Prefers the committee's own record (`raceSignals`) when the feed carries it — that is what actually happened, including a sequence that was started late. Falls back to the scheduled offsets, because a race running its sequence has passed those marks whether or not anyone logged them. The offsets come from `hooter.schedule`, which is what the horn itself uses: a second copy of RRS 26's timing here is a second thing to get wrong. """ if signals and gun_utc_ms is not None: # The committee's record is in absolute time; the phase is relative to the gun. recorded = [((s["tMs"] - gun_utc_ms) / 1000.0, s["signal"]) for s in signals if s.get("signal") in _SEQUENCE_SIGNALS and s.get("tMs") is not None] if recorded: # A record EXISTS, so it is the whole truth for this race — including "nothing has gone # yet". Falling through to the schedule here would let a sequence nobody ran contradict # the one that actually happened. gone = [entry for entry in recorded if entry[0] <= -to_gun_s] return max(gone)[1] if gone else None try: from v3.hooter.schedule import start_sequence_offsets offsets = [(off, sig.value if hasattr(sig, "value") else str(sig)) for off, sig in start_sequence_offsets(pre_start_s)] except Exception: # hooter not importable (isolated unit tests) offsets = [(-pre_start_s, "warning"), (-(pre_start_s - 60), "preparatory"), (-60, "one_minute"), (0, "start")] passed = [(off, sig) for off, sig in offsets if off <= -to_gun_s] return max(passed)[1] if passed else None #: The timed start sequence (RRS 26 / SI 11.1). Recalls and postponements are a different story and #: are narrated separately (`v3/rie/official_events.py`). _SEQUENCE_SIGNALS = frozenset({"warning", "preparatory", "one_minute", "start"}) def program_state(*, gun_utc_ms: int, track_start_utc_ms: int, sec: int, finished: bool, pre_start_s: int = 180, signals: list | None = None) -> dict: """The feed's EXPLICIT presentation phase for one tick — `{"state": "countdown"|"racing"|"result"}`. Clients must never have to infer "is this race live?" from standings alone: a stale finished leaderboard is visually identical to a live one (the "presentation stopped while the boats keep sailing" bug class, found 2026-07-09). Publishing the phase on every delta makes the distinction part of the wire contract: `result` = the shown race is decided (anything static on screen is intentional), `countdown` = pre-gun (the race clock counts down to the start), `racing` = live. The THIS-tick gun/track-start ride along so the client's race clock counts from the CURRENT race's gun. On a single race this equals the header gun; in a continuous pipeline (one fixed session gun so the client never reloads) each heat has its own gun, and the clock resets to it at each hand-over. `toGunS` and `signal` are what let the broadcast say WHERE IN THE RACE it is — "two minutes to the start", "the preparatory signal has gone". Every consumer used to re-derive the time to the gun from the two timestamps, and none of them could name the signal at all. """ to_gun_s = (gun_utc_ms - (track_start_utc_ms + sec * 1000)) / 1000.0 if finished: state = "result" elif to_gun_s > 0: state = "countdown" else: state = "racing" return {"state": state, "gunUtcMs": gun_utc_ms, "trackStartUtcMs": track_start_utc_ms, "toGunS": round(to_gun_s, 1), "signal": last_start_signal(to_gun_s, pre_start_s, signals, gun_utc_ms)} class _Handler(SimpleHTTPRequestHandler): """Serves static broadcast files; intercepts GET /events for the SSE stream.""" def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(_BROADCAST_ROOT), **kwargs) def log_message(self, *args): # keep the console quiet pass def do_GET(self): if self.path.split("?")[0] == _CESIUM_CONFIG_PATH and not _CESIUM_CONFIG_FILE.exists(): return self._serve_cesium_config() if self.path.split("?")[0].rstrip("/") == _EVENTS_PATH: if not self._feed_authorised(): self.send_response(403) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(b'{"error":"feed token required"}') return None return self._serve_events() return super().do_GET() def _serve_cesium_config(self) -> None: """`core/web/cesium-config.js`, generated from the environment when no file is on disk. The photoreal world needs a Cesium ion token, and that token used to reach production by being **on the disk of whoever ran the build**: the file is gitignored, so CI never had it and rendered stylised water, while a `gcloud builds submit` from a workstation baked it into the image and served it. Same tag, same command, a different picture, and nothing anywhere saying which — the image was no longer a function of the commit. So the build context excludes the file (`.gcloudignore`) and the DEPLOYMENT supplies the token, like every other secret. A file on disk still wins, because that is a developer's deliberate local choice; with neither, this 404s and `tiles.js` falls back to stylised water, exactly as it does today.""" token = os.environ.get("VR_CESIUM_ION_TOKEN") or "" if not token: return self.send_error(404, "no cesium-config.js and no VR_CESIUM_ION_TOKEN") body = (b"// Generated from VR_CESIUM_ION_TOKEN by live_feed_server. Not a file on disk:\n" b"// the image must not carry this token, and the deployment must not depend on\n" b"// which machine built it.\n" b"export const CESIUM_ION_TOKEN = " + json.dumps(token).encode() + b";\n") self.send_response(200) self.send_header("Content-Type", "text/javascript") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") # a rotated token must not be cached self.end_headers() self.wfile.write(body) return None def _feed_authorised(self) -> bool: """Is this caller allowed to read the live feed? **Unset `VR_FEED_TOKEN` ⇒ open**, which is what a local run, the simulator and the recorded demo all rely on — and what this service did unconditionally until now. Set it, and the feed becomes a shared-secret surface. Why a token in the URL rather than Cloud Run IAM: the consumer is an **OBS Browser source** on the on-site render machine. A browser cannot mint a Google-signed ID token, so `--no-allow-unauthenticated` would not secure this feed — it would switch the broadcast off. A query parameter is the one channel an OBS source URL actually has. It is weaker than IAM (URLs get logged) and does not pretend otherwise; it is the difference between "anyone who guesses the hostname gets a real-time tactical feed of the championship" and "you need the secret", which is the gap worth closing before a title event. The header form is accepted too, for `curl` and any non-browser consumer.""" want = os.environ.get("VR_FEED_TOKEN") or "" if not want: return True got = "" auth = self.headers.get("Authorization") or "" if auth.lower().startswith("bearer "): got = auth[7:].strip() if not got: qs = self.path.split("?", 1)[1] if "?" in self.path else "" got = (parse_qs(qs).get("token") or [""])[0] return bool(got) and hmac.compare_digest(got, want) def _serve_events(self): srv: LiveFeedServer = self.server # type: ignore[assignment] q = srv._subscribe() try: self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self.send_header("Access-Control-Allow-Origin", "*") # also usable cross-origin if needed self.end_headers() for chunk in srv._backlog(): # catch a late joiner up to "now" self.wfile.write(chunk) self.wfile.flush() while True: # then stream live frames until they leave try: chunk = q.get(timeout=_KEEPALIVE_S) except queue.Empty: # An SSE COMMENT — no event type, so no client listener fires and no feed state # changes. It exists so that SILENCE IS NOT AMBIGUOUS: it keeps intermediaries # (Cloud Run, proxies, the browser's own idle timers) from closing a quiet stream, # and it ticks the renderer's freshness clock, which is what lets the picture tell # "this race is quiet" apart from "the producer is dead". self.wfile.write(b": keepalive\n\n") self.wfile.flush() continue if chunk is None: # sentinel: server shutting down break self.wfile.write(chunk) self.wfile.flush() except ConnectionError: # client navigated away / reconnected — pass # covers reset/broken-pipe/abort (WinError 10053) finally: srv._unsubscribe(q) class LiveFeedServer(ThreadingHTTPServer): """Broadcasts init/delta/result SSE messages to connected renderer clients, and serves the static page. Thread-safe: `publish_*` may be called from the scoring/replay thread while clients stream.""" daemon_threads = True allow_reuse_address = True #: Backlog cap: the newest init (index 0) is always kept, plus at most this many delta/result #: chunks. A bounded heat replays in full for late joiners; an UNBOUNDED session (a full event #: day, the looping demo) would otherwise grow the backlog forever — memory plus an ever-slower #: late-joiner replay. 7200 ≈ two hours of 1 Hz deltas, comfortably more than any heat. MAX_BACKLOG = 7200 #: LIVE broadcasts pass this instead: keep the init + only a few seconds of deltas, so a viewer #: reconnecting across an SSE drop (e.g. Cloud Run's 60-min request cap) snaps straight to NOW #: rather than replaying up to ~2 h of stale history. Replay/VOD keeps the full MAX_BACKLOG. LIVE_BACKLOG = 5 def __init__(self, host: str = "0.0.0.0", port: int = 8765, *, live_backlog: int | None = None): super().__init__((host, port), _Handler) self._clients: set[queue.Queue] = set() self._lock = threading.Lock() self._history: list[bytes] = [] # init + a capped delta/result tail self._max_backlog = live_backlog if live_backlog is not None else self.MAX_BACKLOG @property def port(self) -> int: return self.server_address[1] def handle_error(self, request, client_address): """Stay quiet on client disconnects (browser/OBS reload, EventSource reconnect) — these raise ConnectionError/TimeoutError while reading or writing the socket and are normal, not faults.""" exc = sys.exc_info()[1] if isinstance(exc, (ConnectionError, TimeoutError)): return super().handle_error(request, client_address) # --- client registry (called from handler threads) --- def _subscribe(self) -> queue.Queue: q: queue.Queue = queue.Queue() with self._lock: self._clients.add(q) return q def _unsubscribe(self, q: queue.Queue): with self._lock: self._clients.discard(q) def _backlog(self) -> list[bytes]: with self._lock: return list(self._history) def _fan_out(self, chunk: bytes, *, reset: bool = False): with self._lock: if reset: self._history = [chunk] # a new init starts a fresh race history else: self._history.append(chunk) if len(self._history) > self._max_backlog + 1: # keep the init + the newest tail self._history = [self._history[0]] + self._history[-self._max_backlog:] clients = list(self._clients) for q in clients: q.put(chunk) # --- producer API (called from the scoring/replay thread) --- def start(self, background: bool = True) -> threading.Thread | None: """Begin serving. background=True returns immediately on a daemon thread; False blocks.""" if background: t = threading.Thread(target=self.serve_forever, daemon=True) t.start() return t self.serve_forever() return None def publish_init(self, meta: dict, course: dict, teams: list, params: dict | None = None, media: dict | None = None, video: list | None = None): """Send the static race header to clients (and reset the backlog for a new race). `media`/`video` carry the per-boat tile pictures/clips so the renderer's boat images show in live mode too.""" self._fan_out(_sse("init", {"meta": meta, "course": course, "teams": teams, "params": params or {}, "media": media or {}, "video": video or []}), reset=True) def publish_delta(self, sec: int, standings: dict, events: list, positions: dict, ais: list | None = None, program: dict | None = None, marks: dict | None = None, wind: dict | None = None, onboard: list | None = None): """Push one tick: live frontier positions + scored standings/events (from StreamingScorer). `ais` is the optional AIS context-vessel overlay for this tick (within ~500 m of the boat — an overlay, never scored); omitted/empty when no AIS feed is wired. `program` is the explicit presentation phase from `program_state()`; included on the wire whenever the driver computes it, so clients can't mistake a stale result for a live race. `marks` (role → {lat,lon}) is the optional set of course marks/RC-vessels that are MOVING this tick — the renderer glides them to the new station (e.g. the committee re-setting the course to a shifted wind between races). `wind` ({fromDeg, kts}) is the optional LIVE wind for this tick — the renderer re-points the on-water wind field + sea state (and thus the RC vessels weathervane) as the breeze veers.""" payload = {"sec": sec, "standings": standings, "positions": positions, "events": events, "ais": ais or []} if program is not None: payload["program"] = program if marks: payload["marks"] = marks if wind: payload["wind"] = wind if onboard: payload["rieOnboardDirectives"] = onboard # RIE onboard-manoeuvre inserts (visual sidecar) self._fan_out(_sse("delta", payload)) def publish_result(self, result: dict): """Announce the final result once the race finishes.""" self._fan_out(_sse("result", result)) def stop(self): """Unblock every streaming client and stop serving.""" with self._lock: clients = list(self._clients) for q in clients: q.put(None) self.shutdown() self.server_close()