r"""One boat-cam MP4 per boat for a race the race office has CLOSED, cut from the clips the phones actually uploaded. This is the **onboard footage on its own** — not the 3D replay with camera insets, which is `export_v3_instance.py` + `record_demo.py`. **Nothing is drawn on top.** The device burns the venue-local wall clock into the picture per frame (`MediaTimestampOverlay`), and the style guide is explicit that on a boat-cam that burned-in clock **is** the clock — the producer adds no second one. Which makes the absence load-bearing too: a phone running a build from before that overlay filmed a picture with no time in it at all, and a stamp added here would then be *this tool's* opinion of when the frame was taken rather than the device's. It is not one to have. The record's own times — gun, window, first and last clip — go in the sidecar beside the film, where they are attributed to what actually knows them. (Measured: the TRWC 2026 fleet's clips carry no bar. The overlay landed on 2026-08-01 and the app is installed by hand, so a championship phone can be older than a feature that shipped before it.) python v3/broadcast/demo/export/gen_boatcam_instance.py --instance manual-219 --dry-run python v3/broadcast/demo/export/gen_boatcam_instance.py --instance manual-219 The race is the only argument that matters: the window, the roster, the teams and the venue's zone all come off the record, so the same command works on a championship final and on a simulated heat. Needs `$VR_DATABASE_URL`, `ffmpeg` on PATH, and a way to reach the clips — either `$VR_LOCAL_MEDIA_DIR`/`--media-root` (the office deployment, where the footage is on a disk) or ADC that can READ the media bucket (the cloud one). See `resolve_local`. **No signed URL is involved, and that is why this runs where a replay export does not.** The exporter has to hand the renderer a fetchable URL, so it needs a credential that can *sign* — which org policy makes awkward on a workstation. Here the bytes are read, from a local tree or with plain object-read access, so on the office box no credential is involved at all. **Chronological concatenation, gaps closed.** A phone films a duty cycle, not continuously (`frontVideoDurationMs` every `frontPhotoIntervalMs`), so the source is a string of short clips with real time between them. They are joined in order and the film runs shorter than the race; the burned-in clock is what tells a viewer where the jumps are, and the sidecar records the coverage and the largest gap so nobody has to infer it from the picture. `--real-time` instead pads each gap with black so the film's playhead matches race time — right for lining two boats up side by side, and mostly black on a sparse duty cycle. """ from __future__ import annotations import argparse import json import os import subprocess import sys import tempfile from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE.parents[2] / "ingest")) # store types, media filename parsing sys.path.insert(0, str(_HERE.parent.parent / "live")) # media_map (the clip window) sys.path.insert(0, str(_HERE.parent / "record")) # replay_library (venue-local naming) from export_v3_instance import DEFAULT_PRE_START_S, DEFAULT_TAIL_S, roster_at_gun # noqa: E402 #: Where the films land when `--out` is not given. Beside the replay library's own default rather #: than in the repo: these are hundreds of megabytes of footage, not source. DEFAULT_OUT = _HERE.parents[1] / "_cap" / "boatcams" #: Clip downloads are pure network latency — a few hundred small objects, nothing to compute — so the #: pool is sized for concurrency, the same reasoning as the exporter's signing pool. FETCH_WORKERS = 24 #: A gap longer than this is reported in the sidecar as a break in coverage rather than as the #: ordinary rhythm of a duty cycle. Two seconds is longer than any inter-clip spacing a duty cycle #: produces at the cadences the fleet runs. GAP_REPORT_S = 2.0 @dataclass(frozen=True) class Clip: """One uploaded segment, resolved to what this tool needs: when it starts, where the bytes are.""" start_ms: int end_ms: int | None camera: str gcs_path: str source: str # 'live' | 'clip' — see `--source` slot: str = "" # the mount that filmed it — see `resolve_local` @property def duration_s(self) -> float | None: if not self.end_ms or self.end_ms <= self.start_ms: return None return (self.end_ms - self.start_ms) / 1000.0 def clips_for_boat(store, boat: str, *, t0_ms: int, t1_ms: int, # noqa: ANN001 — Store camera: str = "front", source: str = "auto") -> list[Clip]: """Every clip of `boat` that OVERLAPS `[t0_ms, t1_ms]`, oldest first, one row per file. The store is asked for a deliberately WIDER window — `PRE_RACE_CLIP_MARGIN_MS` back — because a device buffers while it has no signal and uploads on reconnect, so the clips covering the approach and the gun are routinely stamped before the race's own window. That widening is a fetch bound, not the answer, so the overlap is decided here: a clip must actually touch the race to be in the film of it. A segment whose end was never recorded is bounded by its start alone rather than dropped — the store keeps the same rule, in both implementations. A boat is read through ALL its mount-slots (`slots_of_boat`), so a phone re-mounted mid-regatta does not split its own footage in half. """ from media import MediaError, parse_media_filename from media_map import PRE_RACE_CLIP_MARGIN_MS seen: set[str] = set() out: list[Clip] = [] for slot in store.slots_of_boat(boat): for seg in store.video_segments_for(slot, since_ms=t0_ms - PRE_RACE_CLIP_MARGIN_MS, until_ms=t1_ms): if not seg.gcs_path or seg.gcs_path in seen: continue name = str(seg.gcs_path).replace("\\", "/").rsplit("/", 1)[-1] try: _t, cam = parse_media_filename(name) except MediaError: continue # a name this product did not write — never guessed at start = seg.start_ms end = seg.end_ms if seg.end_ms and seg.end_ms > seg.start_ms else None if start > t1_ms or (end or start) < t0_ms: continue seen.add(seg.gcs_path) out.append(Clip(int(start), int(end) if end else None, cam, str(seg.gcs_path), str(seg.source or "clip"), str(slot))) if camera != "both": out = [c for c in out if c.camera == camera] # PER CAMERA, because `auto`'s "prefer the stored file, else take what there is" is a question # about one camera's own footage. Asked of both at once, a front camera with stored clips would # decide that a back camera indexed only as live segments has nothing — dropping a whole camera # on the strength of the other one's bookkeeping. picked: list[Clip] = [] for cam_name in {c.camera for c in out}: picked += _pick_source([c for c in out if c.camera == cam_name], source) picked.sort(key=lambda c: (c.camera, c.start_ms)) return picked def _pick_source(clips: list[Clip], source: str) -> list[Clip]: """`--source`: which of the two recordings of the same seconds to film from. A boat's seconds can be indexed twice — as the low-latency `live` segments the broadcast played and as the store-and-forward `clip` the phone uploaded whole (§19.10). Concatenating both would play the race twice, interleaved. `auto` prefers the store-and-forward file, which is the complete-quality one and the reason it is kept.""" if source in ("live", "clip"): return [c for c in clips if c.source == source] if source == "both": return clips stored = [c for c in clips if c.source == "clip"] return stored or clips def resolve_local(stored_path: str, slot: str, root: Path) -> Path | None: """The clip's bytes inside a LOCAL media tree, or None if this tree does not hold them. **Where the footage lives is a deployment fact.** In the cloud the index points at a bucket; on the office box the same 500 GB sits on a disk under `$VR_LOCAL_MEDIA_DIR` and there is no credential in the picture at all. The rows do not change when the data plane moves — a segment stored in 2026 still names `gs://…` — so the tree is searched by the KEY the row carries, and a hit is proven by the file existing rather than by a rule believed to be right. Three layouts are real in this product and each is tried: * the bucket key kept whole (a faithful mirror); * the key with leading segments folded into one directory name, which is what a mirror made per prefix looks like (`v3/media/trwc2026/…` under `v3_media_trwc2026/…`); * `LocalUploader`'s own scheme, `/`, for clips this deployment wrote. …and each of those with leading segments dropped, for a root pointed inside the tree. The filename carries the device's millisecond, so a tail match cannot quietly find another boat's clip; it can only fail to find one, which is reported. """ if "/" not in stored_path: return None blob = stored_path[len("gs://"):] if stored_path.startswith("gs://") else stored_path if stored_path.startswith("gs://"): blob = blob.partition("/")[2] parts = [p for p in blob.replace("\\", "/").split("/") if p] if not parts: return None candidates: list[Path] = [root.joinpath(*parts)] for k in range(2, min(5, len(parts))): candidates.append(root / "_".join(parts[:k]) / Path(*parts[k:])) for k in range(1, min(5, len(parts))): candidates.append(root.joinpath(*parts[k:])) candidates.append(root / slot.replace(":", "_") / parts[-1]) for path in candidates: if path.is_file(): return path return None def fetch_clips(clips: list[Clip], dest: Path, *, workers: int = FETCH_WORKERS, media_root: Path | None = None) -> dict[str, Path]: """Where each clip's bytes are, as a readable file: `{stored path: local file}`. A local media tree is used IN PLACE — nothing is copied, because the film is assembled by reading these files once and a 500 GB tree is not worth duplicating to do it. Anything the tree does not hold is downloaded from the bucket into `dest`. A clip neither tree nor bucket has is OMITTED rather than substituted: a missing file is a hole in the record and the sidecar counts it, where a placeholder frame would put footage in the film that no camera took.""" got: dict[str, Path] = {} remaining = clips if media_root is not None: remaining = [] for clip in clips: local = resolve_local(clip.gcs_path, clip.slot, media_root) if local is not None: got[clip.gcs_path] = local else: remaining.append(clip) if not remaining: return got from google.cloud import storage dest.mkdir(parents=True, exist_ok=True) client = storage.Client() def one(clip: Clip) -> tuple[str, Path | None]: bucket_name, _, blob_name = clip.gcs_path[len("gs://"):].partition("/") local = dest / f"{clip.start_ms}_{clip.camera}_{blob_name.rsplit('/', 1)[-1]}" try: client.bucket(bucket_name).blob(blob_name).download_to_filename(str(local)) except Exception as exc: # noqa: BLE001 — one absent object must not end the run print(f"[boatcam] missing {clip.gcs_path}: {type(exc).__name__}", flush=True) return clip.gcs_path, None return clip.gcs_path, local with ThreadPoolExecutor(max_workers=workers) as pool: for path, local in pool.map(one, remaining): if local is not None: got[path] = local return got def coverage(clips: list[Clip], *, t0_ms: int, t1_ms: int) -> dict: """What the footage actually covers — the numbers a viewer would otherwise have to guess at. Reported per boat in the sidecar because "the film is 4 minutes and the race was 14" is the first question anyone asks of a duty-cycled camera, and the answer is a property of the RECORD, not of the encode.""" filmed_s = sum(c.duration_s or 0.0 for c in clips) gaps: list[float] = [] prev_end: int | None = None for c in sorted(clips, key=lambda c: c.start_ms): if prev_end is not None and c.start_ms - prev_end > GAP_REPORT_S * 1000: gaps.append(round((c.start_ms - prev_end) / 1000.0, 1)) prev_end = max(prev_end or 0, c.end_ms or c.start_ms) window_s = max(0.0, (t1_ms - t0_ms) / 1000.0) return { "clips": len(clips), "raceWindowS": round(window_s, 1), "filmedS": round(filmed_s, 1), "coverage": round(filmed_s / window_s, 3) if window_s else None, "gapsOver2s": len(gaps), "largestGapS": max(gaps) if gaps else 0.0, "firstClipUtcMs": min((c.start_ms for c in clips), default=None), "lastClipUtcMs": max(((c.end_ms or c.start_ms) for c in clips), default=None), } def _probe(path: Path) -> dict: """The first clip's video stream, as ffprobe reads it — the encode is matched to the source rather than to a constant, so a fleet filming 5 fps at 1280x720 is not silently resampled.""" out = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name,width,height,r_frame_rate", "-of", "json", str(path)], capture_output=True, text=True, check=False) try: s = (json.loads(out.stdout).get("streams") or [{}])[0] except (json.JSONDecodeError, IndexError): return {} num, _, den = str(s.get("r_frame_rate", "")).partition("/") fps = None try: fps = float(num) / float(den or 1) except (TypeError, ValueError, ZeroDivisionError): pass return {"codec": s.get("codec_name"), "width": s.get("width"), "height": s.get("height"), "fps": round(fps, 3) if fps else None} def _concat_file(paths: list[Path], listing: Path) -> None: """The concat demuxer's input. Forward slashes because this file is parsed by ffmpeg, not by the shell, and a Windows path written with backslashes is read as escape sequences.""" lines = ["file '" + str(p).replace("\\", "/") + "'\n" for p in paths] listing.write_text("".join(lines), encoding="utf-8") def encode(paths: list[Path], out: Path, *, work: Path, encode_mode: str = "auto", audio: bool = False, metadata: dict | None = None) -> bool: """Join `paths` into `out`. Returns False if ffmpeg refused. `auto` copies the video stream, which a phone's clips within one race always allow and which is both faster and lossless; `reencode` is for the cases that do not (a mid-race geometry change, or the black filler `--real-time` interleaves). Audio is DROPPED by default: an onboard phone records wind, and clips disagreeing about whether they have a track at all is the ordinary way a concatenation of hundreds of files fails.""" listing = work / (out.stem + ".txt") _concat_file(paths, listing) first = _probe(paths[0]) args = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-f", "concat", "-safe", "0", "-i", str(listing)] if encode_mode == "reencode" or (encode_mode == "auto" and not first.get("codec")): args += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-pix_fmt", "yuv420p"] if first.get("fps"): args += ["-r", str(first["fps"])] else: args += ["-c:v", "copy"] args += ["-c:a", "aac"] if audio else ["-an"] for key, value in (metadata or {}).items(): args += ["-metadata", str(key) + "=" + str(value)] args += ["-movflags", "+faststart", str(out)] proc = subprocess.run(args, capture_output=True, text=True, check=False) if proc.returncode != 0: print("[boatcam] ffmpeg failed: " + proc.stderr.strip()[:400], flush=True) return proc.returncode == 0 def _black_filler(spec: dict, seconds: float, out: Path) -> bool: """`--real-time`: `seconds` of black at the source's own geometry, so the joined film keeps one shape even though the filler itself has to be encoded.""" size = str(spec.get("width") or 1280) + "x" + str(spec.get("height") or 720) args = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "color=c=black:s=" + size + ":r=" + str(spec.get("fps") or 30) + ":d=" + format(max(0.04, seconds), ".3f"), "-c:v", "libx264", "-preset", "veryfast", "-crf", "28", "-pix_fmt", "yuv420p", "-an", str(out)] return subprocess.run(args, capture_output=True, text=True, check=False).returncode == 0 def _teams_for(store, instance_id: str, boats: list[str], at_ms: int) -> dict[str, dict]: # noqa: ANN001 """`{boatRef: {team, sail}}` — the identity the film is named for, resolved exactly the way the feed resolves it (`broadcast_feed.resolve_teams` over the same entries and the same stored team config), never re-derived here.""" from broadcast_feed import resolve_teams cfg = (store.get_broadcast_config(instance_id) or {}).get("teamConfig") out: dict[str, dict] = {} for team in resolve_teams(store, instance_id, boats, at_ms, team_config=cfg): for boat in team.get("boats") or []: ref = boat.get("deviceId") or boat.get("boat") if ref: out[str(ref)] = {"team": team.get("name"), "fullName": team.get("fullName"), "sail": boat.get("sailId")} return out def _sail_of(boat_ref: str) -> str: """The hull number carried in the boat ref (`trwc2026:boat:7` -> `7`) — the fallback when no team config declares a sail id, and never a guess: the ref IS the hull.""" return str(boat_ref).rsplit(":", 1)[-1] def gen_boatcams(store, instance_id: str, *, # noqa: ANN001 — Store out_dir: Path = DEFAULT_OUT, camera: str = "front", source: str = "auto", pre_start_s: int = DEFAULT_PRE_START_S, tail_s: int = DEFAULT_TAIL_S, encode_mode: str = "auto", audio: bool = False, real_time: bool = False, dry_run: bool = False, only_boats: list[str] | None = None, media_root: Path | None = None) -> list[Path]: """Film every boat in `instance_id`. Returns the MP4s written (empty on `--dry-run`).""" from replay_library import _slug, _venue_local from serve_v3_live import apply_broadcast_config, apply_event_config, event_id_for_instance inst = store.get_race_instance(instance_id) if inst is None: raise SystemExit("no such race: " + instance_id) gun = getattr(inst, "official_start_utc_ms", None) stop = getattr(inst, "official_stop_utc_ms", None) if not gun: raise SystemExit(instance_id + " was never armed — there is no race to film") if not stop: raise SystemExit(instance_id + " has no official stop — the race office has not closed it") t0, t1 = int(gun) - pre_start_s * 1000, int(stop) + tail_s * 1000 boats, entries_at = roster_at_gun(store, instance_id, gun_ms=int(gun), stop_ms=int(stop)) if not boats: raise SystemExit(instance_id + " had no boats entered at its gun") if only_boats: wanted = {str(b) for b in only_boats} boats = [b for b in boats if b in wanted or _sail_of(b) in wanted] ids = _teams_for(store, instance_id, boats, entries_at or int(gun)) # The venue's zone and the race's own name, through the SAME overlays the replay applies — a film # of this race must not be filed under a different day or a different name than the replay of it. meta = {"trackStartUtcMs": t0, "gunUtcMs": int(gun), "trackEndUtcMs": t1, "title": getattr(inst, "name", None) or instance_id} meta = apply_event_config(store, event_id_for_instance(store, instance_id), meta=meta, boats=boats) meta, _ = apply_broadcast_config(store, instance_id, meta=meta) when = _venue_local(int(gun), (meta.get("venue") or {}).get("timeZone")) day = when.strftime("%Y-%m-%d") if when else "undated" race_slug = _slug(meta.get("raceName") or meta.get("title") or instance_id, limit=32) out_dir.mkdir(parents=True, exist_ok=True) cameras = ["front", "back"] if camera == "both" else [camera] written: list[Path] = [] for boat in boats: ident = ids.get(boat, {}) team = _slug(ident.get("team") or "", limit=12) sail = ident.get("sail") or _sail_of(boat) all_clips = clips_for_boat(store, boat, t0_ms=t0, t1_ms=t1, camera=camera, source=source) for cam in cameras: clips = [c for c in all_clips if c.camera == cam] stem = "_".join(x for x in (day, race_slug, team, "sail" + str(sail), cam) if x) cov = coverage(clips, t0_ms=t0, t1_ms=t1) print("[boatcam] {} {}: {} clips, {:.0f}s filmed of {:.0f}s ({:.0f}%), " "largest gap {:.0f}s".format(boat, cam, cov["clips"], cov["filmedS"], cov["raceWindowS"], (cov["coverage"] or 0) * 100, cov["largestGapS"]), flush=True) if dry_run or not clips: continue film = _film_one(clips, out_dir / (stem + ".mp4"), coverage_row=cov, boat=boat, ident=ident, cam=cam, meta=meta, instance_id=instance_id, gun_ms=int(gun), encode_mode=encode_mode, audio=audio, real_time=real_time, media_root=media_root) if film: written.append(film) return written def _film_one(clips: list[Clip], out: Path, *, coverage_row: dict, boat: str, ident: dict, cam: str, meta: dict, instance_id: str, gun_ms: int, encode_mode: str, audio: bool, real_time: bool, media_root: Path | None = None) -> Path | None: """Resolve the bytes, join them, and write the film plus its sidecar. One boat, one camera.""" with tempfile.TemporaryDirectory(prefix="boatcam_") as tmp: work = Path(tmp) got = fetch_clips(clips, work / "src", media_root=media_root) have = [c for c in clips if c.gcs_path in got] if not have: print("[boatcam] " + boat + " " + cam + ": no clip bytes could be fetched", flush=True) return None spec = _probe(got[have[0].gcs_path]) paths: list[Path] = [] prev_end: int | None = None for i, clip in enumerate(have): if real_time and prev_end is not None and clip.start_ms - prev_end > 40: filler = work / ("gap" + format(i, "05d") + ".mp4") if _black_filler(spec, (clip.start_ms - prev_end) / 1000.0, filler): paths.append(filler) paths.append(got[clip.gcs_path]) prev_end = clip.end_ms or clip.start_ms # Real-time filler is encoded and the clips are not, so the two cannot be stream-copied # together; say so rather than letting the concat produce a file that stalls at the first gap. mode = "reencode" if real_time else encode_mode venue = meta.get("venue") or {} side = { "instanceId": instance_id, "raceName": meta.get("raceName") or meta.get("title"), "event": meta.get("eventName"), "venue": venue.get("name"), "venueTimeZone": venue.get("timeZone"), "gunUtcMs": gun_ms, "boat": boat, "team": ident.get("team"), "teamFullName": ident.get("fullName"), "sail": ident.get("sail") or _sail_of(boat), "camera": cam, "source": {"kind": "onboard-clips", "fetched": len(have), "indexed": len(clips)}, "sourceVideo": spec, "realTime": real_time, "audio": audio, **coverage_row, } title = " ".join(str(x) for x in (side["raceName"] or instance_id, "—", side["team"] or "", "sail", side["sail"], "(" + cam + ")") if str(x)) ok = encode(paths, out, work=work, encode_mode=mode, audio=audio, metadata={"title": title, "album": meta.get("eventName") or "", "comment": json.dumps(side, ensure_ascii=False)}) if not ok: return None out.with_suffix(".json").write_text(json.dumps(side, ensure_ascii=False, indent=2), encoding="utf-8") print(f"[boatcam] wrote {out.name} ({out.stat().st_size / 1e6:.0f} MB)", flush=True) return out def _main() -> int: ap = argparse.ArgumentParser( description="Build a boat-cam MP4 per boat for one finished race, from the onboard clips " "the phones uploaded. See the module docstring.") ap.add_argument("--instance", required=True, help="race instance id (e.g. manual-219)") ap.add_argument("--out", type=Path, default=DEFAULT_OUT, help="output dir (default " + str(DEFAULT_OUT) + ")") ap.add_argument("--camera", choices=("front", "back", "both"), default="front") ap.add_argument("--media-root", type=Path, default=os.environ.get("VR_LOCAL_MEDIA_DIR"), help="a local media tree to read the clips from instead of the bucket " "(default: $VR_LOCAL_MEDIA_DIR). Anything it does not hold still comes " "from GCS, so a partial mirror is fine") ap.add_argument("--source", choices=("auto", "clip", "live", "both"), default="auto", help="which indexing of the same seconds to film from (default: the " "store-and-forward clip when the boat has any)") ap.add_argument("--boat", action="append", dest="boats", help="film only this boat (a boat ref or a hull number); repeatable") ap.add_argument("--pre-start-s", type=int, default=DEFAULT_PRE_START_S, help="seconds of start sequence to open with " "(default " + str(DEFAULT_PRE_START_S) + ")") ap.add_argument("--tail-s", type=int, default=DEFAULT_TAIL_S, help="seconds past the official stop (default " + str(DEFAULT_TAIL_S) + ")") ap.add_argument("--encode", choices=("auto", "copy", "reencode"), default="auto", dest="encode_mode") ap.add_argument("--audio", action="store_true", help="keep the onboard audio track") ap.add_argument("--real-time", action="store_true", help="pad every gap with black so the playhead matches race time") ap.add_argument("--dry-run", action="store_true", help="report each boat's clip inventory and coverage; write nothing") args = ap.parse_args() dsn = os.environ.get("VR_DATABASE_URL") if not dsn: raise SystemExit("set $VR_DATABASE_URL (e.g. prod via the Cloud SQL proxy)") from postgres_store import PostgresStore, connect store = PostgresStore(connect(dsn)) films = gen_boatcams(store, args.instance, out_dir=args.out, camera=args.camera, source=args.source, pre_start_s=args.pre_start_s, tail_s=args.tail_s, encode_mode=args.encode_mode, audio=args.audio, real_time=args.real_time, dry_run=args.dry_run, only_boats=args.boats, media_root=Path(args.media_root) if args.media_root else None) if not args.dry_run: print("[boatcam] " + str(len(films)) + " film(s) in " + str(args.out), flush=True) return 0 if __name__ == "__main__": raise SystemExit(_main())