""" Which race is ON AIR — the broadcast's hand-over policy. A competition runs races back-to-back: the operator arms the next heat while the current one is still sailing. Two rules follow, and they are the whole of this module: 1. **A new start never cuts away from the race on air.** The broadcast stays with the race it is showing — camera, standings and commentary — until that race has its WINNER (plus a short hold so the finish and the wrap-up actually air) or it closed without one (abandoned, or the race simply ended). Later starts wait their turn. 2. **Moving on is not restarting.** The producer rebuilds the presentation for the next race on the SAME server and the SAME SSE connection; the renderer swaps the race in place (`live/web/live_feed.js` → `onRaceChange`). No process restart, no page reload, no black screen. Both the production live producer (`serve_v3_live.py --follow`) and the sim's operator-driven bridge (`v3/sim/bridge_3d.py`) use this — one policy, so the sim exercises what production does. Pure over injected callables and an injected clock: the whole hand-over runs in tests with no servers, sockets or sleeps. """ from __future__ import annotations import time from collections.abc import Callable #: The start sequence a race goes on air for (SI 11.1 warning → gun). A race is broadcast from #: its ARM, not from its gun — the countdown is the story. PRE_START_MS = 180_000 #: How long a finished race stays on air before the next one goes up — the finish, the result and the #: wrap-up commentary all have to air. Matches the looping-track driver's between-heats result hold. #: #: **Coupled to the client's `params.winnerGraceS`** (40 s; `core/web/app.js`, `overlay_app.js` and #: every manifest producer). The renderer holds the winner gallery for `winnerGraceS` after the #: winner and only then looks for the next race, so this value must stay the SHORTER of the two: #: publish the next race before the client is ready and it is presented over the old race's wrap-up. #: The 15 s margin is the whole safety budget — change either number and check the other. RESULT_HOLD_S = 25 class OnAirQueue: """The race currently ON AIR, and the ones waiting. Releases (a race whose start signal has gone) are `offer`ed as they happen; `take_next` returns the race to put up, or None to stay with the one on air. The on-air race is handed over only when its story is complete: * it has a scored **WINNER** — held a further `winner_hold_s` so the finish airs; or * it **closed without one** (`over=True`: abandoned, or ended with no clean finish), after at least `min_air_s` on air so a race that finished off-screen still gets its airtime. With nothing waiting, the finished race stays up — the broadcast is never blank.""" def __init__(self, *, winner_hold_s: float = RESULT_HOLD_S, min_air_s: float | None = None, now_fn: Callable[[], float] = time.time) -> None: self.winner_hold_s = float(winner_hold_s) self.min_air_s = float(winner_hold_s if min_air_s is None else min_air_s) self._now = now_fn self._pending: list[tuple] = [] self._on_air: str | None = None self._on_air_at: float = 0.0 self._won_at: float | None = None @property def on_air(self) -> str | None: return self._on_air def offer(self, instance: str, *rest) -> bool: """A released race joins the queue (`rest` is carried through to `take_next` untouched — e.g. the gun, or the sim's release base). Returns True if it has to WAIT because something else is on air. A re-arm of an already-waiting race supersedes its earlier release.""" self._pending = [p for p in self._pending if p[0] != instance] self._pending.append((instance, *rest)) return self._on_air is not None and instance != self._on_air def waiting(self) -> list[str]: return [p[0] for p in self._pending] def note_result(self, has_winner: bool) -> None: """Per tick: whether the on-air race's scorer has produced its result yet — the winner call is what starts the hold-on-the-finish clock.""" if has_winner and self._won_at is None: self._won_at = self._now() def done(self, *, over: bool = False) -> bool: """Has the race on air finished its story — a scored winner plus the hold on the finish, or it closed without one (`over`) and has had its minimum airtime? Answered whether or not anything is waiting, because the two cases differ: with a race queued the producer hands over to it; with NOTHING queued it must return to the waiting presentation rather than sit frozen on the result for ever (`release()`). Nothing on air ⇒ trivially done.""" if self._on_air is None: return True now = self._now() if self._won_at is not None: return now - self._won_at >= self.winner_hold_s return over and now - self._on_air_at >= self.min_air_s def release(self) -> None: """Nothing is on air any more — the producer went back to the waiting presentation. The next release goes up immediately instead of queueing behind a race that is no longer showing.""" self._on_air = None self._won_at = None def take_next(self, *, over: bool = False) -> tuple | None: """The next race to put on air (the tuple passed to `offer`), or None to stay put. `over` = the on-air race closed WITHOUT a scored winner.""" if not self._pending or not self.done(over=over): return None nxt = self._pending.pop(0) self._on_air, self._on_air_at, self._won_at = nxt[0], self._now(), None return nxt class StartedRaces: """The operator's races on one course, reported as they GO OFF. `list_instances()` returns the management rows (`store.list_race_instances(...)` or the wire rows of `GET /v3/manage/instances`); `poll()` returns every race whose START SEQUENCE has begun since the last call — gun set, no stop, not abandoned or cancelled, and now within `lead_ms` of the gun — oldest gun first, so the queue airs them in the order they are sailed. `closed()` answers the other half of the hand-over: whether the race on air has been stopped or abandoned by the operator. `lead_ms` is why a race goes on air at the ARM rather than at the gun: the pre-start IS broadcast material (the countdown, the fleet setting up, the line approach), and waiting for the gun would put the race on screen already started. It defaults to the SI pre-start; a race armed further out than that simply waits until its sequence begins. A race whose gun is MOVED after it was reported is reported again (the operator re-armed it); one that is stopped and re-armed likewise. That is what makes it safe to leave the producer running for a whole regatta day.""" #: field aliases so the same watcher reads store dataclasses AND management wire rows _FIELDS = {"id": ("instanceId", "instance_id", "id"), "gun": ("officialStartUtcMs", "official_start_utc_ms"), "stop": ("officialStopUtcMs", "official_stop_utc_ms"), "state": ("raceState", "race_state"), "course": ("courseId", "course_id"), "owner": ("ownerUserId", "owner_user_id"), "name": ("name",)} def __init__(self, list_instances: Callable[[], list], *, course_id: str | None = None, owner: str | None = None, lead_ms: int = PRE_START_MS, now_ms: Callable[[], int] | None = None) -> None: self._list = list_instances #: BOTH scopes are optional, and both are filters rather than requirements. A race carries its #: own course, so a producer following the operator need not be told a course id in advance — #: pass one only to watch a single course when several are racing at once. `owner` is the #: scope that usually matters instead: whose regatta this broadcast is showing. self.course_id = course_id self.owner = owner self.lead_ms = int(lead_ms) self._now_ms = now_ms or (lambda: int(time.time() * 1000)) self._reported: dict[str, int] = {} # instance -> the gun we last reported it with self._rows: dict[str, dict] = {} @classmethod def _get(cls, row, key: str): # noqa: ANN001 — dataclass or wire dict for name in cls._FIELDS[key]: if isinstance(row, dict): if name in row: return row[name] elif hasattr(row, name): return getattr(row, name) return None def _snapshot(self) -> dict[str, dict]: rows = {} for row in (self._list() or []): rid = self._get(row, "id") if rid is None: continue if self.course_id is not None and self._get(row, "course") not in (None, self.course_id): continue if self.owner is not None and self._get(row, "owner") not in (None, self.owner): continue rows[rid] = {"gun": self._get(row, "gun"), "stop": self._get(row, "stop"), "state": self._get(row, "state"), "name": self._get(row, "name")} self._rows = rows return rows def poll(self) -> list[tuple[str, int]]: """`[(instance_id, gun_ms)]` for every race whose start sequence has begun since the last poll, oldest gun first.""" now = self._now_ms() out = [] for rid, row in self._snapshot().items(): gun, stop, state = row["gun"], row["stop"], row["state"] if (gun is None or gun - self.lead_ms > now or stop is not None or state in ("abandoned", "cancelled")): continue if self._reported.get(rid) == gun: # already reported under this gun continue self._reported[rid] = gun out.append((rid, int(gun))) return sorted(out, key=lambda p: p[1]) def closed(self, instance_id: str) -> bool: """True once the operator has stopped or abandoned this race (the last snapshot's view).""" row = self._rows.get(instance_id) if row is None: return False return row["stop"] is not None or row["state"] in ("abandoned", "cancelled") def name_of(self, instance_id: str) -> str | None: row = self._rows.get(instance_id) return row["name"] if row else None