""" V3-native near-live exporter — the near-live-replay backbone's export step, generalized. Reads a finished heat's `[t0, t1]` window straight from a **V3 store** (in-memory or Postgres), scores it with the shared engine, and publishes it as a demo-manifest race so the file-backed renderer / `record_demo.py --race ` can play or record it. This is the store→clip half of the pipeline first proven by `v3/sim/heat_clip.py` (which now imports from here); the sim/provider half lives in the sim, per docs/simulator-and-broadcast.md (sim ↔ server separation). from export_v3_instance import export_instance export_instance(store, race_id="heat-17", instance_id="racemgmt-15317", course_id="…", boats=[…], team_config={…}, title="Heat 17", user="trwc", t0_ms=…, gun_ms=…, t1_ms=…) There is a CLI for the ordinary case — replay a race the race office has already closed: python v3/broadcast/demo/export/export_v3_instance.py --list --owner trwc2026 python v3/broadcast/demo/export/export_v3_instance.py --instance racemgmt-1009 **NOTHING HERE KNOWS ABOUT THE SIMULATOR, and that is the point.** A replay is built from the record the SERVER kept — the track, the entries, the officer's gun and stop, the event config — so a simulated race and a real one are re-rendered by the same command with the same arguments, and the only difference is which race id is passed. The simulator is a client that puts telemetry on the wire; by the time a race is in the store it has left no trace that anything downstream reads. Keep it that way: an option here that means "this was simulated" would be a coupling, and the fact is already carried where it belongs — `meta.simulated`, published with the race by whoever ran it. """ from __future__ import annotations import json import sys from concurrent.futures import ThreadPoolExecutor from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE.parents[2] / "scoring")) # score_race (the shared engine) sys.path.insert(0, str(_HERE.parents[2] / "ingest")) # broadcast_feed + store types sys.path.insert(0, str(_HERE.parent.parent / "live")) # serve_v3_live (meta derivation) RACES_DIR = _HERE.parent / "data" / "races" #: How long a replay's clip URLs stay fetchable. A v4 signature may last **seven days**; a render #: takes 1.5-2 hours and the recorder retries for longer than that after a GPU crash, so the 30-minute #: default the LIVE read plane uses (`read_server._signed_clip_url`) would expire mid-capture and the #: boat-cam tiles would go dark partway through the film — silently, because `app.js` holds the last #: picture on a fetch error. Six days leaves a margin under the ceiling and still says plainly that #: these URLs are perishable: a feed is render input, not an archive of the footage. CLIP_URL_MINUTES = 6 * 24 * 60 #: How many clip URLs to sign at once. Each one is an IAM `signBlob` round trip and nothing else — #: measured 855 ms on this workstation — so the work is pure latency and the pool is sized for #: concurrency, not for cores. A 19-minute race with four boats filming every two seconds is a few #: thousand signatures: about an hour sequentially, a couple of minutes like this. SIGN_WORKERS = 32 def export_media_map(store, boats: list[str], *, t0_ms: int, t1_ms: int, # noqa: ANN001 — Store clip_minutes: int = CLIP_URL_MINUTES) -> dict: """The boat-cam tiles for the replay — the thing the export has never produced. `docs/replay-production.md` blamed the missing tiles on a credential ("a user ADC cannot sign v4 URLs"), and that is true but second: `build_feed` returns `{meta, course, teams, frames}` and the exporter never called `media_from_store` at all, so no exported feed has EVER carried a `media` key. With `feed.media = {}` the renderer's `mediaAt()` returns null and `app.js` collapses every inset. Fixing only the credential would have changed nothing. URLS, NOT FILES. Downloading is the durable choice and it does not scale here: boat 1 alone has 985 clips / 819 MB in `sim2001`'s hour, about 3.7 GB for the four boats and ~600 GB for a championship. So the feed carries signed URLs and is render input with a shelf life — which is also why `CLIP_URL_MINUTES` is days rather than the live plane's minutes. Returns `{}` when nothing can be signed, which is exactly what a workstation without a signing credential gets. That is the same picture as before this existed, so an export never fails for want of a credential — it just cannot add cameras.""" try: from read_server import _iam_signer, _signed_clip_url except Exception: # noqa: BLE001 — no GCS libs ⇒ no tiles, not a crash return {} from media_map import PRE_RACE_CLIP_MARGIN_MS from serve_v3_live import media_from_store # PASS 1 — WHICH clips, without signing any of them. A signature is a network call and this is a # loop: measured 855 ms per IAM `signBlob` round trip, so a 19-minute race with four boats filming # every two seconds is the better part of an hour, and the first attempt at this simply never # finished. `read_server` documents the same trap for `clips.json`. # # The placeholder is a real per-path token so `media_map` can do its own job — dropping clips # outside the race window — and we then sign only what SURVIVED that. Filtering here instead would # be a second implementation of the window, which is the thing the shared `media_map` exists to # prevent. placeholder_of: dict[str, str] = {} def mark(segment): # noqa: ANN001, ANN202 — a VideoSegment return placeholder_of.setdefault(segment.gcs_path, f"\0sign:{len(placeholder_of)}") media = media_from_store(store, boats, t0_ms, clip_url=mark, since_ms=t0_ms - PRE_RACE_CLIP_MARGIN_MS, from_ms=t0_ms, until_ms=t1_ms) path_of = {token: path for path, token in placeholder_of.items()} wanted = {row["file"] for rows in media.values() for row in rows if row.get("file") in path_of} if not wanted: return media # PASS 2 — sign them concurrently. Every call is a round trip to Google and nothing else, so this # is entirely latency; the pool turns an hour into a couple of minutes. Warm the signer first so # the credential refresh happens once, on this thread, rather than in a dozen at the same moment. _iam_signer() signed: dict[str, str] = {} failed = 0 def sign(token: str) -> tuple[str, str | None]: try: return token, _signed_clip_url(path_of[token], minutes=clip_minutes) except Exception: # noqa: BLE001 — see the docstring return token, None print(f" boat-cam: signing {len(wanted)} clip URL(s) ...") with ThreadPoolExecutor(max_workers=SIGN_WORKERS) as pool: for token, url in pool.map(sign, sorted(wanted)): if url: signed[token] = url else: failed += 1 for rows in media.values(): for row in rows: row["file"] = signed.get(row.get("file"), row.get("file")) # Drop the rows whose signature failed rather than leaving a placeholder the renderer would try # to fetch as a relative path. media = {boat: [r for r in rows if not str(r.get("file", "")).startswith("\0sign:")] for boat, rows in media.items()} media = {boat: rows for boat, rows in media.items() if rows} if failed: # SAID OUT LOUD. Absent tiles look like a race that had no cameras, and the difference — # "nobody filmed this" versus "this workstation cannot sign a URL" — is invisible in the film. print(f" boat-cam: {failed} clip(s) could not be signed — those tiles will be absent. " f"A user login has no signing key; authorise a service account that does.") return media def export_instance(store, *, race_id: str, instance_id: str, course_id: str, # noqa: ANN001 — Store boats: list[str], team_config: dict | None, title: str, user: str, t0_ms: int, gun_ms: int, t1_ms: int, source: str = "real", rie: bool = False, races_dir: Path = RACES_DIR, clip_minutes: int = CLIP_URL_MINUTES, mark_frames_step_ms: int | None = None, ghost_boats: list[str] | None = None, entries_at_ms: int | None = None) -> Path: """Store → `build_feed` → `enrich` → a manifest race (`data` inlined, as `record_demo` needs) plus a standalone `.json` for inspection. Existing manifest races are preserved; re-exporting the same `race_id` replaces its entry. Returns the standalone json path. `build_feed` attaches the instance's committee signals (`raceSignals`), so OCS/shorten SCORING flows into the export automatically. `rie=True` additionally routes through the RIE producer so those signals are NARRATED (official-event beats) — the offline counterpart to the live `--rie` path.""" from broadcast_feed import build_feed from score_race import enrich from serve_v3_live import ( apply_broadcast_config, apply_event_config, event_id_for_instance, meta_course_from_store, ) meta, course, _ = meta_course_from_store( store, instance_id=instance_id, course_id=course_id, title=title, user=user, gun_ms=gun_ms, track_start_ms=t0_ms, track_end_ms=t1_ms) # THE SAME OVERLAYS THE LIVE PRODUCER APPLIES, in the same order (event config, then the race's # own broadcast config, instance keys winning). A re-render is the SAME RACE as the one that went # out live, so it must not be a different production: without these the export carried no # `eventName` — the picture's upper-left anchor, and with it the `(Simulated)` the simulator # declares — no hull livery, and no vessel cameras. The race looked right live and arrived in the # replay wearing none of the event's identity, which reads as a rendering fault rather than as a # missing overlay. meta = apply_event_config(store, event_id_for_instance(store, instance_id), meta=meta, boats=boats) meta, stored_cfg = apply_broadcast_config(store, instance_id, meta=meta) feed = build_feed(store, instance_id, boats, meta=meta, course=course, team_config=team_config or (stored_cfg or {}).get("teamConfig"), course_id=course_id, mark_frames_step_ms=mark_frames_step_ms, ghost_boats=ghost_boats, entries_at_ms=entries_at_ms) media = export_media_map(store, boats, t0_ms=t0_ms, t1_ms=t1_ms, clip_minutes=clip_minutes) if media: feed["media"] = media if rie: sys.path.insert(0, str(_HERE.parents[3])) # repo root, for v3.rie from v3.rie.broadcast_bridge import produce_feed scored, _ = produce_feed(feed, mode="rie", scorer=enrich) else: scored = enrich(feed) scored["schema"] = "vr-broadcast-demo/1" scored["source"] = source out = races_dir / f"{race_id.replace('-', '_')}.json" # INDENTED, because this copy exists to be READ. It is the one nobody's code loads — the renderer # plays the manifest's inlined `data`, and this sits beside it so a person can look at what was # scored: the meta the overlays produced, the teams that were resolved, the events the engine # emitted. Written compact it was a single 3 MB line, which an editor will open and no one can # read, so the artefact named "for inspection" could not be inspected. The manifest keeps its # compact form on purpose (see below) — that one is machine input, and large. out.write_text(json.dumps(scored, ensure_ascii=False, indent=2), encoding="utf-8") m, res = scored["meta"], scored.get("result") # record_demo needs the feed INLINED in the manifest entry ("data": …, like the GSYS races); # the standalone json is kept alongside for inspection. entry = {"id": race_id, "title": m["title"], "data": scored, "trackStartUtcMs": m["trackStartUtcMs"], "gunUtcMs": m["gunUtcMs"], "trackEndUtcMs": m["trackEndUtcMs"], "winnerSec": res["winnerSec"] if res else None, "winnerName": res["winnerName"] if res else None} mf_path = races_dir / "manifest.json" manifest = json.loads(mf_path.read_text(encoding="utf-8")) manifest["races"] = [r for r in manifest["races"] if r.get("id") != race_id] + [entry] # ATOMIC, because this file is 22 MB and something is usually READING it. Exporting a race day # is a loop — export, render, export the next — and `record_demo` plays the race it is recording # out of this same manifest. A plain `write_text` truncates and refills over many seconds, so a # reader that arrives mid-write gets a torn file and the render dies on invalid JSON, hours in. # Writing a sibling and renaming makes the swap a single directory operation: a reader sees the # old manifest or the new one, never half of either. # compact, matching the manifest's original format — pretty-printing balloons the 14 MB file tmp = mf_path.with_suffix(".json.tmp") tmp.write_text(json.dumps(manifest, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") tmp.replace(mf_path) print(f"[export] {race_id}: frames={len(scored['frames'])} events={len(scored['events'])} " f"winner={res['winnerName'] if res else None}", flush=True) return out def _fleet_ghosts(store, owner: str, racing: list[str]) -> list[str]: # noqa: ANN001 — Store """The account's OTHER hulls — the boats that are on the water but not in this race. A team-racing championship runs one flight while the next is milling around waiting, and the picture drew an empty sea around four boats. These are real hulls with real fixes, so they are drawn from the same telemetry as the competitors and are not a decoration. The fleet comes from `known_boats()` (every boat ref that has ever written a nav point), scoped to this race's OWNER — never a hardcoded hull list, or a second account's regatta would need an edit here, and TRWC's own numbering would leak into engine-adjacent code. The owner prefix is taken off a slot ref exactly as the caller took it, so the two cannot disagree about what an account is called; boats with no fixes in the window cost nothing, because `boat_frames` simply omits a boat it finds no points for. """ known = getattr(store, "known_boats", None) if known is None: # a store without the method (older mocks) → no ghosts return [] racing_set = set(racing) return [b for b in known() if b not in racing_set and str(b).split(":", 1)[0] == owner] #: How much of the START SEQUENCE a replay opens with, by default. Three minutes, because that is #: where the race is decided — a replay that opens at the gun has thrown away the approach, the #: line-up and the timing that produced the start it then shows. DEFAULT_PRE_START_S = 180 #: A little past the officer's stop, so the finish and the result hold are inside the window. DEFAULT_TAIL_S = 30 def roster_at_gun(store, instance_id: str, *, # noqa: ANN001 — Store gun_ms: int, stop_ms: int) -> tuple[list[str], int | None]: """The boats entered in THIS race at its gun, and the instant the roster was read from. Returns `([], None)` when the race office recorded no roster for the race at all — the caller says what that means for what it was about to do, because nothing here can invent who was sailing. It is a function rather than a paragraph inside the exporter because a second consumer arrived (`gen_boatcam_instance.py`, which films the same four boats): a roster rule that exists twice is one that drifts, and the widening below is exactly the kind of hard-won detail the second copy would be written without. """ gun, stop = int(gun_ms), int(stop_ms) # The roster AT THE GUN — who was actually racing, not who holds the hulls now. Between one # race's stop and the next race's start the same hulls are reallocated, so reading the entries # at export time would export the NEXT race's line-up onto this race's track. entries = store.entries_covering_instance(instance_id, gun) boats = sorted({e.boat for e in entries if e.instance_id == instance_id and not e.is_tombstone}) entries_at_ms = None if not boats: # The gun and the allocation window do not overlap. Measured on TRWC 2026: Race 22's roster # was recorded EIGHT MINUTES after its gun and Race 72's was released 39 seconds before it — # ordinary race-office bookkeeping, and it refused two otherwise complete races. # # Widening is safe HERE and nowhere else, because every row is filtered to THIS instance by # id: the danger the at-the-gun rule guards against is picking up the NEXT race's line-up # after the hulls are reallocated, and a row carrying another race's instance id can never # satisfy that filter. So this cannot import a foreign roster — it can only find this race's # own, recorded slightly off its gun. Races with no roster at all still refuse, because # nothing here can invent who was sailing. # Probes on BOTH sides and close in: a roster can be released seconds before the gun (Race 72 # closed its allocation 39 s early) or opened minutes after it (Race 22, +8 min), so a couple # of coarse points on one side only misses both shapes. for probe in (stop, (gun + stop) // 2, gun - 60_000, gun - 300_000, gun - 900_000, stop + 300_000, stop + 900_000): boats = sorted({e.boat for e in store.entries_covering_instance(instance_id, probe) if e.instance_id == instance_id and not e.is_tombstone}) if boats: # The TEAMS must come from these same rows. Recovering only the boat LIST left the # feed with positions and no teams, and the film rendered as an empty sea -- no # boats, no matchup, no standings. That is worse than the refusal it replaced, # because it looks like a broken renderer instead of missing race-office data. entries_at_ms = probe off_s = (probe - gun) / 1000.0 print(f"[roster] {instance_id}: no roster AT the gun; used this race's own entries " f"from {off_s:+.0f}s ({len(boats)} boats)", flush=True) break return boats, entries_at_ms def export_finished_instance(store, instance_id: str, *, # noqa: ANN001 — Store race_id: str | None = None, pre_start_s: int = DEFAULT_PRE_START_S, tail_s: int = DEFAULT_TAIL_S, rie: bool = False, source: str = "real", races_dir: Path = RACES_DIR, clip_minutes: int = CLIP_URL_MINUTES, mark_frames_step_ms: int | None = None, ghosts: bool = False) -> Path: """Export a race the race office has CLOSED, resolving its whole window from the record. The caller passes an instance id and nothing else: the course, the gun, the stop, the roster and the team display all come off the race itself. That is what makes one command work on a real championship race and on a simulated one — there is no argument in which the two could differ. Refuses a race with no official stop. An open race has no end to replay, and guessing one from the clock would cut the finish off the very replay somebody asked for.""" inst = store.get_race_instance(instance_id) if inst is None: raise SystemExit(f"no such race: {instance_id}") gun = getattr(inst, "official_start_utc_ms", None) stop = getattr(inst, "official_stop_utc_ms", None) if not gun: raise SystemExit(f"{instance_id} was never armed — there is no race to replay") if not stop: raise SystemExit(f"{instance_id} has no official stop — the race office has not closed it") course_id = getattr(inst, "course_id", None) if not course_id: raise SystemExit(f"{instance_id} carries no course") # The roster AT THE GUN — who was actually racing, not who holds the hulls now. Between one # race's stop and the next race's start the same hulls are reallocated, so reading the entries # at export time would export the NEXT race's line-up onto this race's track. boats, entries_at_ms = roster_at_gun(store, instance_id, gun_ms=int(gun), stop_ms=int(stop)) if not boats: raise SystemExit(f"{instance_id} had no boats entered at its gun") cfg = store.get_broadcast_config(instance_id) or {} owner = str(boats[0]).split(":", 1)[0] # the account NAME, off a slot ref (never an owner column) return export_instance( store, race_id=race_id or "".join(c if c.isalnum() else "_" for c in instance_id), instance_id=instance_id, course_id=course_id, boats=boats, team_config=cfg.get("teamConfig"), title=getattr(inst, "name", None) or instance_id, user=owner, t0_ms=int(gun) - pre_start_s * 1000, gun_ms=int(gun), t1_ms=int(stop) + tail_s * 1000, source=source, rie=rie, races_dir=races_dir, clip_minutes=clip_minutes, mark_frames_step_ms=mark_frames_step_ms, ghost_boats=_fleet_ghosts(store, owner, boats) if ghosts else None, entries_at_ms=entries_at_ms) def _resolve_owner(store, dsn: str, owner: str | None) -> str | None: # noqa: ANN001 — Store """`--owner` as the value `race_instances.owner_user_id` actually holds. An account has two spellings and the difference is load-bearing: a slot prefix is the account NAME (`trwc2026:boat:7`) while every owner column is a `users.id`. Comparing one to the other matches nothing, and because this is a FILTER the failure is silent — no rows reads exactly like "this account has no races". So a name is resolved through the accounts registry, and the literal is kept when it cannot be (a store seeded before accounts existed stores the name itself, which is why both spellings are present in one column on production).""" if not owner: return None try: from accounts import PostgresAccounts from postgres_store import connect as _connect from slots import user_id_for_slot_owner return user_id_for_slot_owner(PostgresAccounts(_connect(dsn)), owner) except Exception: # noqa: BLE001 — no registry → the literal was always right return owner def _main() -> int: import argparse import os ap = argparse.ArgumentParser( description="Export a finished race as a manifest race for offline replay/recording. " "Works identically on a real race and a simulated one — see the module docstring.") ap.add_argument("--instance", help="race instance id to export (e.g. racemgmt-1009)") ap.add_argument("--list", action="store_true", help="list closed races instead of exporting") ap.add_argument("--owner", help="with --list: the account (users.id) whose races to list") ap.add_argument("--course", help="with --list: only races on this course") ap.add_argument("--race-id", help="manifest id to write (default: the instance id, sanitised)") ap.add_argument("--pre-start-s", type=int, default=DEFAULT_PRE_START_S, help=f"seconds of start sequence to open with (default {DEFAULT_PRE_START_S} = T-3)") ap.add_argument("--tail-s", type=int, default=DEFAULT_TAIL_S, help=f"seconds past the official stop (default {DEFAULT_TAIL_S})") ap.add_argument("--ghosts", action="store_true", help="also carry the fleet's OTHER hulls (present on the water, not in this " "race) as `ghostFrames`, drawn de-emphasised and never scored") ap.add_argument("--rie", action="store_true", help="narrate through the RIE producer") ap.add_argument("--source", default="real", help="feed's `source` label (default: real)") ap.add_argument("--mark-frames-step-ms", type=int, default=None, help="follow marks that MOVE, sampling every N ms (e.g. 10000). Off by default: " "each sample re-reads every bound buoy's track, measured at 12 s per sample " "over the Cloud SQL proxy. A course of static coordinates costs nothing") ap.add_argument("--clip-minutes", type=int, default=CLIP_URL_MINUTES, help="how long the boat-cam URLs stay fetchable (default ~6 days; v4 caps at 7). " "A render takes hours, so the live plane's 30 min would expire mid-capture") args = ap.parse_args() dsn = os.environ.get("VR_DATABASE_URL") if not dsn: raise SystemExit("set $VR_DATABASE_URL (e.g. prod via the Cloud SQL proxy)") from postgres_store import PostgresStore, connect store = PostgresStore(connect(dsn)) if args.list: owner = _resolve_owner(store, dsn, args.owner) rows = [r for r in store.list_race_instances(owner_user_id=owner) if getattr(r, "official_stop_utc_ms", None) and (not args.course or getattr(r, "course_id", None) == args.course)] rows.sort(key=lambda r: r.official_start_utc_ms or 0) for r in rows: mins = (r.official_stop_utc_ms - r.official_start_utc_ms) / 60000 print(f"{r.id:<24} {getattr(r, 'name', '') or '':<16} " f"{getattr(r, 'course_id', '') or '':<20} {mins:5.1f} min") print(f"{len(rows)} closed race(s)") if not rows and args.owner: # NO ROWS IS NOT AN ANSWER, it is two answers wearing the same face: "this account has no # closed races" and "nothing is stored under that spelling". `race_instances.owner_user_id` # holds a `users.id` for accounts that have one and a literal name for those seeded before # they did — measured on production, both are present in the same column — so a name # passed for a UUID-backed account matches nothing and says nothing. Say which it was. print(f"\nnothing closed under owner {args.owner!r}. Owners that DO have closed races:") seen: dict[str, int] = {} for r in store.list_race_instances(): if getattr(r, "official_stop_utc_ms", None): seen[r.owner_user_id or "?"] = seen.get(r.owner_user_id or "?", 0) + 1 for k, n in sorted(seen.items(), key=lambda kv: -kv[1]): print(f" {k:<40} {n:4d}") return 0 if not args.instance: raise SystemExit("--instance is required (or --list to see what is available)") export_finished_instance(store, args.instance, race_id=args.race_id, pre_start_s=args.pre_start_s, tail_s=args.tail_s, rie=args.rie, source=args.source, clip_minutes=args.clip_minutes, mark_frames_step_ms=args.mark_frames_step_ms, ghosts=args.ghosts) return 0 if __name__ == "__main__": raise SystemExit(_main())