"""The recreational cruise tour scheduler (`v3/broadcast/live/cruise_tour.py`) — the pure sequencer behind the cruise broadcast. What these tests pin is the mode's whole claim: every device is visited every lap (line phones included), a device without fresh video gets the shorter honest visit and never a picture directive, ages ride the beats as data (`None`, never zero), and the spoken lines obey the commentary rules (name from data, no adjacent numbers).""" import re import sys from pathlib import Path _LIVE = Path(__file__).resolve().parents[2] / "live" sys.path.insert(0, str(_LIVE)) from cruise_tour import ( # noqa: E402 SAILING_MIN_KTS, CruiseConfig, TourScheduler, TourStop, boat_display_name, compass_words, sailing_words, visit_line, weather_line, ) #: Small knobs so a whole lap fits in a short simulated run. CFG = CruiseConfig(dwell_s=6, dwell_no_video_s=3, interlude_s=2, opening_pan_s=4, fly_in_s=2, fresh_clip_max_age_s=90, venue_tour_every_laps=2, landmark_dwell_s=4) def _stop(ref, name, *, lat=59.32, lon=18.04, kind="boat", clip_age=10.0, fix_age=2.0, speed=None, slot_label=None): return TourStop(ref=ref, name=name, kind=kind, slot_label=slot_label, lat=lat, lon=lon, fix_age_s=fix_age, clip_age_s=clip_age, speed_kts=speed) def _run(sched, stops, seconds, wind=None): """Drive the scheduler over `seconds` ticks; returns (events, directives) in emit order.""" evs, dirs = [], [] for t in range(seconds): e, d = sched.tick(t, stops, wind) evs += e dirs += d return evs, dirs def test_every_device_is_visited_each_lap_and_the_tour_loops_forever(): stops = [ _stop("o:boat:1", "Alpha", lat=59.320, lon=18.040), _stop("o:boat:2", "Bravo", lat=59.321, lon=18.042), _stop("o:buoy:A", "Start Vessel", kind="vessel", lat=59.322, lon=18.044, slot_label="Buoy A"), ] evs, _dirs = _run(TourScheduler(CFG), stops, 200) visits = [e for e in evs if e["type"] == "tour_visit"] per_lap: dict[int, set] = {} lap = 0 for e in evs: if e["type"] == "tour_lap": lap = e["lap"] elif e["type"] == "tour_visit": per_lap.setdefault(lap, set()).add(e["deviceId"]) assert len(per_lap) >= 2, "the tour must loop — a second lap within the run" # The run ends mid-lap; only COMPLETE laps owe a full round (the last one is still sailing). for lap_no, devs in per_lap.items(): if lap_no == max(per_lap): continue assert devs == {"o:boat:1", "o:boat:2", "o:buoy:A"}, \ "every device — the line phone included — is visited every lap" assert visits, "visits were emitted at all" def test_visit_order_is_boats_by_number_then_buoys_then_marks(): """The operator's checklist order (Ulf): boats in NUMBER order (2 before 10 — numeric, not lexicographic), then the buoys, then the robotic marks. Distance plays no part: the operator ticking devices off must always know who comes next.""" stops = [ _stop("o:buoy:M1", "M1", kind="mark", lat=59.3260, lon=18.0450), _stop("o:boat:10", "Boat 10", lat=59.3200, lon=18.0400), # right next to boat 2… _stop("o:buoy:A", "Start Vessel", kind="vessel", lat=59.3220, lon=18.0440), _stop("o:boat:2", "Bravo", lat=59.3200, lon=18.0401), _stop("o:boat:1", "Alpha", lat=59.3290, lon=18.0700), # …while boat 1 is far away ] order1 = [e["deviceId"] for e in _run(TourScheduler(CFG), stops, 200)[0] if e["type"] == "tour_visit"][:5] order2 = [e["deviceId"] for e in _run(TourScheduler(CFG), stops, 200)[0] if e["type"] == "tour_visit"][:5] assert order1 == order2, "same fleet, same order — reproducible" assert order1 == ["o:boat:1", "o:boat:2", "o:boat:10", "o:buoy:A", "o:buoy:M1"] def test_no_fresh_video_means_short_dwell_and_no_directive(): stops = [ _stop("o:boat:1", "Alpha", clip_age=10.0), # fresh _stop("o:boat:2", "Bravo", clip_age=None), # never filmed _stop("o:boat:3", "Charlie", clip_age=600.0), # stale — beyond freshClipMaxAgeS ] evs, dirs = _run(TourScheduler(CFG), stops, 90) by_dev = {} for e in evs: # FIRST visit per device (lap 1) if e["type"] == "tour_visit": by_dev.setdefault(e["deviceId"], e) dur = {d: by_dev[d]["focusUntilSec"] - by_dev[d]["tSec"] for d in by_dev} assert dur["o:boat:1"] == CFG.fly_in_s + CFG.dwell_s assert dur["o:boat:2"] == CFG.fly_in_s + CFG.dwell_no_video_s assert dur["o:boat:3"] == CFG.fly_in_s + CFG.dwell_no_video_s visited_dirs = {d["boatId"] for d in dirs if d["shotType"] == "tour-visit"} assert visited_dirs == {"o:boat:1"}, "a device without a fresh clip never gets a panel" # …and the panel raises only after the fly-in, ending with the visit d1 = next(d for d in dirs if d["shotType"] == "tour-visit") v1 = by_dev["o:boat:1"] assert d1["tSec"] == v1["tSec"] + CFG.fly_in_s and d1["endSec"] == v1["focusUntilSec"] assert d1["fullFrame"] is False, "the cruise panel is a PiP beside the 3D view, never a cut" def test_segments_never_overlap(): stops = [_stop(f"o:boat:{i}", f"B{i}", lat=59.32 + i * 1e-3) for i in range(4)] evs, dirs = _run(TourScheduler(CFG), stops, 200) visits = [e for e in evs if e["type"] == "tour_visit"] for a, b in zip(visits, visits[1:], strict=False): assert b["tSec"] >= a["focusUntilSec"], "one visit at a time — the next begins after" # One camera at a time is the tour's grammar: the all-cameras 2×2 was removed (Ulf 2026-08-10). assert not any(d["shotType"] == "onboard-quad" for d in dirs), \ "the tour never raises the quad — one device's picture at a time" def test_ages_ride_the_beat_and_null_stays_null(): stops = [_stop("o:boat:1", "Alpha", clip_age=None, fix_age=None, slot_label="Boat 1 · Mast")] evs, _ = _run(TourScheduler(CFG), stops, 40) v = next(e for e in evs if e["type"] == "tour_visit") assert "clipAgeS" not in v or v.get("clipAgeS") is None assert "fixAgeS" not in v or v.get("fixAgeS") is None assert v["slotLabel"] == "Boat 1 · Mast" # a positioned stop with real ages carries them as whole seconds evs2, _ = _run(TourScheduler(CFG), [_stop("o:boat:2", "Bravo", clip_age=31.7, fix_age=4.2)], 40) v2 = next(e for e in evs2 if e["type"] == "tour_visit") assert v2["clipAgeS"] == 31 and v2["fixAgeS"] == 4 def test_spoken_lines_carry_the_name_and_never_two_adjacent_numbers(): # A team-less hull's "name" is its own number — the hard case for the adjacent-numbers rule. for stop in (_stop("o:boat:7", "7", speed=5.2), _stop("o:boat:1", "Alpha", speed=5.2), _stop("o:boat:2", "Bravo", speed=None), _stop("o:buoy:A", "Start Vessel", kind="vessel")): line = visit_line(stop) assert stop.name in line assert not re.search(r"\d\s+\d", line), f"adjacent numbers in {line!r}" assert "making 5 knots" in visit_line(_stop("o:boat:1", "Alpha", speed=5.2)) assert visit_line(_stop("o:boat:1", "Alpha", speed=1.2)).endswith("making 1 knot.") # marks speak their own anemometer, converted upstream to true import dataclasses m = dataclasses.replace(_stop("o:buoy:M", "M1", kind="mark"), wind_from_deg=225.0, wind_kts=6.4) assert visit_line(m) == "The robotic mark reports wind from the southwest at 6 knots." assert weather_line(225.0, 6.4, "Stockholm") == \ "Wind from the southwest at 6 knots over Stockholm." assert compass_words(0) == "north" and compass_words(359) == "north" def test_the_intro_says_what_she_is_doing_and_on_which_tack(): """A boat is introduced by her point of sail and her tack (Ulf 2026-08-11) — beating / reaching / running, on starboard or port. A plain OBSERVATION: both facts are measured (her course over ground, the course's true wind) and the derivation is the shared `race_rules` geometry, so the tour and the scoring engine can never disagree about which tack a boat is on.""" import dataclasses def line(course, wind=0.0, speed=6.0): return visit_line(dataclasses.replace(_stop("o:boat:7", "Alpha", speed=speed), course_deg=course, venue_wind_from_deg=wind)) # wind from the NORTH: sailing NE has it over the port bow, NW over the starboard bow assert line(45) == "This is Alpha, beating on port tack, making 6 knots." assert line(315) == "This is Alpha, beating on starboard tack, making 6 knots." assert line(90) == "This is Alpha, reaching on port tack, making 6 knots." assert line(150) == "This is Alpha, running on port tack, making 6 knots." # …and the same boat under a wind from the SOUTH is on the other tack for the same course assert "starboard tack" in line(90, wind=180.0) def test_a_claim_of_tack_or_point_of_sail_is_withheld_when_it_cannot_be_supported(): """Every way the geometry can fail to support the clause drops it WHOLE — the sentence degrades to the plain introduction rather than inventing a point of sail.""" import dataclasses def line(**kw): base = {"course_deg": 45.0, "venue_wind_from_deg": 0.0} return visit_line(dataclasses.replace(_stop("o:boat:7", "Alpha", speed=kw.pop("speed", 6.0)), **{**base, **kw})) plain = "This is Alpha, making 6 knots." assert line(course_deg=None) == plain, "no course over ground" assert line(venue_wind_from_deg=None) == plain, "no wind on this course" assert line(course_deg=float("nan")) == plain, "a NaN never reaches the prose" assert line(venue_wind_from_deg=float("inf")) == plain # A DRIFTING hull's course is noise that swings through the whole compass: below the threshold # the clause is withheld, or the tour would state a confident tack taken from a wandering fix. assert sailing_words(45.0, 0.0, SAILING_MIN_KTS - 0.1) is None assert sailing_words(45.0, 0.0, SAILING_MIN_KTS) == "beating on port tack" # DEEP RUNNING names no tack: by the lee the boom is on the other side, and nothing we can see # says which. The point of sail still stands — it is the tack that is unsupported. assert sailing_words(178.0, 0.0, 6.0) == "running" assert sailing_words(190.0, 0.0, 6.0) == "running" # a vessel/mark is never given a point of sail (it is moored, and it is not a sailing boat) v = dataclasses.replace(_stop("o:buoy:A", "Start Vessel", kind="vessel"), course_deg=45.0, venue_wind_from_deg=0.0) assert visit_line(v) == "This is Start Vessel." # A sailing boat referred to by NUMBER always wears the "Boat" prefix; a real name never does. assert boat_display_name("10") == "Boat 10" assert boat_display_name(" 7 ") == "Boat 7" assert boat_display_name("Alpha") == "Alpha" assert boat_display_name("") == "" def test_unpositioned_device_skips_this_lap_only(): lost = [_stop("o:boat:1", "Alpha"), _stop("o:boat:2", "Bravo", lat=None, lon=None)] found = [_stop("o:boat:1", "Alpha"), _stop("o:boat:2", "Bravo")] sched = TourScheduler(CFG) evs = [] for t in range(200): e, _ = sched.tick(t, lost if t < 60 else found) evs += e lap_of = {} lap = 0 for e in evs: if e["type"] == "tour_lap": lap = e["lap"] elif e["type"] == "tour_visit": lap_of.setdefault(lap, set()).add(e["deviceId"]) assert "o:boat:2" not in lap_of.get(1, set()), "no finite fix — nothing to aim a camera at" later = [laps for laps in lap_of.values() if "o:boat:2" in laps] assert later, "the device rejoins the tour the moment it reports a position" def test_opening_weather_landmark_tour_and_skyline_bounds(): landmarks = [{"name": "Stadshuset", "lat": 59.3275, "lon": 18.0546, "line": "Stockholm City Hall, on the northern shore."}, {"name": "Västerbron", "lat": 59.3243, "lon": 18.0290, "line": "The Västerbron bridge arches over the western end."}] sched = TourScheduler(CFG, landmarks=landmarks, venue_name="Stockholm") stops = [_stop("o:boat:1", "Alpha"), _stop("o:boat:2", "Bravo")] evs, dirs = [], [] for t in range(260): e, d = sched.tick(t, stops, {"windFromDeg": 225.0, "windKts": 6.0}) evs += e dirs += d # the opening speaks the conditions over the venue w = next(e for e in evs if e["type"] == "tour_weather") assert w["text"] == "Wind from the southwest at 6 knots over Stockholm." assert w["commentaryRole"] == "lead" and w["claimClass"] == "structure" # the guided venue tour runs before lap 1: every landmark, as a poi beat with its line pois = [e for e in evs if e["type"] == "tour_poi"] first_lap = next(e for e in evs if e["type"] == "tour_lap") assert [p["name"] for p in pois[:2]] == ["Stadshuset", "Västerbron"] assert all(p["tSec"] < first_lap["tSec"] for p in pois[:2]) assert pois[0]["poi"] == {"lat": 59.3275, "lon": 18.0546} assert pois[0]["text"].startswith("Stockholm City Hall") assert pois[0]["poiDistM"] > 100, "a building is framed from a landmark stand-off" # …and re-runs every venueTourEveryLaps laps laps = [e["tSec"] for e in evs if e["type"] == "tour_lap"] assert len(laps) >= 3 rerun = [p for p in pois if p["tSec"] > laps[1]] assert rerun, "the landmark tour returns on the configured lap cadence" # the skyline breather is explicitly bounded — no cine may hold the camera forever sky = [e for e in evs if e["type"] == "tour_venue"] assert sky and all(e["cam"] == "venue" and e["camUntilSec"] > e["tSec"] for e in sky) def test_vessel_visits_fly_to_the_reported_fix_as_a_poi(): stops = [_stop("o:buoy:A", "Start Vessel", kind="vessel", lat=59.325, lon=18.05, clip_age=5.0)] evs, dirs = _run(TourScheduler(CFG), stops, 60) v = next(e for e in evs if e["type"] == "tour_visit") assert "focus" not in v, "a vessel is not in the roster — nothing for the scene to track" assert v["poi"] == {"lat": 59.325, "lon": 18.05} assert v["poiDistM"] < 150, "a vessel is framed close, not from a landmark stand-off" assert any(d["boatId"] == "o:buoy:A" for d in dirs), \ "a line phone with fresh video gets its panel like any hull" def test_config_from_meta_reads_camel_case_and_survives_junk(): cfg = CruiseConfig.from_meta({"dwellS": 30, "flyInS": 4, "venueTourEveryLaps": 0, "freshClipMaxAgeS": "soon", "interludeS": -3, "unknown": 1, "excursionEveryLaps": 2}) assert cfg.dwell_s == 30 and cfg.fly_in_s == 4 assert cfg.venue_tour_every_laps == 0 assert cfg.fresh_clip_max_age_s == CruiseConfig.fresh_clip_max_age_s # junk → default assert cfg.interlude_s == CruiseConfig.interlude_s # negative → default assert cfg.excursion_every_laps == 2 assert CruiseConfig.from_meta(None) == CruiseConfig() assert CruiseConfig.from_meta({}).excursion_every_laps == 0, \ "the excursion ships DISARMED — absent reads as off" def test_excursion_lap_delegates_to_the_module_and_wins_a_collision_lap(): """Every `excursion_every_laps` laps the tour hands the picture to `excursion.Excursion` — a continuous flight (moving poi, `poiBearingDeg` on every beat) — and on a lap that owes BOTH the landmark ring and the excursion, the excursion wins. When the flight lands, the ordinary lap resumes with its roll-call. Without an excursion (or with the knob absent) the schedule is untouched — every other test in this file runs that way.""" from excursion import Excursion, ExcursionConfig landmarks = [{"name": "Stadshuset", "lat": 59.3275, "lon": 18.0546, "line": "The hall."}] sites = [{"name": "Alpha House", "lat": 59.3268, "lon": 18.0717, "line": "A line."}] cfg = CruiseConfig(dwell_s=6, dwell_no_video_s=3, interlude_s=2, opening_pan_s=4, fly_in_s=2, venue_tour_every_laps=2, landmark_dwell_s=4, excursion_every_laps=2) exc = Excursion(ExcursionConfig(excursion_speed_mps=30, excursion_target_s=600), sites=sites, centre=(59.3245, 18.0456), home_name="The Bay") sched = TourScheduler(cfg, landmarks=landmarks, venue_name="Stockholm", excursion=exc) stops = [_stop("o:boat:1", "Alpha")] evs = [] for t in range(700): e, _ = sched.tick(t, stops) evs += e laps = [e["tSec"] for e in evs if e["type"] == "tour_lap"] assert len(laps) >= 3, "the tour must keep looping around the flight" # the excursion flies between lap 2 and lap 3 (lap % 2 == 0): moving-aim beats, 1 Hz flight = [e for e in evs if e["type"] == "tour_poi" and "poiBearingDeg" in e and laps[1] < e["tSec"] < laps[2]] assert len(flight) > 30, "the excursion actually flew on its cadence lap" ts = [e["tSec"] for e in flight] assert ts == list(range(ts[0], ts[0] + len(ts))), "one beat every second — never parked" assert any(e.get("text") == "A line." for e in flight), "the site's line is spoken en route" # …and it WON the collision lap: the landmark ring (static poi, no bearing) did not also run ring = [e for e in evs if e["type"] == "tour_poi" and "poiBearingDeg" not in e and laps[1] < e["tSec"] < laps[2]] assert ring == [], "a lap that owes both flies the excursion, not the landmark ring too" # the ring still exists in the schedule (it ran before lap 1, per the opening) assert any(e["type"] == "tour_poi" and "poiBearingDeg" not in e for e in evs) def test_a_lap_does_not_recite_the_fleet_before_showing_it(): """The lap used to open with "On the water today: …" — every device, by name, once per lap. With a dozen phones out that is a list nobody can hold, delivered several times an hour, and it pre-announces exactly what the following minutes say properly: the tour VISITS each device and names it there, on the plate and in the line. It read as filler on air (Ulf 2026-08-11), so the lap now begins at its first visit. The lap BEAT stays — it is what marks a lap boundary for anything counting them, the excursion cadence included — so this asserts the beat is there and silent, not that it is gone.""" stops = [_stop(f"o:boat:{n}", f"Boat {n}", lat=59.32 + n / 1000, lon=18.04) for n in range(1, 13)] evs, _dirs = _run(TourScheduler(CFG), stops, 400) laps = [e for e in evs if e["type"] == "tour_lap"] assert laps, "a lap boundary is still published" for e in laps: assert not e.get("text"), f"a lap must not speak a line-up: {e.get('text')!r}" assert not any("On the water today" in str(e.get("text") or "") for e in evs) # …and the fleet is still named, one device at a time, where the picture shows it. named = {e.get("name") for e in evs if e["type"] == "tour_visit"} assert len(named) >= 2, "the tour still names the devices it visits"