r""" Segment the raw V2 GSYS telemetry (2026-05-23/24) into individual figure-8 races — the discovery half of the real-track rebuild (replacing the synthetic gen_gsys_series). V2 has no trustworthy raceInstance bookkeeping for these flights, so races, gun times and rosters are DERIVED FROM THE TELEMETRY: 1. day_track(dev, day) — one at-sea track per (boat, day): fetch the day's V2 points (downsample to 1 Hz), then trim to the at-sea span (drop dock/transport idling before launch + after haul-out). 2. segment_races(...) — cluster the fleet's at-sea tracks into per-race windows (>=4 boats moving together, split on long idle gaps), confirm each is a FIGURE-8 (the boat track self-crosses / makes two opposite loops), and infer the gun + the [gun-5min, last_finish+2min] presentation window. The geometry/segmentation functions are PURE (unit-tested with synthetic tracks, no network); only day_track / devices_for_day touch Firestore (via v2_reader). Validate end-to-end on the KNOWN May-22 race (race_real.GUN + its 4 devices) before trusting May 23/24. """ from __future__ import annotations import math import sys from dataclasses import dataclass, field from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE.parents[2] / "migration")) # v2_reader sys.path.insert(0, str(_HERE.parents[2] / "scoring")) # score_race.segments_intersect # --- geometry (reference-parameterised; gen_gsys_series' helpers are pinned to the demo centre) ----- _M_PER_DEG_LAT = 111_320.0 def to_xy(lat: float, lon: float, ref: tuple[float, float]) -> tuple[float, float]: """(east, north) metres of (lat, lon) relative to `ref=(lat0, lon0)`.""" return ((lon - ref[1]) * _M_PER_DEG_LAT * math.cos(math.radians(ref[0])), (lat - ref[0]) * _M_PER_DEG_LAT) def _bearing_xy(a: tuple[float, float], b: tuple[float, float]) -> float: return (math.degrees(math.atan2(b[0] - a[0], b[1] - a[1])) + 360.0) % 360.0 def _ang_delta(a: float, b: float) -> float: """Signed shortest a→b in (-180, 180].""" return (b - a + 180.0) % 360.0 - 180.0 def bbox_diag_m(xy: list[tuple[float, float]]) -> float: """Diagonal of the track's bounding box (metres) — how much ground it actually covered.""" if not xy: return 0.0 xs = [p[0] for p in xy] ys = [p[1] for p in xy] return math.hypot(max(xs) - min(xs), max(ys) - min(ys)) # --- at-sea trim (decision 4: one track per boat per day, sailing only) ------------------------------ def at_sea_span(points: list[dict], *, speed_thresh: float = 0.8, idle_gap_s: int = 600) -> list[dict]: """Trim a day's 1 Hz points to the at-sea span: the maximal run between the first and last sustained movement, ignoring idle gaps shorter than `idle_gap_s` (so brief drifting mid-day doesn't split it). `points` are dicts with `t` (ms) and `speed` (m/s), time-ascending. Drops the pre-launch / post- haul-out idling. Returns the trimmed slice (possibly empty).""" moving = [i for i, p in enumerate(points) if (p.get("speed") or 0.0) >= speed_thresh] if not moving: return [] # Collapse moving indices into runs separated by > idle_gap_s of stillness; keep the longest run # (by time span) — that's the day on the water. runs: list[tuple[int, int]] = [] start = prev = moving[0] for i in moving[1:]: gap_ms = points[i]["t"] - points[prev]["t"] if gap_ms > idle_gap_s * 1000: runs.append((start, prev)) start = i prev = i runs.append((start, prev)) lo, hi = max(runs, key=lambda r: points[r[1]]["t"] - points[r[0]]["t"]) return points[lo:hi + 1] # --- figure-8 confirmation from a dense boat track -------------------------------------------------- def track_turning(xy: list[tuple[float, float]], *, min_step_m: float = 3.0) -> tuple[float, int]: """(total_abs_turn_deg, sign_flips) of a track. A digital-8 makes ~two loops of OPPOSITE sense, so the cumulative signed heading change reaches ~2x360 deg AND its running direction reverses once (the cross-over). `min_step_m` skips tiny jitter steps so heading is meaningful.""" pts = _thin_by_distance(xy, min_step_m) if len(pts) < 3: return 0.0, 0 headings = [_bearing_xy(pts[i - 1], pts[i]) for i in range(1, len(pts))] total_abs = 0.0 sign_flips = 0 last_sign = 0 for i in range(1, len(headings)): d = _ang_delta(headings[i - 1], headings[i]) total_abs += abs(d) if abs(d) > 5.0: # ignore near-straight noise for the sign s = 1 if d > 0 else -1 if last_sign and s != last_sign: sign_flips += 1 last_sign = s return total_abs, sign_flips def track_self_crosses(xy: list[tuple[float, float]], *, tol_m: float = 8.0) -> bool: """True if the (decimated) track polyline crosses itself — the figure-8 signature. Decimate first so the O(n^2) segment scan is cheap and dense GPS jitter doesn't create spurious micro-crossings.""" from score_race import segments_intersect # reuse the engine's robust predicate pts = _douglas_peucker(xy, tol_m) segs = [(pts[i], pts[i + 1]) for i in range(len(pts) - 1)] for i in range(len(segs)): for j in range(i + 2, len(segs)): if set(segs[i]) & set(segs[j]): # share an endpoint → touch, not cross continue if segments_intersect(segs[i][0], segs[i][1], segs[j][0], segs[j][1]): return True return False def is_figure8(xy: list[tuple[float, float]]) -> bool: """A boat sailed a figure-8 if it turned ~720 deg with a direction reversal AND the path self-crosses.""" total_abs, flips = track_turning(xy) return total_abs >= 540.0 and flips >= 1 and track_self_crosses(xy) # --- race segmentation ------------------------------------------------------------------------------ @dataclass class RaceCandidate: day: str t0_ms: int # first second of fleet activity in the window t1_ms: int # last second of fleet activity gun_ms: int | None = None last_finish_ms: int | None = None devices: list[str] = field(default_factory=list) @property def track_start_ms(self) -> int: return (self.gun_ms - 300_000) if self.gun_ms is not None else self.t0_ms @property def track_end_ms(self) -> int: end = self.last_finish_ms if self.last_finish_ms is not None else self.t1_ms return end + 120_000 def segment_windows(dev_points: dict[str, list[dict]], *, min_boats: int = 4, move_thresh: float = 0.8, idle_gap_s: int = 300) -> list[tuple[int, int, list[str]]]: """Find per-race windows from the fleet's at-sea points: maximal runs of ≥`min_boats` boats moving in the same second, split where the whole fleet goes idle for > `idle_gap_s`. Returns [(t0_ms, t1_ms, [deviceId,…])] — the roster is the devices active anywhere in the window.""" # bucket every device's moving seconds sec_active: dict[int, set[str]] = {} for dev, pts in dev_points.items(): for p in pts: if (p.get("speed") or 0.0) >= move_thresh: sec_active.setdefault(p["t"] // 1000, set()).add(dev) busy = sorted(s for s, devs in sec_active.items() if len(devs) >= min_boats) if not busy: return [] windows: list[tuple[int, int, list[str]]] = [] start = prev = busy[0] for s in busy[1:]: if (s - prev) > idle_gap_s: windows.append(_window(sec_active, start, prev)) start = s prev = s windows.append(_window(sec_active, start, prev)) return windows def _window(sec_active: dict[int, set[str]], s0: int, s1: int) -> tuple[int, int, list[str]]: roster: set[str] = set() for s in range(s0, s1 + 1): roster |= sec_active.get(s, set()) return s0 * 1000, s1 * 1000, sorted(roster) # --- gun estimation from start-line crossings (decision: estimate the gun; no bookkeeping) ---------- # The fleet starts together by crossing the start line going UPWIND on a 3-minute stopwatch (not # coordinated to wall-clock), so the gun is the moment ≥ most boats surge across the line toward Mark 1. # Validated on the May-22 demo: lands on the start AND catches the OCS boat crossing early. def upwind_crossings(tracks_xy_t: dict[str, list[tuple[tuple[float, float], int]]], start_seg: tuple[tuple[float, float], tuple[float, float]], upwind: tuple[float, float]) -> dict[str, list[int]]: """Per device, the times (ms) it crosses `start_seg` heading upwind (velocity·`upwind` > 0). `tracks_xy_t` is dev → [((x,y), t_ms)]. Reuses the engine's segment-intersection predicate.""" from score_race import segments_intersect out: dict[str, list[int]] = {} for dev, pts in tracks_xy_t.items(): for i in range(1, len(pts)): (a0, _t0), (a1, t1) = pts[i - 1], pts[i] if (segments_intersect(a0, a1, start_seg[0], start_seg[1]) and (a1[0] - a0[0]) * upwind[0] + (a1[1] - a0[1]) * upwind[1] > 0): out.setdefault(dev, []).append(t1) return out def estimate_guns(crossings: dict[str, list[int]], *, min_boats: int = 3, window_ms: int = 45_000, sep_ms: int = 300_000) -> list[tuple[int, list[str]]]: """Estimated gun times: the start of each tight cluster where ≥`min_boats` distinct boats cross upwind within `window_ms`, clusters kept ≥`sep_ms` apart (one gun per race). Returns [(gun_ms, [deviceId,…])] in time order — the mass-start surge; an OCS boat crossing early sits just before its gun.""" allc = sorted((t, d) for d, ts in crossings.items() for t in ts) guns: list[tuple[int, list[str]]] = [] i = 0 while i < len(allc): j, boats = i, set() while j < len(allc) and allc[j][0] - allc[i][0] <= window_ms: boats.add(allc[j][1]) j += 1 if len(boats) >= min_boats and (not guns or allc[i][0] - guns[-1][0] > sep_ms): guns.append((allc[i][0], sorted(boats))) i += 1 return guns def confirm_and_describe(window: tuple[int, int, list[str]], dev_points: dict[str, list[dict]], ref: tuple[float, float], *, top_n: int = 4) -> RaceCandidate | None: """Confirm a window is a figure-8 race and pick its roster. Roster = the `top_n` devices that moved the most (bbox diagonal) in the window — drops a committee/coach phone idling nearby. The window is a race iff ≥2 roster boats each sailed a figure-8 (is_figure8). Returns a RaceCandidate or None.""" t0, t1, devs = window spans: list[tuple[float, str, list[tuple[float, float]]]] = [] for dev in devs: xy = [to_xy(p["lat"], p["lon"], ref) for p in dev_points[dev] if t0 <= p["t"] <= t1 and p.get("lat") is not None] spans.append((bbox_diag_m(xy), dev, xy)) spans.sort(reverse=True, key=lambda s: s[0]) roster = spans[:top_n] if sum(1 for _diag, _dev, xy in roster if is_figure8(xy)) < 2: return None return RaceCandidate(day="", t0_ms=t0, t1_ms=t1, devices=[dev for _d, dev, _xy in roster]) # --- small geometry utils --------------------------------------------------------------------------- def _thin_by_distance(xy: list[tuple[float, float]], min_step_m: float) -> list[tuple[float, float]]: if not xy: return [] out = [xy[0]] for p in xy[1:]: if math.dist(out[-1], p) >= min_step_m: out.append(p) return out def _douglas_peucker(xy: list[tuple[float, float]], tol: float) -> list[tuple[float, float]]: """Ramer–Douglas–Peucker polyline simplification (iterative, no recursion-depth risk on long tracks).""" if len(xy) < 3: return list(xy) keep = [False] * len(xy) keep[0] = keep[-1] = True stack = [(0, len(xy) - 1)] while stack: a, b = stack.pop() dmax, idx = 0.0, -1 for i in range(a + 1, b): d = _point_seg_dist(xy[i], xy[a], xy[b]) if d > dmax: dmax, idx = d, i if idx != -1 and dmax > tol: keep[idx] = True stack.append((a, idx)) stack.append((idx, b)) return [p for p, k in zip(xy, keep, strict=False) if k] def _point_seg_dist(p: tuple[float, float], a: tuple[float, float], b: tuple[float, float]) -> float: ax, ay = a bx, by = b px, py = p dx, dy = bx - ax, by - ay if dx == 0 and dy == 0: return math.dist(p, a) t = max(0.0, min(1.0, ((px - ax) * dx + (py - ay) * dy) / (dx * dx + dy * dy))) return math.dist(p, (ax + t * dx, ay + t * dy)) # --- Firestore wrappers (network; thin) ------------------------------------------------------------- def day_track(client, dev: str, lo_ms: int, hi_ms: int, *, speed_thresh: float = 0.8) -> list[dict]: # noqa: ANN001 """One at-sea, 1 Hz track for `dev` over [lo_ms, hi_ms) — every V2 point downsampled to the last per second, then trimmed to the at-sea span. Each point: {t, lat, lon, speed, course, roll, pitch, azimuth}. Network (v2_reader).""" from v2_reader import stream_points_for_track, tracks_for_device_in_window by_sec: dict[int, dict] = {} for tr in tracks_for_device_in_window(client, dev, from_ms=lo_ms, to_ms=hi_ms): for p in stream_points_for_track(client, tr["trackId"], from_ms=lo_ms, to_ms=hi_ms): if p.get("lat") is None: continue t = int(p.get("gnssTimestamp") or p.get("t") or 0) by_sec[t // 1000] = { "t": (t // 1000) * 1000, "lat": p["lat"], "lon": p.get("lon"), "speed": p.get("speed") or 0.0, "course": p.get("course"), "roll": p.get("roll"), "pitch": p.get("pitch"), "azimuth": p.get("azimuth"), } pts = [by_sec[s] for s in sorted(by_sec)] return at_sea_span(pts, speed_thresh=speed_thresh)