Buckets:
| """cwm3/baselines_dms/extractors/mp_face.py — G1 Google MediaPipe Face Landmarker (Tasks API) as a per-frame DMS baseline (plan §3, 2026-09-14). | |
| Model registry id `mediapipe_face_landmarker` (face_landmarker.task float16/latest, sha256 64184e22…; BlazeFace short-range detector + 478-point | |
| face mesh + 52 blendshapes + 4x4 facial transformation matrix), run with mediapipe 0.10.35 (registry id `pip_mediapipe`) on the CPU | |
| (XNNPACK delegate). Options: RunningMode.IMAGE (stateless, one detect() per frame — no tracking state, so causal frame order, resume | |
| and per-route merging never change a result), num_faces=1, output_face_blendshapes=True, output_facial_transformation_matrixes=True, | |
| detection / presence / tracking confidences at the library defaults (0.5; nothing tuned). | |
| Input the runner hands `infer` the driver crop (video_io.driver_crop: causal pose box + 10 % per side, NATIVE decode: decode_wh=None so the | |
| crop keeps the sensor resolution, 1928x1208 or 1344x760 frames). The pose box spans head to knees, so the face is only 15–30 % of the | |
| crop height and BlazeFace misses it often (0.55 of frames found on a 6-route probe). Head region rule (geometry only, no model, no | |
| label): HEAD = the top square of the crop (rows 0..min(h, w), all columns); the landmarker runs on HEAD; when it finds no face and HEAD | |
| is not the whole crop it runs once more on the full crop (fallback). Probe on the same 6 routes: HEAD 0.80, HEAD->full 0.84. Colour: | |
| the crop is BGR (cv2) -> RGB for mp.Image(SRGB); no other normalisation (the task bundle normalises internally). | |
| Channels (C = 20; every row is NaN = fail 3 when no face is found on either region) | |
| mp_face_conf 1.0 when a face is found (the Tasks API exposes no face score; role conf) | |
| mp_blink_l, mp_blink_r blendshapes eyeBlinkLeft / eyeBlinkRight in [0, 1] (role eye_closed_prob; Left/Right = the SUBJECT's side) | |
| mp_look_{in,out,up,down}_{l,r} blendshapes eyeLook{In,Out,Up,Down}{Left,Right} in [0, 1] (role prob) | |
| mp_squint_{l,r}, mp_wide_{l,r} blendshapes eyeSquint*, eyeWide* in [0, 1] (role prob) | |
| mp_head_yaw, mp_head_pitch, mp_head_roll degrees from the 4x4 facial transformation matrix M (role angle_deg), see below | |
| mp_ear_l, mp_ear_r eye aspect ratio from the eyelid landmarks (role eye_open_ratio), 6-point EAR = (|p2-p6| + |p3-p5|) / (2 |p1-p4|) | |
| in crop pixels; subject's right eye p1..p6 = 33, 160, 158, 133, 153, 144; subject's left eye = 362, 385, 387, 263, 373, 380 | |
| Rotation convention (verified on real frames in --self-test / the contact sheet; scipy 'YXZ' cross-check in --self-test) | |
| M maps canonical-face-model coordinates to MediaPipe's metric camera space: right-handed, +x = image right, +y = image up, +z = toward the | |
| viewer (the face sits at negative z, ~ -60 cm). R = M[:3, :3] is decomposed as R = Ry(yaw) . Rx(b) . Rz(roll) (intrinsic Y-X-Z): | |
| yaw = atan2(R[0,2], R[2,2]) > 0 : the face's forward axis (third column) points to IMAGE RIGHT (the camera's +x) | |
| pitch = asin(R[1,2]) > 0 : forward axis points UP (image up); pitch = -b of the decomposition so that up is positive | |
| roll = atan2(R[1,0], R[1,1]) > 0 : the face rotated counter-clockwise as seen in the image (top of the head towards image LEFT) | |
| A driver looking straight down the road from an image-right seat therefore has a POSITIVE resting yaw (~ +25° on the probe: the camera on the | |
| windscreen is angled towards the driver); the chain reads |yaw - ref| with ref = 0 (rule0) or the route's causal median (ruleref). The matrix is | |
| estimated with a pinhole camera centred on the crop, so the fallback (full-crop) frames carry a few degrees of offset vs HEAD frames (probe: | |
| median -1.5°, p90 |diff| 10°); the share of fallback frames is recorded in measurements.json. | |
| Spans the manifest windows [win_start, win_end] are snapped to the 10 Hz grid and pulled inwards by 0.02 s (grid_spans) before run_route: | |
| run_route stores spans as float32 and video_io.frames_in_spans floors / ceils them without tolerance, which otherwise reads one frame | |
| BEFORE win_start (21 / 50 windows) or AFTER win_end (22 / 50) — found by the 2026-09-14 verification and fixed; exactly the frames | |
| k0 .. k1 = round(10 win_start) .. round(10 win_end) are decoded now, none later than the requested window end. | |
| Outputs results/baselines_dms/phase1/mp_face/{cache/<uid>.npz (RouteCache), measurements.json, sign_convention_contact_sheet.png, README.md} | |
| python baselines_dms/extractors/mp_face.py --self-test # 1 manifest route, 30 frames, temp dir; convention + spec + hygiene checks | |
| python baselines_dms/extractors/mp_face.py --phase1 [--workers 3] # the 50 frozen manifest windows (causal order) + measurements + contact sheet | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| import resource | |
| import subprocess | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| os.environ.setdefault("GLOG_minloglevel", "2") | |
| os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "2") | |
| HERE = Path(__file__).resolve().parent | |
| PKG = HERE.parent | |
| CWM3 = PKG.parent | |
| for p in (str(CWM3), str(PKG), str(HERE)): | |
| if p not in sys.path: | |
| sys.path.insert(0, p) | |
| import cache_schema as cs # noqa: E402 | |
| import video_io as vio # noqa: E402 | |
| import base as B # noqa: E402 | |
| BASELINE = "mp_face" | |
| REGISTRY_ID = "mediapipe_face_landmarker" | |
| PIP_ID = "pip_mediapipe" | |
| PHASE1_DIR = cs.RESULTS_ROOT / "phase1" | |
| OUT_DIR = PHASE1_DIR / BASELINE | |
| MANIFEST = PHASE1_DIR / "manifest_v1.csv" | |
| FACE_PACK = CWM3 / "results" / "targets_face_causal_routes_v2" / "facehist_causal_{regime}_train.npz" | |
| CHANNELS = ["mp_face_conf", "mp_blink_l", "mp_blink_r", | |
| "mp_look_in_l", "mp_look_out_l", "mp_look_up_l", "mp_look_down_l", "mp_look_in_r", "mp_look_out_r", "mp_look_up_r", "mp_look_down_r", | |
| "mp_squint_l", "mp_squint_r", "mp_wide_l", "mp_wide_r", | |
| "mp_head_yaw", "mp_head_pitch", "mp_head_roll", "mp_ear_l", "mp_ear_r"] | |
| BLENDSHAPES = {"mp_blink_l": "eyeBlinkLeft", "mp_blink_r": "eyeBlinkRight", | |
| "mp_look_in_l": "eyeLookInLeft", "mp_look_out_l": "eyeLookOutLeft", "mp_look_up_l": "eyeLookUpLeft", "mp_look_down_l": "eyeLookDownLeft", | |
| "mp_look_in_r": "eyeLookInRight", "mp_look_out_r": "eyeLookOutRight", "mp_look_up_r": "eyeLookUpRight", "mp_look_down_r": "eyeLookDownRight", | |
| "mp_squint_l": "eyeSquintLeft", "mp_squint_r": "eyeSquintRight", "mp_wide_l": "eyeWideLeft", "mp_wide_r": "eyeWideRight"} | |
| ROLES = {"mp_face_conf": "conf", "mp_blink_l": "eye_closed_prob", "mp_blink_r": "eye_closed_prob", | |
| **{k: "prob" for k in BLENDSHAPES if k.startswith(("mp_look", "mp_squint", "mp_wide"))}, | |
| "mp_head_yaw": "angle_deg", "mp_head_pitch": "angle_deg", "mp_head_roll": "angle_deg", "mp_ear_l": "eye_open_ratio", "mp_ear_r": "eye_open_ratio"} | |
| # 6-point EAR landmark indices (MediaPipe 478-mesh): p1 outer corner, p2 p3 upper lid, p4 inner corner, p5 p6 lower lid (Right / Left = the subject's side) | |
| EAR_R = (33, 160, 158, 133, 153, 144) | |
| EAR_L = (362, 385, 387, 263, 373, 380) | |
| NOSE_TIP, EYE_R_OUTER, EYE_L_OUTER = 1, 33, 263 | |
| CI = {c: j for j, c in enumerate(CHANNELS)} | |
| # ----------------------------------------------------------------------------- geometry helpers (pure numpy; unit-tested in --self-test) | |
| def head_region(crop): | |
| """driver crop u8[h,w,3] -> (HEAD view = the top square rows 0..min(h,w), is_sub_region).""" | |
| h, w = crop.shape[:2] | |
| s = min(h, w) | |
| return crop[:s], s < h | |
| def matrix_to_angles(M): | |
| """4x4 (or 3x3) facial transformation matrix -> (yaw, pitch, roll) in degrees with the convention of the module docstring (R = Ry(yaw) Rx(-pitch) Rz(roll)).""" | |
| R = np.asarray(M, np.float64)[:3, :3] | |
| yaw = np.degrees(np.arctan2(R[0, 2], R[2, 2])) | |
| pitch = np.degrees(np.arcsin(np.clip(R[1, 2], -1.0, 1.0))) | |
| roll = np.degrees(np.arctan2(R[1, 0], R[1, 1])) | |
| return float(yaw), float(pitch), float(roll) | |
| def angles_to_matrix(yaw, pitch, roll): | |
| """inverse of matrix_to_angles (degrees) -> 3x3; used by the self-test round trip.""" | |
| a, b, c = np.radians(yaw), -np.radians(pitch), np.radians(roll) | |
| Ry = np.array([[np.cos(a), 0, np.sin(a)], [0, 1, 0], [-np.sin(a), 0, np.cos(a)]]) | |
| Rx = np.array([[1, 0, 0], [0, np.cos(b), -np.sin(b)], [0, np.sin(b), np.cos(b)]]) | |
| Rz = np.array([[np.cos(c), -np.sin(c), 0], [np.sin(c), np.cos(c), 0], [0, 0, 1]]) | |
| return Ry @ Rx @ Rz | |
| def ear_from_points(P): | |
| """P [6,2] pixel coordinates (p1..p6) -> EAR = (|p2-p6| + |p3-p5|) / (2 |p1-p4|); NaN when the eye width is degenerate.""" | |
| P = np.asarray(P, np.float64) | |
| width = np.linalg.norm(P[0] - P[3]) | |
| if not np.isfinite(width) or width < 1e-6: | |
| return float("nan") | |
| return float((np.linalg.norm(P[1] - P[5]) + np.linalg.norm(P[2] - P[4])) / (2.0 * width)) | |
| def landmarks_xy(lms, w, h, idx): | |
| return np.array([[lms[i].x * w, lms[i].y * h] for i in idx], np.float64) | |
| # ----------------------------------------------------------------------------- the extractor | |
| class MPFaceExtractor(B.Extractor): | |
| baseline = BASELINE | |
| channels = CHANNELS | |
| model_ids = [REGISTRY_ID, PIP_ID] | |
| needs_crop = True | |
| decode_wh = None # native frames: the head region keeps the sensor resolution (probe: detection 0.84 native vs 0.76 at 964x604) | |
| batch = 32 | |
| def __init__(self, keep_debug=False): | |
| self.keep_debug = keep_debug | |
| self.debug_rows = [] # per detected face: (yaw, pitch, roll, landmark yaw proxy, eye-line angle deg, stage) -- sign-convention checks only | |
| self.infer_ms = [] # per-frame milliseconds (both regions when the fallback ran) | |
| self.stage_counts = {"head": 0, "full": 0, "none": 0, "no_crop": 0} | |
| self._lmk = None | |
| def setup(self, registry_entry=None): | |
| import cv2 | |
| import mediapipe as mp | |
| from mediapipe.tasks import python as mpp | |
| from mediapipe.tasks.python import vision | |
| from baselines_dms import registry as R | |
| task_path = R.resolve(REGISTRY_ID) # asserts the sha256 first | |
| entry = R.get(REGISTRY_ID) | |
| pip_entry = R.get(PIP_ID) | |
| assert not R.verify(pip_entry), f"registry {PIP_ID}: {R.verify(pip_entry)}" | |
| self.registry_entry = registry_entry or {k: entry.get(k) for k in ("id", "version", "url", "local_path", "sha256", "size_bytes", "license")} | |
| self.weight_sha256 = {task_path.name: entry["sha256"]} | |
| self.lib_versions = {"mediapipe": mp.__version__, "numpy": np.__version__, "opencv": cv2.__version__} | |
| self._mp, self._cv2 = mp, cv2 | |
| self.options = {"running_mode": "IMAGE", "num_faces": 1, "output_face_blendshapes": True, "output_facial_transformation_matrixes": True, | |
| "min_face_detection_confidence": 0.5, "min_face_presence_confidence": 0.5, "min_tracking_confidence": 0.5} | |
| opts = vision.FaceLandmarkerOptions(base_options=mpp.BaseOptions(model_asset_path=str(task_path)), running_mode=vision.RunningMode.IMAGE, num_faces=1, | |
| min_face_detection_confidence=0.5, min_face_presence_confidence=0.5, min_tracking_confidence=0.5, | |
| output_face_blendshapes=True, output_facial_transformation_matrixes=True) | |
| self._lmk = vision.FaceLandmarker.create_from_options(opts) | |
| return self | |
| def teardown(self): | |
| if self._lmk is not None: | |
| try: | |
| self._lmk.close() | |
| except Exception: | |
| pass | |
| self._lmk = None | |
| def _detect(self, img_bgr): | |
| rgb = np.ascontiguousarray(self._cv2.cvtColor(img_bgr, self._cv2.COLOR_BGR2RGB)) | |
| res = self._lmk.detect(self._mp.Image(image_format=self._mp.ImageFormat.SRGB, data=rgb)) | |
| return res if (res.face_landmarks and res.face_blendshapes and res.facial_transformation_matrixes) else None | |
| def _row(self, res, w, h): | |
| row = np.full(self.C, np.nan, np.float32) | |
| row[CI["mp_face_conf"]] = 1.0 | |
| bs = {c.category_name: float(c.score) for c in res.face_blendshapes[0]} | |
| for ch, name in BLENDSHAPES.items(): | |
| row[CI[ch]] = bs.get(name, np.nan) | |
| yaw, pitch, roll = matrix_to_angles(res.facial_transformation_matrixes[0]) | |
| row[CI["mp_head_yaw"]], row[CI["mp_head_pitch"]], row[CI["mp_head_roll"]] = yaw, pitch, roll | |
| lms = res.face_landmarks[0] | |
| row[CI["mp_ear_l"]] = ear_from_points(landmarks_xy(lms, w, h, EAR_L)) | |
| row[CI["mp_ear_r"]] = ear_from_points(landmarks_xy(lms, w, h, EAR_R)) | |
| return row, (yaw, pitch, roll), lms | |
| def infer(self, crops): | |
| out = np.full((len(crops), self.C), np.nan, np.float32) | |
| for i, c in enumerate(crops): | |
| if c is None or c.size == 0 or c.shape[0] < 2 or c.shape[1] < 2: | |
| self.stage_counts["no_crop"] += 1 | |
| continue | |
| t0 = time.time() | |
| head, is_sub = head_region(c) | |
| img, stage = head, "head" | |
| res = self._detect(head) | |
| if res is None and is_sub: | |
| img, stage = c, "full" | |
| res = self._detect(c) | |
| self.infer_ms.append(1000.0 * (time.time() - t0)) | |
| if res is None: | |
| self.stage_counts["none"] += 1 | |
| continue | |
| self.stage_counts[stage] += 1 | |
| h, w = img.shape[:2] | |
| out[i], ypr, lms = self._row(res, w, h) | |
| if self.keep_debug: | |
| nose = landmarks_xy(lms, w, h, (NOSE_TIP,))[0] | |
| re_, le_ = landmarks_xy(lms, w, h, (EYE_R_OUTER, EYE_L_OUTER)) | |
| iod = np.linalg.norm(le_ - re_) | |
| proxy = (nose[0] - 0.5 * (le_[0] + re_[0])) / iod if iod > 1e-6 else np.nan # landmark analogue of the face pack's lm_yaw: > 0 nose right of the eye midpoint | |
| eye_angle = np.degrees(np.arctan2(-(le_[1] - re_[1]), le_[0] - re_[0])) # eye line angle in an image-up frame: > 0 = subject's left eye higher = ccw in the image | |
| self.debug_rows.append((ypr[0], ypr[1], ypr[2], float(proxy), float(eye_angle), stage)) | |
| return out | |
| # channel roles for summarise / the report | |
| def role_spec(): | |
| """{channel: role} with roles angle_deg | eye_closed_prob | eye_open_ratio | prob | conf.""" | |
| return dict(ROLES) | |
| def summarise_spec(): | |
| """summarise.spec_for-style spec (matches summarise.SPECS['mp_face']).""" | |
| import summarise as SM | |
| return {"angle": {"mp_head_yaw": SM.ANGLE_YAW, "mp_head_pitch": SM.ANGLE_PITCH}, | |
| "eye_state": {"mp_blink_l": {"kind": "closed_p", "th": 0.5}, "mp_blink_r": {"kind": "closed_p", "th": 0.5}}, | |
| "prob": ["mp_face_conf", "mp_look_up_l", "mp_look_down_l", "mp_look_in_l", "mp_look_out_l", "mp_look_up_r", "mp_look_down_r", "mp_look_in_r", "mp_look_out_r"], | |
| "embedding": [], "rule": {"yaw": "mp_head_yaw", "pitch": "mp_head_pitch", "closed": ["mp_blink_l", "mp_blink_r"]}, | |
| "chain": {"head_dev": ("mp_head_yaw", "mp_head_pitch"), "closed": ["mp_blink_l", "mp_blink_r"], "gaze_off": None}} | |
| def describe(self): | |
| d = super().describe() | |
| d.update(options=getattr(self, "options", None), head_region="top square of the driver crop (rows 0..min(h,w)); fallback = full crop when no face", | |
| colour="BGR->RGB, no normalisation", roles=self.role_spec()) | |
| return d | |
| # ----------------------------------------------------------------------------- manifest / measurements | |
| def load_manifest(path=MANIFEST): | |
| import pandas as pd | |
| m = pd.read_csv(path) | |
| need = {"uid", "t_anchor", "win_start", "win_end", "regime", "y", "row"} | |
| assert need <= set(m.columns), (need - set(m.columns)) | |
| return m | |
| GRID_EPS_S = 0.02 # span ends are snapped to the 10 Hz grid and pulled inwards by this much (see grid_spans) | |
| def grid_spans(spans, hz=cs.HZ, eps=GRID_EPS_S): | |
| """[[start, end]] -> f64[S,2] with both ends snapped to the hz grid and pulled INWARDS by eps: [k0/hz + eps, k1/hz - eps]. | |
| run_route stores spans as float32 and video_io.frames_in_spans floors / ceils them without tolerance, so a raw end such as 834.2 (f32 | |
| 834.20001) would pull in the frame at 834.3 = one step AFTER the requested window end (and 376.8 -> f32 376.79999 one step before the start). | |
| With eps = 0.02 s (>> the f32 spacing, 0.002 s at 5 h) floor / ceil land exactly on k0 .. k1; cache_schema.covers (tol 0.051) still accepts | |
| the raw windows against the stored spans. Verified 2026-09-14: exact for all 50 manifest windows and for every 0.1 s start up to 18,000 s. | |
| A single-frame request [t, t] gives the (inverted) pair [k/hz + eps, k/hz - eps], for which frames_in_spans yields exactly frame k.""" | |
| a = np.asarray(spans, np.float64).reshape(-1, 2) | |
| k0, k1 = np.round(a[:, 0] * hz), np.round(a[:, 1] * hz) | |
| return np.stack([k0 / hz + eps, k1 / hz - eps], 1) | |
| def spans_by_uid(m): | |
| """uid -> merged decode spans of its manifest windows [win_start, win_end], grid-snapped (grid_spans) so that exactly the frames k0 .. k1 are read.""" | |
| S = {} | |
| for r in m.itertuples(): | |
| S.setdefault(r.uid, []).append([float(r.win_start), float(r.win_end)]) | |
| return {u: cs.merge_spans(grid_spans(v)) for u, v in S.items()} | |
| def dense_lookup(rc): | |
| """RouteCache -> dict k(=round 10 t) -> row index.""" | |
| k = np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64) | |
| return {int(kk): i for i, kk in enumerate(k)} | |
| def _pq(v): | |
| v = np.asarray([x for x in v if np.isfinite(x)], np.float64) | |
| return {"p50": float(np.percentile(v, 50)), "p95": float(np.percentile(v, 95)), "n": int(len(v))} if len(v) else {"p50": None, "p95": None, "n": 0} | |
| def _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": None, "n": int(ok.sum())} | |
| r = spearmanr(a[ok], b[ok]) | |
| return {"rho": float(r.statistic), "p": float(r.pvalue), "n": int(ok.sum())} | |
| def yaw_vs_face_pack(m, caches, channel="mp_head_yaw"): | |
| """pool the 80-step 10 Hz histories (t-7.9..t) of `channel` and the face pack's lm_yaw_dev (Fh8_c[row,:,1]) / lm_yaw (channel 0) over the anchors.""" | |
| out = {"channel": channel, "pack": str(FACE_PACK), "pairs": 0, "per_anchor_rho": []} | |
| A_raw, B_dev, B_yaw, A_c, B_c = [], [], [], [], [] | |
| for regime in sorted(m.regime.unique()): | |
| sub = m[m.regime == regime] | |
| with np.load(str(FACE_PACK).format(regime=regime), allow_pickle=True) as z: | |
| rows = sub.row.to_numpy().astype(np.int64) | |
| assert np.array_equal(z["uid"][rows].astype(str), sub.uid.to_numpy().astype(str)), f"{regime}: manifest row does not point at the pack anchor" | |
| assert np.allclose(z["t"][rows], sub.t_anchor.to_numpy(), atol=1e-6) | |
| ch = [str(c) for c in z["channels"]] | |
| j_dev, j_yaw = ch.index("lm_yaw_dev"), ch.index("lm_yaw") | |
| Fh = z["Fh8_c"][rows].astype(np.float32) | |
| Fv = z["Fh8valid"][rows] | |
| for (i, r), fh, fv in zip(sub.iterrows(), Fh, Fv): | |
| rc = caches.get(r.uid) | |
| if rc is None: | |
| continue | |
| look = dense_lookup(rc) | |
| ka = int(round(r.t_anchor * cs.HZ)) | |
| ks = np.arange(ka - 79, ka + 1) | |
| mine = np.full(80, np.nan, np.float32) | |
| for s, kk in enumerate(ks): | |
| idx = look.get(int(kk)) | |
| if idx is not None and rc.valid[idx]: | |
| mine[s] = float(rc.feat[idx, CI[channel]]) | |
| dev = np.where(fv, fh[:, j_dev], np.nan) | |
| yaw = np.where(fv, fh[:, j_yaw], np.nan) | |
| ok = np.isfinite(mine) & np.isfinite(dev) | |
| out["per_anchor_rho"].append({"idx": int(r.idx), "uid": r.uid, "t_anchor": float(r.t_anchor), "n_pairs": int(ok.sum()), **({"rho": _spearman(mine, dev)["rho"]} if ok.sum() >= 10 else {"rho": None})}) | |
| A_raw.append(mine[ok]) | |
| B_dev.append(dev[ok]) | |
| B_yaw.append(yaw[ok]) | |
| if ok.sum() >= 10: | |
| A_c.append(mine[ok] - np.nanmean(mine[ok])) | |
| B_c.append(dev[ok] - np.nanmean(dev[ok])) | |
| if A_raw: | |
| a, bd, by = np.concatenate(A_raw), np.concatenate(B_dev), np.concatenate(B_yaw) | |
| out["pairs"] = int(len(a)) | |
| out["rho_pooled_vs_lm_yaw_dev"] = _spearman(a, bd) | |
| out["rho_pooled_vs_lm_yaw"] = _spearman(a, by) | |
| out["rho_within_window_centred_vs_lm_yaw_dev"] = _spearman(np.concatenate(A_c), np.concatenate(B_c)) if A_c else None | |
| rr = [d["rho"] for d in out["per_anchor_rho"] if d["rho"] is not None] | |
| out["per_anchor_rho_median"] = float(np.median(rr)) if rr else None | |
| out["per_anchor_rho_share_positive"] = float(np.mean(np.asarray(rr) > 0)) if rr else None | |
| return out | |
| def contact_sheet(m, caches, out_png, ex, n_side=6, tile=300): | |
| """12 frames: the 6 most negative and the 6 most positive valid mp_head_yaw values (at most one per route per side), re-decoded at native | |
| resolution, the HEAD region drawn with the yaw / pitch / roll printed and an arrow along the yaw sign.""" | |
| import cv2 | |
| cand = [] | |
| for u, rc in caches.items(): | |
| for i in np.flatnonzero(rc.valid): | |
| cand.append((float(rc.feat[i, CI["mp_head_yaw"]]), float(rc.feat[i, CI["mp_head_pitch"]]), float(rc.feat[i, CI["mp_head_roll"]]), u, float(rc.t[i]))) | |
| cand.sort(key=lambda x: x[0]) | |
| def pick(seq): | |
| got, used = [], set() | |
| for c in seq: | |
| if c[3] in used: | |
| continue | |
| used.add(c[3]) | |
| got.append(c) | |
| if len(got) == n_side: | |
| break | |
| return got | |
| lo, hi = pick(cand), pick(cand[::-1]) | |
| sel = lo + hi[::-1] | |
| tiles = [] | |
| for yaw, pitch, roll, u, t in sel: | |
| fs = vio.FrameStream(u, grid_spans([[t, t]]), workers=1, decode_wh=None, block=4) # exactly the cached frame k = round(10 t); nothing after it | |
| img = None | |
| for t_, fi_, fr_, bx_ in fs: | |
| if fr_ is not None and abs(t_ - t) < 1e-3: | |
| crop = vio.driver_crop(fr_, bx_) | |
| if crop is not None: | |
| head, is_sub = head_region(crop) | |
| res = ex._detect(head) | |
| img = head if (res is not None or not is_sub) else crop | |
| if res is None and is_sub: | |
| res = ex._detect(crop) | |
| img = img.copy() | |
| if res is not None: | |
| h, w = img.shape[:2] | |
| for idx in (EAR_L + EAR_R + (NOSE_TIP,)): | |
| p = res.face_landmarks[0][idx] | |
| cv2.circle(img, (int(p.x * w), int(p.y * h)), max(2, w // 150), (0, 255, 0), -1) | |
| if img is None: | |
| img = np.zeros((tile, tile, 3), np.uint8) | |
| h, w = img.shape[:2] | |
| s = tile / max(h, w) | |
| img = cv2.resize(img, (max(1, int(round(w * s))), max(1, int(round(h * s)))), interpolation=cv2.INTER_AREA) | |
| canvas = np.zeros((tile + 70, tile, 3), np.uint8) | |
| y0, x0 = (tile - img.shape[0]) // 2, (tile - img.shape[1]) // 2 | |
| canvas[y0:y0 + img.shape[0], x0:x0 + img.shape[1]] = img | |
| cx, cy = tile // 2, tile + 20 | |
| L = int(np.clip(abs(yaw), 5, 60) * 1.8) | |
| cv2.arrowedLine(canvas, (cx, cy), (cx + (L if yaw > 0 else -L), cy), (0, 200, 255), 3, tipLength=0.35) | |
| cv2.putText(canvas, f"yaw {yaw:+.1f}", (6, tile + 48), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA) | |
| cv2.putText(canvas, f"pit {pitch:+.1f} rol {roll:+.1f}", (6, tile + 66), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (200, 200, 200), 1, cv2.LINE_AA) | |
| cv2.putText(canvas, f"{u.split('__')[0][:14]} t={t:.1f}", (6, 14), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 0), 1, cv2.LINE_AA) | |
| tiles.append(canvas) | |
| rows = [np.concatenate(tiles[i:i + 4], 1) for i in range(0, len(tiles), 4)] | |
| sheet = np.concatenate(rows, 0) | |
| banner = np.zeros((36, sheet.shape[1], 3), np.uint8) | |
| cv2.putText(banner, "mp_face sign convention: yaw > 0 = face forward axis points to IMAGE RIGHT (arrow); left 6 = most negative yaw, right 6 = most positive; green = EAR + nose landmarks", | |
| (6, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 255), 1, cv2.LINE_AA) | |
| sheet = np.concatenate([banner, sheet], 0) | |
| Path(out_png).parent.mkdir(parents=True, exist_ok=True) | |
| cv2.imwrite(str(out_png), sheet) | |
| return [{"uid": u, "t": t, "yaw": yaw, "pitch": pitch, "roll": roll} for yaw, pitch, roll, u, t in sel] | |
| def gpu_mem_mb_of_pid(pid): | |
| try: | |
| txt = subprocess.run(["nvidia-smi", "--query-compute-apps=pid,used_memory", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=10).stdout | |
| for line in txt.splitlines(): | |
| parts = [p.strip() for p in line.split(",")] | |
| if len(parts) >= 2 and parts[0].isdigit() and int(parts[0]) == pid: | |
| return float(parts[1]) | |
| except Exception: | |
| pass | |
| return 0.0 | |
| def run_phase1(workers=3, block=50, max_minutes=60.0, out_dir=OUT_DIR, force=False): | |
| import cv2 | |
| t_all = time.time() | |
| m = load_manifest() | |
| spans = spans_by_uid(m) | |
| uids = list(dict.fromkeys(m.uid.tolist())) | |
| cache_dir = Path(out_dir) / "cache" | |
| cache_dir.mkdir(parents=True, exist_ok=True) | |
| ex = MPFaceExtractor(keep_debug=True).setup() | |
| log_lines = [] | |
| def log(msg): | |
| print(msg, flush=True) | |
| log_lines.append(str(msg)) | |
| t_run = time.time() | |
| code, summ = B.run_routes(ex, uids, spans, cache_dir, max_minutes=max_minutes, workers=workers, log=log, block=block, force=force) | |
| wall_run = time.time() - t_run | |
| assert code == B.EXIT_DONE, f"run_routes exit {code}: {summ}" | |
| caches = {u: cs.RouteCache.load(cache_dir / f"{u}.npz") for u in uids} | |
| # ---- frame-level statistics (all cache rows = the union of the manifest windows per route) | |
| fail = np.concatenate([rc.fail for rc in caches.values()]) | |
| valid = fail == 0 | |
| with_box = (fail != 1) & (fail != 2) | |
| hist = {cs.FAIL_CODES[int(k)]: int(v) for k, v in zip(*np.unique(fail, return_counts=True))} | |
| feat = np.concatenate([rc.feat.astype(np.float32) for rc in caches.values()]) | |
| F = feat[valid] | |
| ch_stats = {c: {"mean": float(np.mean(F[:, j])), "std": float(np.std(F[:, j])), "p05": float(np.percentile(F[:, j], 5)), "p50": float(np.percentile(F[:, j], 50)), "p95": float(np.percentile(F[:, j], 95)), | |
| "min": float(F[:, j].min()), "max": float(F[:, j].max())} for j, c in enumerate(CHANNELS)} if len(F) else {} | |
| # ---- per-anchor values at t_anchor (S0 semantics: the frame AT t, else NaN here — no tolerance, descriptive only) | |
| per_anchor = [] | |
| for r in m.itertuples(): | |
| rc = caches[r.uid] | |
| look = dense_lookup(rc) | |
| idx = look.get(int(round(r.t_anchor * cs.HZ))) | |
| rowv = rc.feat[idx].astype(np.float32) if idx is not None else np.full(len(CHANNELS), np.nan, np.float32) | |
| win = (rc.t >= r.win_start - 1e-6) & (rc.t <= r.win_end + 1e-6) | |
| per_anchor.append({"idx": int(r.idx), "uid": r.uid, "regime": r.regime, "y": int(r.y), "t_anchor": float(r.t_anchor), "frame_at_anchor": idx is not None, | |
| "valid_at_anchor": bool(idx is not None and rc.valid[idx]), "fail_at_anchor": (cs.FAIL_CODES[int(rc.fail[idx])] if idx is not None else "no_row"), | |
| "n_frames_window": int(win.sum()), "valid_share_window": float(rc.valid[win].mean()) if win.any() else None, | |
| **{c: (None if not np.isfinite(rowv[CI[c]]) else float(rowv[CI[c]])) for c in ("mp_head_yaw", "mp_head_pitch", "mp_head_roll", "mp_blink_l", "mp_blink_r", "mp_ear_l", "mp_ear_r", "mp_look_out_l", "mp_look_in_r")}}) | |
| # ---- descriptive score-like distribution by y (the y column is used for NOTHING else) | |
| by_y = {} | |
| for yv in (0, 1): | |
| vals = [a["mp_head_yaw"] for a in per_anchor if a["y"] == yv and a["mp_head_yaw"] is not None] | |
| vs = [a["valid_share_window"] for a in per_anchor if a["y"] == yv and a["valid_share_window"] is not None] | |
| by_y[str(yv)] = {"n": int(sum(a["y"] == yv for a in per_anchor)), "yaw_at_anchor_median": (float(np.median(vals)) if vals else None), "valid_share_window_median": (float(np.median(vs)) if vs else None)} | |
| # ---- runtime | |
| infer = _pq(ex.infer_ms) | |
| dec50 = [rc.runtime.get("decode_ms_p50") for rc in caches.values()] | |
| dec95 = [rc.runtime.get("decode_ms_p95") for rc in caches.values()] | |
| route_wall = [rc.runtime.get("wall_s") for rc in caches.values()] | |
| n_frames = int(len(fail)) | |
| ru_self, ru_ch = resource.getrusage(resource.RUSAGE_SELF), resource.getrusage(resource.RUSAGE_CHILDREN) | |
| vram = gpu_mem_mb_of_pid(os.getpid()) | |
| # ---- sign-convention consistency over all detected faces (landmark proxy / eye line; not a tuning) | |
| dbg = np.array([d[:5] for d in ex.debug_rows], np.float64) if ex.debug_rows else np.zeros((0, 5)) | |
| sign = {"n_faces": int(len(dbg)), | |
| "rho_yaw_vs_landmark_yaw_proxy": _spearman(dbg[:, 0], dbg[:, 3]) if len(dbg) else None, | |
| "rho_roll_vs_eye_line_angle": _spearman(dbg[:, 2], dbg[:, 4]) if len(dbg) else None, | |
| "yaw_median_deg": float(np.median(dbg[:, 0])) if len(dbg) else None, "pitch_median_deg": float(np.median(dbg[:, 1])) if len(dbg) else None, "roll_median_deg": float(np.median(dbg[:, 2])) if len(dbg) else None, | |
| "stage_counts": dict(ex.stage_counts)} | |
| # blink <-> EAR pairing check: the subject-side pairing must be the more negative correlation | |
| pair = {"rho_blink_l_vs_ear_l": _spearman(F[:, CI["mp_blink_l"]], F[:, CI["mp_ear_l"]]), "rho_blink_l_vs_ear_r": _spearman(F[:, CI["mp_blink_l"]], F[:, CI["mp_ear_r"]]), | |
| "rho_blink_r_vs_ear_r": _spearman(F[:, CI["mp_blink_r"]], F[:, CI["mp_ear_r"]]), "rho_blink_r_vs_ear_l": _spearman(F[:, CI["mp_blink_r"]], F[:, CI["mp_ear_l"]])} if len(F) else {} | |
| # ---- face pack comparison | |
| fp = yaw_vs_face_pack(m, caches) | |
| # ---- contact sheet | |
| sheet_png = Path(out_dir) / "sign_convention_contact_sheet.png" | |
| sheet = contact_sheet(m, caches, sheet_png, ex) | |
| ex.teardown() | |
| wall_all = time.time() - t_all | |
| meas = {"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": len(uids), "extractor": ex.describe(), "code_sha256": cs.code_sha256(ex.code_files()), | |
| "runner": {"workers": workers, "block": block, "decode_wh": "native", "hz": cs.HZ, "run_routes_summary": {k: summ[k] for k in ("n_done", "n_failed", "n_remaining", "wall_s")}}, | |
| "n_frames": n_frames, "n_frames_with_box": int(with_box.sum()), "valid_share_overall": float(valid.mean()), "valid_share_with_box": float(valid[with_box].mean()) if with_box.any() else None, | |
| "fail_hist": hist, "stage_counts": dict(ex.stage_counts), | |
| "runtime": {"infer_ms_per_frame": infer, "decode_ms_per_frame_route_p50": _pq(dec50), "decode_ms_per_frame_route_p95_max": float(np.nanmax(np.asarray(dec95, np.float64))) if dec95 else None, | |
| "route_wall_s": _pq(route_wall), "wall_s_run_routes": float(wall_run), "wall_s_total_incl_measurements": float(wall_all), | |
| "fps_end_to_end": float(n_frames / wall_run) if wall_run > 0 else None, "peak_rss_mb_main": ru_self.ru_maxrss / 1024.0, "peak_rss_mb_children_max": ru_ch.ru_maxrss / 1024.0, | |
| "vram_mb_peak": vram, "device": "cpu", "omp_num_threads": os.environ.get("OMP_NUM_THREADS")}, | |
| "channel_stats_valid_frames": ch_stats, "per_anchor": per_anchor, "descriptive_by_y": by_y, "sign_convention": sign, "blink_ear_pairing": pair, "face_pack_yaw": fp, | |
| "contact_sheet": {"png": str(sheet_png), "frames": sheet}, "log_tail": log_lines[-60:]} | |
| (Path(out_dir) / "measurements.json").write_text(json.dumps(meas, indent=1, default=str)) | |
| print(json.dumps({k: meas[k] for k in ("n_frames", "valid_share_overall", "valid_share_with_box", "fail_hist", "stage_counts")}, indent=1)) | |
| print(json.dumps({"runtime": meas["runtime"], "sign": sign, "blink_ear_pairing": pair, "face_pack": {k: v for k, v in fp.items() if k != "per_anchor_rho"}}, indent=1, default=str)) | |
| return meas | |
| # ----------------------------------------------------------------------------- self-test | |
| def self_test(workers=2, n_frames=30): | |
| import tempfile | |
| import summarise as SM | |
| rep = {"self_test": "extractors.mp_face", "checks": {}} | |
| # (1) rotation decomposition round trip + scipy cross-check | |
| rng = np.random.default_rng(0) | |
| for _ in range(200): | |
| y, p, r = rng.uniform(-80, 80), rng.uniform(-60, 60), rng.uniform(-80, 80) | |
| y2, p2, r2 = matrix_to_angles(angles_to_matrix(y, p, r)) | |
| assert np.allclose([y, p, r], [y2, p2, r2], atol=1e-6), (y, p, r, y2, p2, r2) | |
| try: | |
| from scipy.spatial.transform import Rotation as Ro | |
| Rm = angles_to_matrix(30.0, 10.0, -5.0) | |
| a, b, c = Ro.from_matrix(Rm).as_euler("YXZ", degrees=True) | |
| assert np.allclose([a, -b, c], [30.0, 10.0, -5.0], atol=1e-6), (a, b, c) | |
| rep["checks"]["scipy_YXZ_cross_check"] = "ok (yaw = Y, pitch = -X, roll = Z of the intrinsic YXZ decomposition)" | |
| except ImportError: | |
| rep["checks"]["scipy_YXZ_cross_check"] = "scipy missing; skipped" | |
| f = angles_to_matrix(30.0, 0.0, 0.0)[:, 2] | |
| assert f[0] > 0 and abs(np.degrees(np.arctan2(f[0], f[2])) - 30) < 1e-9, "yaw > 0 must point the forward axis to +x" | |
| assert angles_to_matrix(0.0, 20.0, 0.0)[1, 2] > 0, "pitch > 0 must point the forward axis to +y (up)" | |
| assert angles_to_matrix(0.0, 0.0, 20.0)[1, 0] > 0, "roll > 0 must rotate +x towards +y (ccw in an image-up frame)" | |
| assert abs(ear_from_points([[0, 0], [1, -1], [2, -1], [3, 0], [2, 1], [1, 1]]) - 4 / 6) < 1e-12 and np.isnan(ear_from_points([[0, 0]] * 6)) | |
| hr, sub = head_region(np.zeros((900, 400, 3), np.uint8)) | |
| assert hr.shape == (400, 400, 3) and sub and not head_region(np.zeros((300, 400, 3), np.uint8))[1] | |
| rep["checks"]["geometry"] = "ok" | |
| gs = grid_spans([[834.2 - 10.2, 834.2], [376.8, 387.0]]) | |
| for (a_, b_), (k0, k1) in zip(cs.merge_spans(gs), ((3768, 3870), (8240, 8342))): # through the f32 cast of run_route (merge_spans sorts by start) | |
| k_ = vio.frames_in_spans([[a_, b_]], cs.HZ, None)[0] | |
| assert k_[0] == k0 and k_[-1] == k1 and len(k_) == k1 - k0 + 1, (a_, b_, k_[0], k_[-1]) | |
| assert cs.spans_cover(cs.merge_spans(gs), [[824.0, 834.2], [376.8, 387.0]]) and list(vio.frames_in_spans(grid_spans([[41.0, 41.0]]), cs.HZ, None)[0]) == [410] | |
| rep["checks"]["grid_spans"] = "ok (f32-safe snapping: exactly k0..k1, no frame after the window end)" | |
| # (2) spec consistency with summarise.SPECS and the roles | |
| spec = MPFaceExtractor.summarise_spec() | |
| ref = SM.SPECS["mp_face"] | |
| assert set(spec["angle"]) == set(ref["angle"]) and set(spec["eye_state"]) == set(ref["eye_state"]) and set(spec["prob"]) == set(ref["prob"]) and spec["rule"] == ref["rule"], (spec, ref) | |
| norm = SM.spec_for("mp_face", CHANNELS, spec) | |
| assert set(norm["angle"]) | set(norm["eye_state"]) | set(norm["prob"]) == set(CHANNELS) | |
| assert set(ROLES) == set(CHANNELS) and set(ROLES.values()) <= {"angle_deg", "eye_closed_prob", "eye_open_ratio", "prob", "logit", "embedding", "conf", "none"} | |
| rep["checks"]["spec"] = "ok" | |
| # (3) hygiene: no forbidden token in this file | |
| sys.path.insert(0, str(CWM3 / "tests")) | |
| import test_no_dms_inputs as TN | |
| src = Path(__file__).read_text() | |
| bad = [t for t in TN.FORBIDDEN if re.search(re.escape(t), src)] | |
| assert not bad, bad | |
| rep["checks"]["hygiene"] = "ok" | |
| # (4) one manifest route, n_frames frames through the real runner into a temp dir | |
| m = load_manifest() | |
| r = m.iloc[0] | |
| span = grid_spans([[r.t_anchor - (n_frames - 1) / cs.HZ, r.t_anchor]]) # exactly n_frames grid steps, the last one AT t_anchor | |
| ex = MPFaceExtractor(keep_debug=True).setup() | |
| with tempfile.TemporaryDirectory() as d: | |
| t0 = time.time() | |
| code, path, info = B.run_route(ex, r.uid, span, d, max_minutes=10, workers=workers, block=16) | |
| wall = time.time() - t0 | |
| rc = cs.RouteCache.load(path) | |
| # grid_spans -> exactly n_frames rows, none after t_anchor (causality), the last one AT t_anchor | |
| assert code == 0 and info["status"] == "done" and rc.n == n_frames and rc.C == len(CHANNELS) and rc.channels == CHANNELS and rc.meta["baseline"] == BASELINE, (code, info, rc.n) | |
| assert int(round(r.t_anchor * cs.HZ)) in dense_lookup(rc) and rc.t.astype(np.float64).max() <= r.t_anchor + 1e-3, (rc.t.max(), r.t_anchor) | |
| code2, _, info2 = B.run_route(ex, r.uid, span, d, max_minutes=10, workers=workers) | |
| assert code2 == 0 and info2["status"] == "exists" | |
| sm = SM.summarise_route(rc, np.array([r.t_anchor]), spec) | |
| assert sm["X"].shape[0] == 1 and sm["row_ok"][0] | |
| F = rc.feat.astype(np.float32)[rc.valid] | |
| dbg = np.array([x[:5] for x in ex.debug_rows], np.float64) if ex.debug_rows else np.zeros((0, 5)) | |
| rep.update(route=r.uid, span=span.tolist(), n_frames=rc.n, valid_share=float(rc.valid.mean()), fail_hist=info["fail_hist"], stage_counts=dict(ex.stage_counts), wall_s=round(wall, 2), | |
| infer_ms=_pq(ex.infer_ms), runtime=rc.runtime, native_wh=rc.meta["native_wh"], decode_wh=rc.meta["decode_wh"], | |
| valid_rows_ranges={c: [float(F[:, j].min()), float(F[:, j].max())] for j, c in enumerate(CHANNELS)} if len(F) else {}, | |
| yaw_vs_landmark_proxy=_spearman(dbg[:, 0], dbg[:, 3]) if len(dbg) else None, resume_status=info2["status"], summarise_K=int(sm["X"].shape[1]), describe=ex.describe()) | |
| if len(F): | |
| assert np.all((F[:, CI["mp_blink_l"]] >= 0) & (F[:, CI["mp_blink_l"]] <= 1)) and np.all(F[:, CI["mp_face_conf"]] == 1.0) | |
| assert np.all(np.abs(F[:, CI["mp_head_yaw"]]) <= 180) and np.all(F[:, CI["mp_ear_l"]] > 0) | |
| ex.teardown() | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return rep | |
| 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("--workers", type=int, default=3) | |
| ap.add_argument("--block", type=int, default=50, help="frames per decode block (native frames are ~7 MB each; 50 keeps <= 2.5 GB in flight with 3 workers)") | |
| ap.add_argument("--max-minutes", type=float, default=60.0) | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--out-dir", default=str(OUT_DIR)) | |
| a = ap.parse_args() | |
| if a.phase1: | |
| run_phase1(workers=min(a.workers, 3), block=a.block, max_minutes=a.max_minutes, out_dir=Path(a.out_dir), force=a.force) | |
| return 0 | |
| self_test(workers=min(a.workers, 2)) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 39.7 kB
- Xet hash:
- 6e39a6106eeb1ee83c24e6bb4a580b38cb2b9b1fcbedcc6d3f42e50937e6b8a7
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.