r""" Auto-detect the GSYS figure-8 races from the real V2 telemetry (decision: auto-detect → user curates). V2 has no usable race bookkeeping and the fleet sailed continuously all day on a 3-minute stopwatch (uncoordinated to wall-clock), so races + gun times are DERIVED FROM THE TRACKS, per day: 1. resolve each of the 4 boats → its device for the day (deviceMounts / _mount_at — handles redeploys) 2. pull 1 Hz tracks over the day window 3. fit the day's marks from where the fleet repeatedly rounded (fit_marks) 4. estimate each gun = the fleet's mass UPWIND start-line crossing (segment_gsys.estimate_guns) 5. confirm a figure-8 follows each gun Prints a candidate race list (date · estimated local start · boats · figure-8 ✓) for the operator to keep/drop before the exporter builds them. Validated on May 22: lands on the demo race (~13:09) and catches the OCS boat. ..\..\..\.venv-demo\Scripts\python.exe export\detect_gsys_races.py --day 2026-05-22 ..\..\..\.venv-demo\Scripts\python.exe export\detect_gsys_races.py --day 2026-05-22 --from 6 --to 22 """ from __future__ import annotations import argparse import math import sys from datetime import UTC, datetime, timedelta from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) sys.path.insert(0, str(_HERE.parents[2] / "migration")) from fit_marks import assign_figure8, rounding_clusters # noqa: E402 from segment_gsys import estimate_guns, is_figure8, to_xy, upwind_crossings # noqa: E402 from v2_reader import ( # noqa: E402 GSYS_USER_ID, device_mounts_for_boat, list_start_groups, list_team_entries, open_firestore, stream_points_for_track, tracks_for_device_in_window, ) CENTER = (59.32119602870078, 18.107211561645112) # course centre (demo digital-8), projection ref VENUE_TZ_H = 2 # CEST = UTC+2 def _ms(dt: datetime) -> int: return int(dt.timestamp() * 1000) def _local(ms: int) -> str: return (datetime.fromtimestamp(ms / 1000, tz=UTC) + timedelta(hours=VENUE_TZ_H)).strftime("%H:%M:%S") def _mount_at(mounts: list[dict], t_ms: int) -> dict | None: """The deviceMount covering t_ms (latest fromUtcMs wins, then narrowest) — mirrors apply_2324.""" cov = [m for m in mounts if int(m.get("fromUtcMs") or 0) <= t_ms and (m.get("toUtcMs") is None or t_ms <= int(m["toUtcMs"]))] if not cov: return None return max(cov, key=lambda m: (int(m.get("fromUtcMs") or 0), -int(m["toUtcMs"]) if m.get("toUtcMs") else float("-inf"))) def sail_to_boat(client) -> dict[str, str]: # noqa: ANN001 """sail number → boatId, read from any GSYS start-group's boat entries (stable across races).""" from v2_reader import list_instances_for_window insts = list_instances_for_window(client, owner_user_id=GSYS_USER_ID, start_ms=_ms(datetime(2026, 5, 1, tzinfo=UTC)), end_ms=_ms(datetime(2026, 6, 1, tzinfo=UTC))) out: dict[str, str] = {} for inst in insts: for g in list_start_groups(client, inst["instanceId"]): for t in list_team_entries(client, inst["instanceId"], g["groupId"]): for b in t.get("boats", []): if b.get("sailId") and b.get("boatId"): out.setdefault(str(b["sailId"]), b["boatId"]) return out _CHUNK_MS = 2 * 60 * 60 * 1000 # fetch the day in 2-hour chunks — a full-day query overruns the # Firestore client and trips its (buggy) retry path. def _fetch_chunk(client, dev: str, lo: int, hi: int, # noqa: ANN001 by_sec: dict[int, tuple[tuple[float, float], int]]) -> None: for tr in tracks_for_device_in_window(client, dev, from_ms=lo, to_ms=hi): for p in stream_points_for_track(client, tr["trackId"], from_ms=lo, to_ms=hi): if p.get("lat") is None: continue t = int(p.get("gnssTimestamp") or p.get("t") or 0) by_sec[t // 1000] = (to_xy(p["lat"], p.get("lon"), CENTER), t) def day_points(client, dev: str, lo: int, hi: int) -> list[tuple[tuple[float, float], int]]: # noqa: ANN001 """1 Hz [(xy, t_ms)] for `dev` over [lo, hi) (last fix per second). No at-sea trim — the whole day is scanned (in chunks, with retry) so every race in it is found.""" by_sec: dict[int, tuple[tuple[float, float], int]] = {} t = lo while t < hi: ce = min(t + _CHUNK_MS, hi) for attempt in range(3): try: _fetch_chunk(client, dev, t, ce, by_sec) break except Exception as e: # noqa: BLE001 — tolerate the client's transient retry-path crash if attempt == 2: print(f" (skipped chunk {_local(t)}–{_local(ce)} for {dev[:8]}: {type(e).__name__})") t = ce return [by_sec[s] for s in sorted(by_sec)] _WIN_MS = int(2.5 * 60 * 60 * 1000) # sliding analysis window — marks are stable within it _STEP_MS = int(1.25 * 60 * 60 * 1000) # 50% overlap so a race near a boundary isn't split def _guns_in_window(tracks_sub: dict[str, list[tuple[tuple[float, float], int]]] ) -> list[tuple[int, list[str]]]: """Fit marks for THIS window's tracks and estimate its guns. Empty if <3 boats or <4 mark clusters.""" active = {s: p for s, p in tracks_sub.items() if len(p) > 50} if len(active) < 3: return [] cents = rounding_clusters([[xy for xy, _t in p] for p in active.values()], top_n=4, radius_m=18.0) if len(cents) < 4: return [] a = assign_figure8(cents) sp, ss = a["start_xy"] mid_xy = ((sp[0] + ss[0]) / 2, (sp[1] + ss[1]) / 2) upw = (a["marks_xy"][0][0] - mid_xy[0], a["marks_xy"][0][1] - mid_xy[1]) n = math.hypot(*upw) or 1.0 return estimate_guns(upwind_crossings(active, (sp, ss), (upw[0] / n, upw[1] / n))) def detect_day(client, day: datetime, from_h: int, to_h: int) -> None: # noqa: ANN001 lo, hi = _ms(day.replace(hour=from_h)), _ms(day.replace(hour=to_h) if to_h < 24 else day + timedelta(days=1)) s2b = sail_to_boat(client) dev_of: dict[str, str] = {} for sail, bid in sorted(s2b.items()): m = _mount_at(device_mounts_for_boat(client, bid), (lo + hi) // 2) if m and m.get("deviceId"): dev_of[sail] = m["deviceId"] print(f"\n=== {day:%Y-%m-%d} window {from_h:02d}:00–{to_h:02d}:00 UTC ===") print("boats→device: " + ", ".join(f"{s}:{d[:8]}" for s, d in sorted(dev_of.items()))) tracks = {sail: day_points(client, dev, lo, hi) for sail, dev in dev_of.items()} for sail, pts in sorted(tracks.items()): print(f" sail {sail}: {len(pts)} pts") if len([t for t in tracks.values() if t]) < 3: print(" <3 boats with telemetry — skipping day") return # slide a window across the day; fit marks + estimate guns PER window (marks shift between sessions), # then merge guns within 5 min (same race seen in two overlapping windows). raw: list[tuple[int, list[str]]] = [] t = lo while t < hi: sub = {s: [(xy, tt) for xy, tt in p if t <= tt < t + _WIN_MS] for s, p in tracks.items()} raw.extend(_guns_in_window(sub)) t += _STEP_MS raw.sort() merged: list[tuple[int, set[str]]] = [] for gun, sails in raw: if merged and gun - merged[-1][0] < 300_000: merged[-1] = (merged[-1][0], merged[-1][1] | set(sails)) else: merged.append((gun, set(sails))) print(f" candidate races ({len(merged)}):") for gun, sails in merged: n_fig8 = sum(1 for p in tracks.values() if is_figure8([xy for xy, t in p if gun <= t <= gun + 1_200_000])) flag = "figure-8 ✓" if n_fig8 >= 2 else f"only {n_fig8} fig8 (review)" print(f" {day:%Y-%m-%d} {_local(gun)} CEST boats={sorted(sails)} {flag}") def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--day", help="YYYY-MM-DD (default: all of 2026-05-22/23/24)") ap.add_argument("--from", dest="from_h", type=int, default=6, help="window start hour UTC (default 6)") ap.add_argument("--to", dest="to_h", type=int, default=22, help="window end hour UTC (default 22)") args = ap.parse_args() client = open_firestore("apps-viewregatta") days = ([datetime.strptime(args.day, "%Y-%m-%d").replace(tzinfo=UTC)] if args.day else [datetime(2026, 5, d, tzinfo=UTC) for d in (22, 23, 24)]) for day in days: detect_day(client, day, args.from_h, args.to_h) return 0 if __name__ == "__main__": sys.exit(main())