Buckets:
| """cwm3/baselines_dms/extractors/l2cs_gaze.py — O3 L2CS-Net (Gaze360 ResNet50) gaze baseline as a DMS-baseline Extractor (plan §3, 2026-09-14). | |
| Model (registry ids `l2cs` = L2CSNet_gaze360.pkl, `git_l2cs` = l2cs package, `git_face_detection` = RetinaFace mobilenet0.25 detector bundled with it) | |
| l2cs.Pipeline(weights, arch='ResNet50', device=cuda:0) is built exactly as upstream; per driver crop we run its RetinaFace detector, keep the LARGEST | |
| face box with score >= 0.5 (Pipeline's default confidence_threshold), cut the face as Pipeline.step does (int bounds, min clipped at 0), BGR -> RGB, | |
| resize 224 x 224, and hand the faces in batches to Pipeline.predict_gaze (its own transform: PIL Resize 448, ImageNet mean/std) -> radians -> degrees. | |
| Pipeline.step itself is not called because it raises on frames without a face (np.stack of an empty list). | |
| Axis naming (IMPORTANT). l2cs.Pipeline names its first output "pitch" and its second "yaw", but in the L2CS code base the first angle is the HORIZONTAL | |
| one: it is gaze2d[0] of the GazeHub Gaze360 labels (= GazeHub yaw), l2cs.utils.gazeto3d puts it in the x component and l2cs.vis.draw_gaze draws it as | |
| the horizontal arrow component (dx = -sin(pitchyaw[0]) cos(pitchyaw[1])). Therefore l2cs_yaw := Pipeline "pitch" (horizontal), l2cs_pitch := Pipeline | |
| "yaw" (vertical). --phase1 verifies this with the contact sheet and with Spearman rho of BOTH raw outputs against the face-pack lm_yaw_dev. | |
| Sign (GazeHub convention): positive l2cs_yaw = gaze toward the image LEFT (camera's left); positive l2cs_pitch = gaze UP (GazeHub pitch = asin(-g_y) with the | |
| image y axis pointing down; l2cs.vis.draw_gaze draws dy = -sin(pitch)). Verified 2026-09-14 on the extreme frames of the Phase-1 caches: the two most negative | |
| l2cs_pitch frames (-61 deg) show the driver looking DOWN at a phone in the lap, the most positive (+8 / +4.5 deg) show the driver looking up / forward at the road. | |
| (The first version of this file documented + = down and drew the contact-sheet arrow with dy = +sin(pitch); values were never affected, only the label.) | |
| Verification fixes (2026-09-14, verifier agent): manifest spans are snapped onto the 10 Hz grid (snap_span_to_grid) so that no frame after win_end is decoded | |
| (the manifest's win_end = t + 0.2 carries float noise such as 650.8000000000001 and video_io.frames_in_spans uses ceil -> one frame at t_anchor + 0.3 s in 22 of 50 | |
| windows of the first run; those rows were trimmed from the caches); CPU device guards; --measure-only keeps raw_timers_ms_per_frame / n_infer_calls of the run. | |
| video_io.frames_in_spans now snaps span ends itself (0.02 s tolerance) and decode_block is index-exact on every header fps (Phase-1 re-run 2026-09-14); the local | |
| snap_span_to_grid stays (redundant but harmless). rho_vs_face_pack also reports the CIRCULAR deviation from the per-route causal median (summarise.circ_dev; | |
| l2cs_yaw is centred near -100 deg and wraps at +-180 on this camera), pooled and per anchor, next to the raw rho. | |
| Channels (f32; NaN row = fail 3 no face) roles for summarise: l2cs_face_conf conf | l2cs_yaw, l2cs_pitch angle_deg | l2cs_box_w none | |
| l2cs_face_conf RetinaFace score of the kept face, (0.5, 1] | |
| l2cs_yaw horizontal gaze angle, degrees, 90 bins x 4 deg - 180 -> (-180, 180); Gaze360 range in practice within +-140 | |
| l2cs_pitch vertical gaze angle, degrees, same range | |
| l2cs_box_w width of the kept face box in NATIVE pixels (frames are decoded at native size; role none) | |
| valid <=> fail 0 : frame decoded, driver box present, a face with score >= 0.5 found, all four outputs finite. No eye-closure output exists (none invented). | |
| decode_wh = None (NATIVE 1928 x 1208 / 1344 x 760 / 816 x 612): the driver face is only ~110-130 px wide at native resolution in a ~440 px driver box, and the | |
| self-test showed the outputs move by up to 27 deg between a 1x and a 2x face, so no resolution is thrown away before the 224 x 224 face resize. | |
| python baselines_dms/extractors/l2cs_gaze.py --self-test [--uid UID] # 1 manifest route, 30 frames (3 s before its anchor) -> results/baselines_dms/phase1/l2cs_gaze/self_test/ | |
| python baselines_dms/extractors/l2cs_gaze.py --phase1 [--workers 3] [--block 60] [--max-minutes 120] [--force] # 50 frozen manifest windows -> cache/, measurements.json, contact sheet | |
| python baselines_dms/extractors/l2cs_gaze.py --measure-only # recompute measurements.json / the contact sheet from existing caches (GPU only for the sheet's face boxes) | |
| Only writes under results/baselines_dms/phase1/l2cs_gaze/ (and results/baselines_dms/cache/l2cs_gaze/ when --out-dir asks for it). No openpilot field is read. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import math | |
| 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(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 registry as REG # noqa: E402 | |
| from base import Extractor, run_route, EXIT_DONE, EXIT_BUDGET, EXIT_FAIL # noqa: E402 | |
| BASELINE = "l2cs_gaze" | |
| CHANNELS = ["l2cs_face_conf", "l2cs_yaw", "l2cs_pitch", "l2cs_box_w"] | |
| ROLES = {"l2cs_face_conf": "conf", "l2cs_yaw": "angle_deg", "l2cs_pitch": "angle_deg", "l2cs_box_w": "none"} | |
| UNITS = {"l2cs_face_conf": "probability (0.5, 1]", "l2cs_yaw": "deg, horizontal, + = image left", "l2cs_pitch": "deg, vertical, + = up", "l2cs_box_w": "native px"} | |
| DECODE_WH = None # native decode (video_io: None = native size); box_coords "frame" == native px | |
| FACE_CONF_TH = 0.5 # l2cs.Pipeline default confidence_threshold | |
| GAZE_SUB_BATCH = 32 # faces per predict_gaze call (448 x 448 ResNet50 inputs) | |
| HORIZONTAL_IS_FIRST_OUTPUT = True # Pipeline.predict_gaze -> (first, second); first ("pitch" in l2cs naming) is the horizontal angle (see module doc) | |
| PHASE1_ROOT = cs.RESULTS_ROOT / "phase1" | |
| OUT_DIR = PHASE1_ROOT / BASELINE | |
| MANIFEST = PHASE1_ROOT / "manifest_v1.csv" | |
| FACE_PACK = CWM3 / "results" / "targets_face_causal_routes_v2" / "facehist_causal_{regime}_train.npz" | |
| LM_YAW_DEV_CH = "lm_yaw_dev" | |
| HIST_STEPS = 80 | |
| MAX_WORKERS = 3 | |
| NEED_VRAM_GB, NEED_RAM_GB, VRAM_HEADROOM_GB = 6.0, 28.0, 4.0 | |
| # ----------------------------------------------------------------------------- extractor | |
| class L2CSGazeExtractor(Extractor): | |
| baseline = BASELINE | |
| channels = CHANNELS | |
| roles = ROLES | |
| needs_crop = True | |
| decode_wh = DECODE_WH | |
| batch = 64 | |
| def spec(self): | |
| """{channel: role} in the task vocabulary (angle_deg | eye_closed_prob | eye_open_ratio | prob | logit | embedding | conf | none).""" | |
| return dict(ROLES) | |
| def summarise_spec(): | |
| """summarise.py-style spec (identical to summarise.SPECS['l2cs']; pass as spec= because meta.baseline is 'l2cs_gaze').""" | |
| return {"angle": {"l2cs_yaw": (20.0, 12.0), "l2cs_pitch": (15.0, 9.0)}, "eye_state": {}, "prob": ["l2cs_face_conf"], "embedding": [], | |
| "rule": {"yaw": "l2cs_yaw", "pitch": "l2cs_pitch", "closed": None}, "chain": {"head_dev": None, "closed": None, "gaze_off": ("l2cs_yaw", "l2cs_pitch")}} | |
| def setup(self, registry_entry=None, device="cuda:0"): | |
| import cv2 | |
| import torch | |
| import torchvision | |
| import face_detection | |
| import l2cs | |
| from l2cs import Pipeline | |
| e_w, e_pkg, e_fd = REG.get("l2cs"), REG.get("git_l2cs"), REG.get("git_face_detection") | |
| weights = REG.resolve("l2cs") # asserts sha256 of L2CSNet_gaze360.pkl | |
| for eid in ("git_l2cs", "git_face_detection"): | |
| probs = REG.verify(REG.get(eid)) | |
| assert not probs, f"registry {eid}: {probs}" | |
| fd_w = Path(face_detection.__file__).resolve().parent / "weights" / "mobilenet0.25_Final.pth" | |
| fd_sha = cs.sha256_file(fd_w) | |
| assert fd_sha == e_fd["sha256"], f"RetinaFace weight sha256 {fd_sha[:16]} != registry {e_fd['sha256'][:16]}" | |
| self.dev = torch.device(device) | |
| self.pipe = Pipeline(weights, arch="ResNet50", device=self.dev, include_detector=True, confidence_threshold=FACE_CONF_TH) | |
| torch.backends.cudnn.benchmark = False # face_detection.load_net turns it on; crop shapes vary -> no per-shape autotuning | |
| self.cv2, self.torch = cv2, torch | |
| self.model_ids = [f"l2cs:{e_w['name']}", f"git_l2cs:{e_pkg['url']}@{e_pkg['version'][:12]}", f"git_face_detection:{e_fd['url']}@{e_fd['version'][:12]}"] | |
| self.weight_sha256 = {"L2CSNet_gaze360.pkl": e_w["sha256"], "mobilenet0.25_Final.pth": fd_sha} | |
| self.lib_versions = {"torch": torch.__version__, "torchvision": torchvision.__version__, "opencv": cv2.__version__, "numpy": np.__version__, | |
| "l2cs": f"0.0.1@{e_pkg['version'][:12]}", "face-detection": f"1.0.5@{e_fd['version'][:12]}", "cuda": str(torch.version.cuda)} | |
| self.registry_entry = {"l2cs": {k: e_w.get(k) for k in ("id", "name", "version", "url", "sha256", "size_bytes", "license")}, | |
| "git_l2cs": {k: e_pkg.get(k) for k in ("id", "name", "version", "url", "license")}, | |
| "git_face_detection": {k: e_fd.get(k) for k in ("id", "name", "version", "url", "sha256", "license")}} | |
| self.timers = {"det_ms": [], "gaze_ms": [], "n_infer_calls": 0, "n_faces": 0, "n_multi_face": 0, "n_largest_not_top_score": 0} | |
| self.n_params_M = round(sum(p.numel() for p in self.pipe.model.parameters()) / 1e6, 3) | |
| self.warmup() | |
| if self.dev.type == "cuda": | |
| torch.cuda.reset_peak_memory_stats(self.dev) | |
| return self | |
| def warmup(self): | |
| """one detector + one gaze pass on synthetic crops (CUDA context, lazy kernel loading) so that the recorded timings are steady-state.""" | |
| self.detect_largest([np.zeros((480, 320, 3), np.uint8)]) | |
| with self.torch.no_grad(): | |
| self.pipe.predict_gaze(np.zeros((2, 224, 224, 3), np.uint8)) | |
| self._sync() | |
| self.timers = {"det_ms": [], "gaze_ms": [], "n_infer_calls": 0, "n_faces": 0, "n_multi_face": 0, "n_largest_not_top_score": 0} | |
| def _sync(self): | |
| if self.dev.type == "cuda": | |
| self.torch.cuda.synchronize(self.dev) | |
| def detect_largest(self, crops): | |
| """crops (u8 BGR, any sizes) -> per crop (box f32[4], score) of the largest face with score >= FACE_CONF_TH, or None. Same-shape crops are batched.""" | |
| out = [None] * len(crops) | |
| groups = {} | |
| for i, c in enumerate(crops): | |
| groups.setdefault(c.shape, []).append(i) | |
| for shape, idx in groups.items(): | |
| dets = self.pipe.detector(np.stack([crops[i] for i in idx])) | |
| for i, d in zip(idx, dets): | |
| best, best_area, n_ok, top_score = None, -1.0, 0, -1.0 | |
| for box, _lm, score in d: # d is sorted by score (descending) by face_detection.post_process | |
| if score < FACE_CONF_TH: | |
| continue | |
| n_ok += 1 | |
| top_score = max(top_score, float(score)) | |
| area = max(float(box[2] - box[0]), 0.0) * max(float(box[3] - box[1]), 0.0) | |
| if area > best_area: | |
| best, best_area = (np.asarray(box, np.float32), float(score)), area | |
| self.timers["n_multi_face"] += int(n_ok > 1) | |
| self.timers["n_largest_not_top_score"] += int(n_ok > 1 and best is not None and best[1] < top_score - 1e-6) # diagnostic: the task rule "largest box" overrode the detector's top score | |
| out[i] = best | |
| return out | |
| def face_crop(self, crop, box): | |
| """Pipeline.step's face cut: int bounds, min clipped at 0 (numpy clips the max), BGR -> RGB, 224 x 224; None when empty.""" | |
| x0, y0, x1, y1 = max(int(box[0]), 0), max(int(box[1]), 0), int(box[2]), int(box[3]) | |
| img = crop[y0:y1, x0:x1] | |
| if img.size == 0 or img.shape[0] < 2 or img.shape[1] < 2: | |
| return None | |
| return self.cv2.resize(self.cv2.cvtColor(img, self.cv2.COLOR_BGR2RGB), (224, 224)) | |
| def _usable(crops): | |
| return [i for i, c in enumerate(crops) if c is not None and c.size > 0 and c.shape[0] >= 2 and c.shape[1] >= 2] | |
| def _gaze_rows(self, faces, keep, out): | |
| """faces (224 x 224 RGB) + keep [(row, score, box_w)] -> fills out rows [score, horizontal, vertical, box_w] (Pipeline.predict_gaze in GAZE_SUB_BATCH batches).""" | |
| for a in range(0, len(faces), GAZE_SUB_BATCH): | |
| with self.torch.no_grad(): | |
| first, second = self.pipe.predict_gaze(np.stack(faces[a:a + GAZE_SUB_BATCH])) | |
| first, second = np.degrees(np.asarray(first, np.float64)).ravel(), np.degrees(np.asarray(second, np.float64)).ravel() | |
| horiz, vert = (first, second) if HORIZONTAL_IS_FIRST_OUTPUT else (second, first) | |
| for j, (i, score, bw) in enumerate(keep[a:a + GAZE_SUB_BATCH]): | |
| out[i] = [score, horiz[j], vert[j], bw] | |
| def infer(self, crops): | |
| out = np.full((len(crops), self.C), np.nan, np.float32) | |
| idx = self._usable(crops) | |
| if not idx: | |
| return out | |
| self.timers["n_infer_calls"] += 1 | |
| t0 = time.time() | |
| dets = self.detect_largest([crops[i] for i in idx]) | |
| self._sync() | |
| t1 = time.time() | |
| faces, keep = [], [] | |
| for i, d in zip(idx, dets): | |
| if d is None: | |
| continue | |
| box, score = d | |
| img = self.face_crop(crops[i], box) | |
| if img is None: | |
| continue | |
| faces.append(img) | |
| keep.append((i, score, float(box[2] - box[0]))) | |
| self._gaze_rows(faces, keep, out) | |
| self._sync() | |
| t2 = time.time() | |
| self.timers["det_ms"].append(1000.0 * (t1 - t0) / len(idx)) | |
| self.timers["gaze_ms"].append(1000.0 * (t2 - t1) / len(idx)) | |
| self.timers["n_faces"] += len(faces) | |
| return out | |
| def infer_with_boxes(self, crops, dets): | |
| """SHARED-DETECTOR path (run_full_extraction.py, 2026-09-14): dets f32[n,5] (x1, y1, x2, y2, score) aligned to `crops`, in crop px, from another RetinaFace | |
| pass on the same crops (openface3.detect_faces: the byte-identical mobilenet0.25 weights, top-scoring anchor, score >= 0.5). Everything downstream is | |
| L2CS's own convention unchanged: face_crop (int bounds, min clipped at 0, BGR -> RGB, 224 x 224), Pipeline.predict_gaze, degrees, axis naming; a | |
| non-finite det or score < FACE_CONF_TH -> NaN row (fail 3). l2cs_box_w = width of the given box (crop = native px).""" | |
| out = np.full((len(crops), self.C), np.nan, np.float32) | |
| idx = self._usable(crops) | |
| if not idx: | |
| return out | |
| dets = np.asarray(dets, np.float32) | |
| self.timers["n_infer_calls"] += 1 | |
| t1 = time.time() | |
| faces, keep = [], [] | |
| for i in idx: | |
| d = dets[i] | |
| if not np.isfinite(d).all() or d[4] < FACE_CONF_TH: | |
| continue | |
| img = self.face_crop(crops[i], d[:4]) | |
| if img is None: | |
| continue | |
| faces.append(img) | |
| keep.append((i, float(d[4]), float(d[2] - d[0]))) | |
| self._gaze_rows(faces, keep, out) | |
| self._sync() | |
| t2 = time.time() | |
| self.timers["det_ms"].append(0.0) | |
| self.timers["gaze_ms"].append(1000.0 * (t2 - t1) / len(idx)) | |
| self.timers["n_faces"] += len(faces) | |
| return out | |
| def teardown(self): | |
| self.pipe = None | |
| if hasattr(self, "torch") and self.dev.type == "cuda": | |
| self.torch.cuda.empty_cache() | |
| def gpu_peak_mb(self): | |
| if self.dev.type != "cuda": | |
| return {"torch_max_allocated_mb": float("nan"), "torch_max_reserved_mb": float("nan"), "device": str(self.dev)} | |
| return {"torch_max_allocated_mb": round(self.torch.cuda.max_memory_allocated(self.dev) / 2**20, 1), "torch_max_reserved_mb": round(self.torch.cuda.max_memory_reserved(self.dev) / 2**20, 1)} | |
| # ----------------------------------------------------------------------------- resources / helpers | |
| def gpu_free_mb(): | |
| try: | |
| s = subprocess.run(["nvidia-smi", "--query-gpu=memory.free", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=20).stdout | |
| return float(s.strip().splitlines()[0]) | |
| except Exception: | |
| return float("nan") | |
| def gpu_process_mb(pid=None): | |
| """VRAM of this process as nvidia-smi sees it (includes the CUDA context); NaN when not listed.""" | |
| pid = pid or os.getpid() | |
| try: | |
| s = subprocess.run(["nvidia-smi", "--query-compute-apps=pid,used_memory", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=20).stdout | |
| for line in s.strip().splitlines(): | |
| p, m = [x.strip() for x in line.split(",")[:2]] | |
| if int(p) == pid: | |
| return float(m) | |
| except Exception: | |
| pass | |
| return float("nan") | |
| def mem_available_gb(): | |
| for line in open("/proc/meminfo"): | |
| if line.startswith("MemAvailable"): | |
| return int(line.split()[1]) / 2**20 | |
| return float("nan") | |
| def check_resources(need_vram_gb=NEED_VRAM_GB, need_ram_gb=NEED_RAM_GB): | |
| free = gpu_free_mb() / 1024.0 | |
| ram = mem_available_gb() | |
| ok = (free >= need_vram_gb + VRAM_HEADROOM_GB) and (ram >= need_ram_gb) | |
| return ok, {"gpu_free_gb": round(free, 1), "need_gpu_gb": need_vram_gb + VRAM_HEADROOM_GB, "mem_available_gb": round(ram, 1), "need_ram_gb": need_ram_gb} | |
| def rss_peaks_mb(): | |
| hwm = float("nan") | |
| for line in open("/proc/self/status"): | |
| if line.startswith("VmHWM"): | |
| hwm = int(line.split()[1]) / 1024.0 | |
| return {"self_max_rss_mb": round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0, 1), "self_vmhwm_mb": round(hwm, 1), | |
| "children_max_rss_mb": round(resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss / 1024.0, 1), | |
| "note": "self = this process (model + up to 2 x workers in-flight native decode blocks); children_max_rss is confounded by fork inheritance (subprocess / spawn children start with the parent RSS before exec) and is NOT the decode workers' footprint"} | |
| def load_manifest(): | |
| import pandas as pd | |
| m = pd.read_csv(MANIFEST) | |
| assert len(m) == 50 and set(["uid", "t_anchor", "win_start", "win_end", "regime", "y", "row"]) <= set(m.columns), (len(m), m.columns.tolist()) | |
| return m | |
| GRID_EPS_S = 2e-3 # snap margin (s): far below cache_schema.SPAN_TOL_S = 0.051 and above the f32 resolution of a span end (~5e-4 s at t ~ 4000 s) | |
| def snap_span_to_grid(a, b, eps=GRID_EPS_S): | |
| """[a, b] (s) -> [k0/10 + eps, k1/10 - eps] with k0 = round(10 a), k1 = round(10 b), so that video_io.frames_in_spans (floor at the start, ceil at the end) decodes | |
| exactly the grid steps k0 .. k1. Without it the float noise of the manifest's win_end = t + 0.2 (e.g. 650.8000000000001 -> ceil(6508.000000000001) = 6509) adds | |
| the frame at win_end + 0.1 s = t_anchor + 0.3 s, and floor(10 win_start) can add one frame before win_start. Existing caches whose spans are the raw window | |
| still `cover` the snapped span (tolerance 0.051 s), so resumability is unchanged.""" | |
| k0, k1 = int(round(float(a) * cs.HZ)), int(round(float(b) * cs.HZ)) | |
| assert k1 >= k0, (a, b) | |
| return [k0 / cs.HZ + eps, k1 / cs.HZ - eps] | |
| def spans_by_uid(m): | |
| S = {} | |
| for r in m.itertuples(): | |
| S.setdefault(r.uid, []).append(snap_span_to_grid(r.win_start, r.win_end)) | |
| return {u: cs.merge_spans(v) for u, v in S.items()} | |
| def pq(v): | |
| v = np.asarray([x for x in v if np.isfinite(x)], np.float64) | |
| return (float(np.percentile(v, 50)), float(np.percentile(v, 95))) if len(v) else (float("nan"), float("nan")) | |
| def value_at(rc, t): | |
| """row of the cache at time t (10 Hz grid) -> (feat f32[C] or NaN, fail code or None).""" | |
| k = int(round(t * cs.HZ)) | |
| kk = np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64) | |
| j = np.searchsorted(kk, k) | |
| if j < len(kk) and kk[j] == k: | |
| return rc.feat[j].astype(np.float32), int(rc.fail[j]) | |
| return np.full(rc.C, np.nan, np.float32), None | |
| def history(rc, t_anchor, steps=HIST_STEPS, arr=None): | |
| """feat f32[steps, C] on the steps t_anchor - (steps-1)/10 .. t_anchor (NaN where absent / invalid); with `arr` (f32[N] aligned to the cache rows) -> f32[steps].""" | |
| kk = np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64) | |
| ka = int(round(t_anchor * cs.HZ)) | |
| want = ka - (steps - 1) + np.arange(steps) | |
| j = np.searchsorted(kk, want) | |
| ok = j < len(kk) | |
| ok[ok] &= kk[j[ok]] == want[ok] | |
| if arr is not None: | |
| h = np.full(steps, np.nan, np.float32) | |
| h[ok] = np.asarray(arr, np.float32)[j[ok]] | |
| return h | |
| H = np.full((steps, rc.C), np.nan, np.float32) | |
| H[ok] = rc.feat[j[ok]].astype(np.float32) | |
| return H | |
| def route_causal_refs(rc, channels=("l2cs_yaw", "l2cs_pitch")): | |
| """per cache row: the route's CAUSAL expanding median of each angle over valid & MOVING frames <= that row (summarise.causal_expanding_median, n_min 25 — the | |
| reference of `ruleref`); NaN while undefined. -> {channel: f32[N]}""" | |
| import summarise as SM | |
| mov = rc.moving if rc.moving is not None else np.ones(rc.n, bool) | |
| return {c: SM.causal_expanding_median(rc.feat[:, CHANNELS.index(c)].astype(np.float32), rc.valid, mov) for c in channels} | |
| # ----------------------------------------------------------------------------- measurements | |
| def spearman(a, b): | |
| from scipy.stats import spearmanr | |
| a, b = np.asarray(a, np.float64), np.asarray(b, np.float64) | |
| ok = np.isfinite(a) & np.isfinite(b) | |
| if ok.sum() < 3: | |
| return {"rho": float("nan"), "p": float("nan"), "n": int(ok.sum())} | |
| r = spearmanr(a[ok], b[ok]) | |
| return {"rho": float(r.statistic), "p": float(r.pvalue), "n": int(ok.sum())} | |
| def rho_vs_face_pack(m, caches): | |
| """Spearman rho between the cache angle channels and lm_yaw_dev (face pack Fh8_c channel 1) over the same 10 Hz history steps of every manifest anchor.""" | |
| import summarise as SM | |
| packs = {} | |
| pooled = {c: [] for c in ("l2cs_yaw", "l2cs_pitch")} | |
| circ = {c: [] for c in pooled} # circular deviation from the per-route causal median (summarise.circ_dev; 2026-09-14) | |
| lm_all, per_anchor = [], [] | |
| ci = {c: CHANNELS.index(c) for c in pooled} | |
| refs_by_uid = {} | |
| for r in m.itertuples(): | |
| if r.regime not in packs: | |
| z = np.load(str(FACE_PACK).format(regime=r.regime), allow_pickle=True) | |
| ch = [str(c) for c in z["channels"]] | |
| assert ch[1] == LM_YAW_DEV_CH, ch | |
| packs[r.regime] = {"uid": z["uid"], "t": z["t"], "Fh8": z["Fh8_c"], "Fv": z["Fh8valid"]} | |
| P = packs[r.regime] | |
| assert str(P["uid"][r.row]) == r.uid and abs(float(P["t"][r.row]) - float(r.t_anchor)) < 1e-6, (r.uid, r.row) | |
| lm = P["Fh8"][r.row, :, 1].astype(np.float32) | |
| lm = np.where(P["Fv"][r.row], lm, np.nan) | |
| rc = caches.get(r.uid) | |
| if rc is None: | |
| continue | |
| H = history(rc, float(r.t_anchor)) | |
| if r.uid not in refs_by_uid: | |
| refs_by_uid[r.uid] = route_causal_refs(rc) | |
| rec = {"idx": int(r.idx), "uid": r.uid, "t_anchor": float(r.t_anchor), "n_pairs": int((np.isfinite(H[:, ci["l2cs_yaw"]]) & np.isfinite(lm)).sum())} | |
| for c, j in ci.items(): | |
| pooled[c].append(H[:, j]) | |
| rec[f"rho_{c}"] = spearman(H[:, j], lm)["rho"] | |
| ref_h = history(rc, float(r.t_anchor), arr=refs_by_uid[r.uid][c]) | |
| d_circ = SM.circ_dev(H[:, j], ref_h) # ((x - ref + 180) mod 360) - 180; NaN before 25 moving valid frames | |
| circ[c].append(d_circ) | |
| rec[f"rho_{c}_circ_centred"] = spearman(d_circ, lm)["rho"] | |
| rec[f"n_pairs_{c}_circ_centred"] = int((np.isfinite(d_circ) & np.isfinite(lm)).sum()) | |
| lm_all.append(lm) | |
| per_anchor.append(rec) | |
| lm_all = np.concatenate(lm_all) if lm_all else np.zeros(0) | |
| # like-for-like: lm_yaw_dev is a deviation from the route reference, l2cs_yaw is absolute (camera geometry offsets of tens of degrees between routes) -> | |
| # also pool the WITHIN-WINDOW demeaned angle (minus the median of the valid steps of that 8 s window), and restrict to the windows where lm_yaw_dev actually moves | |
| dem = {c: [] for c in ci} | |
| lm_rng = [] | |
| for k_, rec in enumerate(per_anchor): | |
| n_ = HIST_STEPS | |
| for c in ci: | |
| h = pooled[c][k_] | |
| dem[c].append(h - (np.nanmedian(h) if np.isfinite(h).any() else np.nan)) | |
| lmw = lm_all[k_ * n_:(k_ + 1) * n_] | |
| rec["lm_yaw_dev_range"] = float(np.nanmax(lmw) - np.nanmin(lmw)) if np.isfinite(lmw).any() else float("nan") | |
| lm_rng.append(rec["lm_yaw_dev_range"]) | |
| lm_rng = np.asarray(lm_rng, np.float64) | |
| thr = float(np.nanmedian(lm_rng)) if len(lm_rng) else float("nan") | |
| moving_w = np.zeros(len(lm_all), bool) | |
| for k_ in range(len(per_anchor)): | |
| moving_w[k_ * HIST_STEPS:(k_ + 1) * HIST_STEPS] = np.isfinite(lm_rng[k_]) and lm_rng[k_] >= thr | |
| out = {"reference": f"results/targets_face_causal_routes_v2/facehist_causal_{{regime}}_train.npz Fh8_c[row, :, 1] = {LM_YAW_DEV_CH} (80 steps t-7.9..t, masked by Fh8valid)", | |
| "n_anchors": len(per_anchor), "pooled": {}, "per_anchor_median_rho": {}, "per_anchor": per_anchor} | |
| out["pooled_within_window_demeaned"] = {} | |
| out["pooled_within_window_demeaned_top_half_lm_range"] = {"lm_yaw_dev_range_threshold_median": thr} | |
| for c in pooled: | |
| out["pooled"][c] = spearman(np.concatenate(pooled[c]) if pooled[c] else np.zeros(0), lm_all) | |
| d_ = np.concatenate(dem[c]) if dem[c] else np.zeros(0) | |
| out["pooled_within_window_demeaned"][c] = spearman(d_, lm_all) | |
| out["pooled_within_window_demeaned_top_half_lm_range"][c] = spearman(d_[moving_w], lm_all[moving_w]) if len(d_) else spearman([], []) | |
| v = np.asarray([a[f"rho_{c}"] for a in per_anchor if a["n_pairs"] >= 20], np.float64) | |
| out["per_anchor_median_rho"][c] = {"median": float(np.nanmedian(v)) if len(v) else float("nan"), "n_anchors_ge20_pairs": int(len(v)), | |
| "share_negative": float(np.nanmean(v < 0)) if len(v) else float("nan")} | |
| out["pooled"]["raw_pipeline_first_output"] = out["pooled"]["l2cs_yaw" if HORIZONTAL_IS_FIRST_OUTPUT else "l2cs_pitch"] | |
| out["pooled"]["raw_pipeline_second_output"] = out["pooled"]["l2cs_pitch" if HORIZONTAL_IS_FIRST_OUTPUT else "l2cs_yaw"] | |
| # circular-centred (2026-09-14): deviation from the route's CAUSAL expanding median over valid & moving frames <= t (the `ruleref` reference), wrapped into | |
| # [-180, 180) by summarise.circ_dev -> like-for-like with lm_yaw_dev (a deviation from a route reference) and immune to the +-180 wrap of l2cs_yaw | |
| out["pooled_circular_centred_route_causal_median"] = {"note": "dev = ((angle - ref_causal_median + 180) mod 360) - 180; ref = summarise.causal_expanding_median over valid & moving " | |
| "frames <= t of the same route (n_min 25); NaN (excluded) while the reference is undefined"} | |
| out["per_anchor_median_rho_circular_centred"] = {} | |
| for c in pooled: | |
| d_ = np.concatenate(circ[c]) if circ[c] else np.zeros(0) | |
| out["pooled_circular_centred_route_causal_median"][c] = spearman(d_, lm_all) | |
| v = np.asarray([a[f"rho_{c}_circ_centred"] for a in per_anchor if a.get(f"n_pairs_{c}_circ_centred", 0) >= 20], np.float64) | |
| out["per_anchor_median_rho_circular_centred"][c] = {"median": float(np.nanmedian(v)) if len(v) else float("nan"), "n_anchors_ge20_pairs": int(len(v)), | |
| "share_negative": float(np.nanmean(v < 0)) if len(v) else float("nan")} | |
| return out | |
| def pick_contact_frames(m, caches, n_side=6): | |
| """per route the valid frame with the most negative / positive l2cs_yaw inside the manifest windows; then the n_side extremes per side over distinct routes.""" | |
| j_yaw, j_conf = CHANNELS.index("l2cs_yaw"), CHANNELS.index("l2cs_face_conf") | |
| neg, pos = [], [] | |
| for u, rc in caches.items(): | |
| f = rc.feat.astype(np.float32) | |
| ok = rc.valid & np.isfinite(f[:, j_yaw]) & (f[:, j_conf] >= 0.8) | |
| if not ok.any(): | |
| continue | |
| ii = np.flatnonzero(ok) | |
| a, b = ii[np.argmin(f[ii, j_yaw])], ii[np.argmax(f[ii, j_yaw])] | |
| neg.append((float(f[a, j_yaw]), u, float(rc.t[a]), int(rc.frame_idx[a]))) | |
| pos.append((float(f[b, j_yaw]), u, float(rc.t[b]), int(rc.frame_idx[b]))) | |
| neg = sorted(neg)[:n_side] | |
| pos = sorted(pos, reverse=True)[:n_side] | |
| return neg, pos | |
| def contact_sheet(ex, m, caches, path, workers=1, tile_h=300): | |
| """12 tiles: 6 most negative (row 1-2 left block) and 6 most positive l2cs_yaw frames; driver crop, RetinaFace box, yaw/pitch/conf text. -> dict.""" | |
| import cv2 | |
| neg, pos = pick_contact_frames(m, caches) | |
| tiles, recs = [], [] | |
| for side, lst in (("neg", neg), ("pos", pos)): | |
| for yaw, u, t, fi in lst: | |
| fs = vio.FrameStream(u, [[t - 0.01, t + 0.01]], workers=workers, decode_wh=DECODE_WH, block=4, box_coords="frame") | |
| fr = box = None | |
| for t_, fi_, fr_, bx_ in fs: | |
| if fi_ == fi: | |
| fr, box = fr_, bx_ | |
| if fr is None: | |
| continue | |
| crop = vio.driver_crop(fr, box) | |
| if crop is None: | |
| continue | |
| crop = np.ascontiguousarray(crop) | |
| det = ex.detect_largest([crop])[0] if ex is not None else None | |
| mir = ex.infer([cv2.flip(crop, 1)])[0] if ex is not None else np.full(4, np.nan, np.float32) # label-free check: a mirrored crop should negate the horizontal angle | |
| rc = caches[u] | |
| feat, _ = value_at(rc, t) | |
| img = crop.copy() | |
| if det is not None: | |
| b = det[0] | |
| cv2.rectangle(img, (max(int(b[0]), 0), max(int(b[1]), 0)), (int(b[2]), int(b[3])), (0, 255, 0), 2) | |
| # gaze arrow from the face centre exactly as l2cs.vis.draw_gaze: dx = -sin(yaw) cos(pitch) (positive yaw -> image left), dy = -sin(pitch) (positive pitch -> UP) | |
| cxy = (int((b[0] + b[2]) / 2), int((b[1] + b[3]) / 2)) | |
| L = 0.8 * float(b[2] - b[0]) | |
| dx, dy = -L * math.sin(math.radians(feat[1])) * math.cos(math.radians(feat[2])), -L * math.sin(math.radians(feat[2])) | |
| cv2.arrowedLine(img, cxy, (int(cxy[0] + dx), int(cxy[1] + dy)), (0, 0, 255), 2, cv2.LINE_AA, tipLength=0.2) | |
| s = tile_h / img.shape[0] | |
| img = cv2.resize(img, (max(int(round(img.shape[1] * s)), 1), tile_h), interpolation=cv2.INTER_AREA) | |
| canvas = np.zeros((tile_h + 58, max(img.shape[1], 260), 3), np.uint8) | |
| canvas[:tile_h, :img.shape[1]] = img | |
| cv2.putText(canvas, f"l2cs_yaw {feat[1]:+.1f} pitch {feat[2]:+.1f}", (4, tile_h + 20), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA) | |
| cv2.putText(canvas, f"conf {feat[0]:.2f} t={t:.1f}s {side}", (4, tile_h + 40), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (200, 200, 200), 1, cv2.LINE_AA) | |
| cv2.putText(canvas, u.split("__")[0][:28] + " " + u.split("__")[1][:6], (4, tile_h + 55), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (160, 160, 160), 1, cv2.LINE_AA) | |
| tiles.append(canvas) | |
| recs.append({"side": side, "uid": u, "t": t, "frame_idx": fi, "l2cs_yaw": float(feat[1]), "l2cs_pitch": float(feat[2]), "l2cs_face_conf": float(feat[0]), | |
| "mirrored_crop_yaw": float(mir[1]), "mirrored_crop_pitch": float(mir[2])}) | |
| if not tiles: | |
| return {"path": None, "n_tiles": 0, "frames": []} | |
| W = max(t_.shape[1] for t_ in tiles) | |
| Hh = max(t_.shape[0] for t_ in tiles) | |
| cols = 4 | |
| rows = int(math.ceil(len(tiles) / cols)) | |
| head = 60 | |
| sheet = np.zeros((head + rows * Hh, cols * W, 3), np.uint8) | |
| cv2.putText(sheet, f"{BASELINE}: sign convention. Tiles 1-6 = most NEGATIVE l2cs_yaw, tiles 7-12 = most POSITIVE (one frame per route, conf >= 0.8). Red arrow = gaze drawn with", | |
| (6, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA) | |
| cv2.putText(sheet, "dx = -sin(yaw) cos(pitch), dy = -sin(pitch) (l2cs.vis.draw_gaze): positive yaw = toward IMAGE LEFT (camera's left), positive pitch = UP. Green = RetinaFace box (largest, score >= 0.5).", | |
| (6, 48), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA) | |
| for k, t_ in enumerate(tiles): | |
| r, c = divmod(k, cols) | |
| y0, x0 = head + r * Hh, c * W | |
| sheet[y0:y0 + t_.shape[0], x0:x0 + t_.shape[1]] = t_ | |
| Path(path).parent.mkdir(parents=True, exist_ok=True) | |
| cv2.imwrite(str(path), sheet) | |
| return {"path": str(path), "n_tiles": len(tiles), "frames": recs} | |
| def measurements(m, caches, route_infos, ex, wall_s, out_dir, sheet_workers=1, extra=None): | |
| j = {c: CHANNELS.index(c) for c in CHANNELS} | |
| n_frames = sum(rc.n for rc in caches.values()) | |
| fail = np.concatenate([rc.fail for rc in caches.values()]) if caches else np.zeros(0, np.uint8) | |
| valid = fail == 0 | |
| had_box = fail != 1 | |
| had_frame = fail != 2 | |
| hist = {cs.FAIL_CODES[int(k)]: int(v) for k, v in zip(*np.unique(fail, return_counts=True))} | |
| dec50, dec95 = pq([ri["runtime"]["decode_ms_p50"] for ri in route_infos if ri.get("runtime")]), pq([ri["runtime"]["decode_ms_p95"] for ri in route_infos if ri.get("runtime")]) | |
| inf50, inf95 = pq([ri["runtime"]["infer_ms_p50"] for ri in route_infos if ri.get("runtime")]), pq([ri["runtime"]["infer_ms_p95"] for ri in route_infos if ri.get("runtime")]) | |
| anchors, F = [], [] | |
| for r in m.itertuples(): | |
| rc = caches.get(r.uid) | |
| if rc is None: | |
| anchors.append({"idx": int(r.idx), "uid": r.uid, "regime": r.regime, "y": int(r.y), "t_anchor": float(r.t_anchor), "cache": False}) | |
| continue | |
| f, code = value_at(rc, float(r.t_anchor)) | |
| H = history(rc, float(r.t_anchor)) | |
| rec = {"idx": int(r.idx), "uid": r.uid, "regime": r.regime, "y": int(r.y), "t_anchor": float(r.t_anchor), "cache": True, "fail_at_anchor": code, | |
| "valid_share_8s": float(np.isfinite(H[:, j["l2cs_yaw"]]).mean()), "mean_abs_yaw_8s": float(np.nanmean(np.abs(H[:, j["l2cs_yaw"]]))) if np.isfinite(H[:, j["l2cs_yaw"]]).any() else None, | |
| "max_abs_yaw_3s": float(np.nanmax(np.abs(H[-30:, j["l2cs_yaw"]]))) if np.isfinite(H[-30:, j["l2cs_yaw"]]).any() else None} | |
| for c in CHANNELS: | |
| rec[c] = (float(f[j[c]]) if np.isfinite(f[j[c]]) else None) | |
| anchors.append(rec) | |
| F.append(rc.feat.astype(np.float32)[rc.valid]) | |
| F = np.concatenate(F) if F else np.zeros((0, len(CHANNELS)), np.float32) | |
| dist = {c: {"p05": float(np.percentile(F[:, j[c]], 5)), "p50": float(np.percentile(F[:, j[c]], 50)), "p95": float(np.percentile(F[:, j[c]], 95)), | |
| "min": float(F[:, j[c]].min()), "max": float(F[:, j[c]].max())} for c in CHANNELS} if len(F) else {} | |
| desc = {} | |
| for reg in ("manual", "engaged", "all"): | |
| for y in (0, 1): | |
| v = [a["max_abs_yaw_3s"] for a in anchors if a.get("cache") and a["max_abs_yaw_3s"] is not None and a["y"] == y and (reg == "all" or a["regime"] == reg)] | |
| desc[f"{reg}_y{y}"] = {"n": len(v), "median_max_abs_yaw_3s": float(np.median(v)) if v else None} | |
| rho = rho_vs_face_pack(m, caches) | |
| # timing percentiles BEFORE the contact sheet (its 12 single-crop mirror checks go through infer() and would otherwise enter the timer lists) | |
| timers = {k: (list(v) if isinstance(v, list) else v) for k, v in ex.timers.items()} if ex is not None else {} | |
| det50, det95 = pq(timers["det_ms"]) if ex is not None else (float("nan"), float("nan")) | |
| gz50, gz95 = pq(timers["gaze_ms"]) if ex is not None else (float("nan"), float("nan")) | |
| sheet = contact_sheet(ex, m, caches, out_dir / "sign_convention_contact_sheet.png", workers=sheet_workers) | |
| M = {"baseline": BASELINE, "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "manifest": str(MANIFEST), "manifest_sha256": cs.sha256_file(MANIFEST), | |
| "n_anchors": int(len(m)), "n_routes": int(m.uid.nunique()), "n_routes_cached": len(caches), "channels": CHANNELS, "roles": ROLES, "units": UNITS, | |
| "decode_wh": "native", "native_wh_by_route": sorted({tuple(rc.meta.get("native_wh") or []) for rc in caches.values()}), | |
| "face_conf_threshold": FACE_CONF_TH, "horizontal_is_first_pipeline_output": HORIZONTAL_IS_FIRST_OUTPUT, | |
| "model_ids": (ex.model_ids if ex else None), "weight_sha256": (ex.weight_sha256 if ex else None), "lib_versions": (ex.lib_versions if ex else None), | |
| "n_params_M_l2cs": (getattr(ex, "n_params_M", None) if ex else None), | |
| "n_frames": int(n_frames), "n_frames_with_box": int(had_box.sum()), "n_frames_decoded": int(had_frame.sum()), | |
| "valid_share": float(valid.mean()) if n_frames else None, "valid_share_with_box": float(valid[had_box].mean()) if had_box.any() else None, | |
| "fail_hist": hist, "n_multi_face_crops": (timers["n_multi_face"] if ex else None), "n_largest_face_not_top_score": (timers["n_largest_not_top_score"] if ex else None), | |
| "n_faces_gazed": (timers["n_faces"] if ex else None), "n_infer_calls": (timers["n_infer_calls"] if ex else None), | |
| "raw_timers_ms_per_frame": ({"det_ms": [round(x, 3) for x in timers["det_ms"]], "gaze_ms": [round(x, 3) for x in timers["gaze_ms"]]} if ex else None), | |
| "per_frame_ms": {"decode_p50_route_median": dec50[0], "decode_p95_route_median": dec95[0], "infer_p50_route_median": inf50[0], "infer_p95_route_median": inf95[0], | |
| "infer_p50_route_max": inf50[1], "det_ms_p50": det50, "det_ms_p95": det95, "gaze_ms_p50": gz50, "gaze_ms_p95": gz95, | |
| "note": "decode/infer: per-route p50/p95 from the cache runtime (per-frame ms per decode block / infer batch); *_route_median = median over routes, *_route_max = p95 over routes. det/gaze: this extractor's per-frame ms per infer() call (RetinaFace on the driver crop / L2CS on the face), pooled over calls."}, | |
| "wall_s": float(wall_s), "fps_end_to_end": (float(n_frames / wall_s) if wall_s else None), | |
| "peak_rss_mb": rss_peaks_mb(), "peak_vram_mb": dict((ex.gpu_peak_mb() if ex else {}), nvidia_smi_process_mb=gpu_process_mb()), | |
| "value_distribution_valid_frames": dist, "descriptive_by_y_max_abs_yaw_3s": desc, "anchors": anchors, "rho_vs_lm_yaw_dev": rho, "contact_sheet": sheet, | |
| "routes": route_infos} | |
| if extra: | |
| M.update(extra) | |
| return M | |
| def write_json(path, obj): | |
| Path(path).parent.mkdir(parents=True, exist_ok=True) | |
| def default(o): | |
| if isinstance(o, (np.floating, np.integer)): | |
| return o.item() | |
| if isinstance(o, np.ndarray): | |
| return o.tolist() | |
| return str(o) | |
| Path(path).write_text(json.dumps(obj, indent=1, default=default, allow_nan=True)) | |
| return path | |
| def load_caches(m, cache_dir): | |
| out = {} | |
| for u in m.uid.unique(): | |
| p = Path(cache_dir) / f"{u}.npz" | |
| if p.exists(): | |
| out[u] = cs.RouteCache.load(p) | |
| return out | |
| # ----------------------------------------------------------------------------- CLI actions | |
| def self_test(uid=None, workers=2, n_frames=30, device="cuda:0"): | |
| m = load_manifest() | |
| r = m.iloc[0] if not uid else m[m.uid == uid].iloc[0] | |
| t_a = float(r.t_anchor) | |
| k_a = int(round(t_a * cs.HZ)) | |
| span = np.array([[(k_a - n_frames + 1) / cs.HZ + 0.01, k_a / cs.HZ - 0.01]], np.float32) # +-0.01 s: exactly n_frames grid steps despite floor / ceil | |
| ok, res = check_resources() | |
| assert ok, f"resources: {res}" | |
| ex = L2CSGazeExtractor().setup(device=device) | |
| # infer contract: None / empty crops -> NaN rows; a crop without a face -> NaN row (fail 3) | |
| z = ex.infer([None, np.zeros((0, 0, 3), np.uint8), np.zeros((64, 64, 3), np.uint8)]) | |
| assert z.shape == (3, 4) and np.isnan(z).all(), z | |
| ex.timers = {"det_ms": [], "gaze_ms": [], "n_infer_calls": 0, "n_faces": 0, "n_multi_face": 0, "n_largest_not_top_score": 0} | |
| d = OUT_DIR / "self_test" | |
| d.mkdir(parents=True, exist_ok=True) | |
| t0 = time.time() | |
| code, path, info = run_route(ex, r.uid, span, d, max_minutes=10, workers=min(workers, MAX_WORKERS), block=16, force=True) | |
| rc = cs.RouteCache.load(path) | |
| assert code == EXIT_DONE and rc.C == 4 and rc.channels == CHANNELS and rc.n == n_frames and rc.meta["baseline"] == BASELINE and rc.meta["decode_wh"] == rc.meta["native_wh"], (code, rc.n, rc.channels, rc.meta.get("decode_wh")) | |
| f = rc.feat.astype(np.float32) | |
| assert rc.valid.any(), "no valid frame in the self-test window" | |
| v = f[rc.valid] | |
| assert np.all((v[:, 0] >= FACE_CONF_TH) & (v[:, 0] <= 1.0)) and np.all(np.abs(v[:, 1:3]) <= 180.0) and np.all(v[:, 3] > 0), "value ranges" | |
| rep = {"self_test": BASELINE, "route": r.uid, "t_anchor": t_a, "span": span.tolist(), "n_frames": rc.n, "fail_hist": info["fail_hist"], "valid_rate": info["valid_rate"], | |
| "runtime": rc.runtime, "det_ms_p50_p95": pq(ex.timers["det_ms"]), "gaze_ms_p50_p95": pq(ex.timers["gaze_ms"]), "wall_s": round(time.time() - t0, 2), | |
| "values_at_anchor": dict(zip(CHANNELS, [float(x) for x in value_at(rc, t_a)[0]])), | |
| "yaw_deg_valid_min_med_max": [float(v[:, 1].min()), float(np.median(v[:, 1])), float(v[:, 1].max())], "pitch_deg_valid_min_med_max": [float(v[:, 2].min()), float(np.median(v[:, 2])), float(v[:, 2].max())], | |
| "box_w_px_med": float(np.median(v[:, 3])), "n_multi_face": ex.timers["n_multi_face"], "peak": dict(rss_peaks_mb(), **ex.gpu_peak_mb()), "describe": ex.describe(), "spec": ex.spec(), | |
| "resources_before": res, "cache": str(path)} | |
| sheet = contact_sheet(ex, m[m.uid == r.uid].head(1), {r.uid: rc}, d / "self_test_sheet.png", workers=1) | |
| rep["sheet"] = sheet["path"] | |
| write_json(d / "self_test.json", rep) | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return 0 | |
| def phase1(workers=3, block=60, max_minutes=120.0, force=False, device="cuda:0", out_dir=OUT_DIR): | |
| out_dir = Path(out_dir) | |
| cache_dir = out_dir / "cache" | |
| (out_dir / "logs").mkdir(parents=True, exist_ok=True) | |
| m = load_manifest() | |
| S = spans_by_uid(m) | |
| ok, res = check_resources() | |
| if not ok: | |
| print(json.dumps({"abort": "resources", **res})) | |
| return EXIT_FAIL | |
| workers = max(1, min(int(workers), MAX_WORKERS)) | |
| ex = L2CSGazeExtractor().setup(device=device) | |
| t_start = time.time() | |
| infos, fails, budget = [], [], False | |
| uids = list(m.uid.unique()) # manifest order (causal frame order inside every route) | |
| for i, u in enumerate(uids): | |
| if time.time() - t_start > max_minutes * 60: | |
| budget = True | |
| break | |
| code, path, info = run_route(ex, u, S[u], cache_dir, max_minutes=max_minutes, workers=workers, block=block, force=force, t_start=t_start) | |
| print(f"[{i + 1}/{len(uids)}] {u[:64]} {info['status']} frames {info.get('n_frames')} valid {None if info.get('valid_rate') is None else round(info['valid_rate'], 3)} wall {info.get('wall_s')}s", flush=True) | |
| (fails if code == EXIT_FAIL else infos).append(info) | |
| if code == EXIT_BUDGET: | |
| budget = True | |
| break | |
| wall = time.time() - t_start | |
| caches = load_caches(m, cache_dir) | |
| for inf in infos: # 'exists' routes carry no runtime -> take it from the cache | |
| if not inf.get("runtime") and inf["uid"] in caches: | |
| inf["runtime"] = caches[inf["uid"]].runtime | |
| inf["fail_hist"] = {cs.FAIL_CODES[int(k)]: int(v) for k, v in zip(*np.unique(caches[inf["uid"]].fail, return_counts=True))} | |
| n_new = sum(int(inf.get("n_new_frames", 0)) for inf in infos) | |
| M = measurements(m, caches, infos + fails, ex, wall, out_dir, sheet_workers=1, | |
| extra={"phase1_run": {"workers": workers, "block": block, "max_minutes": max_minutes, "force": force, "budget_hit": budget, "n_routes_done": len(infos), "n_routes_failed": len(fails), | |
| "n_new_frames_this_run": n_new, "wall_s_this_run": round(wall, 1), "resources_before": res, | |
| "fps_note": "fps_end_to_end = frames in the caches / wall seconds of this run (decode + crop + detect + gaze + write, sequential routes); only meaningful when every route was computed in this run (n_new_frames_this_run == n_frames)"}}) | |
| M["fps_end_to_end"] = float(n_new / wall) if (wall and n_new) else M["fps_end_to_end"] | |
| write_json(out_dir / "measurements.json", M) | |
| write_json(out_dir / "logs" / f"phase1_{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}.json", {"routes": infos, "fails": fails, "wall_s": wall, "budget_hit": budget}) | |
| ex.teardown() | |
| summ = {k: M[k] for k in ("n_frames", "valid_share", "valid_share_with_box", "fail_hist", "fps_end_to_end", "wall_s", "peak_rss_mb", "peak_vram_mb")} | |
| summ["infer_ms_p50_route_median"] = M["per_frame_ms"]["infer_p50_route_median"] | |
| summ["rho_pooled"] = {k: v["rho"] for k, v in M["rho_vs_lm_yaw_dev"]["pooled"].items()} | |
| summ["rho_pooled_circular_centred"] = {k: v["rho"] for k, v in M["rho_vs_lm_yaw_dev"]["pooled_circular_centred_route_causal_median"].items() if isinstance(v, dict) and "rho" in v} | |
| summ["contact_sheet"] = M["contact_sheet"]["path"] | |
| print(json.dumps(summ, indent=1, default=str)) | |
| return EXIT_BUDGET if budget else (EXIT_FAIL if fails else EXIT_DONE) | |
| def measure_only(device="cuda:0", out_dir=OUT_DIR, with_model=True): | |
| out_dir = Path(out_dir) | |
| m = load_manifest() | |
| caches = load_caches(m, out_dir / "cache") | |
| infos = [{"uid": u, "status": "exists", "n_frames": rc.n, "valid_rate": float(rc.valid.mean()), "fail_hist": {cs.FAIL_CODES[int(k)]: int(v) for k, v in zip(*np.unique(rc.fail, return_counts=True))}, | |
| "runtime": rc.runtime, "wall_s": rc.runtime.get("wall_s")} for u, rc in caches.items()] | |
| wall = float(sum((rc.runtime.get("wall_s") or 0.0) for rc in caches.values())) | |
| ex = L2CSGazeExtractor().setup(device=device) if with_model else None | |
| old = json.loads((out_dir / "measurements.json").read_text()) if (out_dir / "measurements.json").exists() else {} | |
| M = measurements(m, caches, infos, ex, wall, out_dir, sheet_workers=1, extra={"phase1_run": old.get("phase1_run"), "measure_only": True, | |
| "fps_note": "measure-only: wall_s = sum of per-route cache wall seconds; fps = n_frames / that"}) | |
| if old: | |
| for k in ("per_frame_ms", "peak_rss_mb", "peak_vram_mb", "n_multi_face_crops", "n_largest_face_not_top_score", "n_faces_gazed", "n_infer_calls", "raw_timers_ms_per_frame", "fps_end_to_end", "wall_s"): | |
| if k in old: | |
| M[k] = old[k] | |
| write_json(out_dir / "measurements.json", M) | |
| print(json.dumps({k: M[k] for k in ("n_frames", "valid_share", "valid_share_with_box", "fail_hist")}, indent=1)) | |
| print(json.dumps(M["rho_vs_lm_yaw_dev"]["pooled"], indent=1)) | |
| print(M["contact_sheet"]["path"]) | |
| return 0 | |
| def main(): | |
| 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("--measure-only", action="store_true") | |
| ap.add_argument("--uid", default="") | |
| ap.add_argument("--workers", type=int, default=3) | |
| ap.add_argument("--block", type=int, default=60) | |
| ap.add_argument("--max-minutes", type=float, default=120.0) | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--device", default="cuda:0") | |
| ap.add_argument("--out-dir", default=str(OUT_DIR)) | |
| a = ap.parse_args() | |
| if a.phase1: | |
| return phase1(a.workers, a.block, a.max_minutes, a.force, a.device, Path(a.out_dir)) | |
| if a.measure_only: | |
| return measure_only(a.device, Path(a.out_dir)) | |
| return self_test(a.uid or None, min(a.workers, 2), device=a.device) | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 48.8 kB
- Xet hash:
- 4116d1ff3a1dadc61d11f1d92433024ba2131a1c597555e66f9cc0b7faa37d5a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.