Buckets:
| """cwm3/baselines_dms/extractors/openface3.py — O2 OpenFace 3.0 (CMU, FG 2025) per-frame extractor of the DMS-baseline pipeline (plan §3, 2026-09-14). | |
| Models (registry id `openface3`; every file sha256-verified by registry.resolve; code = PyPI `openface-test` 0.1.26 imported through baselines_dms/of3_compat.py) | |
| RetinaFace mobilenet0.25 Alignment_RetinaFace.pth face detection on the driver crop -> the TOP-SCORING face (x1, y1, x2, y2, score), as FaceDetector.get_face | |
| STAR 300W (68 pts) Landmark_68.pkl 68 iBUG landmarks in crop pixels from a 256 x 256 warp of the face box (LandmarkDetector.detect_landmarks geometry) | |
| MTL tf_efficientnet_b0 MTL_backbone.pth emotion logits [8], gaze (yaw, pitch) [2], action units [8] from the face box (MultitaskPredictor.predict) | |
| (Landmark_98.pkl is NOT loaded: no channel needs the WFLW points) | |
| Channels (C = 24, prefix of3_; the whole row is NaN when no face reaches the package threshold 0.5 -> fail code 3) | |
| of3_face_conf RetinaFace softmax score of the used face, 0.5 .. 1 role conf | |
| of3_gaze_yaw, of3_gaze_pitch MTL gaze regression; the package returns RADIANS (|value| < pi on every frame, asserted in --phase1), stored in DEGREES role angle_deg | |
| of3_head_yaw/_pitch/_roll the package exposes no head pose -> cv2.solvePnP (ITERATIVE, frontal initial guess) of 6 of the 68 landmarks (iBUG 30 nose tip, | |
| 8 chin, 36 / 45 outer eye corners, 48 / 54 mouth corners) against the standard 6-point anthropometric 3D face model | |
| (Mallick / learnopencv constants, identical to explore/opendriver/odlib.py FACE3D; no 68-point 3D model is available offline); | |
| pinhole f = max(crop_w, crop_h) px, principal point = centre of the detected face box (pose relative to the ray through the | |
| face, no perspective offset from the face position inside the driver crop). DEGREES. Signs (verified on synthetic | |
| projections in --self-test): yaw + = nose toward IMAGE RIGHT; pitch + = nose UP; roll + = top of the head toward IMAGE RIGHT | |
| (clockwise for the viewer). Whether image-right is the driver's left or right depends on the cabin camera (see the contact sheet). | |
| of3_au_01 .. of3_au_08 MTL action-unit head outputs in package order (ME-GraphAU cosine head; no activation applied by the package; observed in 0 .. 1) role prob | |
| of3_eye_open_l, of3_eye_open_r eye aspect ratio (Soukupova & Cech 2016) = (|p2-p6| + |p3-p5|) / (2 |p1-p4|) of iBUG 42-47 (the IMAGE-RIGHT eye, "l" in | |
| iBUG naming) and iBUG 36-41 (the IMAGE-LEFT eye, "r"); dimensionless, typically 0.1 .. 0.45 role eye_open_ratio | |
| of3_emo_<neutral,happy,sad,surprise,fear,disgust,anger,contempt> MTL emotion logits, AffectNet order (stored, role none; not a primary channel) role none | |
| Preprocessing (input = video_io.driver_crop of the causal driver box, +10 %, at NATIVE resolution: decode_wh = None; u8 BGR exactly as decoded) | |
| RetinaFace float32 BGR minus (104, 117, 123), no resize; crops of a batch are zero-padded bottom/right to a common canvas (multiple of 128); PriorBox per canvas | |
| STAR the package's warp (scale = min(w, h) / 200 * 1.05 of the int-truncated face box, centre = box centre) to 256 x 256, BGR (as its training loader), /255 * 2 - 1 | |
| MTL face box (int-truncated, clipped to the crop) -> BGR2RGB -> ToTensor -> Resize 224 x 224 -> ImageNet mean / std (the package's own transform object) | |
| Runtime flags reset after loading (STAR's utility.set_environment turns them on): cudnn.benchmark False, autograd anomaly detection False. fp32 throughout. | |
| Phase-1 stall (first run, 2026-09-14, 10/45 routes): route 11 = TOYOTA_PRIUS route_0-13 has two manifest windows 600 s apart; cli_phase1 handed the MERGED spans | |
| to run_route with 64-frame blocks at native 1928 x 1208, so the block straddling the gap made video_io.decode_block grab ~12,000 HEVC frames in one worker | |
| (minutes without a log line -> the run was killed -> the live pool's 5 semaphores leaked = the resource_tracker warning in phase1_run.log; not a hang in the | |
| models). Fix: ONE run_route call per contiguous span (the cache is extended through run_route's resume path, as ov_action does); independently, | |
| video_io.decode_block now re-seeks across gaps > 60 frames and is index-exact on every header fps (native_fps / seek_corrected in the cache meta). | |
| measure() also reports the eye-state check: Spearman rho of the mean EAR (of3_eye_open_l/r) vs the face pack `eye_open` over the same history steps. | |
| python baselines_dms/extractors/openface3.py --self-test [--uid UID] # synthetic pose-sign check + 1 local TRAIN route, 30 frames (3 s before its anchor) | |
| python baselines_dms/extractors/openface3.py --phase1 [--workers 3] [--max-minutes 60] [--force] | |
| # 50 frozen manifest windows -> results/baselines_dms/phase1/openface3/{cache/,measurements.json,contact_sheet_sign.png} | |
| python baselines_dms/extractors/openface3.py --spec # channel roles / summarise spec as JSON | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import os | |
| import resource | |
| import sys | |
| import time | |
| import warnings | |
| 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 | |
| from base import EXIT_BUDGET, EXIT_DONE, EXIT_FAIL, Extractor, run_route # noqa: E402 | |
| BASELINE = "openface3" | |
| REGISTRY_ID = "openface3" | |
| PHASE1_ROOT = CWM3 / "results" / "baselines_dms" / "phase1" | |
| MANIFEST = PHASE1_ROOT / "manifest_v1.csv" | |
| OUT_DIR = PHASE1_ROOT / BASELINE | |
| FACE_PACK = CWM3 / "results" / "targets_face_causal_routes_v2" | |
| EMOTIONS = ("neutral", "happy", "sad", "surprise", "fear", "disgust", "anger", "contempt") | |
| AU_NAMES = [f"of3_au_{i:02d}" for i in range(1, 9)] | |
| EMO_NAMES = [f"of3_emo_{e}" for e in EMOTIONS] | |
| CHANNELS = ["of3_face_conf", "of3_gaze_yaw", "of3_gaze_pitch", "of3_head_yaw", "of3_head_pitch", "of3_head_roll", *AU_NAMES, "of3_eye_open_l", "of3_eye_open_r", *EMO_NAMES] | |
| ROLES = {"of3_face_conf": "conf", "of3_gaze_yaw": "angle_deg", "of3_gaze_pitch": "angle_deg", "of3_head_yaw": "angle_deg", "of3_head_pitch": "angle_deg", "of3_head_roll": "angle_deg", | |
| **{c: "prob" for c in AU_NAMES}, "of3_eye_open_l": "eye_open_ratio", "of3_eye_open_r": "eye_open_ratio", **{c: "none" for c in EMO_NAMES}} | |
| # summarise.SPECS-format spec: rule constants are the decisions.md ones (yaw 20/12°, pitch 15/9°); the EAR closure threshold 0.2 is the literature constant of | |
| # Soukupova & Cech (2016), not tuned here; the fixed rule's `closed` stays None (decisions.md defines it on an eye-CLOSED probability only); emotion is excluded. | |
| SUMMARISE_SPEC = {"angle": {"of3_head_yaw": (20.0, 12.0), "of3_head_pitch": (15.0, 9.0), "of3_gaze_yaw": (20.0, 12.0), "of3_gaze_pitch": (15.0, 9.0)}, | |
| "eye_state": {"of3_eye_open_l": {"kind": "open_ratio", "th": 0.2}, "of3_eye_open_r": {"kind": "open_ratio", "th": 0.2}}, | |
| "prob": ["of3_face_conf", *AU_NAMES], "embedding": [], | |
| "rule": {"yaw": "of3_head_yaw", "pitch": "of3_head_pitch", "closed": None}, | |
| "chain": {"head_dev": ("of3_head_yaw", "of3_head_pitch"), "closed": None, "gaze_off": ("of3_gaze_yaw", "of3_gaze_pitch")}, | |
| "excluded_channels": EMO_NAMES} | |
| FACE_TH = 0.5 # package: FaceDetector.vis_threshold == LandmarkDetector confidence_threshold | |
| MIN_FACE_PX = 8 # a face box smaller than this after clipping counts as no face | |
| WARP = 256 # STAR input size (Alignment.input_size) | |
| CANVAS_STEP = 128 # RetinaFace batch canvas is padded to a multiple of this (few distinct shapes -> few first-call cuDNN costs of ~1 s each) | |
| FACE3D = {30: (0.0, 0.0, 0.0), 8: (0.0, -330.0, -65.0), 36: (-225.0, 170.0, -135.0), 45: (225.0, 170.0, -135.0), 48: (-150.0, -150.0, -125.0), 54: (150.0, -150.0, -125.0)} | |
| PNP_IDX = sorted(FACE3D) | |
| PNP_3D = np.array([FACE3D[i] for i in PNP_IDX], np.float64) | |
| F_FLIP = np.diag([1.0, -1.0, -1.0]) # model frame (x image-right, y up, z toward the camera) -> OpenCV camera frame (x right, y down, z forward) | |
| EYE_R, EYE_L = list(range(36, 42)), list(range(42, 48)) | |
| CH = {c: i for i, c in enumerate(CHANNELS)} | |
| # ----------------------------------------------------------------------------- geometry helpers (numpy / cv2 only) | |
| def eye_aspect_ratio(p): | |
| """p [6,2] in iBUG order p1..p6 -> (|p2-p6| + |p3-p5|) / (2 |p1-p4|); NaN when the eye width is 0.""" | |
| d = 2.0 * float(np.hypot(*(p[0] - p[3]))) | |
| return float((np.hypot(*(p[1] - p[5])) + np.hypot(*(p[2] - p[4])) ) / d) if d > 0 else float("nan") | |
| def head_pose_deg(pts68, f, cx, cy, face_w): | |
| """68 landmarks (crop px) -> (yaw, pitch, roll) degrees; see the module docstring for the model, the camera and the sign conventions.""" | |
| import cv2 | |
| pts2d = np.ascontiguousarray(pts68[PNP_IDX].astype(np.float64)) | |
| if not np.isfinite(pts2d).all(): | |
| return (np.nan, np.nan, np.nan) | |
| K = np.array([[f, 0.0, cx], [0.0, f, cy], [0.0, 0.0, 1.0]], np.float64) | |
| rvec0 = cv2.Rodrigues(F_FLIP)[0] | |
| tvec0 = np.array([[0.0], [0.0], [f * 450.0 / max(float(face_w), 1.0)]], np.float64) | |
| ok, rvec, tvec = cv2.solvePnP(PNP_3D, pts2d, K, None, rvec0, tvec0, useExtrinsicGuess=True, flags=cv2.SOLVEPNP_ITERATIVE) | |
| if not ok or not np.isfinite(rvec).all() or tvec[2, 0] <= 0: | |
| return (np.nan, np.nan, np.nan) | |
| R = cv2.Rodrigues(rvec)[0] | |
| Rr = F_FLIP @ R # rotation relative to the frontal pose, expressed in the model frame | |
| n, x = Rr[:, 2], Rr[:, 0] # rotated nose direction / ear-to-ear axis | |
| yaw = math.degrees(math.atan2(n[0], n[2])) | |
| pitch = math.degrees(math.atan2(n[1], math.hypot(n[0], n[2]))) | |
| roll = -math.degrees(math.atan2(x[1], x[0])) | |
| return (yaw, pitch, roll) | |
| def _rot(axis, deg): | |
| a = math.radians(deg) | |
| c, s = math.cos(a), math.sin(a) | |
| return {"x": np.array([[1, 0, 0], [0, c, -s], [0, s, c]]), "y": np.array([[c, 0, s], [0, 1, 0], [-s, 0, c]]), "z": np.array([[c, -s, 0], [s, c, 0], [0, 0, 1]])}[axis] | |
| def synthetic_pose_check(f=1000.0, cx=300.0, cy=300.0, z=2000.0, tol_deg=0.5): | |
| """project the 3D model under known rotations and recover them with head_pose_deg -> {case: (expected, recovered)}; asserts the sign conventions.""" | |
| cases = {"yaw+20 (nose to image right)": (_rot("y", 20.0), (20.0, 0.0, 0.0)), "yaw-30": (_rot("y", -30.0), (-30.0, 0.0, 0.0)), | |
| "pitch+15 (nose up)": (_rot("x", -15.0), (0.0, 15.0, 0.0)), "pitch-15": (_rot("x", 15.0), (0.0, -15.0, 0.0)), | |
| "roll+10 (top of head to image right)": (_rot("z", -10.0), (0.0, 0.0, 10.0)), "roll-10": (_rot("z", 10.0), (0.0, 0.0, -10.0)), | |
| "yaw+25 pitch+10": (_rot("y", 25.0) @ _rot("x", -10.0), (25.0, 10.0, None))} | |
| rep = {} | |
| for name, (Rm, exp) in cases.items(): | |
| Xc = (F_FLIP @ Rm @ PNP_3D.T).T + np.array([0.0, 0.0, z]) | |
| uv = np.stack([f * Xc[:, 0] / Xc[:, 2] + cx, f * Xc[:, 1] / Xc[:, 2] + cy], 1) | |
| pts = np.full((68, 2), np.nan) | |
| pts[PNP_IDX] = uv | |
| got = head_pose_deg(pts, f, cx, cy, face_w=float(uv[:, 0].max() - uv[:, 0].min())) | |
| rep[name] = {"expected": exp, "recovered": [round(float(g), 3) for g in got]} | |
| for e, g in zip(exp, got): | |
| assert e is None or abs(e - g) < tol_deg, (name, exp, got) | |
| # frontal face: eyes must project ABOVE the nose (image y down) — the F_FLIP convention | |
| Xc = (F_FLIP @ PNP_3D.T).T + np.array([0.0, 0.0, z]) | |
| v = f * Xc[:, 1] / Xc[:, 2] + cy | |
| assert v[PNP_IDX.index(36)] < v[PNP_IDX.index(30)] < v[PNP_IDX.index(8)], "frontal projection orientation" | |
| return rep | |
| # ----------------------------------------------------------------------------- the extractor | |
| class OpenFace3Extractor(Extractor): | |
| baseline = BASELINE | |
| channels = CHANNELS | |
| model_ids = ["openface3:Alignment_RetinaFace.pth", "openface3:Landmark_68.pkl", "openface3:MTL_backbone.pth", "pip:openface-test==0.1.26"] | |
| needs_crop = True | |
| decode_wh = None # native resolution: RetinaFace / STAR / MTL are resolution-hungry and the models resize internally | |
| batch = 32 | |
| def __init__(self, device="cuda:0"): | |
| self.device = str(device) | |
| self.infer_ms = [] # ms per crop, one entry per infer() call | |
| self.stage_ms = {"det": [], "lm": [], "mtl": [], "pnp_ear": []} | |
| self.n_crops = 0 | |
| self.n_faces = 0 | |
| self.gaze_raw_absmax = 0.0 | |
| self._priors = {} | |
| # -- setup | |
| def setup(self, registry_entry=None): | |
| import importlib.metadata as md | |
| import cv2 | |
| import timm | |
| import torch | |
| import torchvision | |
| import scipy | |
| from baselines_dms import of3_compat | |
| from baselines_dms import registry as R | |
| entry = registry_entry or R.get(REGISTRY_ID) | |
| wdir = R.resolve(REGISTRY_ID) # asserts the sha256 of every file | |
| self.registry_entry = {k: entry.get(k) for k in ("id", "version", "sha256", "local_path", "license")} | |
| of3_compat.apply() | |
| warnings.filterwarnings("ignore", category=FutureWarning) | |
| warnings.filterwarnings("ignore", category=UserWarning) | |
| from openface.face_detection import FaceDetector | |
| from openface.multitask_model import MultitaskPredictor | |
| self.fd = FaceDetector(model_path=str(wdir / "Alignment_RetinaFace.pth"), device=self.device) | |
| self.ld = of3_compat.landmark_detector(device=self.device, landmarks=68, weights_dir=wdir) | |
| self.mt = MultitaskPredictor(model_path=str(wdir / "MTL_backbone.pth"), device=self.device) | |
| torch.backends.cudnn.benchmark = False # STAR's set_environment turned it on; canvas sizes vary | |
| torch.autograd.set_detect_anomaly(False) # idem (debug flag) | |
| torch.set_num_threads(int(os.environ.get("OMP_NUM_THREADS", "4"))) | |
| self.dev = torch.device(self.device) | |
| self.rf_mean = torch.tensor([104.0, 117.0, 123.0], device=self.dev).view(1, 3, 1, 1) | |
| self.weight_sha256 = {f["name"]: f["sha256"] for f in entry.get("files", []) if f["name"] in ("Alignment_RetinaFace.pth", "Landmark_68.pkl", "MTL_backbone.pth")} | |
| self.lib_versions = {"torch": torch.__version__, "torchvision": torchvision.__version__, "timm": timm.__version__, "opencv": cv2.__version__, "numpy": np.__version__, | |
| "scipy": scipy.__version__, "openface-test": md.version("openface-test")} | |
| self.infer([np.full((320, 200, 3), 127, np.uint8)]) # warm-up (no face -> NaN row); timings discarded | |
| self.infer_ms.clear() | |
| for v in self.stage_ms.values(): | |
| v.clear() | |
| self.n_crops = self.n_faces = 0 | |
| return self | |
| def spec(self): | |
| return dict(ROLES) | |
| def summarise_spec(self): | |
| return json.loads(json.dumps(SUMMARISE_SPEC)) | |
| def describe(self): | |
| d = super().describe() | |
| d.update(roles=dict(ROLES), summarise_spec=self.summarise_spec(), face_threshold=FACE_TH, pnp_points=PNP_IDX, pnp_model=FACE3D, device=self.device) | |
| return d | |
| # -- stages | |
| def _sync(self): | |
| import torch | |
| if self.dev.type == "cuda": | |
| torch.cuda.synchronize(self.dev) | |
| def _detect(self, imgs): | |
| """list of u8 BGR crops -> f32[m,5] (x1, y1, x2, y2, score) of the top-scoring RetinaFace box per crop (crop px).""" | |
| import torch | |
| from openface.Pytorch_Retinaface.layers.functions.prior_box import PriorBox | |
| from openface.Pytorch_Retinaface.utils.box_utils import decode | |
| m = len(imgs) | |
| Hc = int(math.ceil(max(im.shape[0] for im in imgs) / float(CANVAS_STEP)) * CANVAS_STEP) | |
| Wc = int(math.ceil(max(im.shape[1] for im in imgs) / float(CANVAS_STEP)) * CANVAS_STEP) | |
| canvas = np.zeros((m, Hc, Wc, 3), np.uint8) | |
| for j, im in enumerate(imgs): | |
| canvas[j, :im.shape[0], :im.shape[1]] = im | |
| x = torch.from_numpy(canvas).to(self.dev).permute(0, 3, 1, 2).float() - self.rf_mean | |
| loc, conf, _landms = self.fd.model(x) | |
| key = (Hc, Wc) | |
| if key not in self._priors: | |
| self._priors[key] = PriorBox(self.fd.cfg, image_size=(Hc, Wc)).forward().to(self.dev) | |
| priors = self._priors[key] | |
| scores = conf[:, :, 1] | |
| best = scores.argmax(1) | |
| sc = scores.gather(1, best[:, None])[:, 0] | |
| boxes = decode(loc[torch.arange(m, device=self.dev), best], priors[best], self.fd.cfg["variance"]) | |
| boxes = boxes * torch.tensor([Wc, Hc, Wc, Hc], device=self.dev, dtype=boxes.dtype) | |
| return torch.cat([boxes, sc[:, None]], 1).cpu().numpy().astype(np.float32) | |
| def _landmarks(self, imgs, boxes): | |
| """crops + int face boxes -> f32[k,68,2] iBUG landmarks in crop px (the package's warp geometry, batched).""" | |
| import torch | |
| al = self.ld.alignment | |
| k = len(imgs) | |
| warps = np.zeros((k, WARP, WARP, 3), np.uint8) | |
| mats = np.zeros((k, 3, 3), np.float64) | |
| for j, (im, (x1, y1, x2, y2)) in enumerate(zip(imgs, boxes)): | |
| scale = min(x2 - x1, y2 - y1) / 200.0 * 1.05 | |
| M = al.getCropMatrix.process(float(scale), float((x2 + x1) / 2.0), float((y2 + y1) / 2.0)) | |
| warps[j] = al.transformPerspective.process(im, M) | |
| mats[j] = M | |
| x = torch.from_numpy(warps).to(al.device).permute(0, 3, 1, 2).float() / 255.0 * 2.0 - 1.0 | |
| out = al.alignment(x) | |
| pts = al.denorm_points(out[-1]).detach().cpu().numpy().astype(np.float64) # [k,68,2] in warp px | |
| res = np.zeros((k, 68, 2), np.float32) | |
| ones = np.ones((68, 1)) | |
| for j in range(k): | |
| inv = np.linalg.inv(mats[j]) | |
| res[j] = (np.concatenate([pts[j], ones], 1) @ inv.T)[:, :2] | |
| return res | |
| def _mtl(self, faces): | |
| import cv2 | |
| import torch | |
| x = torch.stack([self.mt.transform(cv2.cvtColor(f, cv2.COLOR_BGR2RGB)) for f in faces]).to(self.mt.device) | |
| emo, gaze, au = self.mt.model(x) | |
| return emo.float().cpu().numpy(), gaze.float().cpu().numpy(), au.float().cpu().numpy() | |
| # -- contract | |
| def _usable(crops): | |
| return [i for i, c in enumerate(crops) if c is not None and c.ndim == 3 and c.shape[0] >= 16 and c.shape[1] >= 16] | |
| def detect_faces(self, crops): | |
| """crops (u8 BGR driver crops, None allowed) -> f32[n,5] (x1, y1, x2, y2, score) of the top-scoring RetinaFace box per crop in crop px (NaN row for None / | |
| tiny crops). The shared-detector entry point of run_full_extraction.py (2026-09-14): the same boxes are handed back to infer(crops, dets=...) and to | |
| l2cs_gaze.infer_with_boxes, so RetinaFace runs once per frame for both GPU baselines. Batched exactly as infer() batches (one padded canvas per call).""" | |
| import torch | |
| out = np.full((len(crops), 5), np.nan, np.float32) | |
| idx = self._usable(crops) | |
| if idx: | |
| with torch.inference_mode(): | |
| out[idx] = self._detect([np.ascontiguousarray(crops[i]) for i in idx]) | |
| self._sync() | |
| return out | |
| def infer(self, crops, dets=None): | |
| """crops -> f32[n, C]; dets (optional, f32[n,5] aligned to `crops`, from detect_faces on the SAME batch) skips the RetinaFace stage (shared decode runner).""" | |
| import torch | |
| n = len(crops) | |
| out = np.full((n, self.C), np.nan, np.float32) | |
| idx = self._usable(crops) | |
| self.n_crops += len(idx) | |
| if not idx: | |
| return out | |
| t0 = time.time() | |
| imgs = [np.ascontiguousarray(crops[i]) for i in idx] | |
| with torch.inference_mode(): | |
| dets = self._detect(imgs) if dets is None else np.asarray(dets, np.float32)[np.asarray(idx)] | |
| self._sync() | |
| t1 = time.time() | |
| ok = np.isfinite(dets).all(1) & (dets[:, 4] >= FACE_TH) | |
| boxes = np.zeros((len(imgs), 4), np.int64) | |
| for j, im in enumerate(imgs): | |
| if not ok[j]: | |
| continue | |
| H, W = im.shape[:2] | |
| x1, y1, x2, y2 = dets[j, :4].astype(int) # package: det[:4].astype(int) | |
| x1, y1, x2, y2 = max(0, x1), max(0, y1), min(W, x2), min(H, y2) | |
| if x2 - x1 < MIN_FACE_PX or y2 - y1 < MIN_FACE_PX: | |
| ok[j] = False | |
| continue | |
| boxes[j] = (x1, y1, x2, y2) | |
| sel = np.flatnonzero(ok) | |
| self.n_faces += len(sel) | |
| if len(sel): | |
| lm = self._landmarks([imgs[j] for j in sel], boxes[sel]) | |
| self._sync() | |
| t2 = time.time() | |
| emo, gaze, au = self._mtl([imgs[j][boxes[j, 1]:boxes[j, 3], boxes[j, 0]:boxes[j, 2]] for j in sel]) | |
| self._sync() | |
| t3 = time.time() | |
| self.gaze_raw_absmax = max(self.gaze_raw_absmax, float(np.abs(gaze).max())) | |
| for r, j in enumerate(sel): | |
| H, W = imgs[j].shape[:2] | |
| x1, y1, x2, y2 = boxes[j] | |
| yaw, pitch, roll = head_pose_deg(lm[r], float(max(W, H)), (x1 + x2) / 2.0, (y1 + y2) / 2.0, x2 - x1) | |
| out[idx[j]] = [dets[j, 4], math.degrees(gaze[r, 0]), math.degrees(gaze[r, 1]), yaw, pitch, roll, *au[r], | |
| eye_aspect_ratio(lm[r][EYE_L]), eye_aspect_ratio(lm[r][EYE_R]), *emo[r]] | |
| t4 = time.time() | |
| k = len(sel) | |
| self.stage_ms["lm"].append(1000.0 * (t2 - t1) / k) | |
| self.stage_ms["mtl"].append(1000.0 * (t3 - t2) / k) | |
| self.stage_ms["pnp_ear"].append(1000.0 * (t4 - t3) / k) | |
| self.stage_ms["det"].append(1000.0 * (t1 - t0) / len(imgs)) | |
| self.infer_ms.append(1000.0 * (time.time() - t0) / len(imgs)) | |
| return out | |
| def landmarks_for_sheet(self, crop): | |
| """debug helper for the contact sheet: (det f32[5], lm68 f32[68,2]) or (None, None).""" | |
| import torch | |
| with torch.inference_mode(): | |
| d = self._detect([np.ascontiguousarray(crop)])[0] | |
| if not (np.isfinite(d).all() and d[4] >= FACE_TH): | |
| return None, None | |
| H, W = crop.shape[:2] | |
| x1, y1, x2, y2 = d[:4].astype(int) | |
| box = np.array([[max(0, x1), max(0, y1), min(W, x2), min(H, y2)]], np.int64) | |
| if box[0, 2] - box[0, 0] < MIN_FACE_PX or box[0, 3] - box[0, 1] < MIN_FACE_PX: | |
| return None, None | |
| return d, self._landmarks([np.ascontiguousarray(crop)], box)[0] | |
| EXTRACTORS = {BASELINE: OpenFace3Extractor} | |
| # ----------------------------------------------------------------------------- resources / manifest helpers | |
| def resource_guard(need_vram_gb=4.0, need_ram_gb=28.0): | |
| """RTX 5090 is shared: refuse to load when free VRAM < need + 4 GB or MemAvailable < 28 GB (project rule).""" | |
| import subprocess | |
| free_mb = int(subprocess.check_output(["nvidia-smi", "--query-gpu=memory.free", "--format=csv,noheader,nounits"]).decode().split()[0]) | |
| avail_kb = int(next(l for l in open("/proc/meminfo") if l.startswith("MemAvailable")).split()[1]) | |
| rep = {"vram_free_gb": round(free_mb / 1024, 1), "ram_available_gb": round(avail_kb / 2**20, 1), "need_vram_gb": need_vram_gb + 4.0, "need_ram_gb": need_ram_gb} | |
| assert free_mb / 1024 >= need_vram_gb + 4.0, f"not enough free VRAM: {rep}" | |
| assert avail_kb / 2**20 >= need_ram_gb, f"not enough free RAM: {rep}" | |
| return rep | |
| def load_manifest(): | |
| import pandas as pd | |
| M = pd.read_csv(MANIFEST) | |
| assert len(M) == 50 and set(M.regime) == {"manual", "engaged"}, (len(M), set(M.regime)) | |
| return M | |
| def manifest_spans(M): | |
| return {u: cs.merge_spans(g[["win_start", "win_end"]].to_numpy(np.float64)) for u, g in M.groupby("uid", sort=False)} | |
| def pct(v, q): | |
| v = np.asarray([x for x in v if np.isfinite(x)], np.float64) | |
| return float(np.percentile(v, q)) if len(v) else float("nan") | |
| def peak_rss_mb(): | |
| own = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0 | |
| kids = resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss / 1024.0 | |
| return {"main_process_mb": round(own, 1), "largest_decode_worker_mb": round(kids, 1)} | |
| def vram_mb(): | |
| import torch | |
| if not torch.cuda.is_available(): | |
| return {"peak_allocated_mb": None, "peak_reserved_mb": None} | |
| return {"peak_allocated_mb": round(torch.cuda.max_memory_allocated() / 2**20, 1), "peak_reserved_mb": round(torch.cuda.max_memory_reserved() / 2**20, 1)} | |
| def decode_single_frame(uid, t, workers=1): | |
| """-> (frame_bgr native, box frame-coords) of the 10 Hz step k = round(10 t) (None, None when unavailable).""" | |
| k = int(round(t * cs.HZ)) | |
| fs = vio.FrameStream(uid, [[k / cs.HZ, k / cs.HZ]], hz=cs.HZ, workers=workers, decode_wh=None, block=4, box_coords="frame") | |
| for t_, fi_, fr_, bx_ in fs: | |
| if fi_ == k * (cs.FPS // cs.HZ): | |
| return fr_, bx_ | |
| return None, None | |
| # ----------------------------------------------------------------------------- Phase-1 measurements | |
| def face_pack_history(M): | |
| """-> {manifest idx: (k i64[80], lm_yaw_dev f32[80], lm_yaw f32[80], eye_open f32[80])} from facehist_causal_{regime}_train.npz (row `row`; Fh8 step j <-> t - 7.9 + 0.1 j).""" | |
| out = {} | |
| for reg in ("manual", "engaged"): | |
| sub = M[M.regime == reg] | |
| with np.load(FACE_PACK / f"facehist_causal_{reg}_train.npz", allow_pickle=True) as z: | |
| ch = [str(c) for c in z["channels"]] | |
| j_dev, j_yaw, j_eye = ch.index("lm_yaw_dev"), ch.index("lm_yaw"), ch.index("eye_open") | |
| rows = sub.row.to_numpy() | |
| assert np.array_equal(z["uid"][rows], sub.uid.to_numpy()) and np.allclose(z["t"][rows], sub.t_anchor.to_numpy()), f"{reg}: manifest rows do not match the face pack" | |
| Fh = z["Fh8_c"][rows].astype(np.float32) | |
| Fv = z["Fh8valid"][rows] | |
| for i, (idx, ta) in enumerate(zip(sub.idx, sub.t_anchor)): | |
| ka = int(round(float(ta) * cs.HZ)) | |
| k = ka - 79 + np.arange(80) | |
| dev = np.where(Fv[i], Fh[i, :, j_dev], np.nan) | |
| yaw = np.where(Fv[i], Fh[i, :, j_yaw], np.nan) | |
| eye = np.where(Fv[i], Fh[i, :, j_eye], np.nan) | |
| out[int(idx)] = (k, dev, yaw, eye) | |
| return out | |
| def spearman(a, b): | |
| from scipy.stats import spearmanr | |
| a, b = np.asarray(a, np.float64), np.asarray(b, np.float64) | |
| m = np.isfinite(a) & np.isfinite(b) | |
| if m.sum() < 8: | |
| return float("nan"), int(m.sum()) | |
| r = spearmanr(a[m], b[m]).statistic | |
| return float(r), int(m.sum()) | |
| def measure(ex, M, infos, wall_s, cache_dir): | |
| caches = {u: cs.RouteCache.load(cache_dir / f"{u}.npz") for u in M.uid.unique()} | |
| fail_hist = {name: 0 for name in cs.FAIL_CODES.values()} | |
| n_frames = n_box = n_valid = n_valid_box = 0 | |
| G_all = [] | |
| for rc in caches.values(): | |
| for code, cnt in zip(*np.unique(rc.fail, return_counts=True)): | |
| fail_hist[cs.FAIL_CODES[int(code)]] += int(cnt) | |
| hb = np.isfinite(rc.box).all(1) | |
| n_frames += rc.n | |
| n_box += int(hb.sum()) | |
| n_valid += int(rc.valid.sum()) | |
| n_valid_box += int((rc.valid & hb).sum()) | |
| G_all.append(rc.feat.astype(np.float32)[rc.valid]) | |
| G = np.concatenate(G_all) if G_all else np.zeros((0, len(CHANNELS)), np.float32) | |
| ranges = {c: {"p01": round(pct(G[:, j], 1), 3), "p50": round(pct(G[:, j], 50), 3), "p99": round(pct(G[:, j], 99), 3), "min": round(float(G[:, j].min()), 3) if len(G) else None, | |
| "max": round(float(G[:, j].max()), 3) if len(G) else None} for j, c in enumerate(CHANNELS)} | |
| # per-anchor values | |
| main = ["of3_face_conf", "of3_head_yaw", "of3_head_pitch", "of3_head_roll", "of3_gaze_yaw", "of3_gaze_pitch", "of3_eye_open_l", "of3_eye_open_r"] | |
| per_anchor = [] | |
| hist = face_pack_history(M) | |
| pooled = {"head_yaw": ([], []), "gaze_yaw": ([], []), "head_yaw_vs_lm_yaw": ([], []), "gaze_yaw_vs_lm_yaw": ([], []), "ear_mean_vs_facepack_eye_open": ([], [])} | |
| per_anchor_rho = {"head_yaw": [], "gaze_yaw": []} | |
| for _, r in M.iterrows(): | |
| rc = caches[r.uid] | |
| k = np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64) | |
| pos = {int(kk): i for i, kk in enumerate(k)} | |
| ka = int(round(float(r.t_anchor) * cs.HZ)) | |
| i_a = pos.get(ka) | |
| feat = rc.feat.astype(np.float32) | |
| rec = {"idx": int(r.idx), "uid": r.uid, "regime": r.regime, "y": int(r.y), "t_anchor": float(r.t_anchor), "frame_at_anchor": i_a is not None, | |
| "valid_at_anchor": bool(i_a is not None and rc.valid[i_a]), "fail_at_anchor": (cs.FAIL_CODES[int(rc.fail[i_a])] if i_a is not None else "no_frame")} | |
| i_s0 = next((pos[ka - b] for b in range(0, 2) if (ka - b) in pos and rc.valid[pos[ka - b]]), None) # S0 = latest valid within 0.15 s (summarise.S0_TOL_S) | |
| rec["at_anchor"] = {c: (round(float(feat[i_a, CH[c]]), 3) if rec["valid_at_anchor"] else None) for c in main} | |
| rec["s0"] = {c: (round(float(feat[i_s0, CH[c]]), 3) if i_s0 is not None else None) for c in main} | |
| # window [t-8, t] descriptive | |
| w = (k > ka - 80) & (k <= ka) | |
| wv = w & rc.valid | |
| rec["valid_share_8s"] = round(float(wv.sum() / 80.0), 3) | |
| rec["abs_head_yaw_mean_8s"] = round(float(np.abs(feat[wv, CH["of3_head_yaw"]]).mean()), 2) if wv.any() else None | |
| rec["abs_gaze_yaw_mean_8s"] = round(float(np.abs(feat[wv, CH["of3_gaze_yaw"]]).mean()), 2) if wv.any() else None | |
| # Spearman vs the face pack over the 80 history steps | |
| kk, dev, lmy, eye = hist[int(r.idx)] | |
| hy = np.array([feat[pos[x], CH["of3_head_yaw"]] if x in pos else np.nan for x in kk]) | |
| gy = np.array([feat[pos[x], CH["of3_gaze_yaw"]] if x in pos else np.nan for x in kk]) | |
| ear = np.array([0.5 * (feat[pos[x], CH["of3_eye_open_l"]] + feat[pos[x], CH["of3_eye_open_r"]]) if x in pos else np.nan for x in kk]) # eye-state check vs face pack eye_open | |
| for name, mine, ref in (("head_yaw", hy, dev), ("gaze_yaw", gy, dev), ("head_yaw_vs_lm_yaw", hy, lmy), ("gaze_yaw_vs_lm_yaw", gy, lmy), ("ear_mean_vs_facepack_eye_open", ear, eye)): | |
| pooled[name][0].extend(mine.tolist()) | |
| pooled[name][1].extend(ref.tolist()) | |
| rho_h, n_h = spearman(hy, dev) | |
| rho_g, n_g = spearman(gy, dev) | |
| per_anchor_rho["head_yaw"].append(rho_h) | |
| per_anchor_rho["gaze_yaw"].append(rho_g) | |
| rec["rho_head_yaw_vs_lm_yaw_dev_80"] = (round(rho_h, 3) if np.isfinite(rho_h) else None, n_h) | |
| rec["rho_gaze_yaw_vs_lm_yaw_dev_80"] = (round(rho_g, 3) if np.isfinite(rho_g) else None, n_g) | |
| per_anchor.append(rec) | |
| rho = {} | |
| for name, (a, b) in pooled.items(): | |
| r_, n_ = spearman(a, b) | |
| rho[name] = {"rho_pooled": (round(r_, 4) if np.isfinite(r_) else None), "n_pairs": n_} | |
| for name, vals in per_anchor_rho.items(): | |
| v = np.asarray([x for x in vals if np.isfinite(x)]) | |
| rho[name]["per_anchor_median"] = round(float(np.median(v)), 3) if len(v) else None | |
| rho[name]["per_anchor_share_positive"] = round(float((v > 0).mean()), 3) if len(v) else None | |
| rho[name]["n_anchors_defined"] = int(len(v)) | |
| # descriptive by y (never used for selection) | |
| by_y = {} | |
| for yv in (0, 1): | |
| sub = [p for p in per_anchor if p["y"] == yv] | |
| def med(key): | |
| v = np.asarray([p[key] for p in sub if p.get(key) is not None], np.float64) | |
| return round(float(np.median(v)), 2) if len(v) else None | |
| by_y[f"y{yv}"] = {"n": len(sub), "valid_share_8s_median": med("valid_share_8s"), "abs_head_yaw_mean_8s_median": med("abs_head_yaw_mean_8s"), "abs_gaze_yaw_mean_8s_median": med("abs_gaze_yaw_mean_8s"), | |
| "s0_head_yaw_abs_median": (lambda v: round(float(np.median(v)), 2) if len(v) else None)(np.asarray([abs(p["s0"]["of3_head_yaw"]) for p in sub if p["s0"]["of3_head_yaw"] is not None]))} | |
| # runtime | |
| rt = [i["runtime"] for i in infos if i.get("runtime")] | |
| n_new = int(sum(i.get("n_new_frames", 0) for i in infos)) | |
| meas = {"baseline": BASELINE, "manifest": str(MANIFEST.relative_to(CWM3)), "n_anchors": int(len(M)), "n_routes": int(M.uid.nunique()), "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), | |
| "n_frames": int(n_frames), "n_frames_with_box": int(n_box), "n_valid": int(n_valid), "valid_share_overall": round(n_valid / max(n_frames, 1), 4), "valid_share_with_box": round(n_valid_box / max(n_box, 1), 4), | |
| "fail_hist": fail_hist, "n_frames_new_this_run": n_new, "wall_s_routes": round(wall_s, 1), "fps_end_to_end": round(n_new / max(wall_s, 1e-9), 1), | |
| "decode_ms_per_frame": {"route_p50_median": round(float(np.median([r["decode_ms_p50"] for r in rt])), 2) if rt else None, "route_p95_max": round(float(max(r["decode_ms_p95"] for r in rt)), 2) if rt else None, | |
| "note": "per decode block, per worker process (video_io.FrameStream stats), aggregated over the route caches written this run"}, | |
| "infer_ms_per_frame": {"p50": round(pct(ex.infer_ms, 50), 2), "p95": round(pct(ex.infer_ms, 95), 2), "n_batches": len(ex.infer_ms), "batch": ex.batch, | |
| "stages_p50": {k: round(pct(v, 50), 2) for k, v in ex.stage_ms.items()}, "stages_p95": {k: round(pct(v, 95), 2) for k, v in ex.stage_ms.items()}, | |
| "note": "wall ms per crop of infer() incl. host<->device copies (cuda synchronised); det per crop, lm / mtl / pnp_ear per detected face"}, | |
| "n_crops_seen": int(ex.n_crops), "n_faces_detected": int(ex.n_faces), "peak_rss": peak_rss_mb(), "vram": vram_mb(), | |
| "gaze_units": {"raw_abs_max": round(ex.gaze_raw_absmax, 4), "assumed": "radians (raw |value| < pi on every frame; converted to degrees)", "ok": bool(ex.gaze_raw_absmax < math.pi)}, | |
| "channel_ranges_valid_rows": ranges, "spearman_vs_face_pack": rho, "per_anchor": per_anchor, "descriptive_by_y": by_y, | |
| "route_infos": [{k: v for k, v in i.items() if k in ("uid", "status", "n_frames", "n_new_frames", "valid_rate", "fail_hist", "wall_s", "decode_worker_fps")} for i in infos]} | |
| return meas, caches | |
| def contact_sheet(ex, M, caches, out_png, n_side=6, workers=1): | |
| """12 frames: the 6 most negative and 6 most positive of3_head_yaw among valid frames (one per route); face crop + 68 landmarks + printed angles; | |
| also re-runs the extractor on the horizontally flipped crop (sign antisymmetry check). -> dict(selected, flip_check).""" | |
| import cv2 | |
| rows = [] | |
| for u, rc in caches.items(): | |
| f = rc.feat.astype(np.float32) | |
| for i in np.flatnonzero(rc.valid): | |
| rows.append((float(f[i, CH["of3_head_yaw"]]), u, float(rc.t[i]), i)) | |
| rows.sort() | |
| def pick(seq): | |
| got, seen = [], set() | |
| for r in seq: | |
| if r[1] in seen: | |
| continue | |
| seen.add(r[1]) | |
| got.append(r) | |
| if len(got) == n_side: | |
| break | |
| return got | |
| neg, pos = pick(rows), pick(rows[::-1]) | |
| tiles, sel, flips = [], [], [] | |
| TW, TH = 260, 300 | |
| for group, items in (("neg", neg), ("pos", pos)): | |
| for yaw_c, u, t, i in items: | |
| fr, bx = decode_single_frame(u, t, workers) | |
| crop = vio.driver_crop(fr, bx) if fr is not None else None | |
| tile = np.zeros((TH, TW, 3), np.uint8) | |
| rec = {"group": group, "uid": u, "t": round(t, 1), "cache_head_yaw": round(yaw_c, 1)} | |
| if crop is not None: | |
| out2 = ex.infer([crop, np.ascontiguousarray(crop[:, ::-1])]) | |
| d, lm = ex.landmarks_for_sheet(crop) | |
| rec.update(head_yaw=round(float(out2[0, CH["of3_head_yaw"]]), 1), head_pitch=round(float(out2[0, CH["of3_head_pitch"]]), 1), head_roll=round(float(out2[0, CH["of3_head_roll"]]), 1), | |
| gaze_yaw=round(float(out2[0, CH["of3_gaze_yaw"]]), 1), gaze_pitch=round(float(out2[0, CH["of3_gaze_pitch"]]), 1), | |
| flip_head_yaw=round(float(out2[1, CH["of3_head_yaw"]]), 1), flip_gaze_yaw=round(float(out2[1, CH["of3_gaze_yaw"]]), 1), | |
| eye_open_l=round(float(out2[0, CH["of3_eye_open_l"]]), 3), eye_open_r=round(float(out2[0, CH["of3_eye_open_r"]]), 3)) | |
| flips.append((out2[0, CH["of3_head_yaw"]], out2[1, CH["of3_head_yaw"]], out2[0, CH["of3_gaze_yaw"]], out2[1, CH["of3_gaze_yaw"]])) | |
| vis = crop.copy() | |
| if d is not None: | |
| x1, y1, x2, y2 = d[:4].astype(int) | |
| cv2.rectangle(vis, (x1, y1), (x2, y2), (0, 255, 255), 2) | |
| for p in lm: | |
| cv2.circle(vis, (int(round(p[0])), int(round(p[1]))), 2, (0, 255, 0), -1) | |
| for j in PNP_IDX: | |
| cv2.circle(vis, (int(round(lm[j, 0])), int(round(lm[j, 1]))), 4, (255, 0, 255), -1) | |
| # enlarge the face region (x1.8 box) for the tile | |
| cx, cy, s = (x1 + x2) / 2, (y1 + y2) / 2, 0.9 * max(x2 - x1, y2 - y1) | |
| X0, Y0 = int(max(0, cx - s)), int(max(0, cy - s)) | |
| X1, Y1 = int(min(vis.shape[1], cx + s)), int(min(vis.shape[0], cy + s)) | |
| vis = vis[Y0:Y1, X0:X1] | |
| h, w = vis.shape[:2] | |
| sc = min(TW / max(w, 1), (TH - 70) / max(h, 1)) | |
| vis = cv2.resize(vis, (max(1, int(w * sc)), max(1, int(h * sc))), interpolation=cv2.INTER_AREA) | |
| tile[:vis.shape[0], :vis.shape[1]] = vis | |
| txt = [f"{group} t={t:.1f} conf={float(out2[0, CH['of3_face_conf']]):.2f}", f"head y/p/r {rec['head_yaw']:+.0f}/{rec['head_pitch']:+.0f}/{rec['head_roll']:+.0f}", | |
| f"gaze y/p {rec['gaze_yaw']:+.0f}/{rec['gaze_pitch']:+.0f} flip y {rec['flip_head_yaw']:+.0f}/{rec['flip_gaze_yaw']:+.0f}", f"EAR l/r {rec['eye_open_l']:.2f}/{rec['eye_open_r']:.2f}"] | |
| for li, s_ in enumerate(txt): | |
| cv2.putText(tile, s_, (3, TH - 56 + 14 * li), cv2.FONT_HERSHEY_SIMPLEX, 0.38, (255, 255, 255), 1, cv2.LINE_AA) | |
| tiles.append(tile) | |
| sel.append(rec) | |
| while len(tiles) < 2 * n_side: | |
| tiles.append(np.zeros((TH, TW, 3), np.uint8)) | |
| grid = np.concatenate([np.concatenate(tiles[:n_side], 1), np.concatenate(tiles[n_side:2 * n_side], 1)], 0) | |
| header = np.zeros((44, grid.shape[1], 3), np.uint8) | |
| cv2.putText(header, "openface3 sign convention: head yaw + = nose toward IMAGE RIGHT (PnP, 6-pt model); top row = most negative head yaw, bottom row = most positive; one frame per route", | |
| (6, 17), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 255), 1, cv2.LINE_AA) | |
| cv2.putText(header, "gaze yaw / pitch = MTL output (radians -> deg) as-is; 'flip' = the same crop mirrored horizontally (antisymmetric outputs change sign); magenta = the 6 PnP landmarks", | |
| (6, 36), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 255), 1, cv2.LINE_AA) | |
| cv2.imwrite(str(out_png), np.concatenate([header, grid], 0)) | |
| fl = np.asarray(flips, np.float64) | |
| flip_check = None | |
| if len(fl): | |
| hy, hyf, gy, gyf = fl.T | |
| def cc(a, b): | |
| m = np.isfinite(a) & np.isfinite(b) | |
| return round(float(np.corrcoef(a[m], b[m])[0, 1]), 3) if m.sum() >= 3 else None | |
| flip_check = {"n": int(len(fl)), "head_yaw_mean_sum_orig_plus_flip_deg": round(float(np.nanmean(hy + hyf)), 2), "head_yaw_corr_orig_vs_minus_flip": cc(hy, -hyf), | |
| "gaze_yaw_mean_sum_orig_plus_flip_deg": round(float(np.nanmean(gy + gyf)), 2), "gaze_yaw_corr_orig_vs_minus_flip": cc(gy, -gyf), | |
| "note": "an output that is antisymmetric under horizontal mirroring has sum ~ 0 and corr ~ +1"} | |
| return {"selected": sel, "flip_check": flip_check, "png": str(out_png)} | |
| # ----------------------------------------------------------------------------- CLIs | |
| def cli_phase1(workers=3, max_minutes=60.0, force=False, device="cuda:0", log=print): | |
| guard = resource_guard() | |
| M = load_manifest() | |
| spans = manifest_spans(M) | |
| cache_dir = OUT_DIR / "cache" | |
| cache_dir.mkdir(parents=True, exist_ok=True) | |
| t_setup = time.time() | |
| ex = OpenFace3Extractor(device).setup() | |
| setup_s = time.time() - t_setup | |
| log(f"[openface3] setup {setup_s:.1f}s; {len(M)} anchors / {len(spans)} routes; workers {workers}; block 64 native frames") | |
| infos, codes = [], [] | |
| t0 = time.time() | |
| for n_i, u in enumerate(spans): | |
| # one run_route call PER CONTIGUOUS SPAN (see the module docstring, "Phase-1 stall"): no decode block ever straddles the gap between two windows of a | |
| # route; --force applies to the first span only (a forced second call would rebuild the cache and drop the first span's rows), later spans extend it | |
| # through the resume path (RouteCache.merged_with) | |
| for k_sp, sp in enumerate(np.asarray(spans[u], np.float64).reshape(-1, 2)): | |
| code, path, info = run_route(ex, u, [sp], cache_dir, max_minutes=max_minutes, workers=workers, block=64, force=(force and k_sp == 0), log=log, t_start=t0) | |
| info["span"] = [float(sp[0]), float(sp[1])] | |
| infos.append(info) | |
| codes.append(code) | |
| log(f" [{n_i + 1}/{len(spans)} span {k_sp + 1}] {u[:70]} {info['status']} frames {info.get('n_frames')} valid {info.get('valid_rate')} wall {info.get('wall_s')}s") | |
| if code == EXIT_BUDGET: | |
| log("time budget reached; rerun to resume") | |
| break | |
| if codes and codes[-1] == EXIT_BUDGET: | |
| break | |
| wall = time.time() - t0 | |
| if any(c == EXIT_FAIL for c in codes) or any(c == EXIT_BUDGET for c in codes): | |
| (OUT_DIR / "phase1_partial.json").write_text(json.dumps({"codes": codes, "infos": infos}, indent=1, default=str)) | |
| log("phase1 incomplete (failure or budget); measurements not written") | |
| return 3 if EXIT_BUDGET in codes else 4 | |
| meas, caches = measure(ex, M, infos, wall, cache_dir) | |
| meas["setup_s"] = round(setup_s, 1) | |
| meas["resource_guard"] = guard | |
| meas["workers"] = workers | |
| meas["describe"] = ex.describe() | |
| meas["synthetic_pose_check"] = synthetic_pose_check() | |
| meas["contact_sheet"] = contact_sheet(ex, M, caches, OUT_DIR / "contact_sheet_sign.png", workers=1) | |
| meas["vram"] = vram_mb() | |
| meas["peak_rss"] = peak_rss_mb() | |
| (OUT_DIR / "measurements.json").write_text(json.dumps(meas, indent=1, default=str)) | |
| log(json.dumps({k: meas[k] for k in ("n_frames", "valid_share_overall", "valid_share_with_box", "fail_hist", "fps_end_to_end", "infer_ms_per_frame", "vram", "spearman_vs_face_pack", "gaze_units")}, indent=1, default=str)) | |
| log(f"wrote {OUT_DIR / 'measurements.json'} and {meas['contact_sheet']['png']}") | |
| return 0 | |
| def cli_self_test(uid=None, workers=2, device="cuda:0", n_frames=30): | |
| import tempfile | |
| rep = {"self_test": "extractors.openface3", "synthetic_pose_check": synthetic_pose_check()} | |
| rep["resource_guard"] = resource_guard() | |
| M = load_manifest() | |
| r = M.iloc[0] if not uid else M[M.uid == uid].iloc[0] | |
| uid = r.uid | |
| ta = float(r.t_anchor) | |
| span = np.array([[ta - (n_frames - 1) / cs.HZ, ta]], np.float32) | |
| t0 = time.time() | |
| ex = OpenFace3Extractor(device).setup() | |
| rep["setup_s"] = round(time.time() - t0, 1) | |
| with tempfile.TemporaryDirectory() as d: | |
| code, path, info = run_route(ex, uid, span, d, max_minutes=10, workers=workers, block=10) | |
| rc = cs.RouteCache.load(path) | |
| assert code == EXIT_DONE and rc.C == len(CHANNELS) == 24 and rc.channels == CHANNELS, (code, rc.C, rc.channels[:3]) | |
| assert n_frames <= rc.n <= n_frames + 2, (rc.n, n_frames) # floor / ceil of the span ends may add a grid step | |
| assert rc.valid.any(), "no valid frame in the self-test window" | |
| f = rc.feat.astype(np.float32)[rc.valid] | |
| assert np.isfinite(f).all() and (f[:, CH["of3_face_conf"]] >= FACE_TH).all() | |
| rep.update(route=uid, t_anchor=ta, n_frames=rc.n, fail_hist=info["fail_hist"], valid_share=round(float(rc.valid.mean()), 3), runtime=rc.runtime, wall_s=info["wall_s"], | |
| channel_p50={c: round(pct(f[:, j], 50), 3) for j, c in enumerate(CHANNELS)}, infer_ms_p50=round(pct(ex.infer_ms, 50), 2), | |
| stages_ms_p50={k: round(pct(v, 50), 2) for k, v in ex.stage_ms.items()}, gaze_raw_absmax=round(ex.gaze_raw_absmax, 3), vram=vram_mb()) | |
| # flip antisymmetry on the frames of the window (decode again, small) | |
| fs = vio.FrameStream(uid, span, hz=cs.HZ, workers=workers, decode_wh=None, block=10, box_coords="frame") | |
| crops = [vio.driver_crop(fr, bx) for _, _, fr, bx in fs if fr is not None] | |
| crops = [c for c in crops if c is not None][:10] | |
| if crops: | |
| o = ex.infer(crops) | |
| of = ex.infer([np.ascontiguousarray(c[:, ::-1]) for c in crops]) | |
| m = np.isfinite(o[:, CH["of3_head_yaw"]]) & np.isfinite(of[:, CH["of3_head_yaw"]]) | |
| rep["flip_check_10_frames"] = {"n": int(m.sum()), "head_yaw_orig": np.round(o[m, CH["of3_head_yaw"]], 1).tolist(), "head_yaw_flip": np.round(of[m, CH["of3_head_yaw"]], 1).tolist(), | |
| "gaze_yaw_orig": np.round(o[m, CH["of3_gaze_yaw"]], 1).tolist(), "gaze_yaw_flip": np.round(of[m, CH["of3_gaze_yaw"]], 1).tolist()} | |
| rep["describe"] = ex.describe() | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return 0 | |
| 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("--spec", action="store_true") | |
| ap.add_argument("--uid", default="") | |
| ap.add_argument("--workers", type=int, default=3) | |
| ap.add_argument("--max-minutes", type=float, default=60.0) | |
| ap.add_argument("--device", default="cuda:0") | |
| ap.add_argument("--force", action="store_true") | |
| a = ap.parse_args() | |
| if a.spec: | |
| print(json.dumps({"channels": CHANNELS, "roles": ROLES, "summarise_spec": SUMMARISE_SPEC}, indent=1)) | |
| return 0 | |
| if a.phase1: | |
| return cli_phase1(min(a.workers, 3), a.max_minutes, a.force, a.device) | |
| return cli_self_test(a.uid or None, min(a.workers, 3), a.device) | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 47.3 kB
- Xet hash:
- 7a193c32fe323c706f9b96501c56c6de3d6668b69aea70809f9414cf5eb53f0e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.