""" The live producer's race source over **REST** — no database, no store object. The same producer has to run in every combination the operation actually takes: * **live / replay / simulation** — all three are the same client contract. The simulator is a faithful client of these very endpoints, and replay is the same reads with `?t=`; so a producer written against REST cannot tell them apart, which is exactly the point. * **local backend or gcloud** — one `--read-base` URL is the whole difference. * **broadcast on the render PC or in gcloud** — the producer only needs HTTP out, so it runs wherever the render happens to live. Store-backed remains available (and is what a co-located producer should use — it skips the HTTP hop); this module is the deployment-independent path. Everything here is a read the operator console and the guest viewer already make: GET /v3/manage/instances the races + their official gun/stop/state GET /v3/view/{i}/roster?t= the allocation AT the gun → the racing boats GET /v3/view/{i}/teams team display config (names, sails, colours, bibs) GET /v3/view/{i}/course?courseId=&t= course geometry (drifting marks) + venue nav marks GET /v3/view/{i}/wind?courseId=&t= the wind in effect at the gun GET /v3/view/{i}/tracks?from=&to= elapsed track, for a race joined in progress GET /v3/view/{i}/positions live positions (per tick; via `http_fetch_positions`) Only stdlib — the producer image needs no extra dependency for this. """ from __future__ import annotations import json import urllib.error import urllib.parse import urllib.request from collections.abc import Callable DEFAULT_TIMEOUT_S = 8.0 class RestBackend: """Read-only REST client for one backend (`base` = e.g. `http://127.0.0.1:8080` or `https://apps.viewregatta.com`). `token` is sent as a bearer when the instance is not public. Every getter degrades to an empty result rather than raising: a broadcast must not die because one poll timed out — the next tick asks again.""" def __init__(self, base: str, *, token: str | None = None, timeout_s: float = DEFAULT_TIMEOUT_S, on_log: Callable[[str], None] = print) -> None: self.base = base.rstrip("/") self.token = token self.timeout_s = timeout_s self._log = on_log def get(self, path: str, **query) -> dict | list | None: """One GET; `None` on any transport/decode failure (logged once per failure, never raised).""" q = {k: v for k, v in query.items() if v is not None} url = f"{self.base}{path}" + (f"?{urllib.parse.urlencode(q)}" if q else "") req = urllib.request.Request(url) # noqa: S310 — operator-supplied base if self.token: req.add_header("Authorization", f"Bearer {self.token}") try: with urllib.request.urlopen(req, timeout=self.timeout_s) as r: # noqa: S310 return json.loads(r.read().decode("utf-8")) except (urllib.error.URLError, OSError, ValueError, TimeoutError) as e: self._log(f"[rest] GET {path} failed ({e})") return None # --- the race plane --------------------------------------------------------------------------- def instances(self) -> list: """The management rows — what `StartedRaces` reads to know which races have started.""" rows = self.get("/v3/manage/instances") return rows if isinstance(rows, list) else [] def course_of(self, instance_id: str) -> str | None: """The course the RACE says it runs on. A broadcast should not have to be told a course id: the race carries one, and following the operator means following whatever they set.""" for row in self.instances(): if row.get("instanceId") == instance_id: return row.get("courseId") return None def roster(self, instance_id: str, t_ms: int) -> dict: """`{teamId: [boatRef]}` — the operator's allocation AT `t_ms` (pass the gun).""" grid = self.get(f"/v3/view/{instance_id}/roster", t=t_ms) return grid if isinstance(grid, dict) else {} def team_config(self, instance_id: str) -> dict: cfg = self.get(f"/v3/view/{instance_id}/teams") return cfg if isinstance(cfg, dict) else {} def course(self, instance_id: str, course_id: str, t_ms: int) -> dict: geo = self.get(f"/v3/view/{instance_id}/course", courseId=course_id, t=t_ms) return geo if isinstance(geo, dict) else {} def wind(self, instance_id: str, course_id: str, t_ms: int) -> dict: w = self.get(f"/v3/view/{instance_id}/wind", courseId=course_id, t=t_ms) return w if isinstance(w, dict) else {} def frames(self, instance_id: str, *, t_from: int, t_to: int, track_start_ms: int) -> dict: """The elapsed track as engine frames `{"": {boatRef: {lat,lon,speed,course}}}` — the REST counterpart of `reads.boat_frames`, for a race joined IN PROGRESS. `/tracks` returns each boat's points in one call, so this is a single round-trip for the whole fleet. Speed/course are not carried per point on this endpoint, so they are derived from consecutive fixes (what the scorer needs them for is leg/rounding geometry, which is positional).""" from geo import bearing_deg, haversine_m # noqa: PLC0415 — shared with the backend, never re-implemented data = self.get(f"/v3/view/{instance_id}/tracks", **{"from": t_from, "to": t_to}) frames: dict[str, dict] = {} for row in ((data or {}).get("boats") or []): boat = row.get("boat") prev = None for p in (row.get("points") or []): t, lat, lon = p.get("t"), p.get("lat"), p.get("lon") # A no-fix / null-island / out-of-range point must never reach the scorer (it would # poison the leg geometry): skip it and keep the last good one as `prev`. if t is None or lat is None or lon is None or (lat == 0.0 and lon == 0.0): continue if not (-90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0): continue speed = course = 0.0 if prev is not None: dt = (t - prev[0]) / 1000.0 if dt > 0: speed = haversine_m(prev[1], prev[2], lat, lon) / dt / 0.514444 # m/s → kn course = bearing_deg(prev[1], prev[2], lat, lon) prev = (t, lat, lon) sec = (t - track_start_ms) // 1000 if sec < 0: continue frames.setdefault(str(sec), {})[boat] = { "lat": round(lat, 7), "lon": round(lon, 7), "speed": round(speed, 2), "course": round(course, 1)} return frames def rest_presentation_for( backend: RestBackend, *, course_id: str | None = None, group_id: str = "g", title: str | None = None, pre_start_s: int = 180, track_window_ms: int = 86_400_000, user: str = "", ) -> Callable[[str, int], tuple]: """Build the `presentation_for(instance, gun_ms)` the following producer calls at each hand-over, resolved entirely over REST: `(meta, course, boats, team_config, teams)`. The roster, the course and the wind are all read **at the operator's gun**, not at whatever moment the producer happened to look — resolving the allocation at the wrong instant is what leaves a race with no teams, and a race with no teams has no standings, no camera focus and no commentary. `course_id` is optional and is only an OVERRIDE: a race carries its own course, so following the operator does not require anyone to know a course id in advance.""" def presentation_for(instance_id: str, gun_ms: int) -> tuple: roster = backend.roster(instance_id, gun_ms) boats = sorted({b for team in roster.values() for b in team}) team_config = backend.team_config(instance_id) course = backend.course(instance_id, course_id or backend.course_of(instance_id), gun_ms) sl = course.get("startLine") if sl and not course.get("center"): course["center"] = {"lat": (sl["port"]["lat"] + sl["starboard"]["lat"]) / 2, "lon": (sl["port"]["lon"] + sl["starboard"]["lon"]) / 2} track_start = gun_ms - int(pre_start_s) * 1000 meta = {"title": title or (course.get("name") or instance_id), "user": user, "instanceId": instance_id, "groupId": group_id, "openEnded": True, "raceType": "team_race_2", "trackStartUtcMs": track_start, "gunUtcMs": gun_ms, "trackEndUtcMs": track_start + track_window_ms, "frameStepMs": 1000} # `/wind` speaks the read plane's `dirDeg`/`speedKts` (+ optional ambient conditions); the feed # meta speaks the engine's `windFromDeg`/`windKts`. Translate, never guess: a missing key stays # missing rather than becoming a 0° breeze the renderer would point the wind arrow at. wind = backend.wind(instance_id, course_id, gun_ms) or {} if wind.get("dirDeg") is not None: meta["windFromDeg"] = wind["dirDeg"] if wind.get("speedKts") is not None: meta["windKts"] = wind["speedKts"] for src, dst in (("gustKts", "windGustKts"), ("tempC", "tempC"), ("cloudPct", "cloudPct"), ("pressureHpa", "pressureHpa"), ("weatherCode", "weatherCode")): if wind.get(src) is not None: meta[dst] = wind[src] # `teams` in the ENGINE's shape, joined on the boat ref, display from the team config — the # same construction `broadcast_feed.resolve_teams` makes store-side. Names come from config, # never from code (the no-hardcoded-names rule). teams = [] for team_id, team_boats in roster.items(): cfg = team_config.get(team_id, {}) bcfg = cfg.get("boats") or {} teams.append({ "teamId": team_id, "name": cfg.get("name", team_id), "boats": [{"deviceId": b, "sailId": (bcfg.get(b) or {}).get("sailId", ""), "visualId": (bcfg.get(b) or {}).get("visualId", b), "color": (bcfg.get(b) or {}).get("color", "#888888"), "sternM": (bcfg.get(b) or {}).get("sternM", 0)} for b in team_boats], }) return meta, course, boats, team_config, teams return presentation_for