Buckets:
| """cwm3/baselines_dms/video_io.py — the ONE video / time-axis / driver-box path shared by every DMS-baseline extractor (plan §3, 2026-09-14). | |
| Contracts (all verified against data/extract_pose_causal_routes.py, the causal route-pose extractor): | |
| * time base "frame/20": the cabin video dcamera.mp4 is 20 fps and frame-anchored; the 10 Hz feature grid is t = k / 10 <-> frame_idx = 2 k. | |
| The container fps is NEVER used for timing (video_info only reads width / height / frame count). | |
| * uid -> leaf: `leaf_of` (eval/newtest_sample.py:113-115): "<CAR>__<dongle>__<segment>__<route_a-b>" -> "<CAR>/<dongle>/<segment>/<route_a-b>"; | |
| video_path(uid) = /home/USER/BATON/<leaf>/dcamera.mp4 (local only; no network here — missing videos are LISTED by | |
| --missing-leaves for data/fetch_baton_leaves.py, never fetched by this module). | |
| * driver box: results/pose_causal_routes_v2/<uid>.npz preferred, else results/pose_causal_routes/<uid>.npz — keys t, frame_idx, box f32[N,4] | |
| (NATIVE px, causal per-frame expanding-median driver box; NaN undefined), driver_det, moving (vEgo > 2 m/s). Frames outside the pose cache | |
| have no box (NaN) -> extractors that need a crop record fail code 1. | |
| * spans: anchor_spans(uid) = union over the route's s12 anchors (s12_wm/results/targets_{regime}_{split}.npz, per requested regime / split) of | |
| [t - hist_s, t + post_s] (default 8.2 / 0.2 s), merged, clipped to the video; every extractor decodes each route ONCE over these spans. | |
| Only frames <= t_anchor are ever used downstream (summarise.window_index asserts it); the +0.2 s tail only guarantees the frame AT t exists. | |
| * decoding (INDEX-EXACT since 2026-09-14, decode_block): `workers` spawn processes, 2 ffmpeg threads each. cv2's CAP_PROP_POS_FRAMES seek converts the | |
| index to a timestamp with the container HEADER fps, so on recordings whose header fps is not exactly 20 (avg_frame_rate = n_frames / duration; measured on | |
| the Phase-1 manifest: 6 of 45 routes — HYUNDAI_TUCSON 19.9894 / 19.9940 / 19.9914, RIVIAN 20.0002, ELANTRA 20.0001, Unknown__04219e 19.9993) the plain | |
| seek landed +2..+5 frames LATE (Tucson, at frame indices ~10^4) or 1 frame EARLY (Rivian): the image delivered for frame_idx f was f + offset. Now: | |
| - header fps == 20 exactly (39/45 routes; their pts grid is a clean 50 ms lattice, ffprobe-verified): cv2 seek, then the landing index is READ BACK from | |
| the pts clock (CAP_PROP_POS_MSEC / 50 ms) and corrected by grabbing forward or re-seeking earlier and grabbing forward — a landing != target is never | |
| accepted; every grabbed frame's pts is checked against the 50 ms grid (a deviation aborts the cv2 path for that block); | |
| - header fps != 20: decord.VideoReader.get_batch (0.6.0) in the worker — it indexes frames through the file's own pts table (built by a packet scan at | |
| open, cached per worker process), i.e. decode-order exact whatever the header fps or pts stretch (5 of the 6 routes have stretched pts segments, | |
| e.g. 391 x 50.7 ms on Tucson 00000212 — a pts-based cv2 correction alone would be off there). decord RGB -> BGR is bit-identical to cv2's BGR | |
| (max abs diff 0 on the verified frames); the open cost is one full read of the file (~0.1 s warm per 600 MB, ~20 s cold for a 1.7 h route); | |
| - cv2 with the pts correction remains the fallback when decord is unavailable (flagged exact=False in the stream stats). | |
| Verified (`python baselines_dms/video_io.py --verify-exact`, results/baselines_dms/phase1/decode_exactness_check.json): on the three Tucson manifest | |
| routes and one fps == 20 route the frame delivered for index f equals the f-th frame of a continuous decode from frame 0 (max abs pixel diff 0). | |
| A gap of more than RESEEK_GAP_FRAMES (60) video frames between two requested frames of a block is crossed by a fresh (verified) seek instead of grabbing | |
| through it (a 64-frame block straddling two anchor windows 600 s apart used to grab ~12,000 native HEVC frames: 947 ms / frame, the openface3 Phase-1 | |
| stall). The cache meta records native_fps (header fps), seek_corrected = True and decode_stats (decoder used, corrections, offsets) per route. | |
| NOTE the project's FROZEN pose caches (data/extract_pose_causal_routes.py, results/pose_causal_routes[_v2]) were produced with the PLAIN cv2 seek, so on | |
| the fps != 20 routes their driver boxes may be up to ~5 frames = 0.25 s late relative to the corrected frames; the boxes are still used as given (an | |
| expanding-median driver box changes slowly). | |
| iter_frames yields the frames of a route in causal (increasing frame_idx) order as (t, frame_idx, frame_bgr | None, box[4] | NaN); the frame is resized | |
| to `decode_wh` (exact (w, h); None = native) and the box is returned in the coordinates of the RETURNED frame (box_coords="frame") or in native px. | |
| * frames_in_spans snaps span ends within GRID_SNAP_TOL_S = 0.02 s onto the 10 Hz grid (2026-09-14): the ends arrive as float32 (cache schema / merge_spans; | |
| f32(834.2) = 834.20001) and a plain floor / ceil decoded one frame outside the requested span on ~half the Phase-1 windows (21/50 before win_start, | |
| 22/50 after win_end = t_anchor + 0.3 s). The extractors' own grid_spans / snap_span_to_grid / grid_span hacks are now redundant but harmless. | |
| * driver_crop(frame, box, expand=0.10): box expanded by 10 % of its width / height per side (MARGIN of the pose extractors), clipped to the | |
| frame, integer pixel bounds; None when the box is undefined or empty. | |
| No openpilot field is read; no GPU; no network. | |
| python baselines_dms/video_io.py --self-test [--uid UID] # 1 local TRAIN route: boxes, spans, a short decode, a crop, index-exactness on 10 frames | |
| python baselines_dms/video_io.py --missing-leaves {train979,gold,test_val} # -> results/baselines_dms/fetch/missing_<tier>.txt (or *_dup.txt + diff) | |
| python baselines_dms/video_io.py --verify-exact [--uid UID ...] # decode_block vs a continuous decode from frame 0 on the manifest windows of the given routes | |
| # (default: the 3 Tucson fps != 20 routes + 1 fps == 20 route) -> phase1/decode_exactness_check.json | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import sys | |
| import time | |
| from functools import lru_cache | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| BATON = Path("/home/USER/BATON") | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| POSE_V2 = CWM3 / "results" / "pose_causal_routes_v2" | |
| POSE_V1 = CWM3 / "results" / "pose_causal_routes" | |
| FETCH_DIR = CWM3 / "results" / "baselines_dms" / "fetch" | |
| REMOTE = "USER@REMOTE_HOST:/data/backup/BATON" | |
| FPS, HZ = 20, 10 | |
| TIME_BASE = "frame/20: t=k/10, frame_idx=2k" | |
| HIST_S, POST_S = 8.2, 0.2 | |
| MARGIN = 0.10 | |
| DEFAULT_DECODE_WH = (640, 400) | |
| GRID_SNAP_TOL_S = 0.02 # frames_in_spans: a span end within this of a 10 Hz grid point is snapped onto it (span ends are float32 in the cache schema) | |
| RESEEK_GAP_FRAMES = 60 # decode_block: a gap > this many video frames between two requested frames is crossed by a fresh seek, not by grabbing through | |
| PTS_STEP_MS = 1000.0 / FPS # 50 ms: presentation-time step of the frame/20 axis (clean on every header-fps == 20 route of the manifest) | |
| PTS_TOL_MS = 1.0 # tolerance of the pts-clock landing / grid checks of the cv2 path | |
| DECORD_CHUNK = 16 # frames per decord get_batch call (bounds the RGB buffer: 16 native 1928 x 1208 frames ~ 110 MB) | |
| PHASE1_DIR = CWM3 / "results" / "baselines_dms" / "phase1" | |
| ALL_PACKS = tuple(f"{r}_{s}" for r in ("manual", "engaged") for s in ("train", "val", "test")) | |
| TIERS = {"train979": 2, "gold": 3, "test_val": 4} # tier numbers of results/baselines_dms/fetch/missing_tier<N>.txt (fetch agent) / plan §5 | |
| # ----------------------------------------------------------------------------- identifiers / paths | |
| def leaf_of(uid: str) -> str: | |
| """eval/newtest_sample.py:113-115 — '<CAR>__<dongle>__<segment>__<route>' -> '<CAR>/<dongle>/<segment>/<route>' (the route part may contain '__').""" | |
| p = uid.split("__") | |
| return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def video_path(uid: str, root: Path = BATON) -> Path: | |
| return Path(root) / leaf_of(uid) / "dcamera.mp4" | |
| def has_video(uid: str, root: Path = BATON) -> bool: | |
| return video_path(uid, root).exists() | |
| def video_info(path) -> tuple[int, int, int]: | |
| """(W, H, n_frames) from the container header (cv2); fps is deliberately NOT returned (time base is frame/20).""" | |
| import cv2 | |
| cap = cv2.VideoCapture(str(path)) | |
| try: | |
| W, H, n = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)), int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)), int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) | |
| finally: | |
| cap.release() | |
| if n <= 0 or W <= 0 or H <= 0: | |
| raise RuntimeError(f"cannot read the video header of {path}") | |
| return W, H, n | |
| def video_header_fps(path) -> float: | |
| """the container header fps (cv2 CAP_PROP_FPS = ffmpeg avg_frame_rate = n_frames / duration). NEVER used for timing (time base is frame/20); recorded in the | |
| cache meta as `native_fps` and used by decode_block to choose the index-exact decoder (== 20 -> verified cv2 seek, != 20 -> decord).""" | |
| import cv2 | |
| cap = cv2.VideoCapture(str(path)) | |
| try: | |
| fps = float(cap.get(cv2.CAP_PROP_FPS)) | |
| finally: | |
| cap.release() | |
| return fps | |
| # ----------------------------------------------------------------------------- driver boxes (causal route pose) | |
| def pose_path(uid: str): | |
| for d in (POSE_V2, POSE_V1): | |
| p = d / f"{uid}.npz" | |
| if p.exists(): | |
| return p | |
| return None | |
| def load_driver_boxes(uid: str): | |
| """-> dict(t f32[N], frame_idx i64[N], box f32[N,4] native px (NaN undefined), driver_det bool[N], moving bool[N], native_wh, source, time_base) or None.""" | |
| p = pose_path(uid) | |
| if p is None: | |
| return None | |
| with np.load(p, allow_pickle=True) as z: | |
| out = {"t": z["t"].astype(np.float32), "frame_idx": z["frame_idx"].astype(np.int64), "box": z["box"].astype(np.float32), | |
| "driver_det": z["driver_det"].astype(bool), "moving": z["moving"].astype(bool), "native_wh": tuple(int(v) for v in z["native_wh"]), | |
| "source": str(p.parent.name), "time_base": str(z["time_base"]) if "time_base" in z.files else TIME_BASE} | |
| assert out["box"].shape == (len(out["t"]), 4) and np.array_equal(out["frame_idx"], np.round(out["t"].astype(np.float64) * FPS).astype(np.int64)), f"{p}: time base" | |
| return out | |
| def boxes_at(boxes, frame_idx) -> tuple[np.ndarray, np.ndarray]: | |
| """boxes dict (or None) + frame_idx i64[n] -> (box f32[n,4] NaN where no pose row, moving bool[n] False where none).""" | |
| n = len(frame_idx) | |
| B = np.full((n, 4), np.nan, np.float32) | |
| MOV = np.zeros(n, bool) | |
| if boxes is None or len(boxes["frame_idx"]) == 0: | |
| return B, MOV | |
| fi = boxes["frame_idx"] | |
| j = np.searchsorted(fi, frame_idx) | |
| ok = (j < len(fi)) | |
| ok[ok] &= fi[j[ok]] == np.asarray(frame_idx)[ok] | |
| B[ok] = boxes["box"][j[ok]] | |
| MOV[ok] = boxes["moving"][j[ok]] | |
| return B, MOV | |
| # ----------------------------------------------------------------------------- anchors / spans | |
| def anchors_by_uid(packs: tuple = ALL_PACKS) -> dict: | |
| """uid -> sorted f64 anchor times over the given s12 packs ('<regime>_<split>' names).""" | |
| A = {} | |
| for name in packs: | |
| r, s = name.split("_") | |
| with np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True) as z: | |
| uid, t = z["uid"], z["t"].astype(np.float64) | |
| order = np.argsort(uid, kind="stable") | |
| u_s, t_s = uid[order], t[order] | |
| bounds = np.flatnonzero(np.r_[1, u_s[1:] != u_s[:-1], 1]) | |
| for a, b in zip(bounds[:-1], bounds[1:]): | |
| A.setdefault(str(u_s[a]), []).append(t_s[a:b]) | |
| return {u: np.unique(np.concatenate(v)) for u, v in A.items()} | |
| def anchors_of(uid: str, regime_split_packs=ALL_PACKS) -> np.ndarray: | |
| return anchors_by_uid(tuple(regime_split_packs)).get(uid, np.zeros(0, np.float64)) | |
| def spans_from_anchors(ts, hist_s=HIST_S, post_s=POST_S, dur=None, merge_gap=0.0) -> np.ndarray: | |
| """union of [t - hist_s, t + post_s] over anchor times ts -> merged f32[S,2], clipped to [0, dur].""" | |
| ts = np.asarray(ts, np.float64) | |
| if len(ts) == 0: | |
| return np.zeros((0, 2), np.float32) | |
| a = np.maximum(ts - hist_s, 0.0) | |
| b = ts + post_s if dur is None else np.minimum(ts + post_s, float(dur)) | |
| from cache_schema import merge_spans | |
| return merge_spans(np.stack([a, b], 1), gap=merge_gap) | |
| def anchor_spans(uid: str, regime_split_packs=ALL_PACKS, hist_s=HIST_S, post_s=POST_S, dur=None, merge_gap=0.0) -> np.ndarray: | |
| """the decode spans of a route: union over its s12 anchors (of the given packs) of [t - hist_s, t + post_s], merged; f32[S,2].""" | |
| return spans_from_anchors(anchors_of(uid, regime_split_packs), hist_s, post_s, dur, merge_gap) | |
| def frames_in_spans(spans, hz=HZ, n_frames=None, snap_tol_s=GRID_SNAP_TOL_S): | |
| """-> (k i64[n], frame_idx i64[n], t f32[n]) : the hz-grid steps k with k/hz inside a span, frame_idx = k * FPS / hz. A span end within `snap_tol_s` (0.02 s) of | |
| a grid point is SNAPPED onto it (2026-09-14 fix — the ends arrive as float32 from the cache schema, f32(834.2) = 834.20001, and a plain floor / ceil decoded | |
| one frame outside the span on ~half the Phase-1 windows); an end farther from the grid keeps the outer floor (start) / ceil (end). The extractors' | |
| grid_spans / snap_span_to_grid helpers (which pull the ends inwards by <= 0.02 s) are thereby redundant but harmless; an inverted single-frame pair | |
| [k/hz + eps, k/hz - eps] still yields exactly k.""" | |
| assert FPS % hz == 0, f"hz {hz} must divide FPS {FPS}" | |
| step = FPS // hz | |
| tol = snap_tol_s * hz + 1e-3 # in grid steps; the 1e-3 absorbs the float32 resolution of a span end (~2.4e-4 s at t ~ 4000 s) | |
| ks = [] | |
| for a, b in np.asarray(spans, np.float64).reshape(-1, 2): | |
| ka, kb = a * hz, b * hz | |
| k0 = int(round(ka)) if abs(ka - round(ka)) <= tol else int(math.floor(ka)) | |
| k1 = int(round(kb)) if abs(kb - round(kb)) <= tol else int(math.ceil(kb)) | |
| if k1 >= k0: | |
| ks.append(np.arange(k0, k1 + 1)) | |
| k = np.unique(np.concatenate(ks)) if ks else np.zeros(0, np.int64) | |
| k = k[k >= 0] | |
| fi = k * step | |
| if n_frames is not None: | |
| keep = fi < int(n_frames) | |
| k, fi = k[keep], fi[keep] | |
| return k.astype(np.int64), fi.astype(np.int64), (k / hz).astype(np.float32) | |
| # ----------------------------------------------------------------------------- decoding (worker processes) | |
| class _PtsGridError(RuntimeError): | |
| """the cv2 path found a presentation time off the 50 ms grid: its pts-based landing check is not trustworthy here -> the block is decoded with decord.""" | |
| def _runs(fis, max_gap=RESEEK_GAP_FRAMES): | |
| """increasing frame indices -> list of runs; a new run starts where the gap to the previous requested frame exceeds max_gap video frames (re-seek instead of grabbing through).""" | |
| runs = [[fis[0]]] | |
| for f in fis[1:]: | |
| if f - runs[-1][-1] > max_gap: | |
| runs.append([f]) | |
| else: | |
| runs[-1].append(f) | |
| return runs | |
| def _cv2_pts_index(cap): | |
| """decode-order index of the frame cv2 grabbed last, read from its presentation time (exact on the clean 50 ms grid of the header-fps == 20 routes).""" | |
| import cv2 | |
| ms = float(cap.get(cv2.CAP_PROP_POS_MSEC)) | |
| k = ms / PTS_STEP_MS | |
| if abs(k - round(k)) * PTS_STEP_MS > PTS_TOL_MS: | |
| raise _PtsGridError(f"pts {ms:.2f} ms is off the 50 ms grid") | |
| return int(round(k)) | |
| def _cv2_position(cap, target, rep, strict=True): | |
| """position `cap` so that the next grab() delivers decode-order frame `target`: seek, read the landing back from the pts clock, correct by grabbing forward or by | |
| re-seeking earlier and grabbing forward; a landing != target is never accepted. Updates rep[n_seek_corrected / max_landing_offset].""" | |
| import cv2 | |
| if target <= 0: | |
| cap.set(cv2.CAP_PROP_POS_FRAMES, 0) | |
| return | |
| cap.set(cv2.CAP_PROP_POS_FRAMES, int(target)) | |
| last = _cv2_pts_index(cap) if strict else int(round(float(cap.get(cv2.CAP_PROP_POS_MSEC)) / PTS_STEP_MS)) # cv2 believes it just grabbed target - 1 | |
| off = last - (target - 1) | |
| rep["max_landing_offset"] = max(rep["max_landing_offset"], abs(off)) | |
| if off != 0: | |
| rep["n_seek_corrected"] += 1 | |
| rep["landing_offsets"].append(int(off)) | |
| n_re = 0 | |
| while last > target - 1: # landed LATE: re-seek earlier by twice the overshoot plus a growing margin | |
| n_re += 1 | |
| back = 2 * (last - (target - 1)) + 8 * n_re | |
| if target - back <= 0 or n_re > 6: | |
| cap.set(cv2.CAP_PROP_POS_FRAMES, 0) | |
| last = -1 | |
| break | |
| cap.set(cv2.CAP_PROP_POS_FRAMES, int(target - back)) | |
| last = _cv2_pts_index(cap) if strict else int(round(float(cap.get(cv2.CAP_PROP_POS_MSEC)) / PTS_STEP_MS)) | |
| while last < target - 1: # landed EARLY (or re-seeked): grab forward on the verified 50 ms grid | |
| if not cap.grab(): | |
| raise RuntimeError(f"grab failed while positioning at frame {target}") | |
| last += 1 | |
| if strict: | |
| ms = float(cap.get(cv2.CAP_PROP_POS_MSEC)) | |
| if abs(ms - last * PTS_STEP_MS) > PTS_TOL_MS: | |
| raise _PtsGridError(f"pts {ms:.2f} ms at decode index {last} is off the 50 ms grid") | |
| def _cv2_decode(vpath, fis, want, dw, dh, out, got, rep, strict=True): | |
| """cv2 seek + grab / retrieve with the pts-clock landing verification (strict: every grabbed frame must sit on the 50 ms grid, else _PtsGridError).""" | |
| import cv2 | |
| cap = cv2.VideoCapture(str(vpath)) | |
| cap.set(cv2.CAP_PROP_N_THREADS, 2) | |
| try: | |
| runs = _runs(fis) | |
| rep["n_gap_reseeks"] += len(runs) - 1 | |
| for run in runs: | |
| _cv2_position(cap, run[0], rep, strict) | |
| for i in range(run[0], run[-1] + 1): | |
| if not cap.grab(): | |
| return # end of stream: the remaining frames stay got=False (fail code 2) | |
| rep["n_grab"] += 1 | |
| if strict: | |
| ms = float(cap.get(cv2.CAP_PROP_POS_MSEC)) | |
| if abs(ms - i * PTS_STEP_MS) > PTS_TOL_MS: | |
| raise _PtsGridError(f"pts {ms:.2f} ms at decode index {i} is off the 50 ms grid") | |
| if i in want: | |
| ok, fr = cap.retrieve() | |
| if not ok: | |
| return | |
| if fr.shape[1] != dw or fr.shape[0] != dh: | |
| fr = cv2.resize(fr, (dw, dh), interpolation=cv2.INTER_AREA) | |
| out[want[i]] = fr | |
| got[want[i]] = True | |
| finally: | |
| cap.release() | |
| _DECORD_READER = {} # per worker process: {video path: decord.VideoReader}; opening scans every packet of the file once (pts table), so it is kept for the route | |
| def _decord_decode(vpath, fis, want, dw, dh, out, got, rep): | |
| """decord.VideoReader.get_batch: frames addressed through the file's own pts table (decode-order exact on any header fps / pts stretch); RGB -> BGR.""" | |
| import cv2 | |
| import decord | |
| decord.bridge.set_bridge("native") | |
| key = str(vpath) | |
| vr = _DECORD_READER.get(key) | |
| if vr is None: | |
| _DECORD_READER.clear() | |
| vr = _DECORD_READER[key] = decord.VideoReader(key, ctx=decord.cpu(0), num_threads=2) | |
| rep["decoder"] = "decord" | |
| n_vr = len(vr) | |
| idx = [f for f in fis if 0 <= f < n_vr] | |
| for s in range(0, len(idx), DECORD_CHUNK): | |
| sub = idx[s:s + DECORD_CHUNK] | |
| arr = vr.get_batch(sub).asnumpy() | |
| for f, fr in zip(sub, arr): | |
| fr = np.ascontiguousarray(fr[:, :, ::-1]) # RGB -> BGR (bit-identical to cv2's BGR output) | |
| if fr.shape[1] != dw or fr.shape[0] != dh: | |
| fr = cv2.resize(fr, (dw, dh), interpolation=cv2.INTER_AREA) | |
| out[want[f]] = fr | |
| got[want[f]] = True | |
| rep["n_grab"] += 1 | |
| def decode_block(job): | |
| """worker: (vpath, frame_idx list, dw, dh[, native_fps]) -> dict(frames u8[n,dh,dw,3] BGR, got bool[n], n_grab, secs, decoder 'cv2' | 'decord', n_seek_corrected, | |
| max_landing_offset, landing_offsets, n_gap_reseeks, exact, note). INDEX-EXACT (2026-09-14; see the module docstring): header fps == 20 -> cv2 seek whose landing | |
| is read back from the pts clock and corrected (never a landing != target), every grabbed pts checked against the 50 ms grid; header fps != 20 (or a pts off the | |
| grid) -> decord get_batch through the file's pts table; decord unavailable -> cv2 with the pts correction, flagged exact=False. Requested frames more than | |
| RESEEK_GAP_FRAMES apart are reached by a fresh seek, not by grabbing through the gap. n_grab = frames grabbed (cv2) / delivered (decord).""" | |
| import cv2 | |
| cv2.setNumThreads(2) | |
| vpath, fis, dw, dh = job[0], [int(f) for f in job[1]], int(job[2]), int(job[3]) | |
| native_fps = float(job[4]) if len(job) > 4 and job[4] is not None else video_header_fps(vpath) | |
| t0 = time.time() | |
| want = {f: i for i, f in enumerate(fis)} | |
| out = np.zeros((len(fis), dh, dw, 3), np.uint8) | |
| got = np.zeros(len(fis), bool) | |
| rep = dict(n_grab=0, decoder="cv2", n_seek_corrected=0, max_landing_offset=0, landing_offsets=[], n_gap_reseeks=0, exact=True, note="", native_fps=native_fps) | |
| if not fis: | |
| return dict(frames=out, got=got, secs=time.time() - t0, **rep) | |
| use_cv2 = abs(native_fps - FPS) < 1e-6 | |
| if use_cv2: | |
| try: | |
| _cv2_decode(vpath, fis, want, dw, dh, out, got, rep, strict=True) | |
| except _PtsGridError as e: | |
| rep["note"] = f"cv2 path abandoned ({e}); decord" | |
| got[:] = False | |
| rep["n_grab"] = 0 | |
| use_cv2 = False | |
| if not use_cv2: | |
| try: | |
| _decord_decode(vpath, fis, want, dw, dh, out, got, rep) | |
| except Exception as e: # decord missing / failed: cv2 + pts correction (approximate inside stretched pts segments) | |
| rep["note"] = (rep["note"] + f" decord failed ({repr(e)[:80]}); cv2 fallback (not index-exact)").strip() | |
| rep["decoder"] = "cv2_fallback" | |
| rep["exact"] = False | |
| got[:] = False | |
| rep["n_grab"] = 0 | |
| _cv2_decode(vpath, fis, want, dw, dh, out, got, rep, strict=False) | |
| return dict(frames=out, got=got, secs=time.time() - t0, **rep) | |
| class FrameStream: | |
| """iterator over the frames of a route (see iter_frames); exposes native_wh / native_fps / decode_wh / n_frames / stats (incl. the index-exactness record: | |
| decoder, n_seek_corrected, max_landing_offset, n_gap_reseeks, decode_exact) after iteration; decode_meta() -> the dict written into the cache meta.""" | |
| def __init__(self, uid, spans, hz=HZ, workers=6, decode_wh=DEFAULT_DECODE_WH, block=300, vpath=None, boxes="auto", box_coords="frame"): | |
| import multiprocessing as mp | |
| from concurrent.futures import ProcessPoolExecutor | |
| self.uid, self.hz, self.workers, self.block = uid, hz, int(workers), int(block) | |
| self.vpath = Path(vpath) if vpath else video_path(uid) | |
| if not self.vpath.exists(): | |
| raise FileNotFoundError(f"no local cabin video for {uid}: {self.vpath}") | |
| self.W, self.H, self.n_frames = video_info(self.vpath) | |
| self.native_wh = (self.W, self.H) | |
| self.native_fps = video_header_fps(self.vpath) # header fps: recorded in the meta, chooses the exact decoder (never used for timing) | |
| self.decode_wh = tuple(int(v) for v in decode_wh) if decode_wh is not None else (self.W, self.H) | |
| self.k, self.frame_idx, self.t = frames_in_spans(spans, hz, self.n_frames) | |
| self.spans = np.asarray(spans, np.float32).reshape(-1, 2) | |
| self.boxes = load_driver_boxes(uid) if boxes == "auto" else boxes | |
| self.box, self.moving = boxes_at(self.boxes, self.frame_idx) | |
| assert box_coords in ("frame", "native") | |
| self.box_coords = box_coords | |
| self.stats = dict(n_frames_requested=int(len(self.frame_idx)), n_got=0, n_grab=0, decode_worker_s=0.0, decode_ms_per_frame=[], wall_s=0.0, | |
| decoder=[], n_seek_corrected=0, max_landing_offset=0, landing_offsets=[], n_gap_reseeks=0, decode_exact=True, decode_notes=[]) | |
| self._pool_cls, self._ctx = ProcessPoolExecutor, mp.get_context("spawn") | |
| def box_for_frame(self, i): | |
| b = self.box[i] | |
| if self.box_coords == "frame" and np.isfinite(b).all(): | |
| sx, sy = self.decode_wh[0] / self.W, self.decode_wh[1] / self.H | |
| return np.array([b[0] * sx, b[1] * sy, b[2] * sx, b[3] * sy], np.float32) | |
| return b.copy() | |
| def __iter__(self): | |
| N = len(self.frame_idx) | |
| if N == 0: | |
| return | |
| t_wall = time.time() | |
| dw, dh = self.decode_wh | |
| blocks = [np.arange(i, min(i + self.block, N)) for i in range(0, N, self.block)] | |
| with self._pool_cls(max_workers=max(1, self.workers), mp_context=self._ctx) as pool: | |
| futs, bi = [], 0 | |
| def submit_upto(n): | |
| nonlocal bi | |
| while bi < len(blocks) and len(futs) < n: | |
| b = blocks[bi] | |
| futs.append(pool.submit(decode_block, (str(self.vpath), self.frame_idx[b], dw, dh, self.native_fps))) | |
| bi += 1 | |
| submit_upto(2 * max(1, self.workers)) | |
| for b in blocks: | |
| res = futs.pop(0).result() | |
| submit_upto(2 * max(1, self.workers)) | |
| self.stats["n_grab"] += int(res["n_grab"]) | |
| self.stats["decode_worker_s"] += float(res["secs"]) | |
| if res.get("decoder") and res["decoder"] not in self.stats["decoder"]: | |
| self.stats["decoder"].append(res["decoder"]) | |
| self.stats["n_seek_corrected"] += int(res.get("n_seek_corrected", 0)) | |
| self.stats["max_landing_offset"] = max(self.stats["max_landing_offset"], int(res.get("max_landing_offset", 0))) | |
| self.stats["landing_offsets"] += [int(o) for o in res.get("landing_offsets", [])][:20] | |
| self.stats["n_gap_reseeks"] += int(res.get("n_gap_reseeks", 0)) | |
| self.stats["decode_exact"] = bool(self.stats["decode_exact"] and res.get("exact", True)) | |
| if res.get("note"): | |
| self.stats["decode_notes"].append(str(res["note"])[:160]) | |
| self.stats["decode_ms_per_frame"].append(1000.0 * res["secs"] / max(len(b), 1)) | |
| for j, i in enumerate(b): | |
| fr = res["frames"][j] if res["got"][j] else None | |
| if fr is not None: | |
| self.stats["n_got"] += 1 | |
| yield float(self.t[i]), int(self.frame_idx[i]), fr, self.box_for_frame(i) | |
| self.stats["wall_s"] = time.time() - t_wall | |
| def decode_meta(self): | |
| """-> {native_fps, seek_corrected, decode_stats{decoder, n_seek_corrected, max_landing_offset, landing_offsets, n_gap_reseeks, decode_exact, notes}} for the cache meta.""" | |
| s = self.stats | |
| return {"native_fps": float(self.native_fps), "seek_corrected": True, | |
| "decode_stats": {"decoder": list(s["decoder"]), "n_seek_corrected": int(s["n_seek_corrected"]), "max_landing_offset": int(s["max_landing_offset"]), | |
| "landing_offsets": list(s["landing_offsets"])[:20], "n_gap_reseeks": int(s["n_gap_reseeks"]), "decode_exact": bool(s["decode_exact"]), | |
| "notes": list(s["decode_notes"])[:5], "n_grab": int(s["n_grab"])}} | |
| def iter_frames(uid, spans, hz=HZ, workers=6, decode_wh=DEFAULT_DECODE_WH, block=300, vpath=None, boxes="auto", box_coords="frame"): | |
| """yield (t, frame_idx, frame_bgr u8[dh,dw,3] | None, box f32[4] | NaN) for every hz-grid frame inside `spans`, in causal (increasing frame_idx) order. | |
| decode_wh: exact (w, h) resize (INTER_AREA) or None for native; boxes: 'auto' = load_driver_boxes(uid) (None -> all NaN) or a dict of the same form; | |
| box_coords 'frame' (default; scaled to the returned frame) | 'native'. Use FrameStream(...) directly when the header (native_wh) or the decode | |
| statistics are needed after iteration.""" | |
| return iter(FrameStream(uid, spans, hz, workers, decode_wh, block, vpath, boxes, box_coords)) | |
| # ----------------------------------------------------------------------------- crops | |
| def expand_box(box, W, H, expand=MARGIN): | |
| """[x0,y0,x1,y1] -> integer bounds expanded by `expand` x width / height per side and clipped to [0,W] x [0,H] (data/extract_pose_causal.expand_box).""" | |
| x0, y0, x1, y1 = [float(v) for v in box] | |
| w, h = x1 - x0, y1 - y0 | |
| return [int(max(0, round(x0 - expand * w))), int(max(0, round(y0 - expand * h))), int(min(W, round(x1 + expand * w))), int(min(H, round(y1 + expand * h)))] | |
| def driver_crop(frame, box, expand=MARGIN): | |
| """frame u8[H,W,3] + box (frame coordinates) -> the driver crop u8[h,w,3] (a view) or None when the box is NaN / degenerate.""" | |
| if frame is None or box is None or not np.all(np.isfinite(box)): | |
| return None | |
| H, W = frame.shape[:2] | |
| x0, y0, x1, y1 = expand_box(box, W, H, expand) | |
| if x1 - x0 < 2 or y1 - y0 < 2: | |
| return None | |
| return frame[y0:y1, x0:x1] | |
| # ----------------------------------------------------------------------------- missing-leaf lists (for data/fetch_baton_leaves.py; nothing is fetched here) | |
| def _shim_labels(): | |
| for p in (CWM3, CWM3 / "train", CWM3 / "data", CWM3 / "eval", CWM3 / "common", CWM2 / "eval", CWM2 / "data", CWM2, CWM2 / "baselines" / "legacy_cockpitwm"): | |
| if str(p) not in sys.path: | |
| sys.path.insert(0, str(p)) | |
| def missing_leaves(tier: str, root: Path = BATON): | |
| """-> (missing_leaves sorted, counts {leaf: n_rows}, source_tag, all_leaves_of_tier). Tiers (plan §5): train979 = routes of the 979 TRAIN warning-label anchors | |
| (data.labels.train_label_table); gold = routes of the clean human gold (cwm2/eval/warning_gold.build_gold, clean_test + clean_val, union over legacy_first | |
| seeds 0/1/2 and the s12 rule); test_val = every route of the s12 test / val packs not already in train979 / gold. 'missing' = no complete | |
| <root>/<leaf>/dcamera.mp4 (exists with the remote size of results/baton_leaves.json when known; a partially transferred file counts as missing).""" | |
| assert tier in TIERS, f"tier must be one of {sorted(TIERS)}" | |
| _shim_labels() | |
| sizes = {} | |
| lj = CWM3 / "results" / "baton_leaves.json" | |
| if lj.exists(): | |
| sizes = {k: v.get("dcamera_bytes") for k, v in json.load(open(lj)).get("leaves", {}).items()} | |
| def has_cabin(leaf): | |
| """local dcamera.mp4 exists AND (when results/baton_leaves.json knows the remote size) has that size — an in-flight rsync --partial --inplace file does not count.""" | |
| f = Path(root) / leaf / "dcamera.mp4" | |
| if not f.exists(): | |
| return False | |
| need = sizes.get(leaf) | |
| return need is None or f.stat().st_size == need | |
| if tier == "train979": | |
| from data.labels import train_label_table | |
| T = train_label_table(("consensus",)) | |
| cnt = T.assign(leaf=T.uid.map(leaf_of)).groupby("leaf").size().to_dict() | |
| leaves = set(cnt) | |
| return sorted(l for l in leaves if not has_cabin(l)), cnt, "train_label_anchors_979", leaves | |
| if tier == "gold": | |
| import pandas as pd | |
| import warning_gold as WG | |
| EP = WG.episode_table() | |
| lab = EP[EP.label != "unsure"].reset_index(drop=True) | |
| packs = {(r, s): WG.load_pack(r, s) for r in ("manual", "engaged") for s in ("test", "val")} | |
| Gs = [] | |
| for seed in (0, 1, 2): | |
| legacy = WG.merge_legacy(seed) | |
| if legacy is not None: | |
| Gs.append(WG.build_gold(lab, packs, legacy, "legacy_first")) | |
| Gs.append(WG.build_gold(lab, packs, None, "s12")) | |
| GA = pd.concat(Gs, ignore_index=True) | |
| GA = GA[GA.clean_test | GA.clean_val] | |
| E = GA.drop_duplicates("idx") | |
| cnt = E.assign(leaf=E.uid.map(leaf_of)).groupby("leaf").size().to_dict() | |
| leaves = {leaf_of(u) for u in GA.uid.unique()} | |
| return sorted(l for l in leaves if not has_cabin(l)), cnt, "clean_gold_test_val", leaves | |
| # test_val | |
| cnt = {} | |
| for r in ("manual", "engaged"): | |
| for s in ("test", "val"): | |
| with np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True) as z: | |
| uu, cc = np.unique(z["uid"].astype(str), return_counts=True) | |
| for u, c in zip(uu, cc): | |
| cnt[leaf_of(u)] = cnt.get(leaf_of(u), 0) + int(c) | |
| leaves = set(cnt) | |
| excl = missing_leaves("train979", root)[3] | missing_leaves("gold", root)[3] | |
| return sorted(l for l in leaves if not has_cabin(l) and l not in excl), cnt, "s12_test_val_packs", leaves | |
| def write_leaf_list(path, leaves, counts, tier, tag, header): | |
| path = Path(path) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| with open(path, "w") as f: | |
| f.write(f"# {header} ({time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}); pull {REMOTE}/<leaf>/{{dcamera,qcamera}}.mp4 -> {BATON}/<leaf>/ via data/fetch_baton_leaves.py --leaves-file " | |
| f"(rsync -a --partial --inplace, sha256-verified; never --delete); written by baselines_dms/video_io.py --missing-leaves {tier}\n") | |
| f.write("# BATON/<vehicle>/<dongle>/<segment>/<route_leaf>\tn_uid_rows\ttier\tsource (first field = the path under the BATON root, local and remote)\n") | |
| for l in sorted(leaves): | |
| f.write(f"BATON/{l}\t{counts.get(l, 0)}\t{TIERS[tier]}\t{tag}\n") | |
| return path | |
| def read_leaf_list(path): | |
| out = [] | |
| for line in Path(path).read_text().splitlines(): | |
| if not line.strip() or line.startswith("#"): | |
| continue | |
| first = line.split("\t")[0].strip() | |
| out.append(first[6:] if first.startswith("BATON/") else first) | |
| return set(out) | |
| def cli_missing_leaves(tier, out_dir=FETCH_DIR): | |
| """writes missing_<tier>.txt unless the fetch agent's missing_tier<N>.txt (or missing_<tier>.txt) already exists -> missing_<tier>_dup.txt + a set diff.""" | |
| miss, cnt, tag, all_leaves = missing_leaves(tier) | |
| header = {"train979": "tier 2: leaves without local dcamera.mp4 among the routes of the 979 TRAIN warning-label anchors (data.labels.train_label_table)", | |
| "gold": "tier 3: leaves without local dcamera.mp4 among the routes of the clean human gold (warning_gold.build_gold, clean_test + clean_val)", | |
| "test_val": "tier 4: leaves without local dcamera.mp4 among ALL routes of the s12 test/val packs, excluding leaves already in tier 2 or 3"}[tier] | |
| out_dir = Path(out_dir) | |
| existing = [p for p in (out_dir / f"missing_tier{TIERS[tier]}.txt", out_dir / f"missing_{tier}.txt") if p.exists()] | |
| dest = out_dir / (f"missing_{tier}_dup.txt" if existing else f"missing_{tier}.txt") | |
| write_leaf_list(dest, miss, cnt, tier, tag, header) | |
| rep = {"tier": tier, "tier_no": TIERS[tier], "n_leaves": len(all_leaves), "n_missing": len(miss), "missing_rows": int(sum(cnt.get(l, 0) for l in miss)), "written": str(dest)} | |
| if existing: | |
| theirs = read_leaf_list(existing[0]) | |
| mine = set(miss) | |
| rep["diff_against"] = str(existing[0]) | |
| rep["only_in_mine"] = sorted(mine - theirs) | |
| rep["only_in_theirs"] = sorted(theirs - mine) | |
| rep["identical"] = mine == theirs | |
| print(json.dumps(rep, indent=1)) | |
| return rep | |
| # ----------------------------------------------------------------------------- self-test | |
| def _self_test(uid=None, workers=2): | |
| from cache_schema import merge_spans | |
| assert leaf_of("A__b__c__route_0-1") == "A/b/c/route_0-1" and leaf_of("A__b__c__r__x") == "A/b/c/r__x" | |
| k, fi, t = frames_in_spans([[1.0, 2.0], [1.5, 3.05]], HZ, 1000) | |
| assert k[0] == 10 and k[-1] == 31 and np.array_equal(fi, 2 * k) and abs(t[3] - 1.3) < 1e-6 and len(k) == 22 | |
| # 2026-09-14 snapping: f32 span ends, an inverted eps pair, and the ends of the Phase-1 windows land exactly on k0 .. k1 | |
| k32 = frames_in_spans(np.array([[834.2 - 10.2, 834.2]], np.float32), HZ, None)[0] | |
| assert k32[0] == 8240 and k32[-1] == 8342 and len(k32) == 103, (k32[0], k32[-1], len(k32)) | |
| assert list(frames_in_spans([[41.02, 40.98]], HZ, None)[0]) == [410] and list(frames_in_spans([[41.002, 40.998]], HZ, None)[0]) == [410] | |
| assert list(frames_in_spans([[650.8000000000001 - 10.2, 650.8000000000001]], HZ, None)[0])[-1] == 6508 and frames_in_spans([[376.8, 387.0]], HZ, None)[0][0] == 3768 | |
| assert list(_runs([100, 102, 104, 400, 402], 60)) == [[100, 102, 104], [400, 402]] | |
| sp = spans_from_anchors([10.0, 12.0, 40.0], 8.2, 0.2, dur=45.0) | |
| assert sp.shape == (2, 2) and abs(sp[0, 0] - 1.8) < 1e-6 and abs(sp[0, 1] - 12.2) < 1e-6 and abs(sp[1, 1] - 40.2) < 1e-6 | |
| assert len(merge_spans(sp)) == 2 | |
| fr = np.zeros((400, 640, 3), np.uint8) | |
| c = driver_crop(fr, np.array([100, 100, 300, 350], np.float32)) | |
| assert c.shape == (300, 240, 3), c.shape # +10 % per side: x 80..320, y 75..375 | |
| assert driver_crop(fr, np.array([np.nan] * 4)) is None | |
| rep = {"self_test": "video_io", "unit_checks": "ok"} | |
| if uid is None: | |
| # smallest local TRAIN route with a pose cache | |
| cands = [] | |
| for u in anchors_by_uid(("manual_train", "engaged_train")): | |
| if has_video(u) and pose_path(u) is not None: | |
| cands.append((video_path(u).stat().st_size, u)) | |
| if not cands: | |
| print(json.dumps(dict(rep, route="none local (video + pose cache); decode test skipped"))) | |
| return rep | |
| uid = min(cands)[1] | |
| boxes = load_driver_boxes(uid) | |
| W, H, n = video_info(video_path(uid)) | |
| ts = anchors_of(uid) | |
| spans_all = anchor_spans(uid, dur=n / FPS) | |
| # decode only the first anchor's window (a smoke test, not a route pass) | |
| span = spans_from_anchors(ts[:1], HIST_S, POST_S, dur=n / FPS) | |
| fs = FrameStream(uid, span, workers=workers, decode_wh=DEFAULT_DECODE_WH, block=50) | |
| n_box = n_crop = 0 | |
| last_fi = -1 | |
| shapes = set() | |
| for t_, fi_, fr_, bx_ in fs: | |
| assert fi_ > last_fi and fi_ == int(round(t_ * FPS)) and fi_ % 2 == 0 | |
| last_fi = fi_ | |
| if fr_ is not None: | |
| shapes.add(fr_.shape) | |
| if np.isfinite(bx_).all(): | |
| n_box += 1 | |
| cr = driver_crop(fr_, bx_) | |
| n_crop += cr is not None | |
| rep.update(route=uid, leaf=leaf_of(uid), native_wh=[W, H], n_video_frames=n, n_anchors=int(len(ts)), n_spans_full_route=int(len(spans_all)), | |
| span_seconds_full_route=float(np.sum(spans_all[:, 1] - spans_all[:, 0])), pose_source=(boxes or {}).get("source"), pose_frames=(0 if boxes is None else int(len(boxes["t"]))), | |
| smoke_span=span.tolist(), frames_requested=fs.stats["n_frames_requested"], frames_got=fs.stats["n_got"], frames_grabbed=fs.stats["n_grab"], frame_shapes=[list(s) for s in shapes], | |
| frames_with_box=n_box, crops=n_crop, decode_worker_fps=round(fs.stats["n_grab"] / max(fs.stats["decode_worker_s"], 1e-9), 1), wall_s=round(fs.stats["wall_s"], 2), workers=workers) | |
| assert fs.stats["n_got"] == fs.stats["n_frames_requested"] > 0 and shapes == {(DEFAULT_DECODE_WH[1], DEFAULT_DECODE_WH[0], 3)} | |
| rep["decode_meta"] = fs.decode_meta() | |
| # index-exactness smoke check: decode_block on frames 100 .. 118 (native) vs a continuous decode from frame 0 | |
| import cv2 | |
| targets = list(range(100, 120, 2)) | |
| res = decode_block((str(video_path(uid)), targets, W, H, fs.native_fps)) | |
| cap = cv2.VideoCapture(str(video_path(uid))) | |
| ref = {} | |
| for i in range(targets[-1] + 1): | |
| ok, fr = cap.read() | |
| assert ok, i | |
| if i in targets: | |
| ref[i] = fr | |
| cap.release() | |
| diffs = [int(np.abs(res["frames"][j].astype(np.int16) - ref[f].astype(np.int16)).max()) for j, f in enumerate(targets)] | |
| rep["index_exact_smoke"] = {"targets": targets, "decoder": res["decoder"], "max_abs_diff": max(diffs), "n_seek_corrected": res["n_seek_corrected"], "landing_offsets": res["landing_offsets"]} | |
| assert res["got"].all() and max(diffs) == 0, diffs | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return rep | |
| # ----------------------------------------------------------------------------- index-exactness verification (Phase-1 evidence) | |
| VERIFY_DEFAULT_UIDS = ("driver_139__route_24__route_0-8", "driver_139__route_40__route_0-8", | |
| "driver_139__route_45__route_0-37", "driver_86__route_2__route_0-8") | |
| def _plain_seek_offset(vpath, f, ref, search=8): | |
| """which continuous-decode frame does the PLAIN cv2 seek (CAP_PROP_POS_FRAMES = f, then read) deliver? -> delivered - f, or None when not within +-search of f.""" | |
| import cv2 | |
| cap = cv2.VideoCapture(str(vpath)) | |
| cap.set(cv2.CAP_PROP_POS_FRAMES, int(f)) | |
| ok, fr = cap.read() | |
| cap.release() | |
| if not ok: | |
| return None | |
| for d in sorted(range(-search, search + 1), key=abs): | |
| g = f + d | |
| if g in ref and np.array_equal(fr, ref[g]): | |
| return d | |
| return None | |
| def verify_index_exact(uids=VERIFY_DEFAULT_UIDS, manifest=PHASE1_DIR / "manifest_v1.csv", out_json=PHASE1_DIR / "decode_exactness_check.json", n_plain=6, log=print): | |
| """per route: decode_block (exactly as the workers run it, native size, header fps of the route) on ALL 10 Hz frames of its manifest windows vs ONE continuous | |
| cv2 decode from frame 0; records max |pixel diff| (must be 0), the decoder used, the corrections applied and the offset of the PLAIN cv2 seek on n_plain frames.""" | |
| import csv | |
| import cv2 | |
| rows = list(csv.DictReader(open(manifest))) | |
| rep = {"written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "manifest": str(manifest), "routes": []} | |
| for uid in uids: | |
| wins = [[float(r["win_start"]), float(r["win_end"])] for r in rows if r["uid"] == uid] | |
| vp = video_path(uid) | |
| W, H, n = video_info(vp) | |
| fps = video_header_fps(vp) | |
| k, fis, _ = frames_in_spans(wins, HZ, n) | |
| fis = [int(x) for x in fis] | |
| sample = sorted({fis[int(round(x))] for x in np.linspace(0, len(fis) - 1, n_plain)}) | |
| t0 = time.time() | |
| res = decode_block((str(vp), fis, W, H, fps)) | |
| t_block = time.time() - t0 | |
| want = set(fis) | |
| keep_near = set() | |
| for s in sample: | |
| keep_near.update(range(s - 8, s + 9)) | |
| cap = cv2.VideoCapture(str(vp)) | |
| ref = {} | |
| t0 = time.time() | |
| for i in range(fis[-1] + 9): | |
| ok, fr = cap.read() | |
| if not ok: | |
| break | |
| if i in want or i in keep_near: | |
| ref[i] = fr | |
| cap.release() | |
| t_cont = time.time() - t0 | |
| diffs = [int(np.abs(res["frames"][j].astype(np.int16) - ref[f_].astype(np.int16)).max()) if f_ in ref else None for j, f_ in enumerate(fis)] | |
| plain = {int(s): _plain_seek_offset(vp, s, ref) for s in sample} | |
| r = {"uid": uid, "native_wh": [W, H], "n_video_frames": n, "header_fps": round(fps, 5), "windows": wins, "n_frames_checked": len(fis), "frame_idx_range": [fis[0], fis[-1]], | |
| "decoder": res["decoder"], "exact_flag": res["exact"], "got_all": bool(res["got"].all()), | |
| "max_abs_pixel_diff": (max(d for d in diffs if d is not None) if any(d is not None for d in diffs) else None), "n_frames_without_reference": sum(d is None for d in diffs), | |
| "n_seek_corrected": res["n_seek_corrected"], "landing_offsets_corrected": res["landing_offsets"][:10], "n_gap_reseeks": res["n_gap_reseeks"], | |
| "plain_cv2_seek_offset_by_frame": plain, "decode_block_s": round(t_block, 2), "continuous_decode_s": round(t_cont, 2), "note": res["note"]} | |
| rep["routes"].append(r) | |
| log(json.dumps({k_: r[k_] for k_ in ("uid", "header_fps", "decoder", "n_frames_checked", "max_abs_pixel_diff", "n_seek_corrected", "plain_cv2_seek_offset_by_frame", "decode_block_s")}, default=str)) | |
| rep["all_exact"] = all(r["max_abs_pixel_diff"] == 0 and r["got_all"] and r["n_frames_without_reference"] == 0 for r in rep["routes"]) | |
| Path(out_json).parent.mkdir(parents=True, exist_ok=True) | |
| Path(out_json).write_text(json.dumps(rep, indent=1, default=str)) | |
| log(f"all_exact={rep['all_exact']} -> {out_json}") | |
| return rep | |
| if __name__ == "__main__": | |
| import argparse | |
| sys.path.insert(0, str(Path(__file__).resolve().parent)) | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--self-test", action="store_true") | |
| ap.add_argument("--uid", default=[], nargs="*") | |
| ap.add_argument("--workers", type=int, default=2) | |
| ap.add_argument("--missing-leaves", default="", choices=["", *TIERS]) | |
| ap.add_argument("--fetch-dir", default=str(FETCH_DIR)) | |
| ap.add_argument("--verify-exact", action="store_true", help="decode_block vs a continuous decode from frame 0 on the manifest windows (default: 3 Tucson fps != 20 routes + 1 fps == 20 route)") | |
| a = ap.parse_args() | |
| uids = [u for u in a.uid if u] | |
| if a.verify_exact: | |
| verify_index_exact(tuple(uids) if uids else VERIFY_DEFAULT_UIDS) | |
| if a.missing_leaves: | |
| cli_missing_leaves(a.missing_leaves, Path(a.fetch_dir)) | |
| if a.self_test or not (a.missing_leaves or a.verify_exact): | |
| _self_test(uids[0] if uids else None, a.workers) | |
Xet Storage Details
- Size:
- 45.8 kB
- Xet hash:
- b12e423a6705493541d6f3342ccc683796c31bead5ca87634e2fe4d5033177da
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.