r""" Generate a synthetic boat-cam MP4 per boat per race for the regenerated GSYS 2026-05-23/24 series (`data/races/manifest.json`). The flights are synthetic (no real footage), so this paints a per-second **telemetry HUD** frame from each boat's track and encodes the sequence at **1 fps** (the manifest's frame cadence, `frameStepMs=1000`): * header — team name + sail number, gun-relative clock (gun+M:SS) * readout — boat speed (knots) and heading (deg), big * minimap — the race course (marks + start/finish lines) with every boat's dot each second, this boat highlighted white * footer — absolute wall time (venue-local CEST = UTC+2) + race title 12 races x 4 boats = 48 videos. Frames are streamed straight to ffmpeg as rawvideo over stdin (no intermediate PNGs), so the whole run needs only ffmpeg + Pillow. Output: data/boatcam//_.mp4 (local files; not wired into V3). ..\..\..\.venv-demo\Scripts\python.exe export\gen_boatcam_videos.py ..\..\..\.venv-demo\Scripts\python.exe export\gen_boatcam_videos.py --only gsys01 --boats 1 # quick test """ from __future__ import annotations import argparse import json import math import subprocess import sys from datetime import UTC, datetime, timedelta from pathlib import Path from PIL import Image, ImageDraw, ImageFont HERE = Path(__file__).resolve().parent DATA = HERE.parent / "data" DEFAULT_MANIFEST = DATA / "races" / "manifest.json" DEFAULT_OUT = DATA / "boatcam" W, H = 1280, 720 FPS = 1 VENUE_TZ_OFFSET_H = 2 # CEST (Europe/Stockholm, summer) — matches the discovery tooling MS_PER_S = 1000 MS2KN = 1.943844 # speed is metres/sec in the feed; show knots # --- fonts --------------------------------------------------------------------------------------- def _font(size: int, *, bold: bool = False) -> ImageFont.FreeTypeFont: """A scalable TrueType font — Arial on Windows, else Pillow's bundled default scaled up.""" for name in (("arialbd.ttf", "arial.ttf") if bold else ("arial.ttf",)): for base in (Path("C:/Windows/Fonts"), Path("/usr/share/fonts")): p = base / name if p.exists(): return ImageFont.truetype(str(p), size) return ImageFont.load_default(size=size) # --- colour helpers ------------------------------------------------------------------------------ def _hex(rgb: str) -> tuple[int, int, int]: s = rgb.lstrip("#") return (int(s[0:2], 16), int(s[2:4], 16), int(s[4:6], 16)) def _mix(c: tuple[int, int, int], f: float) -> tuple[int, int, int]: """Scale a colour toward black (`f<1`) or white (`f>1`, clamped).""" return tuple(max(0, min(255, round(v * f))) for v in c) # type: ignore[return-value] # --- minimap projection -------------------------------------------------------------------------- class Projector: """Equirectangular lat/lon -> pixel mapping for one race's minimap, fit to the bounding box of the course marks + every boat position over the whole race (so nothing ever leaves the frame), north up, equal metric aspect (longitude compressed by cos(lat)).""" def __init__(self, latlons: list[tuple[float, float]], box: tuple[int, int, int, int]): lats = [p[0] for p in latlons] lons = [p[1] for p in latlons] self.mlon = 111_320.0 * math.cos(math.radians((min(lats) + max(lats)) / 2.0)) self.mlat = 111_320.0 self._lon_min = min(lons) self._lat_min = min(lats) xm = [(lo - self._lon_min) * self.mlon for lo in lons] ym = [(la - self._lat_min) * self.mlat for la in lats] self._xspan = max(xm) or 1.0 self._yspan = max(ym) or 1.0 x0, y0, x1, y1 = box pad = 14 bw, bh = (x1 - x0 - 2 * pad), (y1 - y0 - 2 * pad) self._scale = min(bw / self._xspan, bh / self._yspan) # centre the drawing inside the box self._ox = x0 + pad + (bw - self._xspan * self._scale) / 2.0 self._oy = y0 + pad + (bh - self._yspan * self._scale) / 2.0 def xy(self, lat: float, lon: float) -> tuple[float, float]: xm = (lon - self._lon_min) * self.mlon ym = (lat - self._lat_min) * self.mlat px = self._ox + xm * self._scale py = self._oy + (self._yspan - ym) * self._scale # invert: north up return (px, py) # --- per-race rendering -------------------------------------------------------------------------- def _course_geometry(course: dict) -> tuple[list[tuple[float, float]], list[tuple]]: """`(mark_points, gate_lines)` from the feed course block, for minimap drawing.""" marks = [(float(w["lat"]), float(w["lon"])) for w in course.get("waypoints") or []] lines = [] for line in (course.get("startLine"), course.get("finishLine")): if isinstance(line, dict) and line.get("port") and line.get("starboard"): p, s = line["port"], line["starboard"] lines.append(((float(p["lat"]), float(p["lon"])), (float(s["lat"]), float(s["lon"])))) return marks, lines def _frames_sorted(feed: dict) -> list[tuple[int, dict]]: """`[(sec, {dev: pt})]` ordered by second.""" return sorted(((int(s), fr) for s, fr in feed["frames"].items()), key=lambda x: x[0]) def _fmt_gun_rel(sec_abs_ms: int, gun_ms: int) -> str: d = round((sec_abs_ms - gun_ms) / 1000) sign = "+" if d >= 0 else "-" d = abs(d) return f"gun {sign}{d // 60}:{d % 60:02d}" def render_frame(draw_img: Image.Image, *, feed: dict, sec: int, fr: dict, dev: str, team_name: str, sail: str, team_rgb: tuple[int, int, int], proj: Projector, marks, lines, title: str, fonts: dict, t0_ms: int, step_ms: int, gun_ms: int) -> None: d = ImageDraw.Draw(draw_img) bg = _mix(team_rgb, 0.22) d.rectangle((0, 0, W, H), fill=bg) accent = _mix(team_rgb, 1.25) white = (240, 240, 245) dim = (170, 175, 185) # header bar d.rectangle((0, 0, W, 96), fill=_mix(team_rgb, 0.4)) d.text((40, 48), f"{team_name.upper()} #{sail}", font=fonts["h1"], fill=white, anchor="lm") pt = fr.get(dev) or {} abs_ms = t0_ms + sec * step_ms d.text((W - 40, 48), _fmt_gun_rel(abs_ms, gun_ms), font=fonts["h2"], fill=accent, anchor="rm") # big readouts (left column) spd_kn = (pt.get("speed") or 0.0) * MS2KN hdg = pt.get("course") d.text((40, 230), "SPEED", font=fonts["lbl"], fill=dim, anchor="lm") spd_txt = f"{spd_kn:0.1f}" d.text((40, 300), spd_txt, font=fonts["big"], fill=white, anchor="lm") d.text((40 + d.textlength(spd_txt, font=fonts["big"]) + 16, 330), "kn", font=fonts["unit"], fill=dim, anchor="lm") d.text((40, 430), "HEADING", font=fonts["lbl"], fill=dim, anchor="lm") hdg_txt = "--" if hdg is None else f"{hdg:0.0f}" d.text((40, 500), hdg_txt, font=fonts["big"], fill=white, anchor="lm") d.text((40 + d.textlength(hdg_txt, font=fonts["big"]) + 16, 530), "deg", font=fonts["unit"], fill=dim, anchor="lm") # minimap (right side box) box = (W - 470, 150, W - 40, 580) d.rectangle(box, outline=_mix(team_rgb, 0.7), width=2, fill=_mix(team_rgb, 0.30)) for a, b in lines: d.line((*proj.xy(*a), *proj.xy(*b)), fill=(220, 210, 120), width=3) for m in marks: mx, my = proj.xy(*m) d.ellipse((mx - 5, my - 5, mx + 5, my + 5), fill=(235, 170, 60)) # every boat's current dot; this boat last + highlighted others = [dd for dd in fr if dd != dev] for dd in others: p = fr.get(dd) or {} if p.get("lat") is None: continue bx, by = proj.xy(p["lat"], p["lon"]) d.ellipse((bx - 6, by - 6, bx + 6, by + 6), fill=dim) if pt.get("lat") is not None: bx, by = proj.xy(pt["lat"], pt["lon"]) d.ellipse((bx - 11, by - 11, bx + 11, by + 11), fill=accent, outline=white, width=3) # footer local = datetime.fromtimestamp(abs_ms / 1000, tz=UTC) + timedelta(hours=VENUE_TZ_OFFSET_H) d.text((40, H - 42), local.strftime("%Y-%m-%d %H:%M:%S") + " CEST", font=fonts["foot"], fill=dim, anchor="lm") d.text((W - 40, H - 42), title, font=fonts["foot"], fill=dim, anchor="rm") def encode_boat(out_path: Path, *, feed: dict, dev: str, team_name: str, sail: str, team_rgb: tuple[int, int, int], proj: Projector, marks, lines, title: str, fonts: dict, gun_ms: int) -> int: """Stream one boat's per-second HUD frames straight into ffmpeg (rawvideo stdin) -> H.264 MP4 at 1 fps. Returns the number of frames written.""" meta = feed["meta"] t0_ms = int(meta["trackStartUtcMs"]) step_ms = int(meta.get("frameStepMs") or 1000) frames = _frames_sorted(feed) out_path.parent.mkdir(parents=True, exist_ok=True) cmd = [ "ffmpeg", "-y", "-loglevel", "error", "-f", "rawvideo", "-pixel_format", "rgb24", "-video_size", f"{W}x{H}", "-framerate", str(FPS), "-i", "-", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "20", "-preset", "veryfast", "-movflags", "+faststart", str(out_path), ] proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE) img = Image.new("RGB", (W, H)) n = 0 try: for sec, fr in frames: render_frame(img, feed=feed, sec=sec, fr=fr, dev=dev, team_name=team_name, sail=sail, team_rgb=team_rgb, proj=proj, marks=marks, lines=lines, title=title, fonts=fonts, t0_ms=t0_ms, step_ms=step_ms, gun_ms=gun_ms) proc.stdin.write(img.tobytes()) n += 1 proc.stdin.close() except BrokenPipeError: pass err = proc.stderr.read().decode("utf-8", "replace") rc = proc.wait() if rc != 0: raise RuntimeError(f"ffmpeg failed ({rc}) for {out_path.name}:\n{err}") return n def build_projector(feed: dict, marks, lines) -> Projector: pts = list(marks) + [e for ln in lines for e in ln] for fr in feed["frames"].values(): for p in fr.values(): if p.get("lat") is not None: pts.append((p["lat"], p["lon"])) return Projector(pts, (W - 470, 150, W - 40, 580)) def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--manifest", default=str(DEFAULT_MANIFEST)) ap.add_argument("--out", default=str(DEFAULT_OUT)) ap.add_argument("--only", default=None, help="render just this race id (e.g. gsys01)") ap.add_argument("--boats", type=int, default=None, help="cap boats per race (quick test)") args = ap.parse_args() manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8")) out_root = Path(args.out) fonts = { "h1": _font(46, bold=True), "h2": _font(40, bold=True), "big": _font(150, bold=True), "lbl": _font(34, bold=True), "unit": _font(34), "foot": _font(26), } races = [r for r in manifest["races"] if args.only is None or r["id"] == args.only] if not races: print(f"no races matched --only {args.only!r}", file=sys.stderr) return 1 total = 0 for race in races: rid, feed = race["id"], race["data"] title = race["title"] gun_ms = int(feed["meta"]["gunUtcMs"]) marks, lines = _course_geometry(feed["course"]) proj = build_projector(feed, marks, lines) boats = [(b, t) for t in feed["teams"] for b in t["boats"]] if args.boats is not None: boats = boats[: args.boats] for b, team in boats: sail = str(b["sailId"]) team_rgb = _hex(b.get("color") or team.get("color") or "#888888") out_path = out_root / rid / f"{team['teamId']}_{sail}.mp4" n = encode_boat(out_path, feed=feed, dev=b["deviceId"], team_name=team["name"], sail=sail, team_rgb=team_rgb, proj=proj, marks=marks, lines=lines, title=title, fonts=fonts, gun_ms=gun_ms) total += 1 print(f" {rid}/{team['teamId']}_{sail}.mp4 ({n} frames, {n}s)") print(f"wrote {total} boat-cam videos under {out_root}") return 0 if __name__ == "__main__": sys.exit(main())