"""The guided excursion (`v3/broadcast/live/excursion.py`) — the continuous narrated city flight. What these tests pin is the mode's whole claim (Ulf 2026-08-10): the aim point MOVES every single tick (never parks — a long text slows a leg, never stops it), a story route flies whole and in chapter order, eligibility is computed from the radius and the time budget (never tagged on the data), the stand-off bearing crosses the compass seam by the shortest delta, and the recall abandons the route for a single graceful home leg.""" import math import sys from pathlib import Path _LIVE = Path(__file__).resolve().parents[2] / "live" sys.path.insert(0, str(_LIVE)) from cruise_tour import _speech_seconds # noqa: E402 from excursion import ( # noqa: E402 BLEND_S, Excursion, ExcursionConfig, _bearing_deg, _dist_m, _wrap_delta, ) CENTRE = (59.3245, 18.0456) CFG = ExcursionConfig(excursion_speed_mps=30, excursion_standoff_m=300, excursion_target_s=600, recall_speed_mps=70, recall_budget_s=60) def _site(name, lat, lon, line="", tour=None): s = {"name": name, "lat": lat, "lon": lon} if line: s["line"] = line if tour: s["tour"] = tour return s #: ~1.1 km apart along the shore — comfortably inside every default budget. NEAR = [_site("Alpha House", 59.3268, 18.0717, line="A first line about the Alpha House."), _site("Bravo Hall", 59.3280, 18.0916, line="A second line, about the Bravo Hall.")] def _fly(exc, t0=0, limit=5000): """begin() at t0 and tick to completion; returns the emitted beats.""" exc.begin(t0) beats = [] for t in range(t0, t0 + limit): beats += exc.tick(t) if exc.done: break assert exc.done, "the route must complete within the test horizon" return beats def test_geometry_helpers(): assert abs(_dist_m(59.0, 18.0, 60.0, 18.0) - 110_574) < 600 # one degree of latitude assert abs(_bearing_deg(59.0, 18.0, 60.0, 18.0) - 0.0) < 0.01 # due north assert abs(_bearing_deg(59.0, 18.0, 59.0, 18.1) - 90.0) < 0.01 # due east assert _wrap_delta(350.0, 10.0) == 20.0 # across the seam, short way assert _wrap_delta(10.0, 350.0) == -20.0 def test_the_aim_point_moves_every_tick_and_never_faster_than_the_knob(): beats = _fly(Excursion(CFG, sites=NEAR, centre=CENTRE, home_name="The Bay")) assert len(beats) > 30 for a, b in zip(beats, beats[1:], strict=False): step = _dist_m(a["poi"]["lat"], a["poi"]["lon"], b["poi"]["lat"], b["poi"]["lon"]) assert step > 0.0, "the tour NEVER parks — the aim point moves every tick" assert step <= CFG.excursion_speed_mps * 1.01 + 0.01, "…and never outruns the speed knob" def test_wire_shape_of_every_beat(): beats = _fly(Excursion(CFG, sites=NEAR, centre=CENTRE, home_name="The Bay")) for b in beats: assert b["type"] == "tour_poi" and b["shotType"] == "boat-feature" assert b["claimClass"] == "structure" and b["commentaryRole"] == "lead" assert b["poiDistM"] == CFG.excursion_standoff_m assert b["focusUntilSec"] == b["tSec"] + 3, "a stalled feed must release the camera" assert 0.0 <= b["poiBearingDeg"] < 360.0 # the plate names the site being approached, and the venue's own name on the glide home assert beats[0]["name"] in ("Alpha House", "Bravo Hall") assert beats[-1]["name"] == "The Bay" def test_each_narrated_site_speaks_exactly_once_and_the_line_fits_its_leg(): beats = _fly(Excursion(CFG, sites=NEAR, centre=CENTRE)) texts = [b["text"] for b in beats if "text" in b] assert sorted(texts) == sorted(s["line"] for s in NEAR), "one line per site, exactly once" # a line lands so it FINISHES before the site is reached: after a text beat, its site's name # stays on the plate for at least the speech time for tb in (b for b in beats if "text" in b): tail = [b for b in beats if b["tSec"] >= tb["tSec"] and b.get("name") == tb.get("name")] assert len(tail) >= _speech_seconds(tb["text"]), "the leg is never shorter than its line" def test_the_route_ends_at_home(): beats = _fly(Excursion(CFG, sites=NEAR, centre=CENTRE)) last = beats[-1]["poi"] assert _dist_m(last["lat"], last["lon"], *CENTRE) <= CFG.excursion_speed_mps * 2, \ "the final beat is within a tick or two of home — the excursion always returns" def test_a_story_route_flies_whole_in_chapter_order_never_interleaved(): # Chapters deliberately laid so nearest-neighbour WOULD interleave the free site between # them if the story unit could be broken. sites = [_site("Free A", 59.3268, 18.0717, line="About the free site A."), _site("Ch 1", 59.3296, 18.0785, line="Chapter one.", tour="story"), _site("Ch 2", 59.3258, 18.0745, line="Chapter two.", tour="story"), _site("Ch 3", 59.3280, 18.0916, line="Chapter three.", tour="story")] beats = _fly(Excursion(CFG, sites=sites, centre=CENTRE)) order = [] for b in beats: n = b.get("name") if n and (not order or order[-1] != n): order.append(n) idx = [order.index(f"Ch {i}") for i in (1, 2, 3)] assert idx == sorted(idx) and idx[2] - idx[0] == 2, \ f"chapters in listed order, contiguous — got {order}" def test_rotation_is_deterministic_and_varies_the_start(): exc = Excursion(CFG, sites=NEAR, centre=CENTRE) first_of = [] for _ in range(2): beats = _fly(exc, t0=0) first_of.append(next(b["name"] for b in beats if b.get("name"))) assert first_of[0] != first_of[1], "consecutive excursions rotate the starting site" # …but the SAME counter over the same catalogue always plans the same route (no randomness) a = Excursion(CFG, sites=NEAR, centre=CENTRE)._plan_route(0) b = Excursion(CFG, sites=NEAR, centre=CENTRE)._plan_route(0) assert a == b def test_radius_excludes_a_unit_whole_and_is_never_a_tag(): far = _site("Far Isle", 59.3345, 17.5448, line="Thirty kilometres west.") half_far = [_site("Near Ch", 59.3268, 18.0717, tour="mix", line="Near chapter."), _site("Far Ch", 59.3345, 17.5448, tour="mix", line="Far chapter.")] exc = Excursion(CFG, sites=[*NEAR, far, *half_far], centre=CENTRE, radius_m=CFG.recall_radius_m()) assert CFG.recall_radius_m() == 70 * 60 names = {b.get("name") for b in _fly(exc)} assert "Far Isle" not in names, "a site beyond the recall radius never flies on a race day" assert "Near Ch" not in names and "Far Ch" not in names, \ "one far chapter grounds the WHOLE story route — a unit is eligible whole or not at all" assert {"Alpha House", "Bravo Hall"} <= names def test_the_time_budget_grounds_a_unit_too_long_for_it(): far = _site("Far Isle", 59.3345, 17.5448, line="Thirty kilometres west.") # ~2000 s round trip exc = Excursion(CFG, sites=[*NEAR, far], centre=CENTRE) # radius UNBOUNDED for _ in range(4): # every rotation position names = {b.get("name") for b in _fly(exc)} assert "Far Isle" not in names, "a unit that cannot fit excursionTargetS never flies" roomy = ExcursionConfig(excursion_speed_mps=30, excursion_target_s=4000) names = {b.get("name") for b in _fly(Excursion(roomy, sites=[*NEAR, far], centre=CENTRE))} assert "Far Isle" in names, "…and flies the day an event raises the budget — data, not code" def test_the_standoff_bearing_crosses_the_seam_by_the_shortest_delta(): # Travel bearings ~170° then ~190°, so the camera bearing blends 350° → 10° ACROSS the # compass seam. Observed over the two named legs only (the U-turn home is a legitimate big # bank, not a seam case — the renderer's spring is what smooths its sweep). s1 = _site("South One", CENTRE[0] - 0.0178, CENTRE[1] + 0.0061) s2 = _site("South Two", CENTRE[0] - 0.0356, CENTRE[1]) exc = Excursion(CFG, sites=[s1, s2], centre=CENTRE) beats = [b for b in _fly(exc) if b.get("name") in ("South One", "South Two")] bearings = [b["poiBearingDeg"] for b in beats] for a, b in zip(bearings, bearings[1:], strict=False): assert abs(_wrap_delta(a, b)) <= (20.0 / BLEND_S) + 0.2, \ f"a 20° turn blends in small shortest-way steps — never {a}→{b}" assert not any(90.0 < x < 270.0 for x in bearings), \ "the 350°→10° blend crosses the seam directly — never the long way through south" assert any(x > 340.0 for x in bearings) and any(x < 20.0 for x in bearings), \ "…and both sides of the seam were actually visited" def test_recall_turns_for_home_with_a_raised_cap_and_no_more_narration(): exc = Excursion(CFG, sites=NEAR, centre=CENTRE, home_name="The Bay") exc.begin(0) for t in range(12): # well inside the first leg exc.tick(t) at = exc.tick(12)[0]["poi"] exc.recall(13) beats = [] for t in range(13, 13 + CFG.recall_budget_s + 5): beats += exc.tick(t) if exc.done: break assert exc.done, "a site inside the radius is home within the recall budget" assert all("text" not in b for b in beats), "the recall speaks no new line" assert all(b["poiSpeedMps"] == int(70 * 1.5) for b in beats), \ "the home run carries the raised spring cap" d0 = _dist_m(at["lat"], at["lon"], *CENTRE) dN = _dist_m(beats[-1]["poi"]["lat"], beats[-1]["poi"]["lon"], *CENTRE) assert dN < d0, "…and it flies HOME" # idempotent, and a no-op when not flying exc.recall(999) assert exc.done done_before = Excursion(CFG, sites=NEAR, centre=CENTRE) done_before.recall(0) # never begun — nothing to do assert done_before.done def test_update_sites_holds_last_on_garbage_and_honours_a_deliberate_empty(): exc = Excursion(CFG, sites=NEAR, centre=CENTRE) junk = [{"name": "NaN Isle", "lat": float("nan"), "lon": 18.0}, {"name": "", "lat": 59.32, "lon": 18.0}, {"lat": 59.32, "lon": 18.0}] assert exc.update_sites(junk) is False assert exc.sites, "a publish whose every entry is invalid keeps the working catalogue" exc.begin(0) assert exc.flying changed = exc.update_sites([_site("New Place", 59.3268, 18.0717, line="New.")]) assert changed and exc.flying, "a swap never touches the route already flying" beats = [] for t in range(5000): beats += exc.tick(t) if exc.done: break assert not any(b.get("name") == "New Place" for b in beats), \ "the running excursion finishes on the plan it took off with" exc.begin(9000) assert any(b.get("name") == "New Place" for b in exc.tick(9000)), \ "…and the NEXT excursion flies the new catalogue" assert exc.update_sites([]) is True assert exc.sites == [], "a deliberate empty list retires the tour — that is data's right" def test_available_and_config_from_meta(): assert not Excursion(CFG, sites=NEAR, centre=None).available(), "no home point — no flight" assert not Excursion(CFG, sites=[], centre=CENTRE).available() assert Excursion(CFG, sites=NEAR, centre=CENTRE).available() cfg = ExcursionConfig.from_meta({"excursionSpeedMps": 40, "excursionTargetS": 900, "recallSpeedMps": "fast", "unknown": 1, "excursionStandoffM": -5}) assert cfg.excursion_speed_mps == 40 and cfg.excursion_target_s == 900 assert cfg.recall_speed_mps == ExcursionConfig.recall_speed_mps # junk → default assert cfg.excursion_standoff_m == ExcursionConfig.excursion_standoff_m # negative → default assert ExcursionConfig.from_meta(None) == ExcursionConfig() assert math.isclose(ExcursionConfig.from_meta({"recallSpeedMps": 50, "recallBudgetS": 30}) .recall_radius_m(), 1500.0)