""" Boat-cam OBS supervisor — drives `OBSController.reconcile_boat_sources` from the live race. Watches which boats are currently *sending* (a position fix and/or a fresh media clip on each tick) and keeps OBS holding exactly one boatcam Browser source per live boat — adding one the moment a boat appears, removing it after a grace period of silence. Anti-flap and on-air protection live here so the controller (director_obs.py) stays a thin obs-websocket wrapper: * grace period — a boat is "active" while it has been seen within the last `grace_s` seconds, so a normal burst-y GPS/upload gap doesn't churn sources (add on first sight, remove only after `grace_s` of silence). * on-air guard — never remove a boat whose full-screen scene is the current OBS program scene (defer until it's off-air), so a live cut never goes black. Wire it from the live driver (live/serve_live.py `--obs`): call `observe(sec, device_ids)` then `tick(sec)` each frame. Pure logic (active-set, grace, protect) is unit-tested with a fake controller; the OBS calls themselves want a live OBS to verify (see director_obs.py). """ from __future__ import annotations from collections.abc import Callable, Iterable class BoatSourceSupervisor: def __init__(self, obs, base_url: str, *, grace_s: float = 45.0, per_boat_scenes: bool = True, grid: bool = True, grid_scene: str = "Boat Cams", label_of: Callable[[str], str] | None = None, canvas: tuple[int, int] = (1920, 1080), query: str = "", composite: str = ""): self.obs = obs self.base_url = base_url self.grace_s = grace_s self.per_boat_scenes = per_boat_scenes self.grid = grid self.grid_scene = grid_scene self.canvas = canvas self.label_of = label_of or (lambda did: did) #: Extra query on every boat URL — how the sources are put on the LIVE HLS stream rather than #: the delayed-clip surface. Re-read per tick by the caller, because the instance changes when #: the producer moves to the next heat. self.query = query #: The COMPOSITE — the 3D broadcast — tiled in the grid beside the cameras, so the panel shows #: the programme and its sources together instead of only the parts. self.composite = composite self._last_seen: dict[str, float] = {} # device_id -> sec last seen sending def observe(self, sec: float, device_ids: Iterable[str]): """Record that these boats are sending as of race-second `sec`.""" for did in device_ids: self._last_seen[did] = sec def active(self, sec: float) -> dict[str, str]: """Boats seen within the grace window → {device_id: label}.""" return {did: self.label_of(did) for did, t in self._last_seen.items() if sec - t <= self.grace_s} def protect_set(self) -> set[str]: """Boats whose full-screen scene is currently on the program output — never auto-removed.""" if not self.per_boat_scenes: return set() try: prog = self.obs.current_program_scene() except Exception: return set() # can't tell → don't block removals if not prog: return set() return {did for did in self._last_seen if self.obs.boat_scene_name(self.label_of(did)) == prog} def set_grid_scene(self, name: str) -> None: """Rename the grid scene live — e.g. to the two teams now racing ("Boat JPN & GBR"). Renames the existing OBS scene in place so its tiled items follow the heat; a no-op when unchanged, and if the old scene is absent the next tick() simply creates it under `name`.""" if not name or name == self.grid_scene: return try: self.obs.rename_scene(self.grid_scene, name) # keeps the tiled items; no-op if old missing except Exception: # never let a rename hiccup break the loop pass self.grid_scene = name def tick(self, sec: float) -> dict: """Reconcile OBS sources to the active set as of `sec`. Returns the controller's change dict.""" return self.obs.reconcile_boat_sources( self.active(sec), self.base_url, per_boat_scenes=self.per_boat_scenes, grid=self.grid, grid_scene=self.grid_scene, canvas=self.canvas, protect=self.protect_set(), query=self.query, composite=self.composite) def grid_scene_for_teams(boats: Iterable[tuple]) -> str: """Grid scene name for the heat: ``"Boat & "`` for the two teams racing, ordered by each team's LOWEST sail number — stable within a heat, so a lead change never renames the scene. `boats` is an iterable of ``(deviceId, sailId, teamName)``. Falls back to "Boat Cams" when no team is known.""" low: dict[str, int] = {} for _did, sail, team in boats: if not team: continue try: s = int(sail) except (TypeError, ValueError): s = 999 low[team] = min(low.get(team, 999), s) teams = sorted(low, key=lambda t: low[t])[:2] return "Boat " + " & ".join(teams) if teams else "Boat Cams" def run_from_feed(obs, feed_url: str, base_url: str, *, grace_s: float = 12.0, clean: bool = True, min_tick_s: float = 5.0, log=print, read_base: str = "", lag_ms: float | None = None, composite: str = "", clips: bool = False) -> None: """Drive the boat-cam supervisor from a live SSE race feed: keep OBS holding one boat-cam per boat currently racing, in a grid scene named after the two teams (``grid_scene_for_teams``). Obsolete boats drop after ``grace_s``. Runs until the feed closes or the process is interrupted. The cameras are put on the LIVE HLS STREAM, locked to the same presentation clock as the 3D scene: the feed header names the race, and `read_base`/`lag_ms` say where the media is served from and how far behind now to play. **Without this the sources fall back to the delayed-clip surface**, where a boat shows a picture only when a finished clip happens to cover the current window — which reads as "most of the cameras are black". The race changes under us as the producer moves to the next heat, so the query is re-read from every header and the sources are re-pointed. `clips` keeps the sources on the DELAYED-CLIP surface instead. Needed against a server that has not fragmented its clips (`VR_MEDIA_FRAGMENT` off — the default, and every local sim run): a plain MP4 cannot be an HLS segment, so those playlists are all gaps and an HLS source would show a perfectly black tile. The clip surface plays either container. `composite` (e.g. the 3D broadcast's scene name) is tiled in the grid beside the cameras, so the panel shows the programme together with the sources feeding it.""" import json import time import urllib.parse import urllib.request if clean: # fresh start: drop stale boatcam inputs + "Boat *" scenes for inp in sorted(obs._input_names()): if inp.startswith("boatcam:"): try: obs.remove_input(inp) except Exception: pass for sc in sorted(obs._scene_names()): if sc.startswith("Boat "): try: obs.remove_scene(sc) except Exception: pass from obs_boatcams import boat_query # sibling: the one definition of the live-HLS query labels: dict[str, str] = {} sup = BoatSourceSupervisor(obs, base_url, grace_s=grace_s, composite=composite, label_of=lambda did: labels.get(did, did)) last_tick, last_key, boats = 0.0, None, [] req = urllib.request.Request(feed_url, headers={"Accept": "text/event-stream"}) with urllib.request.urlopen(req) as resp: # noqa: S310 — trusted local/broadcast feed for raw in resp: line = raw.decode("utf-8", "replace").strip() if line.startswith("data:"): try: delta = json.loads(line[5:].strip()) except Exception: continue inst = ((delta.get("meta") or {}).get("instanceId") or (delta.get("meta") or {}).get("instance")) if inst: # a header — it names the race the cameras must play q = "" if clips else boat_query(str(inst), read_base, lag_ms) if q != sup.query: sup.query = q log(f"[obs] cameras follow {inst}: {q or '(delayed-clip surface)'}") found = (delta.get("standings") or {}).get("boats") or [] if found: boats = [(b.get("deviceId"), b.get("sailId"), b.get("teamName")) for b in found if b.get("deviceId")] now = time.monotonic() if boats and now - last_tick >= min_tick_s: for did, sail, _team in boats: # the team is on the page's nameplate, not here # The scene name is the HULL, and only the hull. A hull number is painted on the # boat and never changes; the TEAM sailing it changes every heat. Putting the team # in the scene name made every rotation rename every scene — which is the whole # reason the controller carries a two-pass relabel dance through temporary names, # and why an operator's muscle memory for "the tile in slot 3" broke twice an hour. # Nothing is lost: the page's own nameplate carries flag, sail AND team, drawn # from the feed, so it is both more informative and always current. labels[did] = str(sail) devs = [d0 for d0, _s, _t in boats] sup.observe(now, devs) sup.set_grid_scene(grid_scene_for_teams(boats)) changes = sup.tick(now) key = (sup.grid_scene, tuple(sorted(devs))) if key != last_key or any(changes.values()): log(f"[obs] grid='{sup.grid_scene}' active={len(devs)} added={changes['added']} " f"removed={changes['removed']} relabelled={changes['relabelled']} " f"repointed={changes.get('repointed', [])}") last_key = key last_tick = now # --- FIXED SLOTS: a panel whose shape never changes ------------------------------------------- def slot_order(boats) -> list[str]: """Device ids in a STABLE order — by sail number, then id. Which tile a boat lands in must not depend on the order the feed happened to list it in, or a boat would hop between tiles from one tick to the next while the operator was looking at it. """ def key(row): _did, sail, _team = row try: return (0, int(sail), "") except (TypeError, ValueError): return (1, 0, str(sail)) return [r[0] for r in sorted(boats, key=key)] def slot_letter(i: int) -> str: """A, B, C, D … — a tile's PERMANENT name. Letters, not numbers, because the boats are numbered: a tile called "Boat Cam 1" showing hull 3 reads as a contradiction at a glance, and an operator calling "take cam 2" would be ambiguous. A letter can only ever mean a position.""" return chr(ord("A") + i) if i < 26 else f"A{i - 25}" #: The scene that shows every camera at once — the whole race's cameras in one picture. PANEL = "Race cam" #: One camera, on one boat. Distinct from PANEL because they are different things: the panel is the #: race, a tile is a boat. Both are house style, not product facts, so both are parameters. TILE = "Boat cam" def slot_scene(i: int, tile: str = TILE) -> str: return f"{tile} {slot_letter(i)}" def slot_input(i: int, tile: str = TILE) -> str: return f"{tile.lower().replace(' ', '')}:slot:{slot_letter(i)}" def run_slots_from_feed(obs, feed_url: str, base_url: str, *, slots: int = 4, panel: str = PANEL, tile: str = TILE, grid_scene: str | None = None, composite: str = "", read_base: str = "", lag_ms: float | None = None, clips: bool = False, media_base: str = "", min_tick_s: float = 5.0, canvas=(1920, 1080), log=print) -> None: """Drive a panel of FIXED tiles: `Race cam A..N` always exist, and a heat change only re-points them at the boats now racing. The scene holding all of them is `panel` itself. The alternative — a source per boat, created and destroyed as heats rotate — means the OBS layout is a different shape every few minutes. Scene names come and go, so an operator's muscle memory for "the tile in slot 3" is worth nothing, hotkeys bound to a scene break when that scene is deleted, and every rotation creates browser sources that have to load from scratch mid-show. Fixed slots make the panel a constant. Nothing is created or removed after the first pass; a rotation is a URL change on a page that is already running. WHO is in a tile is answered by the page's own nameplate — flag, sail and team, drawn from the feed — which is more than a scene name could carry and is always current. `media_base` moves the clip fetches to ANOTHER ORIGIN than the feed. Every tile holds a permanent SSE connection, a browser allows ~6 per origin, and OBS shares one network stack across all its browser sources — so four cameras plus the programme page saturate the pool and every clip request queues forever, showing an empty tile with no error. Measured, not theorised: blanking the other four pages made the fifth play immediately. `composite` is OFF by default: the panel is the cameras. The produced picture is its own scene and the operator cuts to it; tiling it inside the camera panel makes the panel show one thing that is not a camera, and costs a tile. """ import json import time import urllib.parse import urllib.request from obs_boatcams import boat_query grid_scene = grid_scene or panel # the scene that shows every boat at once for i in range(slots): # build the panel once, then leave it alone obs.ensure_scene(slot_scene(i, tile)) obs.ensure_scene(grid_scene) existing = obs._input_names() for i in range(slots): if slot_input(i, tile) not in existing: obs.create_browser_input(slot_scene(i, tile), slot_input(i, tile), f"{base_url}?feed=live", *canvas) for scene in (slot_scene(i, tile), grid_scene): if obs._scene_item_id(scene, slot_input(i, tile)) is None: obs.add_scene_item(scene, slot_input(i, tile)) if composite and obs._scene_item_id(grid_scene, composite) is None: obs.add_scene_item(grid_scene, composite) tiles = ([composite] if composite else []) + [slot_input(i, tile) for i in range(slots)] try: placed = obs.tile_scene(grid_scene, tiles, canvas) log(f"[obs] tiled {placed} of {len(tiles)} in {grid_scene!r}") except Exception as e: # noqa: BLE001 — tiling is cosmetic log(f"[obs] could not tile {grid_scene!r}: {type(e).__name__}") log(f"[obs] {slots} fixed tiles ({slot_scene(0, tile)}..{slot_scene(slots - 1, tile)}) " f"in {grid_scene!r}") shown: dict[int, str] = {} # slot -> the URL it is currently on query, boats, last_tick = "", [], 0.0 req = urllib.request.Request(feed_url, headers={"Accept": "text/event-stream"}) with urllib.request.urlopen(req) as resp: # noqa: S310 — trusted local/broadcast feed for raw in resp: line = raw.decode("utf-8", "replace").strip() if line.startswith("data:"): try: frame = json.loads(line[5:].strip()) except Exception: # noqa: BLE001 — a keepalive or partial line continue inst = ((frame.get("meta") or {}).get("instanceId") or (frame.get("meta") or {}).get("instance")) if inst: query = "" if clips else boat_query(str(inst), read_base, lag_ms) found = (frame.get("standings") or {}).get("boats") or [] if found: boats = [(b.get("deviceId"), b.get("sailId"), b.get("teamName")) for b in found if b.get("deviceId")] now = time.monotonic() if not boats or now - last_tick < min_tick_s: continue last_tick = now order = slot_order(boats) for i in range(slots): did = order[i] if i < len(order) else None # A slot with no boat keeps its LAST picture rather than going blank: a 3-boat heat in # a 4-tile panel should not stare back with an error card. if did is None: continue url = (f"{base_url}?feed=live&boat={did}" + (f"&mediaBase={urllib.parse.quote(media_base, safe='')}" if media_base else "") + (f"&{query}" if query else "")) if shown.get(i) == url: continue obs.set_browser_source_url(slot_input(i, tile), url) obs.refresh_browser_input(slot_input(i, tile)) shown[i] = url log(f"[obs] {slot_scene(i, tile)} -> {did}") def main() -> None: import argparse from director_obs import OBSController # sibling module; lazy so importing this needs no live OBS ap = argparse.ArgumentParser(description="Dynamic OBS boat-cams driven by a live race feed") ap.add_argument("--feed", default="http://localhost:8766/events", help="SSE race feed URL") ap.add_argument("--base", default=None, help="bare boatcam.html URL (?feed=live&boat= is appended per boat). Unset " "falls back to $VR_PRODUCER_BASE + /core/web/boatcam.html, then to the " "local dev default; VR_RACE_DAY=1 refuses a loopback base") ap.add_argument("--obs-host", default="localhost") ap.add_argument("--obs-port", type=int, default=4455) ap.add_argument("--obs-password", default="") ap.add_argument("--grace", type=float, default=12.0, help="seconds a boat lingers after it stops racing (anti-flap)") ap.add_argument("--no-clean", action="store_true", help="keep any existing boatcam sources on start") ap.add_argument("--read-base", default="", help="origin serving the boat-cam media (e.g. https://apps.viewregatta.com). The " "scene may render on the on-site machine while the media comes from gcloud; " "this is the only thing that differs between the two deployments") ap.add_argument("--lag", type=float, default=22000, metavar="MS", help="MILLISECONDS behind now to play — the shared presentation clock the 3D scene " "uses (config.js liveLagMs), so the cameras and the programme show the same " "instant. Same unit as the page's `?lag=` and obs_boatcams' --lag") ap.add_argument("--clips", action="store_true", help="keep the cameras on the delayed-clip surface instead of the live HLS " "stream. Required against a server that does not fragment its clips " "(VR_MEDIA_FRAGMENT off) — an HLS source there plays nothing at all") ap.add_argument("--composite", default="", help="name of a programme scene to ALSO tile in the panel. Off by default — the " "panel is the cameras, and the produced picture is its own scene") ap.add_argument("--panel", default=PANEL, help=f"the scene showing every camera at once (default {PANEL!r})") ap.add_argument("--tile", default=TILE, help=f"one boat's camera (default {TILE!r}) — the tiles are ' A', " "' B' … A panel is the race; a tile is a boat") ap.add_argument("--media-base", default="", help="fetch the tiles' clips from THIS origin instead of the page's own. The feed's " "SSE streams occupy the browser's per-origin connection budget, so serving the " "media from the same host leaves nothing for the video and every tile sits " "empty. Not needed in production, where clips are signed GCS URLs") ap.add_argument("--slots", type=int, metavar="N", help="run a PERMANENT panel of N tiles (Boat Cam A, B, C …) instead of one source " "per boat. Nothing is created or removed as heats rotate — a rotation only " "re-points tiles that are already running, so the layout, the scene names " "and any hotkeys bound to them stay put. Who is in a tile is answered by the " "page's own nameplate (flag, sail, team), which a scene name cannot carry") a = ap.parse_args() from producer_base import RaceDayBaseError, resolve try: a.base, _src = resolve(a.base, default="http://localhost:8766/core/web/boatcam.html", path="/core/web/boatcam.html") except RaceDayBaseError as e: print(f"[obs] {e}", flush=True) return 2 obs = OBSController(host=a.obs_host, port=a.obs_port, password=a.obs_password).connect() print(f"[obs] connected {a.obs_host}:{a.obs_port}; following {a.feed}", flush=True) print(f"[obs] boatcam base: {a.base} (from the {_src})", flush=True) if a.slots: run_slots_from_feed(obs, a.feed, a.base, slots=a.slots, panel=a.panel, tile=a.tile, composite=a.composite, read_base=a.read_base, lag_ms=a.lag, clips=a.clips, media_base=a.media_base, log=lambda m: print(m, flush=True)) return run_from_feed(obs, a.feed, a.base, grace_s=a.grace, clean=not a.no_clean, read_base=a.read_base, lag_ms=a.lag, composite=a.composite, clips=a.clips, log=lambda m: print(m, flush=True)) if __name__ == "__main__": main()