r""" Generic team-race exporter for the V3 broadcast demo. Reads the race-specific settings (Firestore instance, window, device->boat mapping, teams, title, director params) from race_real.py, pulls the boat tracks for the window, scores them with the engine, and writes data/races/real.json + manifest.json. Read-only on Firestore. To export a different race, edit race_real.py only. ..\.venv-demo\Scripts\python.exe export\export_team_race.py """ import json import os import sys from datetime import UTC, datetime, timedelta from pathlib import Path from google.cloud import firestore from google.cloud.firestore_v1.base_query import FieldFilter as FF _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) # for race_real # the scoring engine is the shared v3/scoring package, three levels up from demo/export sys.path.insert(0, str(_HERE.parents[2] / "scoring")) sys.path.insert(0, str(_HERE.parents[2] / "ingest")) # for wind_archive (§17) sys.path.insert(0, str(_HERE.parents[3])) # repo root, for v3.rie from race_real import GUN, INSTANCE, PARAMS, PROJECT, TEAMS, TITLE, TRACK_END, TRACK_START, USER, WIND # noqa: E402 from score_race import enrich # noqa: E402 from wind_archive import fetch_archive_hourly, nearest_sample # noqa: E402 from v3.rie.broadcast_bridge import produce_feed # noqa: E402 def _resolve_wind(center: dict | None, group_wind_deg): # noqa: ANN001 """Wind for the race `meta`, **archive-first**: the real ERA5 wind nearest the gun at the course centre wins; only if that can't be reached (offline, or race younger than ERA5's ~5-day lag) does it fall back to the offline `race_real.WIND`, then the start-group's recorded `windDeg`. Returns `(from_deg, kts, gust_kts)` — any may be `None` (None from_deg ⇒ wind-dependent commentary stays disabled).""" if center is not None: s = nearest_sample(fetch_archive_hourly(center["lat"], center["lon"], TRACK_START, TRACK_END), GUN) if s is not None: print(f"wind: ERA5 archive {s['dir_deg']:.0f}° {s['speed_kts']:.1f} kn at gun") return s["dir_deg"], s["speed_kts"], s["gust_kts"] from_deg = WIND.get("fromDeg") if WIND.get("fromDeg") is not None else group_wind_deg if from_deg is not None: print(f"wind: offline fallback {from_deg:.0f}° {WIND.get('kts')} kn") return from_deg, WIND.get("kts"), None db = firestore.Client(project=PROJECT) device_first = {} # Boat photos/videos live in this GCS bucket as /.jpg|.mp4 — they # are NOT stored on the track points (the app uploads them on a separate endpoint), so we discover # them by listing the bucket per device, windowed to the race. Matches the V2 backend default # ({project}-viewregatta-photos); override with PHOTOS_BUCKET if your deployment uses another. PHOTOS_BUCKET = os.environ.get("PHOTOS_BUCKET") or f"{PROJECT}-viewregatta-photos" def latlon(g): return {"lat": g.get("lat"), "lon": g.get("lon")} if g else None def device_for_boat(boat_id): if not boat_id: return None for m in db.collection("deviceMounts").where(filter=FF("boatId", "==", boat_id)).stream(): md = m.to_dict() or {} if (md.get("fromUtcMs") or 0) <= GUN and (md.get("toUtcMs") is None or GUN <= md.get("toUtcMs")): return md.get("deviceId") return None def pull_frames(devs): frames = {} for dev in devs: if not dev: continue n = 0 for tr in db.collection("tracks").where(filter=FF("deviceId", "==", dev)).stream(): td = tr.to_dict() or {} if td.get("deletedAt") is not None: continue if (td.get("endUtcMs") or 0) < TRACK_START or (td.get("startUtcMs") or 0) > TRACK_END: continue for bdoc in (db.collection("tracks").document(tr.id).collection("batches") .order_by("fromUtcMs").stream()): bb = bdoc.to_dict() or {} if (bb.get("toUtcMs") or 0) < TRACK_START: continue if (bb.get("fromUtcMs") or 0) > TRACK_END: break for p in bb.get("points", []): t = p.get("t") or 0 if t < TRACK_START or t > TRACK_END or p.get("lat") is None: continue sec = (t - TRACK_START) // 1000 rec = {"lat": round(p["lat"], 7), "lon": round(p.get("lon"), 7), "speed": round(p.get("speed", 0) or 0, 2), "course": round(p.get("course", 0) or 0, 1)} frames.setdefault(str(sec), {})[dev] = rec device_first.setdefault(dev, rec) n += 1 print(f" device {dev[:12]}… {n} points") return frames def _fname_to_ms(name): """Parse the UTC timestamp embedded in a photo/video blob name: /YYYYMMDDTHHMMSSZ[_cam].ext.""" try: stem = name.rsplit("/", 1)[-1].split("_")[0].split(".")[0] return int(datetime.strptime(stem, "%Y%m%dT%H%M%SZ").replace(tzinfo=UTC).timestamp() * 1000) except Exception: return None def download_media(devs): """Build each boat's picture/video timeline for the tile slots by LISTING the GCS photos bucket (photos aren't on the track points) and DOWNLOADING the in-window blobs into demo/data/media// (gitignored, like the audio). Returns {deviceId: [{tSec, file, video}]} sorted by time, `file` relative to demo/data/media/. Resilient: missing bucket / failed fetch is logged and skipped.""" media_root = _HERE.parent / "data" / "media" try: from google.cloud import storage bucket = storage.Client(project=PROJECT).bucket(PHOTOS_BUCKET) except Exception as e: print(f" media: storage unavailable ({e}); skipping pictures") return {} # day prefixes spanning the window, so we list /YYYYMMDD… not the device's whole history days, d0 = [], datetime.fromtimestamp(TRACK_START / 1000, tz=UTC).date() d1 = datetime.fromtimestamp(TRACK_END / 1000, tz=UTC).date() while d0 <= d1: days.append(d0.strftime("%Y%m%d")) d0 += timedelta(days=1) out, total = {}, 0 for dev in devs: lst = [] try: blobs = [b for day in days for b in bucket.list_blobs(prefix=f"{dev}/{day}")] except Exception as e: print(f" media list failed for {dev[:12]}: {e}") continue for blob in blobs: ts = _fname_to_ms(blob.name) if ts is None or ts < TRACK_START or ts > TRACK_END: continue base = blob.name.rsplit("/", 1)[-1] rel = f"{dev}/{base}" local_abs = media_root.joinpath(*rel.split("/")) if not local_abs.exists(): local_abs.parent.mkdir(parents=True, exist_ok=True) try: blob.download_to_filename(str(local_abs)) except Exception as e: print(f" media fetch failed {blob.name}: {e}") continue lst.append({"tSec": round((ts - TRACK_START) / 1000, 3), "file": rel, "video": base.lower().endswith(".mp4")}) total += 1 if lst: lst.sort(key=lambda m: m["tSec"]) out[dev] = lst print(f" media {dev[:12]}… {len(lst)} in window") print(f" media: {total} pictures/videos across {len(out)} boats (bucket {PHOTOS_BUCKET})") return out def main(): inst = db.collection("raceInstances").document(INSTANCE).get().to_dict() or {} grp = list(db.collection("raceInstances").document(INSTANCE).collection("startGroups").stream())[0] gd = grp.to_dict() or {} teams, devs = [], [] for tm in TEAMS: boats = [] for b in tm["boats"]: devs.append(b["dev"]) boats.append({"deviceId": b["dev"], "sailId": b["sail"], "visualId": f'{tm["name"]} {b["sail"]}', "boatId": None, "color": tm["color"], "sternM": 0}) team = {"teamId": tm["teamId"], "name": tm["name"], "boats": boats} if tm.get("nameLang"): team["nameLang"] = tm["nameLang"] # BCP-47 pronunciation hint for the TTS (e.g. English name) teams.append(team) print("pulling tracks…") frames = pull_frames(devs) present = {d for fr in frames.values() for d in fr} for tm in teams: tm["boats"] = [b for b in tm["boats"] if b["deviceId"] in present] teams = [t for t in teams if t["boats"]] # course (digital 8) cdata = db.collection("courses").document(inst.get("courseId")).get().to_dict() or {} sl, fl = cdata.get("startLine") or {}, (cdata.get("finishLine") or cdata.get("startLine") or {}) wps = [] for w in cdata.get("waypoints") or []: ll = ({"lat": w["lat"], "lon": w["lon"]} if w.get("lat") is not None else None) if ll is None: dev = w.get("deviceId") or w.get("portDeviceId") or w.get("stbdDeviceId") if dev and dev in device_first: ll = {"lat": device_first[dev]["lat"], "lon": device_first[dev]["lon"]} if ll: wps.append({"name": w.get("name") or "", "type": w.get("type"), "rounding": w.get("rounding"), "lat": ll["lat"], "lon": ll["lon"]}) pts = [latlon(sl.get("port")), latlon(sl.get("starboard")), latlon(fl.get("port")), latlon(fl.get("starboard"))] pts += [{"lat": w["lat"], "lon": w["lon"]} for w in wps] pts = [p for p in pts if p and p.get("lat") is not None] center = ({"lat": sum(p["lat"] for p in pts) / len(pts), "lon": sum(p["lon"] for p in pts) / len(pts)} if pts else None) course = { "name": cdata.get("name") or "digital 8", "center": center, "startLine": {"port": latlon(sl.get("port")), "starboard": latlon(sl.get("starboard"))}, "finishLine": {"port": latlon(fl.get("port")), "starboard": latlon(fl.get("starboard"))}, "waypoints": wps, } wind_from, wind_kts, wind_gust = _resolve_wind(center, gd.get("windDeg")) feed = { "schema": "vr-broadcast-demo/1", "source": "real", "meta": {"title": TITLE, "user": USER, "instanceId": INSTANCE, "groupId": grp.id, "openEnded": False, "raceType": "team_race_2", "trackStartUtcMs": TRACK_START, "gunUtcMs": GUN, "trackEndUtcMs": TRACK_END, "frameStepMs": 1000, # Wind: archive-first (real ERA5 at the gun); else offline race_real.WIND fallback; # else the start-group's recorded windDeg; else None (wind-dependent commentary off). "windFromDeg": wind_from, "windKts": wind_kts, **({"windGustKts": wind_gust} if wind_gust is not None else {})}, "course": course, "teams": teams, "frames": frames, "video": [], # per-boat picture/video timeline for the tile slots (listed from GCS + downloaded locally) "media": download_media(sorted(present)), } # Opt-in AI bake via the event-agnostic `ai_bake` (the ONE author+editor path every event shares). # `--author` (L4) rewrites the colour TEXT, safety-validated; `--ai` (L2) frames it (keep/drop + # Part D shot type). With no ANTHROPIC_API_KEY/SDK both are deterministic no-ops, so either is always # safe to pass; a real bake is a SNAPSHOT (non-deterministic) written into real.json. from ai_bake import apply_ai_bake use_ai = "--ai" in sys.argv use_author = "--author" in sys.argv # Producer swap seam (Phase 0): `enrich` is the byte-identical default; `--rie` opts into the RIE # producer, which auto-falls-back to `enrich` when it declines a heat. `commentarySource` records # which producer actually scored the feed. rie_mode = "rie" if "--rie" in sys.argv else "enrich" feed_scored, provenance = produce_feed(feed, mode=rie_mode, scorer=enrich) print(f"producer: commentarySource={provenance['commentarySource']}") scored, audit = apply_ai_bake(feed_scored, feed, author=use_author, editor=use_ai) out_dir = _HERE.parent / "data" / "races" out_dir.mkdir(parents=True, exist_ok=True) if audit: with open(out_dir / "ai_choices_real.json", "w", encoding="utf-8") as f: # committed audit json.dump({k: audit[k] for k in ("beats", "choices", "authored") if k in audit}, f, ensure_ascii=False, indent=1) if "choices" in audit: n_drop = sum(1 for c in audit["choices"] if not c["keep"]) print(f"ai: {len(audit['beats'])} colour beats -> {n_drop} dropped, {len(audit['beats']) - n_drop} kept") if "authored" in audit: print(f"author: {sum(1 for d in audit['authored'].values() if d)}/{len(audit['beats'])} colour lines re-authored") with open(out_dir / "real.json", "w", encoding="utf-8") as f: json.dump(scored, f, ensure_ascii=False, separators=(",", ":")) m, res = scored["meta"], scored.get("result") # feed path is relative to THIS manifest file (resolved against the manifest URL in the page), # so it's independent of where core/web is served from. manifest = {"races": [{"id": "real", "title": m["title"], "feed": "real.json", "trackStartUtcMs": m["trackStartUtcMs"], "gunUtcMs": m["gunUtcMs"], "trackEndUtcMs": m["trackEndUtcMs"], "winnerSec": res["winnerSec"] if res else None, "winnerName": res["winnerName"] if res else None}], "params": PARAMS} with open(out_dir / "manifest.json", "w", encoding="utf-8") as f: json.dump(manifest, f, ensure_ascii=False, indent=1) print("teams:", [(t["name"], [b["sailId"] for b in t["boats"]]) for t in teams]) print(f"frames={len(frames)} events={len(scored['events'])} " f"result={res['winnerName'] if res else None} center={center} waypoints={len(wps)}") if __name__ == "__main__": main()