""" Demo driver — replay the recorded race LIVE over the LiveFeedServer (B-B manual end-to-end). Ties the two live seams together: it feeds the demo race frame-by-frame through `StreamingScorer` (B-A) and publishes each tick over `LiveFeedServer` (B-B) as SSE, exactly as the production live service will. Point a browser (or OBS browser source) at the printed URL to watch it render live off the stream — same renderer, same BroadcastFeed contract as the file-backed demo. Run (from anywhere, with .venv-demo): .venv-demo/Scripts/python v3/broadcast/live/serve_live.py [feed.json] [--port 8765] [--speed 5] `--speed` is the feed rate (x realtime); default 1 = watchable live (the ~12 min race screens in ~12 min, smooth, with audio). A higher --speed delivers frames faster for a quick preview, but then the live view skips forward to stay at the edge (jumpy by nature) — use 1 to actually watch it. """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) sys.path.insert(0, str(_HERE.parent.parent / "ingest")) # for the optional AIS overlay from ai_live_editor import LiveAIEditor # noqa: E402 from live_feed_server import LiveFeedServer # noqa: E402 from streaming_score import StreamingScorer # noqa: E402 def _centroid(positions: dict) -> tuple[float, float] | None: """The mean `(lat, lon)` of a tick's boat positions — the reference point the AIS overlay follows. `None` when the tick carries no positioned boats.""" pts = [(p["lat"], p["lon"]) for p in positions.values() if p.get("lat") is not None and p.get("lon") is not None] if not pts: return None return (sum(la for la, _ in pts) / len(pts), sum(lo for _, lo in pts) / len(pts)) _DEFAULT_FEED = _HERE.parent / "demo" / "data" / "races" / "real.json" def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("feed", nargs="?", default=str(_DEFAULT_FEED), help="enriched/raw feed JSON to replay") ap.add_argument("--port", type=int, default=8765) ap.add_argument("--speed", type=float, default=1.0, help="feed rate x realtime (1 = watchable live; higher = fast preview, but the live " "view then skips/jumps to stay at the edge)") ap.add_argument("--ai", action="store_true", help="run the live bounded-LLM editor on each tick's colour beats (Haiku; no-op without a key)") ap.add_argument("--author", action="store_true", help="ALSO run the L4 author live (rewrite colour TEXT, validated). Reviewer-gated — off by default") ap.add_argument("--ais", action="store_true", help="overlay AIS context vessels within 500 m via aisstream.io (needs $AISSTREAM_API_KEY)") ap.add_argument("--ais-radius", type=float, default=500.0, help="AIS overlay radius in metres") ap.add_argument("--obs", action="store_true", help="drive OBS to auto add/remove one boatcam Browser source per live boat (needs OBS + obs-websocket)") 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("--obs-grace", type=float, default=45.0, help="seconds of silence before a boat's OBS source is removed (anti-flap)") args = ap.parse_args() data = json.loads(Path(args.feed).read_text(encoding="utf-8")) meta, course, teams = data["meta"], data["course"], data["teams"] frames, params = data["frames"], data.get("params", {}) secs = sorted(int(s) for s in frames) srv = LiveFeedServer(host="0.0.0.0", port=args.port) srv.start(background=True) print(f"Live broadcast server on :{srv.port}") print(f"Open: http://localhost:{srv.port}/core/web/index.html?feed=live") srv.publish_init(meta, course, teams, params, media=data.get("media"), video=data.get("video")) use_ai = args.ai or args.author # the author also needs the captured beats sc = StreamingScorer(meta, course, teams, capture_ai=use_ai) _live_author = None if args.author: from ai_author import AIAuthor _live_author = AIAuthor(model="claude-haiku-4-5") # live author is Haiku (the offline bake stays Opus) editor = LiveAIEditor(meta, course, teams, author=_live_author, verbose=True) if use_ai else None if use_ai: _on = "editor" + (" + author" if args.author else "") print(f"live AI {_on}: ON (Haiku; deterministic fallback without ANTHROPIC_API_KEY)" + (" [author is REVIEWER-GATED — enable only post-approval]" if args.author else "")) # Optional: overlay AIS context vessels within `--ais-radius` of the boat (aisstream.io). The # provider hands the feed the latest tick's centroid so its bounding box follows the boat. ais_feed = None ais_center: dict = {"pos": None} if args.ais: import os from ais_feed import AISFeed key = os.environ.get("AISSTREAM_API_KEY", "") if not key: print("AIS overlay: OFF - set $AISSTREAM_API_KEY to enable") else: ais_feed = AISFeed(key, lambda: ais_center["pos"], radius_m=args.ais_radius) ais_feed.start() print(f"AIS overlay: ON (aisstream.io, {args.ais_radius:.0f} m radius)") # Optional: auto add/remove one OBS boatcam Browser source per live boat (see obs_boat_sources.py). sup = None if args.obs: import logging from director_obs import OBSController from obs_boat_sources import BoatSourceSupervisor logging.getLogger("obsws_python").setLevel(logging.CRITICAL) # don't dump its own connect traceback labels = {b["deviceId"]: str(b.get("sailId") or b["deviceId"]) for t in teams for b in t.get("boats", [])} base = f"http://localhost:{srv.port}/core/web/boatcam.html" try: obs = OBSController(host=args.obs_host, port=args.obs_port, password=args.obs_password).connect() prog = obs.current_program_scene() # round-trip self-test: proves a request works obs.remove_stale_boat_inputs(set(labels), verbose=True) # drop other-race leftovers; reuse this race's sup = BoatSourceSupervisor(obs, base, grace_s=args.obs_grace, label_of=lambda d: labels.get(d, d)) print(f"OBS boatcam sources: ON (grace {args.obs_grace:.0f}s; per-boat scenes + 'Boat Cams' grid)") print(f" connected; current program scene = {prog!r}. New 'Boat ' scenes + a 'Boat Cams'" " grid scene will appear in OBS's Scenes list as boats come on the air.") except Exception as e: # Any OBS connection problem (not running / WebSocket off / wrong port / auth+no password / # bad password) must NOT kill the broadcast - the live page works without OBS. obsws raises # its own OBSSDKError for auth, so catch broadly here. # (ASCII-only text: this prints to the Windows console, which is cp1252 - no arrows/em dashes.) print(f"OBS boatcam sources: OFF - could not connect to OBS at {args.obs_host}:{args.obs_port} ({e}).") print(" Check OBS is running with Tools > WebSocket Server Settings enabled; if authentication" " is on, pass the password via --obs-password. Then re-run with --obs.") dt = 1.0 / max(args.speed, 0.01) last = {"standings": None, "positions": {}} _obs_err_shown = False # print the first OBS reconcile failure in full try: for sec in secs: positions = frames[str(sec)] delta = sc.push_frame(sec, positions) events = editor.edit(delta["events"], delta["aiBeats"], sc.ai_context()) if editor else delta["events"] ais_list: list = [] if ais_feed is not None: center = _centroid(positions) if center is not None: ais_center["pos"] = center ais_list = ais_feed.targets_near(center[0], center[1]) srv.publish_delta(sec, delta["standings"], events, positions, ais_list) last = {"standings": delta["standings"], "positions": positions} if sup: # mirror the live boat set into OBS sources try: sup.observe(sec, positions.keys()) changes = sup.tick(sec) for did in changes.get("added", []): # show what got created, with its scene name print(f"OBS +source: boat {labels.get(did, did)} -> scene " f"{OBSController.boat_scene_name(labels.get(did, did))!r} + 'Boat Cams' grid") for did in changes.get("removed", []): print(f"OBS -source: boat {labels.get(did, did)} (silent > {args.obs_grace:.0f}s)") except Exception: # never let an OBS hiccup kill the broadcast, if not _obs_err_shown: # but show the FIRST failure in full so it's fixable import traceback print("OBS reconcile error (continuing without further OBS updates this run):") traceback.print_exc() _obs_err_shown = True time.sleep(dt) tail = sc.finish() # flush the last lag-seconds of commentary tail_events = editor.edit(tail["events"], tail["aiBeats"], sc.ai_context()) if editor else tail["events"] if tail_events: srv.publish_delta(secs[-1], last["standings"], tail_events, last["positions"]) if sc.result(): srv.publish_result(sc.result()) print("replay complete — still serving (clients can reconnect for the full backlog). Ctrl-C to stop.") while True: time.sleep(1) except KeyboardInterrupt: print("\nstopping") finally: if ais_feed is not None: ais_feed.stop() srv.stop() if __name__ == "__main__": main()