Buckets:
| """cwm3/baselines_dms/extractors/videomae2.py — O4: VideoMAEv2 distilled ViT-S/16 (K710, FROZEN) window embeddings of the driver crop (plan §3, 2026-09-14). | |
| Model registry id `videomaev2` = vit_s_k710_dl_from_giant.pth (sha256 24fb7168…, 44,334,609 B; HF OpenGVLab/VideoMAE2@706cc172 distill/), loaded strictly | |
| into the vendored definition baselines_dms/vendor/videomaev2_model.py (upstream models/modeling_finetune.py @29eab1e8). ViT-S/16, tubelet 2, | |
| 16 frames, embed 384, depth 12, heads 6. Feature = forward_features() = fc_norm(mean over the 8 x 14 x 14 tokens) -> 384-d, i.e. the pre-logit | |
| feature; the K710 head is never used. Forward in fp16 on cuda:0; no fine-tuning, no gradient. | |
| Windows the 10 Hz frame grid (t = k / 10, frame_idx = 2 k) is decoded once per route; for EVERY 5 Hz grid time t_end (k even) inside the requested spans | |
| the clip is the L = 16 frames at 5 Hz ending at t_end: t_end - 3.0, -2.8, ..., t_end (3.2 s causal window; every second 10 Hz frame). A LEAD-IN of | |
| 3.0 s before each requested span is decoded (causal: earlier frames only) so that windows ending near the span start are complete; --no-lead-in | |
| disables it. win_valid <=> all 16 frames decoded AND a (non-degenerate) driver box on every one of them. | |
| Input per frame: video_io.driver_crop (causal route box, +10 % per side) of the frame decoded at an aspect-preserving 600-px-high size (958x600 for | |
| 1928x1208, 1062x600 for 1344x760, 800x600 for 816x612; INTER_AREA), letterboxed to 224 x 224 (long side -> 224, INTER_AREA down / INTER_LINEAR up, | |
| centred, padding = ImageNet mean colour RGB (124, 116, 104) so it normalises to ~0), BGR -> RGB, / 255, then (x - mean) / std with the ImageNet | |
| mean (0.485, 0.456, 0.406) / std (0.229, 0.224, 0.225) used by the VideoMAEv2 repo (dataset/kinetics.py: ClipToTensor + Normalize). Tensor | |
| [B, 3, 16, 224, 224] fp16. | |
| Caches WINDOW cache <out>/windows/<uid>.npz (cache_schema.WindowCache: win_t_end f32[M], emb f16[M,384], win_valid bool[M], win_len 16; the roles | |
| spec of the baseline is {"emb": "embedding"}; summarise.summarise_windows consumes it) and a PER-FRAME cache <out>/cache/<uid>.npz | |
| (cache_schema.RouteCache, 10 Hz, over the decoded spans incl. lead-in) whose channels are cheap crop-geometry descriptors (role none): | |
| vmae2_crop_w / vmae2_crop_h (driver crop size in DECODED px, meta.decode_wh) and vmae2_lb_scale (224 / long side). Its valid / fail codes are the | |
| per-frame availability that the windows are built from (fail 1 no box / 2 no frame). | |
| No openpilot field is read; no BATON frame leaves the machine; no network (weights come from the registry, sha256-asserted). | |
| python baselines_dms/extractors/videomae2.py --self-test [--uid UID] # 1 local TRAIN route, 30 frames (+ lead-in): shapes, validity, determinism, skip | |
| python baselines_dms/extractors/videomae2.py --phase1 [--workers 3 --batch 32 --force --no-lead-in] | |
| -> results/baselines_dms/phase1/videomae2/{cache,windows}/<uid>.npz, measurements.json, contact_sheet_sign.png, README.md | |
| python baselines_dms/extractors/videomae2.py --uids <file> --out-dir <dir> [--packs ... --hist-s 8.2 --post-s 0.2] # anchor-span route runner (later phases) | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import resource | |
| import subprocess | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| HERE = Path(__file__).resolve().parent | |
| PKG = HERE.parent | |
| CWM3 = PKG.parent | |
| for p in (str(CWM3), str(PKG), str(HERE)): | |
| if p not in sys.path: | |
| sys.path.insert(0, p) | |
| import cache_schema as cs # noqa: E402 | |
| import video_io as vio # noqa: E402 | |
| import base # noqa: E402 (extractors/base.py: Extractor ABC, fail_codes_for, EXTRACTORS) | |
| BASELINE = "videomae2" | |
| REGISTRY_ID = "videomaev2" | |
| PREFIX = "vmae2_" | |
| EMB_CH = "emb" | |
| D_EMB = 384 | |
| L = 16 # frames per window | |
| WIN_HZ = 5 # window frame rate | |
| STEP10 = cs.HZ // WIN_HZ # = 2 : 10 Hz steps between window frames | |
| WIN_SPAN_S = (L - 1) / WIN_HZ # = 3.0 s between the first and the last frame of a window | |
| IMG = 224 | |
| MEAN_RGB = (0.485, 0.456, 0.406) # VideoMAEv2 repo dataset/kinetics.py -> video_transforms.Normalize(mean, std) after ClipToTensor (/255) | |
| STD_RGB = (0.229, 0.224, 0.225) | |
| PAD_RGB = tuple(int(round(m * 255)) for m in MEAN_RGB) # (124, 116, 104): letterbox padding normalises to ~0 | |
| DECODE_H = 600 # aspect-preserving decode height (crop long side stays >= 224 for every native size) | |
| OUT_PHASE1 = cs.RESULTS_ROOT / "phase1" / BASELINE | |
| MANIFEST = cs.RESULTS_ROOT / "phase1" / "manifest_v1.csv" | |
| FACE_PACK_DIR = CWM3 / "results" / "targets_face_causal_routes_v2" | |
| WIN_FAIL = {0: "ok", 1: "no_box", 2: "no_frame", 4: "non_finite", 5: "not_decoded"} # per-window reasons (5 = a needed frame lies before the decoded spans / the video) | |
| # ----------------------------------------------------------------------------- preprocessing | |
| def decode_wh_for(native_wh, h=DECODE_H): | |
| """aspect-preserving decode size with even width and height `h` (never upsamples beyond native).""" | |
| W, H = int(native_wh[0]), int(native_wh[1]) | |
| h = min(h, H) | |
| w = int(round(W * h / H)) | |
| w += w % 2 | |
| return (w, h) | |
| def letterbox_rgb(crop_bgr, size=IMG, pad_rgb=PAD_RGB): | |
| """u8 BGR crop -> (u8 RGB [size, size, 3] letterboxed, scale). Long side -> size, centred, padding = ImageNet mean colour.""" | |
| import cv2 | |
| h, w = crop_bgr.shape[:2] | |
| s = size / float(max(h, w)) | |
| nw, nh = max(1, int(round(w * s))), max(1, int(round(h * s))) | |
| r = cv2.resize(crop_bgr, (nw, nh), interpolation=(cv2.INTER_AREA if s < 1.0 else cv2.INTER_LINEAR)) | |
| out = np.empty((size, size, 3), np.uint8) | |
| out[:, :, 0], out[:, :, 1], out[:, :, 2] = pad_rgb[2], pad_rgb[1], pad_rgb[0] # BGR canvas | |
| x0, y0 = (size - nw) // 2, (size - nh) // 2 | |
| out[y0:y0 + nh, x0:x0 + nw] = r | |
| return np.ascontiguousarray(out[:, :, ::-1]), float(s) # -> RGB | |
| # ----------------------------------------------------------------------------- extractor | |
| class VideoMAE2Extractor(base.Extractor): | |
| """frozen VideoMAEv2 ViT-S window embeddings (infer_windows) + per-frame crop geometry (infer, the base per-frame contract).""" | |
| baseline = BASELINE | |
| channels = [PREFIX + "crop_w", PREFIX + "crop_h", PREFIX + "lb_scale"] | |
| model_ids = [REGISTRY_ID] | |
| weight_sha256 = {} | |
| lib_versions = {} | |
| needs_crop = True | |
| decode_wh = decode_wh_for((1928, 1208)) # default; the window runner picks decode_wh_for(native_wh) per route | |
| batch = 32 # windows per forward | |
| emb_channel = EMB_CH | |
| D = D_EMB | |
| win_len = L | |
| win_hz = WIN_HZ | |
| device = "cuda:0" | |
| def spec(self): | |
| """channel roles for summarise.SPECS (embedding | none).""" | |
| return {EMB_CH: "embedding", **{c: "none" for c in self.channels}} | |
| def setup(self, registry_entry=None, device="cuda:0"): | |
| import cv2 | |
| import timm | |
| import torch | |
| from baselines_dms import registry as R | |
| sys.path.insert(0, str(PKG / "vendor")) | |
| import videomaev2_model as vm | |
| t0 = time.time() | |
| entry = registry_entry if registry_entry is not None else R.get(REGISTRY_ID) | |
| path = R.resolve(entry["id"]) # sha256 of the checkpoint asserted here | |
| self.registry_entry = {k: entry.get(k) for k in ("id", "name", "version", "url", "local_path", "sha256", "size_bytes", "license")} | |
| self.weight_sha256 = {Path(entry["local_path"]).name: entry["sha256"]} | |
| self.lib_versions = {"torch": torch.__version__, "timm": timm.__version__, "opencv": cv2.__version__, "numpy": np.__version__} | |
| self.device = device | |
| m = vm.vit_small_patch16_224(num_classes=710, all_frames=L, tubelet_size=2) | |
| ck = torch.load(path, map_location="cpu", weights_only=False) | |
| res = m.load_state_dict(ck["module"], strict=True) | |
| self.load_result = str(res) | |
| self.n_params = int(sum(p.numel() for p in m.parameters())) | |
| self.model = m.to(device).half().eval() | |
| for p in self.model.parameters(): | |
| p.requires_grad_(False) | |
| self._mean = torch.tensor(MEAN_RGB, dtype=torch.float16, device=device).view(1, 3, 1, 1, 1) | |
| self._std = torch.tensor(STD_RGB, dtype=torch.float16, device=device).view(1, 3, 1, 1, 1) | |
| self.infer_windows(np.zeros((1, L, IMG, IMG, 3), np.uint8)) # warm-up (CUDA context / kernels) so that the measured per-window ms exclude it | |
| torch.cuda.synchronize() if str(device).startswith("cuda") else None | |
| self.load_s = time.time() - t0 | |
| return self | |
| def teardown(self): | |
| import torch | |
| self.model = None | |
| torch.cuda.empty_cache() | |
| # per-frame contract (base.run_route): crop geometry only — the embedding needs the window runner below | |
| def infer(self, crops): | |
| out = np.full((len(crops), self.C), np.nan, np.float32) | |
| for i, c in enumerate(crops): | |
| if c is None or c.size == 0: | |
| continue | |
| h, w = c.shape[:2] | |
| out[i] = [w, h, IMG / float(max(h, w))] | |
| return out | |
| def infer_windows(self, clips): | |
| """clips u8 RGB [B, L, 224, 224, 3] (letterboxed) -> f32 [B, 384] fc_norm(mean-pooled tokens); fp16 forward; NaN-free unless the model overflowed.""" | |
| import torch | |
| x = torch.from_numpy(np.ascontiguousarray(clips)).to(self.device, non_blocking=True) | |
| with torch.inference_mode(): | |
| x = x.permute(0, 4, 1, 2, 3).half().div_(255.0) # B, 3, T, H, W in [0, 1] | |
| x = (x - self._mean) / self._std | |
| f = self.model.forward_features(x) | |
| return f.float().cpu().numpy() | |
| base.EXTRACTORS[BASELINE] = VideoMAE2Extractor | |
| # ----------------------------------------------------------------------------- window runner (one route, decoded once) | |
| def merge64(spans, snap_tol=1e-6): | |
| """cache_schema.merge_spans semantics in float64 (merge_spans returns f32, whose rounding can shift a 0.2 s grid boundary by one step). Span bounds within | |
| `snap_tol` s of the 10 Hz grid are snapped onto it: video_io.frames_in_spans decodes up to ceil(end * 10), so an end carrying float noise such as | |
| 650.8000000000001 (= t_anchor + 0.2 in the manifest) would otherwise decode one frame AFTER the requested span end (verification fix, 2026-09-14).""" | |
| a = np.asarray(spans, np.float64).reshape(-1, 2) | |
| if len(a) == 0: | |
| return np.zeros((0, 2), np.float64) | |
| k = np.round(a * cs.HZ) | |
| a = np.where(np.abs(k / cs.HZ - a) < snap_tol, k / cs.HZ, a) | |
| a = a[np.argsort(a[:, 0], kind="stable")] | |
| out = [list(a[0])] | |
| for s_, e_ in a[1:]: | |
| if s_ <= out[-1][1]: | |
| out[-1][1] = max(out[-1][1], e_) | |
| else: | |
| out.append([s_, e_]) | |
| return np.asarray(out, np.float64) | |
| def window_grid(spans, n_video_frames, tol=1e-6): | |
| """5 Hz window END steps k5 with k5 / 5 INSIDE `spans` (tolerance 1e-6 s; a 0.2 s-grid span [T-10, T+0.2] gives exactly 52 windows) and 4 k5 < n_video_frames | |
| -> (k5 i64[M], t_end f32[M]); the window at k5 uses the 10 Hz steps 2 k5 - 2 i, i = 0..L-1.""" | |
| ks = [np.arange(int(np.ceil(a * WIN_HZ - tol)), int(np.floor(b * WIN_HZ + tol)) + 1) for a, b in np.asarray(spans, np.float64).reshape(-1, 2)] | |
| k5 = np.unique(np.concatenate(ks)) if ks else np.zeros(0, np.int64) | |
| k5 = k5[(k5 >= 0) & (k5 * (cs.FPS // WIN_HZ) < int(n_video_frames))] | |
| return k5.astype(np.int64), (k5 / WIN_HZ).astype(np.float32) | |
| def lead_in_spans(spans, n_video_frames, lead_in=True): | |
| """decode spans (float64): each requested span extended CAUSALLY so that the first window inside it has its 16 frames decoded (3.0 s lead-in, aligned to the | |
| grid) and its end covers the last window's frame; lead_in=False keeps the requested spans (the first 15 windows of a span are then incomplete).""" | |
| spans = merge64(spans) | |
| if len(spans) == 0: | |
| return spans | |
| out = [] | |
| for a, b in spans: | |
| k5, _ = window_grid([[a, b]], n_video_frames) | |
| if len(k5) == 0: | |
| out.append([a, b]) | |
| continue | |
| a_ext = max(0.0, (STEP10 * (int(k5[0]) - (L - 1))) / cs.HZ) if lead_in else a | |
| out.append([min(a, a_ext), max(b, (STEP10 * int(k5[-1])) / cs.HZ)]) | |
| return merge64(out) | |
| class _StreamChain: | |
| """one video_io.FrameStream PER contiguous decode span, iterated in order (video_io.decode_block grabs every frame between the first and the last index of a | |
| block, so a block straddling two spans would decode the whole gap between two anchors); exposes the concatenated t / frame_idx / box / moving and summed stats.""" | |
| def __init__(self, uid, spans, hz, workers, decode_wh, block, vpath=None): | |
| self.streams = [vio.FrameStream(uid, [sp], hz=hz, workers=workers, decode_wh=decode_wh, block=block, vpath=vpath, box_coords="frame") for sp in np.asarray(spans, np.float64).reshape(-1, 2)] | |
| f0 = self.streams[0] | |
| self.vpath, self.native_wh, self.decode_wh, self.n_frames, self.boxes = f0.vpath, f0.native_wh, f0.decode_wh, f0.n_frames, f0.boxes | |
| self.native_fps = f0.native_fps | |
| self.t = np.concatenate([f.t for f in self.streams]) if self.streams else np.zeros(0, np.float32) | |
| self.frame_idx = np.concatenate([f.frame_idx for f in self.streams]) | |
| self.box = np.concatenate([f.box for f in self.streams]) | |
| self.moving = np.concatenate([f.moving for f in self.streams]) | |
| assert np.all(np.diff(self.frame_idx) > 0), "decode spans must be disjoint and sorted" | |
| def __iter__(self): | |
| for f in self.streams: | |
| yield from f | |
| def stats(self): | |
| return {"decode_ms_per_frame": sum((f.stats["decode_ms_per_frame"] for f in self.streams), []), "n_grab": sum(f.stats["n_grab"] for f in self.streams), | |
| "decode_worker_s": sum(f.stats["decode_worker_s"] for f in self.streams), "n_got": sum(f.stats["n_got"] for f in self.streams)} | |
| def decode_meta(self): | |
| """video_io.FrameStream.decode_meta aggregated over the per-span streams (native_fps, seek_corrected, decode_stats).""" | |
| ms = [f.decode_meta() for f in self.streams] | |
| ds = [m["decode_stats"] for m in ms] | |
| return {"native_fps": float(self.native_fps), "seek_corrected": True, | |
| "decode_stats": {"decoder": sorted({d for x in ds for d in x["decoder"]}), "n_seek_corrected": int(sum(x["n_seek_corrected"] for x in ds)), | |
| "max_landing_offset": int(max([x["max_landing_offset"] for x in ds] or [0])), "landing_offsets": sum((x["landing_offsets"] for x in ds), [])[:20], | |
| "n_gap_reseeks": int(sum(x["n_gap_reseeks"] for x in ds)), "decode_exact": bool(all(x["decode_exact"] for x in ds)), | |
| "notes": sum((x["notes"] for x in ds), [])[:5], "n_grab": int(sum(x["n_grab"] for x in ds))}} | |
| class WindowAssembler: | |
| """streaming builder of the 16-frame 5 Hz clips — the loop body of run_route_windows factored out (2026-09-14) so that run_full_extraction.py can feed it frames | |
| from the shared decode. feed(k, frame_present, crop) once per 10 Hz step k in causal order (crop = video_io.driver_crop of the DECODED frame, None when no | |
| box / no frame); windows end at the 5 Hz steps `k5` (window_grid) and use the 10 Hz steps 2 k5 - 2 i, i = 0..L-1; a step of the clip that was never fed counts | |
| as not decoded (win fail 5). finish() -> (emb f32[M,D] NaN on invalid, win_valid, win_fail, infer_ms, pre_ms) with the fp16-overflow rule of run_route_windows.""" | |
| def __init__(self, ex, k5, batch=None): | |
| self.ex, self.batch = ex, int(batch or ex.batch) | |
| self.k5 = np.asarray(k5, np.int64) | |
| self.M = int(len(self.k5)) | |
| self.k_end_to_j = {int(STEP10 * k): j for j, k in enumerate(self.k5)} | |
| self.emb = np.full((self.M, ex.D), np.nan, np.float32) | |
| self.win_fail = np.full(self.M, 5, np.uint8) | |
| self.buf = {} # 10 Hz step k -> (status 0 ok / 1 no box / 2 no frame, u8 RGB [224,224,3] | None) | |
| self.pend_j, self.pend_clips = [], [] | |
| self.infer_ms, self.pre_ms = [], [] | |
| def flush(self): | |
| if not self.pend_j: | |
| return | |
| import torch | |
| t0 = time.time() | |
| rows = self.ex.infer_windows(np.stack(self.pend_clips)) | |
| torch.cuda.synchronize() if str(self.ex.device).startswith("cuda") else None | |
| self.infer_ms.append(1000.0 * (time.time() - t0) / len(self.pend_j)) | |
| self.emb[np.asarray(self.pend_j)] = rows | |
| self.pend_j.clear() | |
| self.pend_clips.clear() | |
| def feed(self, k, frame_present, crop): | |
| if k % STEP10: # 5 Hz grid only: preprocess once, reuse in up to L windows | |
| return | |
| if not frame_present: | |
| self.buf[k] = (2, None) | |
| elif crop is None: | |
| self.buf[k] = (1, None) | |
| else: | |
| t0 = time.time() | |
| pre, _ = letterbox_rgb(crop) | |
| self.pre_ms.append(1000.0 * (time.time() - t0)) | |
| self.buf[k] = (0, pre) | |
| j = self.k_end_to_j.get(k) | |
| if j is not None: | |
| need = [k - STEP10 * q for q in range(L - 1, -1, -1)] # oldest .. newest | |
| st = [self.buf.get(kk, (5, None))[0] for kk in need] | |
| code = max(st) | |
| self.win_fail[j] = code | |
| if code == 0: | |
| self.pend_j.append(j) | |
| self.pend_clips.append(np.stack([self.buf[kk][1] for kk in need])) | |
| if len(self.pend_j) >= self.batch: | |
| self.flush() | |
| for kk in [kk for kk in self.buf if kk < k - STEP10 * (L - 1)]: | |
| del self.buf[kk] | |
| def finish(self): | |
| self.flush() | |
| win_valid = self.win_fail == 0 | |
| emb = self.emb | |
| emb[~win_valid] = np.nan | |
| nonfin = win_valid & ~np.isfinite(emb).all(1) | |
| if nonfin.any(): # fp16 overflow etc.: the window is not usable | |
| win_valid[nonfin] = False | |
| self.win_fail[nonfin] = 4 | |
| emb[nonfin] = np.nan | |
| return emb, win_valid, self.win_fail, self.infer_ms, self.pre_ms | |
| def _children_rss_mb(): | |
| """current RSS (MB) of the decode worker processes: (max single child, total) — sampled, since RUSAGE_CHILDREN.ru_maxrss inherits the parent's RSS at fork.""" | |
| try: | |
| import psutil | |
| v = [c.memory_info().rss / 2**20 for c in psutil.Process().children(recursive=True)] | |
| return (max(v) if v else 0.0, sum(v)) | |
| except Exception: | |
| return (float("nan"), float("nan")) | |
| def run_route_windows(ex: VideoMAE2Extractor, uid, spans, cache_dir, window_dir, workers=3, block=50, lead_in=True, force=False, log=print, vpath=None): | |
| """decode the route's spans once (+ lead-in; one stream per contiguous span), write <cache_dir>/<uid>.npz (RouteCache, per-frame geometry / availability) and | |
| <window_dir>/<uid>.npz (WindowCache, 384-d embedding per 5 Hz window end inside `spans`). -> (code 0 done | 4 failed, {"cache": path, "windows": path}, info).""" | |
| import cv2 | |
| import torch | |
| t_route = time.time() | |
| cache_dir, window_dir = Path(cache_dir), Path(window_dir) | |
| p_rc, p_wc = cache_dir / f"{uid}.npz", window_dir / f"{uid}.npz" | |
| spans = merge64(spans) | |
| info = {"uid": uid, "status": "", "n_frames": 0, "n_windows": 0, "wall_s": 0.0} | |
| paths = {"cache": p_rc, "windows": p_wc} | |
| if p_rc.exists() and p_wc.exists() and not force: | |
| try: | |
| rc, wc = cs.RouteCache.load(p_rc), cs.WindowCache.load(p_wc) | |
| if rc.channels == list(ex.channels) and wc.D == ex.D and wc.covers(spans) and rc.covers(spans): | |
| info.update(status="exists", n_frames=rc.n, n_windows=wc.m, wall_s=round(time.time() - t_route, 2)) | |
| return base.EXIT_DONE, paths, info | |
| except Exception as e: | |
| log(f"{uid[:50]}: cannot read existing caches ({repr(e)[:80]}); recomputing") | |
| try: | |
| vp = Path(vpath) if vpath else vio.video_path(uid) | |
| W, H, n_video = vio.video_info(vp) | |
| except (FileNotFoundError, RuntimeError) as e: | |
| info.update(status="no_video", error=str(e)) | |
| return base.EXIT_FAIL, paths, info | |
| dec_spans = lead_in_spans(spans, n_video, lead_in) | |
| k5, t_end = window_grid(spans, n_video) | |
| M = len(k5) | |
| dwh = decode_wh_for((W, H)) | |
| try: | |
| fs = _StreamChain(uid, dec_spans, cs.HZ, workers, dwh, block, vpath) | |
| except FileNotFoundError as e: | |
| info.update(status="no_video", error=str(e)) | |
| return base.EXIT_FAIL, paths, info | |
| N = len(fs.frame_idx) | |
| feat = np.full((N, ex.C), np.nan, np.float32) | |
| had_frame = np.zeros(N, bool) | |
| had_box = np.isfinite(fs.box).all(1) | |
| asm = WindowAssembler(ex, k5) # the streaming clip builder (shared with run_full_extraction.py) | |
| rss_child_max, rss_child_tot = 0.0, 0.0 | |
| for i, (t_, fi_, fr_, bx_) in enumerate(fs): | |
| if i % 25 == 0: | |
| a_, b_ = _children_rss_mb() | |
| rss_child_max, rss_child_tot = max(rss_child_max, a_), max(rss_child_tot, b_) | |
| k = int(round(float(t_) * cs.HZ)) | |
| crop = None | |
| if fr_ is not None: | |
| had_frame[i] = True | |
| crop = vio.driver_crop(fr_, bx_) | |
| if crop is not None: | |
| h, w = crop.shape[:2] | |
| feat[i] = [w, h, IMG / float(max(h, w))] | |
| asm.feed(k, fr_ is not None, crop) | |
| emb, win_valid, win_fail, infer_ms, pre_ms = asm.finish() | |
| # per-frame cache (geometry / availability) | |
| fail = base.fail_codes_for(feat, had_frame, had_box, True) | |
| valid = fail == 0 | |
| feat[~valid] = np.nan | |
| code_sha = cs.code_sha256(ex.code_files()) | |
| extra = dict(pose_source=(fs.boxes or {}).get("source", "none"), needs_crop=True, crop_expand=vio.MARGIN, hz=cs.HZ, partial=False, video_path=str(fs.vpath), | |
| n_video_frames=int(fs.n_frames), registry_entry=getattr(ex, "registry_entry", None), lead_in_s=(WIN_SPAN_S if lead_in else 0.0), requested_spans=spans.tolist(), | |
| decode_h=DECODE_H, letterbox=f"long side -> {IMG}, centred, pad RGB {PAD_RGB}", colour="BGR->RGB, /255, ImageNet mean {} std {}".format(MEAN_RGB, STD_RGB), | |
| win_len=L, win_hz=WIN_HZ, win_span_s=WIN_SPAN_S, frame_stride_10hz=STEP10, emb_dim=ex.D, feature="forward_features = fc_norm(mean over tokens), pre-logit, fp16", | |
| n_params=getattr(ex, "n_params", None), win_fail_codes=WIN_FAIL, channel_roles=ex.spec(), **fs.decode_meta()) | |
| meta = cs.make_meta(ex.baseline, ex.model_ids, ex.weight_sha256, ex.lib_versions, fs.decode_wh, fs.native_wh, code_sha, "local", **extra) | |
| wall = time.time() - t_route | |
| runtime_f = cs.make_runtime(fs.stats["decode_ms_per_frame"], [], N, wall_s=wall) | |
| runtime_f.update(preprocess_ms_p50=float(np.percentile(pre_ms, 50)) if pre_ms else float("nan"), preprocess_ms_p95=float(np.percentile(pre_ms, 95)) if pre_ms else float("nan"), | |
| infer_unit="none (embeddings live in the window cache)") | |
| rc = cs.RouteCache(uid, fs.t[:N].astype(np.float32), fs.frame_idx[:N].astype(np.int64), fs.box[:N].astype(np.float32), valid, fail, feat.astype(np.float16), list(ex.channels), | |
| cs.merge_spans(dec_spans), meta, runtime_f, moving=fs.moving[:N].astype(bool)) | |
| rc.save(p_rc) | |
| runtime_w = cs.make_runtime(fs.stats["decode_ms_per_frame"], infer_ms, N, wall_s=wall) | |
| wfail_hist = {WIN_FAIL.get(int(a), str(int(a))): int(b) for a, b in zip(*np.unique(win_fail, return_counts=True))} | |
| runtime_w.update(infer_unit="window (ms per 16-frame window, batched)", n_windows=int(M), n_windows_valid=int(win_valid.sum()), batch=int(ex.batch), | |
| preprocess_ms_p50=runtime_f["preprocess_ms_p50"], preprocess_ms_p95=runtime_f["preprocess_ms_p95"], win_fail_hist=wfail_hist) | |
| wc = cs.WindowCache(uid, t_end.astype(np.float32), emb.astype(np.float16), win_valid, L, cs.merge_spans(spans), dict(meta, cache_kind="windows"), runtime_w) | |
| wc.save(p_wc) | |
| fail_hist = {cs.FAIL_CODES[int(a)]: int(b) for a, b in zip(*np.unique(fail, return_counts=True))} | |
| info.update(status="done", n_frames=int(N), n_frames_got=int(had_frame.sum()), n_frames_box=int(had_box.sum()), valid_rate=float(valid.mean()) if N else None, fail_hist=fail_hist, | |
| n_windows=int(M), n_windows_valid=int(win_valid.sum()), win_valid_rate=float(win_valid.mean()) if M else None, win_fail_hist=wfail_hist, | |
| decode_ms_per_frame=[float(v) for v in fs.stats["decode_ms_per_frame"]], infer_ms_per_window=[float(v) for v in infer_ms], preprocess_ms=[float(v) for v in pre_ms], | |
| decode_worker_fps=round(fs.stats["n_grab"] / max(fs.stats["decode_worker_s"], 1e-9), 1), n_grab=int(fs.stats["n_grab"]), decode_wh=list(fs.decode_wh), native_wh=list(fs.native_wh), | |
| rss_child_max_mb=round(rss_child_max, 1), rss_children_total_mb=round(rss_child_tot, 1), n_streams=len(fs.streams), | |
| spans_requested=spans.tolist(), spans_decoded=cs.merge_spans(dec_spans).tolist(), wall_s=round(wall, 2), runtime=runtime_w) | |
| return base.EXIT_DONE, paths, info | |
| def run_routes_windows(ex, uids, spans_by_uid, out_dir, workers=3, lead_in=True, force=False, log=print): | |
| out_dir = Path(out_dir) | |
| t0 = time.time() | |
| recs, fails = [], [] | |
| for u in uids: | |
| code, paths, info = run_route_windows(ex, u, spans_by_uid[u], out_dir / "cache", out_dir / "windows", workers=workers, lead_in=lead_in, force=force, log=log) | |
| log(f"{u[:60]} {info['status']} frames {info.get('n_frames')} windows {info.get('n_windows')} win_valid {None if info.get('win_valid_rate') is None else round(info['win_valid_rate'], 3)} wall {info.get('wall_s')}s") | |
| (fails if code == base.EXIT_FAIL else recs).append(info) | |
| summ = {"n_requested": len(uids), "n_done": len(recs), "n_failed": len(fails), "wall_s": round(time.time() - t0, 1), "fails": fails[:20], "baseline": ex.baseline} | |
| return (base.EXIT_FAIL if fails else base.EXIT_DONE), summ, recs | |
| # ----------------------------------------------------------------------------- resources | |
| def gpu_free_mb(): | |
| try: | |
| out = subprocess.run(["nvidia-smi", "--query-gpu=memory.free", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=20).stdout | |
| return float(out.strip().splitlines()[0]) | |
| except Exception: | |
| return float("nan") | |
| def mem_available_gb(): | |
| for line in open("/proc/meminfo"): | |
| if line.startswith("MemAvailable"): | |
| return float(line.split()[1]) / 1e6 | |
| return float("nan") | |
| def check_resources(need_vram_gb=3.0, margin_gb=4.0, need_ram_gb=28.0): | |
| """project rule: free VRAM >= need + 4 GB and MemAvailable >= 28 GB before any GPU load.""" | |
| free_mb, avail = gpu_free_mb(), mem_available_gb() | |
| ok = (free_mb >= (need_vram_gb + margin_gb) * 1024) and (avail >= need_ram_gb) | |
| rep = {"gpu_free_mb": free_mb, "mem_available_gb": round(avail, 1), "need_vram_gb": need_vram_gb + margin_gb, "need_ram_gb": need_ram_gb, "ok": bool(ok)} | |
| if not ok: | |
| raise SystemExit(f"resource check failed: {json.dumps(rep)}") | |
| return rep | |
| def peak_rss_mb(recs=()): | |
| """peak RSS (MB): main process from getrusage; decode workers from the RSS samples taken during the frame loops (max single worker, max total over workers).""" | |
| me = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0 | |
| cm = max([r.get("rss_child_max_mb", 0.0) for r in recs] + [0.0]) | |
| ct = max([r.get("rss_children_total_mb", 0.0) for r in recs] + [0.0]) | |
| return {"main_mb": round(me, 1), "decode_worker_max_mb_sampled": round(cm, 1), "decode_workers_total_mb_sampled": round(ct, 1)} | |
| # ----------------------------------------------------------------------------- phase 1 | |
| def _pq(v): | |
| v = np.asarray([x for x in v if np.isfinite(x)], np.float64) | |
| return {"p50": float(np.percentile(v, 50)), "p95": float(np.percentile(v, 95)), "n": int(len(v))} if len(v) else {"p50": None, "p95": None, "n": 0} | |
| def load_face_hist(regime, rows, uids, ts): | |
| """face pack rows -> dict(lm_yaw_dev f32[n,80] (NaN invalid), t_hist f64[n,80], lm0 f32[n]); alignment uid / t asserted (channel 1 = lm_yaw_dev).""" | |
| p = FACE_PACK_DIR / f"facehist_causal_{regime}_train.npz" | |
| with np.load(p, allow_pickle=True) as z: | |
| ch = [str(c) for c in z["channels"]] | |
| j = ch.index("lm_yaw_dev") | |
| rows = np.asarray(rows, np.int64) | |
| u, t = z["uid"][rows], z["t"][rows].astype(np.float64) | |
| assert np.array_equal(u.astype(str), np.asarray(uids).astype(str)) and np.allclose(t, ts, atol=1e-6), f"{p.name}: manifest rows not aligned with the face pack" | |
| Fh = z["Fh8_c"][rows][:, :, j].astype(np.float32) | |
| Fv = z["Fh8valid"][rows] | |
| F0 = z["F0_c"][rows][:, j].astype(np.float32) | |
| F0v = z["F0valid"][rows] | |
| row_ok = z["row_ok"][rows] | |
| Fh[~Fv] = np.nan | |
| F0[~F0v] = np.nan | |
| t_hist = ts[:, None] - (79 - np.arange(80))[None, :] / cs.HZ | |
| return {"lm_yaw_dev": Fh, "t_hist": t_hist, "lm0": F0, "row_ok": row_ok, "file": str(p)} | |
| def contact_sheet(items, out_png, ex, workers=1): | |
| """items: list of dict(uid, t_end, label_lines) -> 2 x 6 sheet of the letterboxed 224 crop the model saw at each window END frame.""" | |
| import cv2 | |
| tiles = [] | |
| for it in items: | |
| tile = np.full((IMG + 62, IMG, 3), 32, np.uint8) | |
| try: | |
| W, H, n = vio.video_info(vio.video_path(it["uid"])) | |
| fs = vio.FrameStream(it["uid"], [[it["t_end"], it["t_end"]]], hz=cs.HZ, workers=workers, decode_wh=decode_wh_for((W, H)), block=4, box_coords="frame") | |
| for t_, fi_, fr_, bx_ in fs: | |
| if abs(t_ - it["t_end"]) < 1e-3 and fr_ is not None: | |
| crop = vio.driver_crop(fr_, bx_) | |
| if crop is not None: | |
| pre, _ = letterbox_rgb(crop) | |
| tile[:IMG] = pre[:, :, ::-1] | |
| except Exception as e: | |
| it["label_lines"] = list(it["label_lines"]) + [f"decode err {str(e)[:18]}"] | |
| for r, line in enumerate(it["label_lines"][:4]): | |
| cv2.putText(tile, line, (3, IMG + 13 + 12 * r), cv2.FONT_HERSHEY_SIMPLEX, 0.36, (255, 255, 255), 1, cv2.LINE_AA) | |
| tiles.append(tile) | |
| while len(tiles) < 12: | |
| tiles.append(np.full((IMG + 62, IMG, 3), 32, np.uint8)) | |
| rows = [np.concatenate(tiles[r * 6:(r + 1) * 6], 1) for r in range(2)] | |
| sheet = np.concatenate(rows, 0) | |
| hdr = np.full((44, sheet.shape[1], 3), 16, np.uint8) | |
| cv2.putText(hdr, f"{BASELINE}: embedding model, no angle channel. Top row = 6 most NEGATIVE lm_yaw_dev (face pack) at the window end, bottom = 6 most POSITIVE.", (6, 16), | |
| cv2.FONT_HERSHEY_SIMPLEX, 0.42, (255, 255, 255), 1, cv2.LINE_AA) | |
| cv2.putText(hdr, "Printed: lm_yaw_dev (IOD-normalised, causal ref), PC1 score of the 384-d embedding (unsupervised, sign arbitrary), uid[:16] @ t_end. Tile = model input (224 letterbox, RGB->shown BGR).", (6, 34), | |
| cv2.FONT_HERSHEY_SIMPLEX, 0.38, (200, 200, 200), 1, cv2.LINE_AA) | |
| out = np.concatenate([hdr, sheet], 0) | |
| Path(out_png).parent.mkdir(parents=True, exist_ok=True) | |
| cv2.imwrite(str(out_png), out) | |
| return out_png | |
| def phase1(out_dir=OUT_PHASE1, manifest=MANIFEST, workers=3, batch=32, lead_in=True, force=False, log=print): | |
| import pandas as pd | |
| import torch | |
| from scipy.stats import spearmanr | |
| import summarise as SM | |
| torch.set_num_threads(int(os.environ.get("OMP_NUM_THREADS", "4"))) | |
| out_dir = Path(out_dir) | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| res = check_resources() | |
| log(f"resources ok: {json.dumps(res)}") | |
| Mf = pd.read_csv(manifest) | |
| assert len(Mf) == 50 and {"uid", "t_anchor", "win_start", "win_end", "regime", "y", "row"} <= set(Mf.columns), Mf.columns | |
| torch.cuda.reset_peak_memory_stats() | |
| ex = VideoMAE2Extractor() | |
| ex.batch = int(batch) | |
| ex.setup() | |
| vram_after_load = torch.cuda.memory_allocated() / 2**20 | |
| log(f"model loaded in {ex.load_s:.1f}s: {ex.load_result}; params {ex.n_params/1e6:.2f} M; VRAM {vram_after_load:.0f} MB") | |
| uids = list(dict.fromkeys(Mf.uid)) | |
| spans_by_uid = {u: merge64(Mf.loc[Mf.uid == u, ["win_start", "win_end"]].to_numpy(np.float64)) for u in uids} | |
| t_run = time.time() | |
| code, summ, recs = run_routes_windows(ex, uids, spans_by_uid, out_dir, workers=min(int(workers), 3), lead_in=lead_in, force=force, log=log) | |
| wall_run = time.time() - t_run | |
| torch.cuda.synchronize() | |
| vram = {"after_load_mb": round(vram_after_load, 1), "peak_allocated_mb": round(torch.cuda.max_memory_allocated() / 2**20, 1), "peak_reserved_mb": round(torch.cuda.max_memory_reserved() / 2**20, 1)} | |
| done = [r for r in recs if r["status"] == "done"] | |
| exists = [r for r in recs if r["status"] == "exists"] | |
| # ---------------- per-frame / per-window aggregates over the caches (works for fresh and pre-existing caches) | |
| fail_hist, wfail_hist = {}, {} | |
| n_frames = n_frames_box = n_valid = n_valid_box = 0 | |
| n_frames_in_win = n_valid_in_win = 0 | |
| n_win = n_win_valid = 0 | |
| dec_ms, inf_ms, pre_ms = [], [], [] | |
| E_all, T_all, U_all = [], [], [] | |
| for u in uids: | |
| rc = cs.RouteCache.load(out_dir / "cache" / f"{u}.npz") | |
| wc = cs.WindowCache.load(out_dir / "windows" / f"{u}.npz") | |
| for a, b in zip(*np.unique(rc.fail, return_counts=True)): | |
| fail_hist[cs.FAIL_CODES[int(a)]] = fail_hist.get(cs.FAIL_CODES[int(a)], 0) + int(b) | |
| hb = np.isfinite(rc.box).all(1) | |
| n_frames += rc.n | |
| n_frames_box += int(hb.sum()) | |
| n_valid += int(rc.valid.sum()) | |
| n_valid_box += int((rc.valid & hb).sum()) | |
| inw = np.zeros(rc.n, bool) | |
| for a, b in spans_by_uid[u]: | |
| inw |= (rc.t >= a - 1e-3) & (rc.t <= b + 1e-3) | |
| n_frames_in_win += int(inw.sum()) | |
| n_valid_in_win += int((rc.valid & inw).sum()) | |
| n_win += wc.m | |
| n_win_valid += int(wc.win_valid.sum()) | |
| for k_, v_ in (wc.runtime.get("win_fail_hist") or {}).items(): | |
| wfail_hist[k_] = wfail_hist.get(k_, 0) + int(v_) | |
| E_all.append(wc.emb.astype(np.float32)[wc.win_valid]) | |
| T_all.append(wc.win_t_end.astype(np.float64)[wc.win_valid]) | |
| U_all += [u] * int(wc.win_valid.sum()) | |
| for r in done: | |
| dec_ms += r["decode_ms_per_frame"] | |
| inf_ms += r["infer_ms_per_window"] | |
| pre_ms += r["preprocess_ms"] | |
| E = np.concatenate(E_all) if E_all else np.zeros((0, ex.D), np.float32) | |
| T = np.concatenate(T_all) if T_all else np.zeros(0) | |
| U = np.asarray(U_all) | |
| # ---------------- unsupervised structure of the embedding: PCA over all valid phase-1 windows (no labels) | |
| pca = {} | |
| pc1 = np.full(len(E), np.nan, np.float32) | |
| if len(E) > 10: | |
| mu = E.mean(0) | |
| Uu, S, Vt = np.linalg.svd(E - mu, full_matrices=False) | |
| v1 = Vt[0] * np.sign(Vt[0][np.argmax(np.abs(Vt[0]))]) # data-only sign convention: largest-|loading| dim positive | |
| pc1 = ((E - mu) @ v1).astype(np.float32) | |
| evr = (S ** 2) / np.sum(S ** 2) | |
| pca = {"n_windows": int(len(E)), "explained_var_ratio_pc1_3": [float(x) for x in evr[:3]], "emb_norm_mean": float(np.linalg.norm(E, axis=1).mean()), | |
| "emb_norm_std": float(np.linalg.norm(E, axis=1).std()), "sign_rule": "PC1 oriented so that its largest-|loading| embedding dim has a positive loading"} | |
| # distance of each window to its route's mean embedding (cosine) — a sign-free "atypical pose" proxy | |
| dist_route = np.full(len(E), np.nan, np.float32) | |
| for u in np.unique(U): | |
| m = U == u | |
| if m.sum() >= 2: | |
| mu_r = E[m].mean(0) | |
| En = E[m] / np.maximum(np.linalg.norm(E[m], axis=1, keepdims=True), 1e-6) | |
| dist_route[m] = 1.0 - En @ (mu_r / max(np.linalg.norm(mu_r), 1e-6)) | |
| # ---------------- face pack alignment: lm_yaw_dev at the window end times inside each anchor's 8 s history | |
| face = {r: load_face_hist(r, Mf.loc[Mf.regime == r, "row"].to_numpy(), Mf.loc[Mf.regime == r, "uid"].to_numpy(), Mf.loc[Mf.regime == r, "t_anchor"].to_numpy(np.float64)) | |
| for r in ("manual", "engaged")} | |
| pairs_pc1, pairs_lm, pairs_lm_mean, pairs_dist, pair_meta = [], [], [], [], [] | |
| per_anchor = [] | |
| win_by_uid = {u: cs.WindowCache.load(out_dir / "windows" / f"{u}.npz") for u in uids} | |
| for r in ("manual", "engaged"): | |
| sub = Mf[Mf.regime == r].reset_index(drop=True) | |
| fh = face[r] | |
| for i, a in sub.iterrows(): | |
| wc = win_by_uid[a.uid] | |
| ew = SM.summarise_windows(wc, [float(a.t_anchor)]) | |
| te = wc.win_t_end.astype(np.float64) | |
| rec = {"idx": int(a.idx), "uid": a.uid, "regime": r, "y": int(a.y), "t_anchor": float(a.t_anchor), "n_win_8s": int(ew["n_win_8s"][0]), | |
| "anchor_emb_valid": bool(np.isfinite(ew["emb_anchor"][0]).all()), "lm_yaw_dev_at_anchor": (None if not np.isfinite(fh["lm0"][i]) else float(fh["lm0"][i]))} | |
| if rec["anchor_emb_valid"] and len(E) > 10: | |
| ea = ew["emb_anchor"][0] | |
| j = int(np.searchsorted(te, float(a.t_anchor) + 1e-6, side="right")) - 1 | |
| while j >= 0 and not wc.win_valid[j]: | |
| j -= 1 | |
| rec["anchor_win_t_end"] = float(te[j]) if j >= 0 else None | |
| rec["pc1_anchor"] = float((ea - mu) @ v1) | |
| rec["emb_norm_anchor"] = float(np.linalg.norm(ea)) | |
| em = ew["emb_mean_8s"][0] | |
| rec["cos_anchor_vs_mean8s"] = float(np.dot(ea, em) / max(np.linalg.norm(ea) * np.linalg.norm(em), 1e-6)) if np.isfinite(em).all() else None | |
| per_anchor.append(rec) | |
| # window ends inside the anchor's 10 Hz history t - 7.9 .. t -> lm_yaw_dev at t_end (and mean over the 16 window frames) | |
| sel = np.flatnonzero(wc.win_valid & (te >= a.t_anchor - 7.9 - 1e-6) & (te <= a.t_anchor + 1e-6)) | |
| for j in sel: | |
| q = int(round((te[j] - (a.t_anchor - 7.9)) * cs.HZ)) | |
| if not (0 <= q < 80): | |
| continue | |
| lm = fh["lm_yaw_dev"][i, q] | |
| if not np.isfinite(lm): | |
| continue | |
| qs = [q - STEP10 * s for s in range(L) if q - STEP10 * s >= 0] | |
| lm_m = np.nanmean(fh["lm_yaw_dev"][i, qs]) if len(qs) else np.nan | |
| g = np.flatnonzero((U == a.uid) & (np.abs(T - te[j]) < 1e-3)) | |
| if len(g) != 1: | |
| continue | |
| pairs_pc1.append(float(pc1[g[0]])) | |
| pairs_dist.append(float(dist_route[g[0]])) | |
| pairs_lm.append(float(lm)) | |
| pairs_lm_mean.append(float(lm_m)) | |
| pair_meta.append((a.uid, float(te[j]), int(a.idx))) | |
| rho = {"n_pairs": len(pairs_lm), "note": "embedding model: no yaw channel. PC1 = first principal component of the 384-d embedding over all valid phase-1 windows (unsupervised, " | |
| "sign arbitrary — fixed by a data-only rule); dist_route = 1 - cos(emb, route mean emb). Pairs = valid windows ending inside an anchor's 8 s face-pack history " | |
| "with a finite lm_yaw_dev at the window end time; nothing is fitted to lm_yaw_dev or to y."} | |
| if len(pairs_lm) >= 10: | |
| P, Lm, Lmm, Dd = map(np.asarray, (pairs_pc1, pairs_lm, pairs_lm_mean, pairs_dist)) | |
| ok1 = np.isfinite(P) & np.isfinite(Lm) | |
| rho["spearman_pc1_vs_lm_yaw_dev_at_t_end"] = float(spearmanr(P[ok1], Lm[ok1])[0]) | |
| ok2 = np.isfinite(P) & np.isfinite(Lmm) | |
| rho["spearman_pc1_vs_lm_yaw_dev_window_mean"] = float(spearmanr(P[ok2], Lmm[ok2])[0]) | |
| rho["spearman_pc1_vs_abs_lm_yaw_dev"] = float(spearmanr(P[ok1], np.abs(Lm[ok1]))[0]) | |
| ok3 = np.isfinite(Dd) & np.isfinite(Lm) | |
| rho["spearman_dist_route_vs_abs_lm_yaw_dev"] = float(spearmanr(Dd[ok3], np.abs(Lm[ok3]))[0]) if ok3.sum() >= 10 else None | |
| rho["lm_yaw_dev_pairs_mean_std"] = [float(Lm[ok1].mean()), float(Lm[ok1].std())] | |
| # descriptive score distribution by y (never used for tuning): PC1 at anchor / cos to 8 s mean | |
| desc = {} | |
| pa = pd.DataFrame(per_anchor) | |
| if "pc1_anchor" in pa: | |
| for y_ in (0, 1): | |
| s = pa[pa.y == y_] | |
| desc[f"y{y_}"] = {"n": int(len(s)), "pc1_anchor_mean": float(s.pc1_anchor.mean()), "pc1_anchor_std": float(s.pc1_anchor.std()), | |
| "cos_anchor_vs_mean8s_mean": float(s.cos_anchor_vs_mean8s.dropna().mean()), "n_win_8s_mean": float(s.n_win_8s.mean())} | |
| # ---------------- contact sheet: 6 most negative / 6 most positive lm_yaw_dev window ends (one per uid where possible) | |
| sheet_path = None | |
| if len(pairs_lm) >= 12: | |
| order = np.argsort(np.asarray(pairs_lm)) | |
| def pick(idx_iter): | |
| seen, out = set(), [] | |
| for q in idx_iter: | |
| u_ = pair_meta[q][0] | |
| if u_ in seen: | |
| continue | |
| seen.add(u_) | |
| out.append(q) | |
| if len(out) == 6: | |
| break | |
| return out | |
| neg, pos = pick(order), pick(order[::-1]) | |
| items = [{"uid": pair_meta[q][0], "t_end": pair_meta[q][1], "label_lines": [f"lm_yaw_dev {pairs_lm[q]:+.3f}", f"PC1 {pairs_pc1[q]:+.2f} d_route {pairs_dist[q]:.3f}", | |
| f"{pair_meta[q][0][:16]}", f"t_end {pair_meta[q][1]:.1f}s idx {pair_meta[q][2]}"]} for q in neg + pos] | |
| sheet_path = str(contact_sheet(items, out_dir / "contact_sheet_sign.png", ex, workers=1)) | |
| rss = peak_rss_mb(done) | |
| meas = { | |
| "baseline": BASELINE, "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "manifest": str(manifest), "manifest_sha256": cs.sha256_file(manifest), "out_dir": str(out_dir), | |
| "args": {"workers": min(int(workers), 3), "batch": int(batch), "lead_in": bool(lead_in), "force": bool(force), "device": ex.device, "decode_h": DECODE_H}, | |
| "model": {"registry_id": REGISTRY_ID, "weight_sha256": ex.weight_sha256, "registry_entry": ex.registry_entry, "load_state_dict": ex.load_result, "n_params": ex.n_params, | |
| "load_s": round(ex.load_s, 2), "lib_versions": ex.lib_versions, "code_sha256": cs.code_sha256(ex.code_files()), "emb_dim": ex.D, "win_len": L, "win_hz": WIN_HZ, | |
| "win_span_s": WIN_SPAN_S, "channel_roles": ex.spec(), "per_frame_channels": list(ex.channels)}, | |
| "preprocessing": {"crop": "video_io.driver_crop(frame, causal route box, expand 0.10)", "decode": f"aspect-preserving, height {DECODE_H} px (INTER_AREA)", "letterbox": f"long side -> {IMG}, centred, pad RGB {PAD_RGB}", | |
| "colour": "BGR -> RGB, / 255", "normalise": {"mean": MEAN_RGB, "std": STD_RGB, "source": "VideoMAEv2 repo dataset/kinetics.py (ClipToTensor + Normalize, ImageNet)"}, "dtype": "fp16", "tensor": [None, 3, L, IMG, IMG]}, | |
| "routes": {"n_uids": len(uids), "n_done_now": len(done), "n_pre_existing": len(exists), "n_failed": summ["n_failed"], "fails": summ["fails"]}, | |
| "n_frames": int(n_frames), "n_frames_in_manifest_windows": int(n_frames_in_win), "n_frames_with_box": int(n_frames_box), | |
| "valid_share": {"per_frame_overall": float(n_valid / max(n_frames, 1)), "per_frame_on_frames_with_box": float(n_valid_box / max(n_frames_box, 1)), | |
| "per_frame_in_manifest_windows": float(n_valid_in_win / max(n_frames_in_win, 1)), "window_overall": float(n_win_valid / max(n_win, 1))}, | |
| "n_windows": int(n_win), "n_windows_valid": int(n_win_valid), "fail_hist_per_frame": fail_hist, "fail_hist_per_window": wfail_hist, | |
| "timing": {"decode_ms_per_frame": _pq(dec_ms), "infer_ms_per_window": _pq(inf_ms), "preprocess_ms_per_5hz_frame": _pq(pre_ms), "wall_s_routes_loop": round(wall_run, 1), | |
| "frames_per_s_end_to_end": float(sum(r["n_frames"] for r in done) / max(wall_run, 1e-9)), "windows_per_s_end_to_end": float(sum(r["n_windows"] for r in done) / max(wall_run, 1e-9)), | |
| "note": "end-to-end = decoded 10 Hz frames (incl. lead-in) / wall of the route loop (decode + crop + letterbox + forward + cache write); routes already cached are excluded"}, | |
| "memory": {"peak_rss": rss, "vram": vram, "resources_at_start": res}, | |
| "per_anchor": per_anchor, "descriptive_by_y": desc, "pca": pca, "rho": rho, | |
| "sign_convention": "n/a: 384-d embedding, no head / gaze angle channel. The contact sheet shows the model INPUT for the extreme lm_yaw_dev windows with PC1 printed; PC1 sign is a data-only convention.", | |
| "contact_sheet": sheet_path, "per_route": [{k: r[k] for k in ("uid", "status", "n_frames", "n_windows", "n_windows_valid", "win_fail_hist", "fail_hist", "wall_s", "decode_wh", "native_wh", "n_streams", "n_grab", "decode_worker_fps", "rss_child_max_mb") if k in r} for r in recs], | |
| } | |
| (out_dir / "measurements.json").write_text(json.dumps(meas, indent=1, default=str)) | |
| write_readme(meas, out_dir / "README.md") | |
| ex.teardown() | |
| log(f"phase1 done: frames {n_frames} windows {n_win} win_valid {n_win_valid} wall {wall_run:.1f}s -> {out_dir/'measurements.json'}") | |
| return meas | |
| def write_readme(m, path): | |
| t, v, rho, vr = m["timing"], m["valid_share"], m["rho"], m["memory"]["vram"] | |
| sha = m["model"]["weight_sha256"] | |
| f = lambda x, d=3: ("n/a" if x is None or (isinstance(x, float) and not np.isfinite(x)) else f"{x:.{d}f}") | |
| L_ = [f"# {BASELINE} — VideoMAEv2 distilled ViT-S/16 (K710) frozen window embeddings: Phase-1 integration run", "", | |
| f"Written by `baselines_dms/extractors/videomae2.py --phase1` at {m['utc']} on the frozen manifest `{Path(m['manifest']).name}` (sha256 {m['manifest_sha256'][:12]}…, 50 TRAIN anchors, windows [t−10 s, t+0.2 s]).", | |
| "Descriptive integration measurements only (runnability, validity, throughput, sign / alignment diagnostics); nothing is tuned on labels — the `y` column is used only for the descriptive score distribution.", "", | |
| "## Model", "", | |
| f"- registry id `{REGISTRY_ID}`: `{list(sha)[0]}` sha256 `{list(sha.values())[0][:16]}…` ({m['model']['registry_entry']['size_bytes']} B; {m['model']['registry_entry']['version']})", | |
| f"- definition: vendored `baselines_dms/vendor/videomaev2_model.py` (upstream models/modeling_finetune.py @29eab1e8, two timm import lines adapted); `vit_small_patch16_224(num_classes=710, all_frames=16, tubelet_size=2)`; load_state_dict strict: `{m['model']['load_state_dict']}`; {m['model']['n_params']/1e6:.2f} M parameters", | |
| f"- feature: `forward_features()` = fc_norm(mean over the 8×14×14 tokens) → **384-d pre-logit feature**; the K710 head is never used; fp16 forward on {m['args']['device']}; frozen (no fine-tuning, no gradient)", | |
| f"- libs: " + ", ".join(f"{k} {v_}" for k, v_ in m['model']['lib_versions'].items()) + f"; code sha256 {m['model']['code_sha256'][:12]}…", "", | |
| "## Channels", "", | |
| "| cache | channel | role | units / range |", "|---|---|---|---|", | |
| "| `windows/<uid>.npz` (WindowCache) | `emb` f16[M,384] | embedding | unbounded real (fc_norm output; typical L2 norm " + (f"{m['pca'].get('emb_norm_mean', float('nan')):.1f} ± {m['pca'].get('emb_norm_std', float('nan')):.1f}" if m['pca'] else 'n/a') + ") |", | |
| "| `windows/<uid>.npz` | `win_t_end` f32[M] | — | s on the frame/20 axis, 5 Hz grid (k/5); window = [win_t_end − 3.0, win_t_end] |", | |
| "| `windows/<uid>.npz` | `win_valid` bool[M] | — | True ⇔ all 16 frames decoded AND a non-degenerate driver box on every one |", | |
| "| `cache/<uid>.npz` (RouteCache, 10 Hz) | `vmae2_crop_w`, `vmae2_crop_h` | none | driver-crop size in DECODED px (meta.decode_wh; +10 % expanded, clipped) |", | |
| "| `cache/<uid>.npz` | `vmae2_lb_scale` | none | 224 / long side of the crop (< 1 = downscaled) |", "", | |
| "Spec returned by `VideoMAE2Extractor.spec()`: `" + json.dumps(m['model']['channel_roles']) + "` (matches `summarise.SPECS['videomae2']['embedding'] == ['emb']`).", "", | |
| "## Windows and preprocessing", "", | |
| f"- window grid: every 5 Hz grid time t_end inside the requested spans; clip = the 16 frames at 5 Hz ending at t_end (t_end − 3.0 … t_end, every second frame of the 10 Hz decode; 3.2 s causal window). A causal lead-in of {WIN_SPAN_S:.1f} s before each span is decoded (`lead_in={m['args']['lead_in']}`) so the first windows are complete.", | |
| f"- decode: `video_io.FrameStream` (cv2 seek + grab, ≤ {m['args']['workers']} worker processes, blocks of 50 frames), ONE stream per contiguous decode span (a block straddling two anchor spans would otherwise grab every frame of the gap), aspect-preserving resize to height {DECODE_H} px (958×600 / 1062×600 / 800×600 for the three native sizes), INTER_AREA.", | |
| "- crop: `video_io.driver_crop(frame, causal route box, expand 0.10)` (box from results/pose_causal_routes[_v2], scaled to the decoded frame).", | |
| f"- letterbox: long side → 224, centred, padding = ImageNet mean colour RGB {PAD_RGB} (normalises to ≈ 0); INTER_AREA when downscaling, INTER_LINEAR when upscaling.", | |
| f"- colour / normalisation: BGR → RGB, /255, then (x − mean)/std with mean {MEAN_RGB}, std {STD_RGB} — the ImageNet constants of the VideoMAEv2 repo (`dataset/kinetics.py`: `ClipToTensor` + `video_transforms.Normalize`). Tensor [B, 3, 16, 224, 224] fp16; batch {m['args']['batch']} windows.", | |
| "- meaning of valid: per frame (RouteCache) `valid` ⇔ frame decoded and a driver box present (fail 1 no box / 2 no frame); per window `win_valid` ⇔ all 16 frames valid and the fp16 feature finite (window reasons: ok / no_box / no_frame / not_decoded = a needed frame precedes the decoded spans or the video start).", "", | |
| "## Measured (this run)", "", | |
| f"- routes {m['routes']['n_uids']} (computed now {m['routes']['n_done_now']}, pre-existing {m['routes']['n_pre_existing']}, failed {m['routes']['n_failed']}); 10 Hz frames decoded **{m['n_frames']}** (of which {m['n_frames_in_manifest_windows']} inside the manifest windows; {m['n_frames_with_box']} with a box); windows **{m['n_windows']}**, valid **{m['n_windows_valid']}**", | |
| f"- valid share: per frame overall {f(v['per_frame_overall'])}, on frames with a box {f(v['per_frame_on_frames_with_box'])}, inside the manifest windows {f(v['per_frame_in_manifest_windows'])}; **windows {f(v['window_overall'])}**", | |
| f"- fail-code histogram per frame: `{json.dumps(m['fail_hist_per_frame'])}`; per window: `{json.dumps(m['fail_hist_per_window'])}`", | |
| f"- decode ms / frame p50 {f(t['decode_ms_per_frame']['p50'], 2)} p95 {f(t['decode_ms_per_frame']['p95'], 2)} (per decode block, worker time); letterbox ms / 5 Hz frame p50 {f(t['preprocess_ms_per_5hz_frame']['p50'], 3)} p95 {f(t['preprocess_ms_per_5hz_frame']['p95'], 3)}; **infer ms / window p50 {f(t['infer_ms_per_window']['p50'], 2)} p95 {f(t['infer_ms_per_window']['p95'], 2)}** (batched fp16, incl. H2D + normalisation)", | |
| f"- wall {t['wall_s_routes_loop']} s for the route loop → **{t['frames_per_s_end_to_end']:.1f} frames/s end-to-end** ({t['windows_per_s_end_to_end']:.1f} windows/s); {t['note']}", | |
| f"- memory: peak RSS main process {m['memory']['peak_rss']['main_mb']} MB (incl. the face-pack load for the diagnostics), decode workers (sampled every 25 frames) max single {m['memory']['peak_rss']['decode_worker_max_mb_sampled']} MB / total {m['memory']['peak_rss']['decode_workers_total_mb_sampled']} MB; VRAM after load {vr['after_load_mb']} MB, **peak allocated {vr['peak_allocated_mb']} MB**, peak reserved {vr['peak_reserved_mb']} MB", | |
| "", "## Alignment diagnostic vs the face pack (lm_yaw_dev)", "", | |
| f"- {rho['note']}", | |
| f"- n pairs {rho['n_pairs']}; Spearman ρ(PC1, lm_yaw_dev at t_end) = **{f(rho.get('spearman_pc1_vs_lm_yaw_dev_at_t_end'))}**; ρ(PC1, window-mean lm_yaw_dev) = {f(rho.get('spearman_pc1_vs_lm_yaw_dev_window_mean'))}; ρ(PC1, |lm_yaw_dev|) = {f(rho.get('spearman_pc1_vs_abs_lm_yaw_dev'))}; ρ(1 − cos to route mean, |lm_yaw_dev|) = {f(rho.get('spearman_dist_route_vs_abs_lm_yaw_dev'))}", | |
| f"- PCA over {m['pca'].get('n_windows')} valid windows: explained variance PC1–3 {[round(x, 3) for x in m['pca'].get('explained_var_ratio_pc1_3', [])]}; {m['pca'].get('sign_rule')}", | |
| f"- sign convention: {m['sign_convention']}", | |
| f"- contact sheet: `{Path(m['contact_sheet']).name if m['contact_sheet'] else 'n/a'}` (top row 6 most negative lm_yaw_dev window ends, bottom 6 most positive; each tile is the exact 224 letterboxed model input of the window's last frame)", | |
| "", "## Per-anchor values at t_anchor (descriptive)", "", | |
| "`per_anchor` in measurements.json: n_win_8s (valid windows with T − 8 < win_t_end ≤ T), anchor_win_t_end (last valid window ≤ T within 0.5 s, `summarise.summarise_windows`), pc1_anchor, emb_norm_anchor, cos_anchor_vs_mean8s, lm_yaw_dev_at_anchor (face pack F0). " | |
| f"Descriptive distribution by y (not used for anything): `{json.dumps(m['descriptive_by_y'])}`.", "", | |
| "## Files", "", | |
| "- `cache/<uid>.npz` RouteCache (45 routes), `windows/<uid>.npz` WindowCache (45 routes), `measurements.json`, `contact_sheet_sign.png`, this README.", | |
| "- Reproduce: `python baselines_dms/extractors/videomae2.py --phase1` (skips routes whose caches already cover the manifest spans; `--force` recomputes).", ""] | |
| Path(path).write_text("\n".join(L_)) | |
| return path | |
| # ----------------------------------------------------------------------------- self-test | |
| def _self_test(uid=None, workers=2, log=print): | |
| import tempfile | |
| import torch | |
| rep = {"self_test": "extractors.videomae2", "steps": []} | |
| # unit checks (no video, no GPU) | |
| lb, s = letterbox_rgb(np.full((300, 100, 3), (10, 20, 30), np.uint8)) | |
| assert lb.shape == (IMG, IMG, 3) and abs(s - IMG / 300) < 1e-9 and tuple(lb[0, 0]) == PAD_RGB and tuple(lb[IMG // 2, IMG // 2]) == (30, 20, 10), (lb[0, 0], lb[IMG // 2, IMG // 2]) | |
| assert (lb[:, :70] == np.array(PAD_RGB, np.uint8)).all() and (lb[:, -70:] == np.array(PAD_RGB, np.uint8)).all(), "letterbox padding must be the mean colour" | |
| assert decode_wh_for((1928, 1208)) == (958, 600) and decode_wh_for((1344, 760)) == (1062, 600) and decode_wh_for((816, 612)) == (800, 600) | |
| k5, t5 = window_grid([[10.0, 12.0]], 10000) | |
| assert k5[0] == 50 and k5[-1] == 60 and len(k5) == 11 and abs(t5[3] - 10.6) < 1e-6 | |
| k5b, _ = window_grid([[376.8, 387.0]], 10 ** 6) | |
| assert len(k5b) == 52 and k5b[0] == 1884 and k5b[-1] == 1935, (len(k5b), k5b[0], k5b[-1]) | |
| assert len(window_grid([[222.2, 232.39999999999998]], 10 ** 6)[0]) == 52 | |
| ext = lead_in_spans([[10.0, 12.0]], 10000) | |
| assert ext.shape == (1, 2) and abs(ext[0, 0] - 7.0) < 1e-6 and abs(ext[0, 1] - 12.0) < 1e-6, ext | |
| assert abs(lead_in_spans([[1.0, 2.0]], 10000)[0, 0]) < 1e-6 | |
| assert abs(lead_in_spans([[10.0, 12.0]], 10000, lead_in=False)[0, 0] - 10.0) < 1e-9 | |
| rep["steps"].append({"unit": "letterbox / decode size / window grid / lead-in ok"}) | |
| check_resources() | |
| ex = VideoMAE2Extractor().setup() | |
| rep["model"] = {"load": ex.load_result, "params_M": round(ex.n_params / 1e6, 3), "load_s": round(ex.load_s, 2), "weight_sha256": ex.weight_sha256, "spec": ex.spec()} | |
| assert "All keys matched" in ex.load_result | |
| # per-frame contract + determinism on synthetic clips | |
| g = ex.infer([np.zeros((50, 30, 3), np.uint8), None]) | |
| assert g.shape == (2, 3) and np.isnan(g[1]).all() and g[0, 0] == 30 and g[0, 1] == 50 | |
| rng = np.random.default_rng(0) | |
| clip = rng.integers(0, 255, (2, L, IMG, IMG, 3), np.uint8) | |
| e1, e2 = ex.infer_windows(clip), ex.infer_windows(clip[:1]) | |
| assert e1.shape == (2, ex.D) and np.isfinite(e1).all() and np.abs(e1[0] - e2[0]).max() < 1e-1, np.abs(e1[0] - e2[0]).max() | |
| rep["steps"].append({"synthetic": "infer_windows (2,384) finite; batch-1 vs batch-2 max abs diff %.4f" % float(np.abs(e1[0] - e2[0]).max())}) | |
| # 1 local TRAIN route, 30 frames at 10 Hz (+ 3.0 s lead-in) -> 15 windows | |
| if uid is None: | |
| cands = [(vio.video_path(u).stat().st_size, u) for u in vio.anchors_by_uid(("manual_train", "engaged_train")) if vio.has_video(u) and vio.pose_path(u) is not None] | |
| if not cands: | |
| rep["route"] = "none local; skipped" | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return rep | |
| uid = min(cands)[1] | |
| ts = vio.anchors_of(uid) | |
| W, H, n = vio.video_info(vio.video_path(uid)) | |
| k_end = int(round((float(ts[len(ts) // 2]) if len(ts) else 30.0) * WIN_HZ)) * STEP10 # 10 Hz step of the last frame, on the 5 Hz grid | |
| span = np.array([[(k_end - 29) / cs.HZ, k_end / cs.HZ]], np.float64) # 30 frames at 10 Hz -> 15 window ends | |
| with tempfile.TemporaryDirectory() as d: | |
| t0 = time.time() | |
| code, paths, info = run_route_windows(ex, uid, span, Path(d) / "cache", Path(d) / "windows", workers=workers, block=20) | |
| assert code == 0 and info["status"] == "done", info | |
| rc, wc = cs.RouteCache.load(paths["cache"]), cs.WindowCache.load(paths["windows"]) | |
| k5, _ = window_grid(span, n) | |
| assert wc.m == len(k5) == 15 and wc.D == ex.D and wc.win_len == L and rc.covers(span) and wc.covers(span), (wc.m, len(k5)) | |
| n_req = len(vio.frames_in_spans(span, cs.HZ, n)[0]) | |
| assert 30 <= n_req <= 31 and rc.n >= n_req + 29, (n_req, rc.n) | |
| E = wc.emb.astype(np.float32) | |
| assert np.isfinite(E[wc.win_valid]).all() and np.isnan(E[~wc.win_valid]).all() | |
| rep["route"] = {"uid": uid, "native_wh": [W, H], "decode_wh": info["decode_wh"], "span": span.tolist(), "spans_decoded": info["spans_decoded"], "n_frames_10hz": rc.n, "n_frames_requested": n_req, | |
| "fail_hist": info["fail_hist"], "n_windows": wc.m, "n_windows_valid": int(wc.win_valid.sum()), "win_fail_hist": info["win_fail_hist"], | |
| "emb_norm_mean": float(np.linalg.norm(E[wc.win_valid], axis=1).mean()) if wc.win_valid.any() else None, | |
| "cos_consecutive_mean": float(np.mean([np.dot(E[i], E[i + 1]) / (np.linalg.norm(E[i]) * np.linalg.norm(E[i + 1])) for i in range(wc.m - 1) if wc.win_valid[i] and wc.win_valid[i + 1]])) if wc.win_valid.sum() > 1 else None, | |
| "runtime": wc.runtime, "wall_s": round(time.time() - t0, 2)} | |
| # skip path | |
| code2, _, info2 = run_route_windows(ex, uid, span, Path(d) / "cache", Path(d) / "windows", workers=workers) | |
| assert code2 == 0 and info2["status"] == "exists", info2 | |
| # determinism of a real window: recompute the last window from the same frames | |
| if wc.win_valid.any(): | |
| j = int(np.flatnonzero(wc.win_valid)[-1]) | |
| t_end = float(wc.win_t_end[j]) | |
| fs = vio.FrameStream(uid, [[t_end - WIN_SPAN_S, t_end]], hz=cs.HZ, workers=workers, decode_wh=decode_wh_for((W, H)), block=40, box_coords="frame") | |
| frames = {int(round(t_ * cs.HZ)): (fr_, bx_) for t_, fi_, fr_, bx_ in fs} | |
| need = [int(round(t_end * cs.HZ)) - STEP10 * q for q in range(L - 1, -1, -1)] | |
| clip = np.stack([letterbox_rgb(vio.driver_crop(*frames[k]))[0] for k in need])[None] | |
| e = ex.infer_windows(clip)[0] | |
| diff = float(np.abs(e - E[j]).max()) | |
| rep["steps"].append({"recompute_last_window": {"t_end": t_end, "max_abs_diff_vs_cache_f16": diff}}) | |
| assert diff < 0.1, diff | |
| rep["steps"].append({"rerun": info2["status"]}) | |
| rep["vram_peak_mb"] = round(torch.cuda.max_memory_allocated() / 2**20, 1) | |
| rep["describe"] = ex.describe() | |
| ex.teardown() | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return rep | |
| # ----------------------------------------------------------------------------- CLI | |
| def main(): | |
| import argparse | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--self-test", action="store_true") | |
| ap.add_argument("--phase1", action="store_true") | |
| ap.add_argument("--uid", default="") | |
| ap.add_argument("--uids", default="", help="file with route uids (one per line): anchor-span runner (packs / hist-s / post-s)") | |
| ap.add_argument("--out-dir", default="") | |
| ap.add_argument("--manifest", default=str(MANIFEST)) | |
| ap.add_argument("--workers", type=int, default=3) | |
| ap.add_argument("--batch", type=int, default=32) | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--no-lead-in", action="store_true") | |
| ap.add_argument("--hist-s", type=float, default=vio.HIST_S) | |
| ap.add_argument("--post-s", type=float, default=vio.POST_S) | |
| ap.add_argument("--packs", default=",".join(vio.ALL_PACKS)) | |
| a = ap.parse_args() | |
| a.workers = min(a.workers, 3) | |
| if a.phase1: | |
| phase1(Path(a.out_dir) if a.out_dir else OUT_PHASE1, Path(a.manifest), workers=a.workers, batch=a.batch, lead_in=not a.no_lead_in, force=a.force) | |
| return 0 | |
| if a.self_test or not (a.uids or a.uid): | |
| _self_test(a.uid or None, min(a.workers, 2)) | |
| return 0 | |
| import torch | |
| torch.set_num_threads(int(os.environ.get("OMP_NUM_THREADS", "4"))) | |
| check_resources() | |
| ex = VideoMAE2Extractor().setup() | |
| ex.batch = a.batch | |
| uids = [a.uid] if a.uid else [l.strip() for l in open(a.uids) if l.strip()] | |
| out_dir = Path(a.out_dir) if a.out_dir else cs.RESULTS_ROOT / "cache" / BASELINE | |
| packs = tuple(a.packs.split(",")) | |
| spans = {u: vio.anchor_spans(u, packs, a.hist_s, a.post_s) for u in uids} | |
| code, summ, recs = run_routes_windows(ex, uids, spans, out_dir, workers=a.workers, lead_in=not a.no_lead_in, force=a.force) | |
| (out_dir / "logs").mkdir(parents=True, exist_ok=True) | |
| (out_dir / "logs" / f"run_{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}.json").write_text(json.dumps({"summary": summ, "routes": [{k: r.get(k) for k in ("uid", "status", "n_frames", "n_windows", "n_windows_valid", "wall_s")} for r in recs]}, indent=1, default=str)) | |
| ex.teardown() | |
| return code | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 62.8 kB
- Xet hash:
- 9444767815baf1679dab9a2a4d55d8046b72dcfcfaf3337655b09cce643b76f2
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.