"""The per-boat boat-cam film (`gen_boatcam_instance`) — clip selection and where the bytes are. Nothing here encodes anything: ffmpeg is the easy half. What is worth pinning is the two places this tool can be silently wrong — pick the wrong clips, or fail to find footage that is sitting on the disk — because both fail the same way, as a film that is simply shorter than the race with no error anywhere. """ import sys from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE.parents[1] / "export")) sys.path.insert(0, str(_HERE.parents[3] / "ingest")) sys.path.insert(0, str(_HERE.parents[2] / "live")) import gen_boatcam_instance as G # noqa: E402 from store import InMemoryStore, VideoSegment # noqa: E402 BOAT = "u1:boat:9" SLOT = f"{BOAT}:Mast" BLOB = "v3/media/u1/boat/9/Mast" T0, T1 = 1_000_000, 1_100_000 # a 100 s window def _store(*segments: tuple[int, int | None, str, str]) -> InMemoryStore: """`(start, end, camera, source)` rows, one file each. The `source` is in the path because a live segment and the store-and-forward clip of the same seconds are two different objects.""" s = InMemoryStore() s.upsert_slot(SLOT, None) for i, (start, end, camera, source) in enumerate(segments): s.add_video_segment(VideoSegment( slot=SLOT, start_ms=start, end_ms=end, source=source, rendition=camera, gcs_path=f"gs://bucket/{BLOB}/20260816T1212{i:02d}.000Z_{camera}.mp4")) return s def test_a_clip_overlapping_the_window_is_kept_even_when_it_started_before_it(): """The boats film through the pre-start and a device uploads what it buffered, so the clip covering the opening instant routinely STARTS outside the window.""" store = _store((T0 - 3_000, T0 + 2_000, "front", "clip"), # straddles the opening (T0 + 50_000, T0 + 52_000, "front", "clip"), # inside (T1 + 10_000, T1 + 12_000, "front", "clip")) # after — a later race's footage clips = G.clips_for_boat(store, BOAT, t0_ms=T0, t1_ms=T1) assert [c.start_ms for c in clips] == [T0 - 3_000, T0 + 50_000] def test_a_clip_whose_end_was_never_recorded_is_still_found(): """`end_utc` is nullable and a SQL predicate on a null drops the row silently. The overlap is therefore decided in Python — a clip that exists, is indexed, and never appears is the one failure mode with nothing to see anywhere.""" store = _store((T0 + 10_000, None, "front", "clip")) assert len(G.clips_for_boat(store, BOAT, t0_ms=T0, t1_ms=T1)) == 1 def test_only_the_asked_for_camera_is_filmed(): store = _store((T0 + 1_000, T0 + 3_000, "front", "clip"), (T0 + 1_000, T0 + 3_000, "back", "clip")) assert {c.camera for c in G.clips_for_boat(store, BOAT, t0_ms=T0, t1_ms=T1)} == {"front"} assert {c.camera for c in G.clips_for_boat(store, BOAT, t0_ms=T0, t1_ms=T1, camera="both")} == {"front", "back"} def test_the_same_seconds_indexed_twice_are_filmed_once(): """A boat's seconds can be indexed as the live segments the broadcast played AND as the store-and-forward file. Filming both plays the race twice, interleaved.""" store = _store((T0 + 1_000, T0 + 3_000, "front", "live"), (T0 + 1_000, T0 + 3_000, "front", "clip")) assert [c.source for c in G.clips_for_boat(store, BOAT, t0_ms=T0, t1_ms=T1)] == ["clip"] # …and a boat that has ONLY live segments is still filmed, rather than dropped for having the # wrong kind of index. live_only = _store((T0 + 1_000, T0 + 3_000, "front", "live")) assert len(G.clips_for_boat(live_only, BOAT, t0_ms=T0, t1_ms=T1)) == 1 def test_one_cameras_bookkeeping_never_decides_for_the_other(): """`auto` asks "is there a stored file?" of each camera separately. Asked of both at once, a front camera with stored clips would drop a back camera indexed only as live segments.""" store = _store((T0 + 1_000, T0 + 3_000, "front", "clip"), (T0 + 1_000, T0 + 3_000, "back", "live")) clips = G.clips_for_boat(store, BOAT, t0_ms=T0, t1_ms=T1, camera="both") assert {(c.camera, c.source) for c in clips} == {("front", "clip"), ("back", "live")} def test_coverage_reports_the_hole_rather_than_the_film_length(): """"The film is 4 minutes and the race was 14" is the first question anyone asks of a duty-cycled camera, and it is a property of the RECORD, not of the encode.""" clips = [G.Clip(T0, T0 + 10_000, "front", "a", "clip", SLOT), G.Clip(T0 + 60_000, T0 + 70_000, "front", "b", "clip", SLOT)] cov = G.coverage(clips, t0_ms=T0, t1_ms=T1) assert cov["filmedS"] == 20.0 assert cov["raceWindowS"] == 100.0 assert cov["coverage"] == 0.2 assert cov["gapsOver2s"] == 1 assert cov["largestGapS"] == 50.0 # --- where the bytes are ------------------------------------------------------------------------- def _write(root: Path, rel: str) -> Path: p = root / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(b"\0") return p def test_a_local_tree_is_searched_in_every_layout_this_product_writes(tmp_path): """Three layouts are real — the key kept whole, the key with leading segments folded into one directory (a mirror made per prefix, which is what the office box holds), and `LocalUploader`'s own `/`. Each is proven by the file existing, not by a rule believed to be right.""" stored = f"gs://bucket/{BLOB}/clip.mp4" whole = tmp_path / "whole" want = _write(whole, f"{BLOB}/clip.mp4") assert G.resolve_local(stored, SLOT, whole) == want folded = tmp_path / "folded" # v3/media/u1/… under v3_media_u1/… want = _write(folded, "v3_media_u1/boat/9/Mast/clip.mp4") assert G.resolve_local(stored, SLOT, folded) == want inside = tmp_path / "inside" # a root pointed further down the tree want = _write(inside, "boat/9/Mast/clip.mp4") assert G.resolve_local(stored, SLOT, inside) == want uploader = tmp_path / "uploader" # LocalUploader: slot, ':' -> '_' want = _write(uploader, "u1_boat_9_Mast/clip.mp4") assert G.resolve_local(stored, SLOT, uploader) == want def test_a_tree_that_does_not_hold_the_clip_says_so(tmp_path): """None, so the caller falls back to the bucket. A partial mirror must not read as a total one.""" _write(tmp_path, f"{BLOB}/other.mp4") assert G.resolve_local(f"gs://bucket/{BLOB}/clip.mp4", SLOT, tmp_path) is None def test_a_local_tree_is_used_in_place_and_never_copied(tmp_path): """A 500 GB media tree is not worth duplicating to read it once — and a copy step is a second place the run can fail on disk space.""" root = tmp_path / "media" want = _write(root, f"{BLOB}/clip.mp4") clip = G.Clip(T0, T0 + 2_000, "front", f"gs://bucket/{BLOB}/clip.mp4", "clip", SLOT) dest = tmp_path / "dest" assert G.fetch_clips([clip], dest, media_root=root) == {clip.gcs_path: want} assert not dest.exists() # nothing was staged