"""The recreational CRUISE tour — the producer-side visit scheduler. A cruise broadcast has no race to score: the picture pans the venue, visits every device one by one with a fly-in, shows each boat's own camera as a large panel, and presents nothing but the boat's NAME (plus a small verification strip — the mode exists so an operator can watch the tour and confirm, device by device, that the position is live, the video arrives, and nothing is mixed up). This module is the schedule and nothing else: pure functions of (second, device snapshot), no clock, no store, no server — the `_Cruise` driver in `serve_v3_live.py` feeds it once a second and publishes whatever it returns over the ordinary delta channel. It is deliberately a PRODUCER module. The renderer already knows how to honour a focus beat (`focus` + `focusUntilSec` + `shotType`), an onboard directive (`rieOnboardDirectives`) and a spoken line (`text`); the tour only composes those. The broadcast presents, the producer decides — and what the tour decides is visible in the beats it emits, which is what the tests pin. Truthfulness rules the scheduler enforces itself: * a device with no FRESH clip gets a shorter visit and NO picture directive — the panel is simply absent, never invented; * a device with no finite fix is skipped THIS LAP only (a camera cannot be aimed at a position nobody has), and rejoins the tour the moment it reports again; * ages ride the beat as data (`fixAgeS`/`clipAgeS`), `None` when the reading never arrived — the renderer draws an em dash, never a zero. """ from __future__ import annotations import math from dataclasses import dataclass #: Spoken compass words for a wind FROM-bearing — 8-point, because "the wind is from the #: south-southwest" reads as instrument talk while "from the southwest" is how a person says it. _COMPASS8 = ("north", "northeast", "east", "southeast", "south", "southwest", "west", "northwest") def compass_words(deg: float) -> str: """The spoken 8-point compass name of a bearing (0/360 → "north").""" return _COMPASS8[round((deg % 360) / 45) % 8] def boat_display_name(raw: str) -> str: """A sailing boat referred to by NUMBER is always "Boat {n}" (Ulf 2026-08-10) — a bare "10" on a plate or in a spoken line is a number, not a name (the same judgement `commentary_i18n.boat_reference` makes when a team label is a bare number). A real name ("Alpha") passes through untouched. Applied where the tour resolves its names, so the plate, the panel caption, the roll-call and the spoken line cannot disagree.""" s = str(raw or "").strip() return f"Boat {s}" if s.isdigit() else s @dataclass(frozen=True) class CruiseConfig: """The tour's knobs. Defaults here; an event overrides via `presentation.cruise` (camelCase), carried to the driver on `meta.cruise` — event data, never a renderer constant.""" dwell_s: int = 20 #: hold at a device WITH fresh video (beyond the fly-in) dwell_no_video_s: int = 8 #: hold at a device without — the honest shorter visit interlude_s: int = 8 #: idle venue pan between visits opening_pan_s: int = 25 #: the opening high pan before the first visit fly_in_s: int = 6 #: camera travel onto the device before its panel raises fresh_clip_max_age_s: int = 90 #: a clip older than this is not "the camera working now" venue_tour_every_laps: int = 3 #: guided landmark tour cadence (0 = opening only) landmark_dwell_s: int = 12 #: minimum hold at a venue landmark #: Guided EXCURSION cadence (the continuous city flight, `excursion.Excursion`) — 0 = OFF, #: which is the shipped default: the whole feature arms by event data, never by deploy. excursion_every_laps: int = 0 _KEYS = {"dwellS": "dwell_s", "dwellNoVideoS": "dwell_no_video_s", "interludeS": "interlude_s", "openingPanS": "opening_pan_s", "flyInS": "fly_in_s", "freshClipMaxAgeS": "fresh_clip_max_age_s", "venueTourEveryLaps": "venue_tour_every_laps", "landmarkDwellS": "landmark_dwell_s", "excursionEveryLaps": "excursion_every_laps"} @classmethod def from_meta(cls, block: dict | None) -> CruiseConfig: """The event's `presentation.cruise` block → a config; unknown keys ignored, non-numeric values ignored (a malformed knob falls back to the default rather than crashing the show).""" kwargs = {} for wire, field_name in cls._KEYS.items(): v = (block or {}).get(wire) if isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) and v >= 0: kwargs[field_name] = int(v) return cls(**kwargs) @dataclass(frozen=True) class TourStop: """One device (or robotic mark) as the tour sees it THIS tick — a snapshot, rebuilt by the driver every second from the store, never cached here.""" ref: str #: boat ref (`owner:boat:7`) or slot ref (`owner:buoy:A`) name: str #: the presented name (BoatMeta.name → teamConfig → hull number) kind: str = "boat" #: "boat" | "vessel" | "mark" slot_label: str | None = None #: "Boat 7 · Mast" / "Buoy A" — the verification strip's slot lat: float | None = None lon: float | None = None fix_age_s: float | None = None #: None = never positioned clip_age_s: float | None = None #: None = never filmed speed_kts: float | None = None #: spoken in the intro when it is a real number wind_from_deg: float | None = None #: a mark that carries an anemometer reports its reading wind_kts: float | None = None #: Course over ground (degrees TRUE) and the course's resolved wind — together they say what the #: boat is DOING (beating / reaching / running, and on which tack). Kept as the two RAW facts: #: the derivation belongs to the presenter (`sailing_words`), not to the driver that reads them. #: Deliberately NOT `wind_from_deg`, which is the mark's own anemometer and whose presence is #: what licenses "the robotic mark REPORTS wind…" — a forecast must never be able to claim that. course_deg: float | None = None venue_wind_from_deg: float | None = None def positioned(self) -> bool: return (isinstance(self.lat, (int, float)) and isinstance(self.lon, (int, float)) and math.isfinite(self.lat) and math.isfinite(self.lon)) #: The tour's reading order (Ulf 2026-08-10): the BOATS first, then the BUOYS (the line phones on #: buoy slots), then the robotic MARKS — each group in number order. A checklist is read in a #: known order, not flown by proximity: the operator ticking devices off must know who comes next. _KIND_RANK = {"boat": 0, "vessel": 1, "mark": 2} def stop_order(s: TourStop) -> tuple: """Sort key: kind (boats → buoys → marks), then the ref's own NUMBER (numerically, so Boat 2 precedes Boat 10), letters after numbers within a kind (buoy A, B … after any numbered one).""" tail = s.ref.rsplit(":", 1)[-1] if tail.isdigit(): return (_KIND_RANK.get(s.kind, 3), 0, int(tail), tail) return (_KIND_RANK.get(s.kind, 3), 1, 0, tail) #: Below this a hull is DRIFTING, not sailing. A course over ground derived from a near-stationary #: GNSS fix is noise that swings through the whole compass, so a confident "beating on port tack" #: taken from it is a wrong claim stated as a plain observation — the one thing the tour must not do. SAILING_MIN_KTS = 1.5 #: Within this of DEAD DOWNWIND the tack is the BOOM's to state, not the course's: a boat sailing by #: the lee is on the opposite tack to the one her track suggests, and nothing we can see says which. #: So deep running is presented as running, and simply does not name a tack. (Beating and reaching #: have no such ambiguity — there the wind's side and the boom's side are the same side.) SAILING_BY_THE_LEE_TWA = 165.0 #: What the boat is doing, by point of sail. `race_rules.point_of_sail` answers in the geometry's #: words (upwind / reach / downwind); a broadcast says what a sailor says. _POINT_OF_SAIL_VERB = {"upwind": "beating", "reach": "reaching", "downwind": "running"} def sailing_words(course_deg, wind_from_deg, speed_kts) -> str | None: # noqa: ANN001 """"beating on starboard tack" — what a boat is DOING, from her course against the true wind (Ulf 2026-08-11). A plain OBSERVATION: both facts are measured, and the derivation is the geometry every sailor reads off the same two numbers — never an inference about intent. `None` (and the line simply says less) whenever the geometry cannot support the claim: a missing or non-finite course or wind, or a hull too slow for her course to mean anything (`SAILING_MIN_KTS`). The shared `v3/scoring/race_rules` helpers do the maths, so the tour and the scoring engine can never disagree about which tack a boat is on; imported lazily, and a scoring path that is not there costs the clause rather than the show.""" ok = (isinstance(course_deg, (int, float)) and not isinstance(course_deg, bool) and isinstance(wind_from_deg, (int, float)) and not isinstance(wind_from_deg, bool) and math.isfinite(course_deg) and math.isfinite(wind_from_deg) and isinstance(speed_kts, (int, float)) and math.isfinite(speed_kts) and speed_kts >= SAILING_MIN_KTS) if not ok: return None try: from race_rules import ang_diff, point_of_sail, tack_of # noqa: PLC0415 except ImportError: # noqa: S110 — the clause is optional, the show is not return None cog, wind = float(course_deg), float(wind_from_deg) verb = _POINT_OF_SAIL_VERB.get(point_of_sail(cog, wind)) if not verb: return None if ang_diff(cog, wind) > SAILING_BY_THE_LEE_TWA: return verb # deep running: the boom's tack, not ours to state return f"{verb} on {tack_of(cog, wind)} tack" def visit_line(stop: TourStop) -> str: """The spoken introduction. A word always separates the name from any number that follows (root CLAUDE.md: never let two numbers sit adjacent — a name may itself end in a digit).""" if stop.kind == "mark": if stop.wind_from_deg is not None and stop.wind_kts is not None: return (f"The robotic mark reports wind from the {compass_words(stop.wind_from_deg)} " f"at {_kts_words(stop.wind_kts)}.") return f"This is the robotic mark {stop.name}." kts = stop.speed_kts if (isinstance(kts, (int, float)) and math.isfinite(kts) and round(kts) >= 1 and stop.kind == "boat"): # …and WHAT SHE IS DOING, when the course and the wind can say so: "This is Boat 7, beating # on starboard tack, making 6 knots." The clause is dropped whole when the geometry is not # there, so the sentence degrades to the plain introduction rather than inventing a point of # sail. Both clauses put a word between the name and the number the rule above protects. doing = sailing_words(stop.course_deg, stop.venue_wind_from_deg, kts) if doing: return f"This is {stop.name}, {doing}, making {_kts_words(kts)}." return f"This is {stop.name}, making {_kts_words(kts)}." return f"This is {stop.name}." def _kts_words(kts: float) -> str: n = round(kts) return f"{n} knot" if n == 1 else f"{n} knots" def weather_line(wind_from_deg: float, wind_kts: float, venue_name: str | None) -> str: where = f" over {venue_name}" if venue_name else "" return (f"Wind from the {compass_words(wind_from_deg)} " f"at {_kts_words(wind_kts)}{where}.") # # THE ROLL CALL IS GONE. Every lap used to open with "On the water today: Alpha, Bravo and Charlie." # — the whole fleet, recited, once per lap, for as long as the tour ran. With a dozen devices that is # a long list of names a viewer cannot hold, delivered several times an hour, and it says in advance # exactly what the next few minutes are about to say properly: the tour VISITS each device in turn # and names it then, on the plate and in the line. A list that pre-announces a list is filler, and it # read as filler on air ("jävligt tjatigt" — Ulf, 2026-08-11). # # So the lap begins at its first visit. Nothing is lost: who is on the water is still stated, one # name at a time, beside the picture of that device — which is the form a viewer can actually use. #: How far the camera stands off a point of interest. A building is framed from a distance a #: boat-sized object would vanish at, so the producer says which it is and the renderer only flies. POI_DIST_LANDMARK_M = 240 POI_DIST_VESSEL_M = 70 class TourScheduler: """The sequencer: one segment at a time, forever. `tick(now_sec, stops, wind)` is called once a second (idempotent — extra calls inside a running segment return nothing) and answers `(events, directives)` to publish at this second. The choreography: OPENING pan (+ spoken weather) → guided landmark tour → per lap: visit → interlude → … → skyline pan → next lap (the landmark tour re-runs every `venue_tour_every_laps` laps) Visit order is the operator's own checklist (Ulf 2026-08-10): the boats in NUMBER order, then the buoys, then the robotic marks — `stop_order`. Reproducible, and the operator ticking devices off always knows who comes next. """ def __init__(self, cfg: CruiseConfig, *, landmarks: list[dict] | None = None, venue_name: str | None = None, excursion=None) -> None: # noqa: ANN001 — excursion.Excursion self.cfg = cfg #: `[{name, lat, lon, line}]` — venue facts from the event definition, never authored here. self.landmarks = [lm for lm in (landmarks or []) if isinstance(lm.get("lat"), (int, float)) and isinstance(lm.get("lon"), (int, float))] self.venue_name = venue_name #: The continuous city flight (`excursion.Excursion`), owned by the driver and DELEGATED #: to on excursion laps — this scheduler only decides WHEN it flies, never where. self.excursion = excursion self._state = "start" self._until: int | None = None #: the second the current segment runs to self._lap = 0 self._visited: set[str] = set() #: refs visited this lap self._lm_queue: list[dict] = [] #: landmarks still to visit in the current tour # --- the one public entry ------------------------------------------------------------------- def tick(self, now_sec: int, stops: list[TourStop], wind: dict | None = None) -> tuple[list[dict], list[dict]]: """Advance the tour to `now_sec`. Returns `(events, directives)` — empty while the current segment is still running. `wind` is the feed-meta shape (`windFromDeg`/`windKts`).""" now_sec = int(now_sec) if self._until is not None and now_sec < self._until: return [], [] if self._state == "start": return self._opening(now_sec, wind) return self._advance(now_sec, stops) # --- segments ------------------------------------------------------------------------------- def _opening(self, t: int, wind: dict | None) -> tuple[list[dict], list[dict]]: self._state = "opened" self._until = t + self.cfg.opening_pan_s events = [] if wind and wind.get("windFromDeg") is not None and wind.get("windKts") is not None: events.append(self._beat(t, "tour_weather", text=weather_line(wind["windFromDeg"], wind["windKts"], self.venue_name))) return events, [] def _advance(self, t: int, stops: list[TourStop]) -> tuple[list[dict], list[dict]]: state = self._state if state == "opened": return self._begin_landmarks(t) if self.landmarks else self._begin_lap(t, stops) if state == "landmark": return self._next_landmark(t, stops) if state in ("lap", "interlude"): return self._next_visit(t, stops) if state == "visit": # the visit just ended — an idle-pan breather before the next flight self._state = "interlude" self._until = t + self.cfg.interlude_s return [], [] if state == "skyline": return self._begin_lap(t, stops) if state == "skyline_then_landmarks": return self._begin_landmarks(t) if state == "skyline_then_excursion": return self._begin_excursion(t, stops) if state == "excursion": return self._tick_excursion(t, stops) raise AssertionError(f"unknown tour state {state!r}") def _begin_landmarks(self, t: int) -> tuple[list[dict], list[dict]]: self._state = "landmark" self._lm_queue = list(self.landmarks) self._until = t # _next_landmark emits the first stop immediately return self._next_landmark(t, []) def _begin_excursion(self, t: int, stops: list[TourStop]) -> tuple[list[dict], list[dict]]: """Hand the picture to the continuous city flight (`excursion.Excursion.begin`); with nothing eligible to fly the lap simply resumes — an excursion that cannot happen must never stall the tour.""" self.excursion.begin(t) if self.excursion.done: return self._begin_lap(t, stops) self._state = "excursion" return self._tick_excursion(t, stops) def _tick_excursion(self, t: int, stops: list[TourStop]) -> tuple[list[dict], list[dict]]: beats = self.excursion.tick(t) if self.excursion.done: return self._begin_lap(t, stops) self._until = t + 1 # one moving-aim beat per second — never parked return beats, [] def _next_landmark(self, t: int, stops: list[TourStop]) -> tuple[list[dict], list[dict]]: if not self._lm_queue: return self._begin_lap(t, stops) lm = self._lm_queue.pop(0) line = str(lm.get("line") or "") dwell = max(self.cfg.landmark_dwell_s, self.cfg.fly_in_s + _speech_seconds(line)) self._until = t + dwell ev = self._beat(t, "tour_poi", name=str(lm.get("name") or ""), text=line, poi={"lat": float(lm["lat"]), "lon": float(lm["lon"])}, poiDistM=POI_DIST_LANDMARK_M, focusUntilSec=t + dwell, shotType="boat-feature") return [ev], [] def _begin_lap(self, t: int, stops: list[TourStop]) -> tuple[list[dict], list[dict]]: self._lap += 1 self._visited = set() if not stops: # An empty water says nothing over and over: the pan keeps running, the tour waits. self._state = "lap" self._until = t + max(10, self.cfg.interlude_s) return [], [] # Straight to the first device — see the note on the removed roll call above. The lap beat # stays (silent), because it is what marks a lap boundary for anything counting them, and # dropping it would change the excursion cadence as a side effect of a wording decision. self._state = "lap" self._until = t return [self._beat(t, "tour_lap", lap=self._lap)], [] def _next_visit(self, t: int, stops: list[TourStop]) -> tuple[list[dict], list[dict]]: nxt = self._pick(stops) if nxt is None: return self._skyline(t) self._visited.add(nxt.ref) self._state = "visit" fresh = (nxt.clip_age_s is not None and nxt.clip_age_s <= self.cfg.fresh_clip_max_age_s) dwell = self.cfg.dwell_s if fresh else self.cfg.dwell_no_video_s end = t + self.cfg.fly_in_s + dwell self._until = end ev = self._beat(t, "tour_visit", deviceId=nxt.ref, name=nxt.name, slotLabel=nxt.slot_label, fixAgeS=_age(nxt.fix_age_s), clipAgeS=_age(nxt.clip_age_s), text=visit_line(nxt), focusUntilSec=end, shotType="boat-feature") if nxt.kind == "boat": ev["focus"] = [nxt.ref] # the scene tracks the moving hull else: # a vessel/mark is not in the roster — the camera flies to its reported fix ev["poi"] = {"lat": nxt.lat, "lon": nxt.lon} ev["poiDistM"] = POI_DIST_VESSEL_M directives = [] if fresh: directives.append({"tSec": t + self.cfg.fly_in_s, "endSec": end, "boatId": nxt.ref, "shotType": "tour-visit", "fullFrame": False, "priority": 1, "reason": "tour", "name": nxt.name}) return [ev], directives def _pick(self, stops: list[TourStop]) -> TourStop | None: """The next unvisited POSITIONED stop in checklist order (`stop_order`); a stop with no finite fix skips this lap only — a camera cannot be aimed at a position nobody has.""" todo = [s for s in stops if s.ref not in self._visited and s.positioned()] return min(todo, key=stop_order) if todo else None # (The lap used to end on an all-cameras 2×2 quad; Ulf removed it 2026-08-10 — the lap goes # straight to the skyline breather. One camera at a time is the tour's whole grammar.) def _skyline(self, t: int) -> tuple[list[dict], list[dict]]: """The between-laps breather: the venue establishing camera pans the city skyline (`cam: "venue"`, aimed along `venue.cityBearingDeg` by the renderer). `camUntilSec` bounds it explicitly — without it the renderer holds a cine until the next cam event, i.e. forever.""" self._state = "skyline" hold = max(self.cfg.interlude_s, 10) self._until = t + hold # Re-run the guided landmark tour every N laps (0 = opening only); the EXCURSION rides its # own cadence and WINS a collision lap — the city flight is the rarer, longer segment, and # a lap that owed both would otherwise never fly it. every = self.cfg.venue_tour_every_laps every_x = self.cfg.excursion_every_laps if (self.excursion is not None and every_x > 0 and self._lap % every_x == 0 and self.excursion.available()): self._state = "skyline_then_excursion" elif self.landmarks and every > 0 and self._lap % every == 0: self._state = "skyline_then_landmarks" ev = self._beat(t, "tour_venue", cam="venue", camUntilSec=t + hold) return [ev], [] # --- helpers -------------------------------------------------------------------------------- def _beat(self, t: int, kind: str, **fields) -> dict: """Every tour beat is a `structure`-class claim spoken by the Caller — facts, no tactics.""" ev = {"tSec": int(t), "type": kind, "claimClass": "structure", "commentaryRole": "lead"} ev.update({k: v for k, v in fields.items() if v is not None}) return ev def _age(v: float | None) -> int | None: """Ages ride the wire as whole seconds; `None` stays `None` (an em dash, never a zero).""" return None if v is None else max(0, int(v)) def _speech_seconds(text: str) -> int: """The same conservative words-per-second model the renderer's camera pacing uses (`estimatedSpeechSeconds`, app.js) — so a segment is never shorter than its own sentence.""" words = len([w for w in str(text or "").split() if w]) return max(3, math.ceil(words / 2.7 + 0.8))