r""" Fit the 4 course marks (buoys) of the GSYS figure-8 from the real boat tracks (decision 5) — the marks weren't recorded in V2, so we infer them from where the fleet repeatedly rounded, then nudge each so no boat's track passes within 2 m of it. Per DAY (marks are stable within a day, re-laid between days): 1. rounding_clusters — pool every boat's sharp-turn apexes and cluster them; the 4 densest clusters are the buoys (the whole fleet rounds each one, lap after lap). 2. assign_figure8 — split the 4 into the 2 windward (west) + 2 leeward (east) marks and lay them out in the figure-8 waypoint order gen_gsys_series.Course.feed_course() expects, with the start/finish line on the leeward pair. 3. clear_marks — push each mark off the tracks until the minimum boat-to-mark distance ≥ 2 m. Pure (unit-tested with synthetic tracks); takes/returns plain lat-lon, no Firestore. """ from __future__ import annotations import math import sys from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) # segment_gsys geometry helpers from segment_gsys import _ang_delta, _bearing_xy, _thin_by_distance, to_xy # noqa: E402 _M_PER_DEG_LAT = 111_320.0 def to_latlon(x: float, y: float, ref: tuple[float, float]) -> tuple[float, float]: """Inverse of segment_gsys.to_xy.""" return (ref[0] + y / _M_PER_DEG_LAT, ref[1] + x / (_M_PER_DEG_LAT * math.cos(math.radians(ref[0])))) # --- rounding detection + clustering --------------------------------------------------------------- def turn_apexes(xy: list[tuple[float, float]], *, step_m: float = 5.0, min_angle_deg: float = 35.0, wind_from: float | None = None, beat_cone_deg: float = 60.0, leg_span: int = 4) -> list[tuple[float, float]]: """Vertices where the track turns hard — a boat rounding a mark. Thin to ~`step_m` spacing first so the per-vertex turn angle is meaningful (not GPS jitter). When `wind_from` is given, DROP **tacking** apexes — a tack is a hard turn whose incoming AND outgoing legs are both close-hauled *toward* the wind (within `beat_cone_deg` of `wind_from`), so it is a beat zig-zag, not a mark rounding. This de-noises the windward leg, where the tacking apexes would otherwise swamp the real windward-mark cluster (the leeward run turns are clean).""" pts = _thin_by_distance(xy, step_m) out = [] for i in range(1, len(pts) - 1): turn = abs(_ang_delta(_bearing_xy(pts[i - 1], pts[i]), _bearing_xy(pts[i], pts[i + 1]))) if turn < min_angle_deg: continue if wind_from is not None: j0, j1 = max(0, i - leg_span), min(len(pts) - 1, i + leg_span) h_in = _bearing_xy(pts[j0], pts[i]) h_out = _bearing_xy(pts[i], pts[j1]) if abs(_ang_delta(h_in, wind_from)) < beat_cone_deg and abs(_ang_delta(h_out, wind_from)) < beat_cone_deg: continue # both legs beating toward the wind -> a tack, not a rounding out.append(pts[i]) return out def cluster(points: list[tuple[float, float]], radius_m: float = 15.0) -> list[tuple[tuple[float, float], int]]: """Greedy radius clustering (no deps). Returns [(centroid, member_count)] sorted by count desc.""" clusters: list[list[tuple[float, float]]] = [] for p in points: for c in clusters: cx = sum(q[0] for q in c) / len(c) cy = sum(q[1] for q in c) / len(c) if math.dist(p, (cx, cy)) <= radius_m: c.append(p) break else: clusters.append([p]) out = [((sum(q[0] for q in c) / len(c), sum(q[1] for q in c) / len(c)), len(c)) for c in clusters] out.sort(key=lambda cn: cn[1], reverse=True) return out def rounding_clusters(tracks_xy: list[list[tuple[float, float]]], *, top_n: int = 4, radius_m: float = 15.0, **apex_kw) -> list[tuple[float, float]]: """The `top_n` densest rounding centroids across all boats' tracks — the buoys.""" apexes = [a for xy in tracks_xy for a in turn_apexes(xy, **apex_kw)] return [centroid for centroid, _n in cluster(apexes, radius_m)[:top_n]] # --- figure-8 assignment --------------------------------------------------------------------------- def assign_figure8(centroids: list[tuple[float, float]]) -> dict: """Lay 4 mark centroids out as a figure-8: the 2 westernmost are the windward gate (Mark 1 pair), the 2 easternmost the leeward / start-finish pair. Within each pair, the more-northerly is `port`. Returns {marks_xy:[6], start_xy:(port,stbd), finish_xy:(port,stbd)} mirroring the wp0..wp5 order (west, west, east, east, west, west) gen_gsys_series uses (so the route self-crosses → figure8).""" if len(centroids) < 4: raise ValueError(f"need 4 mark clusters, got {len(centroids)}") by_x = sorted(centroids[:4], key=lambda p: p[0]) west = sorted(by_x[:2], key=lambda p: -p[1]) # north (port) first east = sorted(by_x[2:], key=lambda p: -p[1]) wp = [west[0], west[1], east[0], east[1], west[0], west[1]] # figure-8 waypoint order return {"marks_xy": wp, "start_xy": (east[0], east[1]), "finish_xy": (east[0], east[1])} # --- ≥2 m clearance -------------------------------------------------------------------------------- def clear_marks(marks_xy: list[tuple[float, float]], tracks_xy: list[list[tuple[float, float]]], *, min_m: float = 2.0, max_iter: int = 30) -> list[tuple[float, float]]: """Push each mark off the tracks until no boat point is within `min_m` — directly away from the nearest encroaching point, iterated. Approximates the boat mid-ship line by the track polyline.""" allpts = [p for xy in tracks_xy for p in xy] if not allpts: return list(marks_xy) out = [] for m in marks_xy: mk = (m[0], m[1]) for _ in range(max_iter): near = min(allpts, key=lambda p: math.dist(p, mk)) d = math.dist(near, mk) if d >= min_m: break dx, dy = mk[0] - near[0], mk[1] - near[1] n = math.hypot(dx, dy) or 1.0 push = (min_m - d) + 0.10 mk = (mk[0] + dx / n * push, mk[1] + dy / n * push) out.append(mk) return out # --- top-level: tracks → feed `course` block ------------------------------------------------------- def build_course_block(tracks_latlon: list[list[tuple[float, float]]], ref: tuple[float, float], *, name: str = "digital 8", min_clear_m: float = 2.0, wind_from: float | None = None, cluster_radius_m: float = 18.0, **apex_kw) -> dict: """Fit the figure-8 course (waypoints + start/finish lines) from the day's boat tracks (lat-lon), in the shape gen_gsys_series.Course.feed_course() produces. `ref` is the projection origin (course centre). Marks are cleared to `min_clear_m` from every track. Pass `wind_from` to drop tacking apexes (recommended — the windward gate is otherwise lost in the beat's zig-zag noise).""" tracks_xy = [[to_xy(lat, lon, ref) for lat, lon in t] for t in tracks_latlon] cents = rounding_clusters(tracks_xy, radius_m=cluster_radius_m, wind_from=wind_from, **apex_kw) marks = clear_marks(assign_figure8(cents)["marks_xy"], tracks_xy, min_m=min_clear_m) a = assign_figure8(marks) # re-split the cleared marks wps = [{"name": "", "type": "rounding", "rounding": "port", "lat": round(to_latlon(x, y, ref)[0], 7), "lon": round(to_latlon(x, y, ref)[1], 7)} for x, y in a["marks_xy"]] def _line(pair): (px, py), (sx, sy) = pair pll, sll = to_latlon(px, py, ref), to_latlon(sx, sy, ref) return {"port": {"lat": round(pll[0], 7), "lon": round(pll[1], 7)}, "starboard": {"lat": round(sll[0], 7), "lon": round(sll[1], 7)}} pts = [to_latlon(x, y, ref) for x, y in a["marks_xy"]] center = {"lat": sum(p[0] for p in pts) / len(pts), "lon": sum(p[1] for p in pts) / len(pts)} return {"name": name, "center": center, "startLine": _line(a["start_xy"]), "finishLine": _line(a["finish_xy"]), "waypoints": wps} def min_clearance_m(course_block: dict, tracks_latlon: list[list[tuple[float, float]]], ref: tuple[float, float]) -> float: """Smallest boat-to-mark distance (m) for a built course — a QA number for the exporter to print.""" marks = [to_xy(w["lat"], w["lon"], ref) for w in course_block["waypoints"]] allpts = [to_xy(lat, lon, ref) for t in tracks_latlon for lat, lon in t] return min((math.dist(p, m) for m in marks for p in allpts), default=float("inf"))