Buckets:
| """cwm3/baselines_dms/summarise.py — per-route feature cache (10 Hz) -> anchor summaries on the s12 5 Hz grid, CAUSAL windows only (plan §3 "锚点汇总", 2026-09-14). | |
| Alignment / causality contract | |
| * anchors = rows of s12_wm/results/targets_{regime}_{split}.npz (uid, t on a 0.2 s grid); write_summary writes one row per pack row, uid / t equal | |
| to the pack (asserted), `row_ok` False (and every quantity NaN) where the route has no cache or no frame at t (missing policy: NaN where undefined). | |
| * the route cache is placed on a dense 10 Hz grid k = round(10 t) (dense_series; absent / invalid frames -> NaN); an anchor at T reads the 80 | |
| steps k(T) - 79 .. k(T) (t - 7.9 .. t) — `window_index` asserts t_frame <= T + 1e-6 for every used frame; NOTHING after T is ever read | |
| (tests/test_baselines_dms.py plants a spike at T + 0.1 s and checks every column is unchanged). | |
| * S0 = the latest valid frame <= T within S0_TOL_S = 0.15 s (i.e. the frame at T, else the one 0.1 s earlier), NaN otherwise. | |
| Feature blocks (per channel role; the roles of a baseline's channels come from a small spec dict, see SPECS / spec_for): | |
| angle {ch}_share_off_{1,3,8}s share of valid steps in the OFF state, OFF = hysteresis on |dev| with the CIRCULAR deviation dev = ((x - ref + 180) mod 360) - 180 | |
| (`circ_dev`, degrees; 2026-09-14 fix: L2CS yaw wraps near +-180 on the off-axis camera, so 179 vs -179 is 2°, not 358°) | |
| (on `on`° / off `off`°, > 1 s invalid gap resets; train/warning_features_ext.hysteresis_rows semantics, re-implemented here | |
| without importing that module); rules.py (rule0 / ruleref) uses the same |circ_dev| | |
| {ch}_dwell_since_on_s seconds since the last valid NOT-off step = length of the current off spell (0 when on-road now; cap 8 s; NaN when no valid step) | |
| {ch}_max_abs_3s (of |circ_dev|), {ch}_mean_8s, {ch}_std_8s (of the raw angle) over valid steps (NaN when none) | |
| ... computed twice: ref = 0 (prefix `{ch}_`) and ref = the route's CAUSAL expanding median over MOVING valid frames <= t (prefix `{ch}_refm_`, | |
| NaN before REF_N_MIN = 25 moving frames; build_kin_causal.expanding_median_moving semantics, the pool being the route itself) | |
| eye_state {ch}_closed_share_3s, {ch}_longest_closure_s_8s, {ch}_n_closures_8s, {ch}_min_open_3s closed = p > th (kind closed_p) or ratio < th (kind open_ratio) | |
| prob {ch}_mean_3s, {ch}_mean_8s, {ch}_min_3s | |
| embedding (window caches, summarise_windows): emb_anchor[D] = the last valid window with win_t_end <= T (within ANCHOR_TOL_S), emb_mean_8s[D] over | |
| valid windows with T - 8 < win_t_end <= T, n_win_8s | |
| Also written by write_summary: valid_share_8s, n_valid_8s, the fixed-rule columns of rules.py (rule0 / ruleref and their components, chain-style | |
| att@3s), `thresholds` json (every constant), `source_cache_sha256` json {uid: sha256 of the cache file}, hz 10, hist_s 8, baseline, spec json. | |
| Spec lookup (2026-09-14, layer-2 launch): summarise_route resolves the spec by the CACHE's meta.baseline (the extractor's BASELINE constant), so SPECS must | |
| carry every extractor name: `l2cs_gaze` is an ALIAS of the original `l2cs` entry (same dict; the extractor writes meta.baseline = "l2cs_gaze" while the | |
| spec was first written under `l2cs`), `openface3` mirrors extractors/openface3.SUMMARISE_SPEC verbatim (asserted equal in --self-test when the extractor | |
| module is importable), and the WINDOW baselines (videomae2 / ov_action) declare `default_role: "ignore"`: their per-frame cache channels (crop geometry / | |
| encoder units / logits) are NOT summarised as `prob` blocks — X stays empty and the anchor quantity is the window embedding (fit_readout.summary_features | |
| then uses emb_anchor ‖ emb_mean_8s ‖ n_win_8s, as documented there). `excluded_channels` (openface3 emotions) are likewise never defaulted to `prob`. | |
| Nothing here reads a video, a model or any openpilot field. | |
| python baselines_dms/summarise.py --self-test # synthetic cache: causality + shapes | |
| python baselines_dms/summarise.py --baseline l2cs --regime manual --split val [--cache-dir ...] [--out ...] | |
| """ | |
| from __future__ import annotations | |
| import bisect | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| PKG = Path(__file__).resolve().parent | |
| CWM3 = PKG.parent | |
| EXPLORE = CWM3.parent | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| if str(PKG) not in sys.path: | |
| sys.path.insert(0, str(PKG)) | |
| import cache_schema as cs # noqa: E402 | |
| HZ = 10 | |
| HIST_S = 8.0 | |
| HIST = int(round(HIST_S * HZ)) | |
| S0_TOL_S = 0.15 | |
| ANCHOR_TOL_S = 0.5 # embedding: the anchor window may end up to 0.5 s before T (window grid coarser than 10 Hz) | |
| GAP_RESET = 10 # > 1 s of invalid steps resets the hysteresis state | |
| DWELL_CAP_S = 8.0 | |
| REF_N_MIN = 25 | |
| WIN = {"1s": 10, "3s": 30, "8s": 80} | |
| ROLES = ("angle", "eye_state", "prob", "embedding") | |
| ANGLE_YAW = {"on": 20.0, "off": 12.0} | |
| ANGLE_PITCH = {"on": 15.0, "off": 9.0} # decisions.md fixes pitch on 15°; off = 0.6 x on, the ratio of the yaw pair (recorded in `thresholds`) | |
| # default channel roles per baseline (channel names of plan §3; a concrete extractor registers / checks its spec against these). "rule": the channels | |
| # rules.py reads (yaw / pitch angle, eye-closure probability or list averaged, gaze-off angle pair); "chain": the three legacy attention channels. | |
| SPECS = { | |
| "l2cs": {"angle": {"l2cs_yaw": ANGLE_YAW, "l2cs_pitch": ANGLE_PITCH}, "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")}}, | |
| "ov_gaze": {"angle": {"ov_head_yaw": ANGLE_YAW, "ov_head_pitch": ANGLE_PITCH, "ov_gaze_yaw": ANGLE_YAW, "ov_gaze_pitch": ANGLE_PITCH}, | |
| "eye_state": {"ov_eye_closed_l": {"kind": "closed_p", "th": 0.5}, "ov_eye_closed_r": {"kind": "closed_p", "th": 0.5}}, "prob": ["ov_face_conf"], "embedding": [], | |
| "rule": {"yaw": "ov_head_yaw", "pitch": "ov_head_pitch", "closed": ["ov_eye_closed_l", "ov_eye_closed_r"]}, | |
| "chain": {"head_dev": ("ov_head_yaw", "ov_head_pitch"), "closed": ["ov_eye_closed_l", "ov_eye_closed_r"], "gaze_off": ("ov_gaze_yaw", "ov_gaze_pitch")}}, | |
| "mp_face": {"angle": {"mp_head_yaw": ANGLE_YAW, "mp_head_pitch": 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}}, | |
| "videomae2": {"angle": {}, "eye_state": {}, "prob": [], "embedding": ["emb"], "rule": None, "chain": None, "default_role": "ignore"}, | |
| "ov_action": {"angle": {}, "eye_state": {}, "prob": [], "embedding": ["logits"], "rule": None, "chain": None, "default_role": "ignore"}, | |
| # mirror of extractors/openface3.SUMMARISE_SPEC (constants: decisions.md yaw 20/12°, pitch 15/9°; EAR closure 0.2 = Soukupova & Cech 2016; emotions excluded; | |
| # the fixed rule's `closed` stays None because decisions.md defines it on an eye-CLOSED probability only). --self-test asserts equality with the extractor. | |
| "openface3": {"angle": {"of3_head_yaw": ANGLE_YAW, "of3_head_pitch": ANGLE_PITCH, "of3_gaze_yaw": ANGLE_YAW, "of3_gaze_pitch": ANGLE_PITCH}, | |
| "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"] + [f"of3_au_{i:02d}" for i in range(1, 9)], "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": [f"of3_emo_{e}" for e in ("neutral", "happy", "sad", "surprise", "fear", "disgust", "anger", "contempt")]}, | |
| } | |
| SPECS["l2cs_gaze"] = SPECS["l2cs"] # alias: the L2CS extractor's BASELINE / cache dir / meta.baseline is "l2cs_gaze" (same dict object) | |
| SPEC_ALIASES = {"l2cs_gaze": "l2cs"} | |
| def spec_for(baseline, channels, spec=None): | |
| """normalised spec: every role present; channels not named by any role default to `prob` (unless the spec says `default_role: "ignore"` — window baselines — | |
| or lists them in `excluded_channels`); angle thresholds as {on, off}; roles restricted to existing channels.""" | |
| s = dict(spec if spec is not None else SPECS.get(baseline, {})) | |
| ch = list(channels) | |
| default_role = s.get("default_role", "prob") | |
| assert default_role in ("prob", "ignore"), default_role | |
| excluded = set(s.get("excluded_channels", [])) | |
| out = {"angle": {}, "eye_state": {}, "prob": [], "embedding": list(s.get("embedding", [])), "rule": s.get("rule"), "chain": s.get("chain"), | |
| "default_role": default_role, "excluded_channels": sorted(excluded)} | |
| for c, th in dict(s.get("angle", {})).items(): | |
| if c in ch: | |
| th = {"on": float(th[0]), "off": float(th[1])} if isinstance(th, (tuple, list)) else {"on": float(th["on"]), "off": float(th["off"])} | |
| assert th["on"] > th["off"] > 0, (c, th) | |
| out["angle"][c] = th | |
| for c, es in dict(s.get("eye_state", {})).items(): | |
| if c in ch: | |
| es = dict(es) | |
| assert es.get("kind", "closed_p") in ("closed_p", "open_ratio"), es | |
| out["eye_state"][c] = {"kind": es.get("kind", "closed_p"), "th": float(es.get("th", 0.5))} | |
| named = set(out["angle"]) | set(out["eye_state"]) | |
| out["prob"] = [c for c in s.get("prob", []) if c in ch] | |
| if default_role == "prob": | |
| out["prob"] += [c for c in ch if c not in named and c not in s.get("prob", []) and c not in excluded] | |
| return out | |
| # ----------------------------------------------------------------------------- window primitives (local re-implementations; no import of train/) | |
| def circ_dev(x, ref=0.0): | |
| """CIRCULAR deviation of angles in DEGREES: ((x - ref + 180) mod 360) - 180, in [-180, 180); NaN propagates. Every angle_deg channel of every baseline goes through | |
| this before the hysteresis / max_abs (angle_block) and before rule0 / ruleref (rules.off_state) — 2026-09-14 fix: L2CS yaw is centred near -100° on the off-axis | |
| cabin camera and wraps at +-180°, so a linear x - ref read 358° for a 2° difference (tests/test_baselines_dms.py::test_angle_deviation_is_circular).""" | |
| x = np.asarray(x, np.float64) | |
| return (np.mod(x - np.asarray(ref, np.float64) + 180.0, 360.0) - 180.0).astype(np.float32) | |
| def hysteresis_rows(x, on, off, gap_reset=GAP_RESET): | |
| """x [n, T] (NaN = invalid) -> state [n, T] in {0, 1, NaN}: per-row causal hysteresis, state (re-)initialised as (v >= on) at the first valid step and after | |
| a run of > gap_reset invalid steps; v >= on -> 1, v < off -> 0, else hold (train/warning_features_ext.hysteresis_rows semantics).""" | |
| x = np.asarray(x, np.float32) | |
| n, T = x.shape | |
| S = np.full((n, T), np.nan, np.float32) | |
| cur = np.zeros(n, np.float32) | |
| init = np.ones(n, bool) | |
| gap = np.full(n, gap_reset + 1, np.int64) | |
| for i in range(T): | |
| v = x[:, i] | |
| fin = np.isfinite(v) | |
| reset = fin & (init | (gap > gap_reset)) | |
| cur = np.where(reset, (v >= on).astype(np.float32), np.where(fin & (v >= on), 1.0, np.where(fin & (v < off), 0.0, cur))).astype(np.float32) | |
| init = init & ~fin | |
| gap = np.where(fin, 0, gap + 1) | |
| S[:, i] = np.where(fin, cur, np.nan) | |
| return S | |
| def _share(mask_true, valid, k): | |
| """share of the valid steps among the last k that are True; NaN when no valid step.""" | |
| v = valid[:, -k:].sum(1) | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| return np.where(v > 0, (mask_true[:, -k:] & valid[:, -k:]).sum(1) / np.maximum(v, 1), np.nan).astype(np.float32) | |
| def _dwell_since(state_true, valid, cap=DWELL_CAP_S, hz=HZ): | |
| """seconds since the last valid step where state_true holds (0 if that is the last step); cap when never; NaN when no valid step.""" | |
| n, T = valid.shape | |
| idx = np.where(state_true & valid, np.arange(T)[None, :], -1).max(1) | |
| out = np.where(idx >= 0, (T - 1 - idx) / hz, cap).astype(np.float32) | |
| out[valid.sum(1) == 0] = np.nan | |
| return np.minimum(out, cap).astype(np.float32) | |
| def _longest_run_s(state_true, valid, hz=HZ): | |
| """longest run of consecutive valid True steps -> seconds; NaN when no valid step.""" | |
| n, T = valid.shape | |
| best = np.zeros(n, np.int64) | |
| cur = np.zeros(n, np.int64) | |
| for i in range(T): | |
| on = state_true[:, i] & valid[:, i] | |
| cur = np.where(on, cur + 1, 0) | |
| best = np.maximum(best, cur) | |
| out = (best / hz).astype(np.float32) | |
| out[valid.sum(1) == 0] = np.nan | |
| return out | |
| def _n_runs(state_true, valid): | |
| """number of run starts of (valid & True) — a run is broken by a valid False step OR an invalid step; NaN when no valid step.""" | |
| on = state_true & valid | |
| starts = on & ~np.concatenate([np.zeros((len(on), 1), bool), on[:, :-1]], 1) | |
| out = starts.sum(1).astype(np.float32) | |
| out[valid.sum(1) == 0] = np.nan | |
| return out | |
| def _nan_reduce(x, k, fn): | |
| """fn over the last k steps of x [n,T] ignoring NaN; NaN when all NaN.""" | |
| xs = x[:, -k:] | |
| any_ = np.isfinite(xs).any(1) | |
| out = np.full(len(x), np.nan, np.float32) | |
| if any_.any(): | |
| with np.errstate(all="ignore"): | |
| out[any_] = fn(xs[any_], 1) | |
| return out | |
| def causal_expanding_median(x, valid, moving, n_min=REF_N_MIN): | |
| """per-frame causal reference: median of x over frames j <= i with valid & moving & finite (frame i included); NaN while fewer than n_min such frames.""" | |
| arr = [] | |
| out = np.full(len(x), np.nan, np.float32) | |
| last = np.nan | |
| for i in range(len(x)): | |
| v = x[i] | |
| if valid[i] and moving[i] and np.isfinite(v): | |
| bisect.insort(arr, float(v)) | |
| n = len(arr) | |
| if n >= n_min: | |
| last = arr[n // 2] if n % 2 else 0.5 * (arr[n // 2 - 1] + arr[n // 2]) | |
| out[i] = last | |
| return out | |
| # ----------------------------------------------------------------------------- dense grid / windows | |
| def dense_series(cache: cs.RouteCache): | |
| """cache -> dict(k0, M, G f32[M,C] (NaN absent / invalid), V bool[M], F bool[M] frame present, MOV bool[M] (True everywhere when the cache has no `moving`)).""" | |
| k = np.round(cache.t.astype(np.float64) * HZ).astype(np.int64) | |
| k0 = int(k.min()) if len(k) else 0 | |
| M = int(k.max() - k0 + 1) if len(k) else 0 | |
| C = cache.C | |
| G = np.full((M, C), np.nan, np.float32) | |
| V = np.zeros(M, bool) | |
| F = np.zeros(M, bool) | |
| MOV = np.zeros(M, bool) | |
| G[k - k0] = cache.feat.astype(np.float32) | |
| V[k - k0] = cache.valid | |
| F[k - k0] = cache.fail != 2 | |
| MOV[k - k0] = True if cache.moving is None else cache.moving | |
| G[~V] = np.nan | |
| return {"k0": k0, "M": M, "G": G, "V": V, "F": F, "MOV": MOV, "channels": list(cache.channels)} | |
| def anchor_steps(k0, anchors_t, hz=HZ): | |
| ka = np.round(np.asarray(anchors_t, np.float64) * hz).astype(np.int64) | |
| assert np.allclose(ka / hz, anchors_t, atol=1e-6), "anchors must lie on the 10 Hz grid" | |
| return ka - k0 | |
| def window_index(k0, M, anchors_t, hist=HIST, hz=HZ): | |
| """-> idx i64[n, hist] into the dense series (-1 outside) for the steps t - (hist-1)/hz .. t; ASSERTS every used frame time <= t_anchor + 1e-6.""" | |
| ka = anchor_steps(k0, anchors_t, hz) | |
| idx = ka[:, None] + np.arange(-hist + 1, 1)[None, :] | |
| ok = (idx >= 0) & (idx < M) | |
| t_frame = (idx + k0) / hz | |
| assert np.all(t_frame[ok] <= (np.asarray(anchors_t, np.float64)[:, None] + 1e-6).repeat(hist, 1)[ok]), "window_index: a frame after the anchor was selected" | |
| return np.where(ok, idx, -1) | |
| def gather(G, V, idx): | |
| ok = idx >= 0 | |
| ic = np.clip(idx, 0, max(len(G) - 1, 0)) | |
| Gh = np.where(ok[..., None], G[ic], np.nan).astype(np.float32) | |
| Vh = np.where(ok, V[ic], False) | |
| return Gh, Vh | |
| def s0_index(k0, M, V, anchors_t, tol_s=S0_TOL_S, hz=HZ): | |
| """index of the latest valid step <= t within tol_s (-1 none).""" | |
| ka = anchor_steps(k0, anchors_t, hz) | |
| out = np.full(len(ka), -1, np.int64) | |
| for back in range(int(np.floor(tol_s * hz + 1e-9)), -1, -1): # farthest first so that the nearest valid wins | |
| j = ka - back | |
| ok = (j >= 0) & (j < M) | |
| ok[ok] &= V[j[ok]] | |
| out[ok] = j[ok] | |
| return out | |
| # ----------------------------------------------------------------------------- feature blocks | |
| def angle_block(xh, Vh, on, off, ref_h=None, prefix=""): | |
| x = np.where(Vh, xh, np.nan).astype(np.float32) | |
| dev = np.abs(circ_dev(x, 0.0 if ref_h is None else ref_h)) # circular (degrees): 179 vs -179 -> 2, never 358 | |
| st = hysteresis_rows(dev, on, off) | |
| sv = np.isfinite(st) | |
| off_ = sv & (st > 0.5) | |
| not_off = sv & (st < 0.5) | |
| X = np.stack([_share(off_, sv, WIN["1s"]), _share(off_, sv, WIN["3s"]), _share(off_, sv, WIN["8s"]), _dwell_since(not_off, sv), | |
| _nan_reduce(dev, WIN["3s"], np.nanmax), _nan_reduce(x, WIN["8s"], np.nanmean), _nan_reduce(x, WIN["8s"], np.nanstd)], 1).astype(np.float32) | |
| names = [f"{prefix}share_off_1s", f"{prefix}share_off_3s", f"{prefix}share_off_8s", f"{prefix}dwell_since_on_s", f"{prefix}max_abs_3s", f"{prefix}mean_8s", f"{prefix}std_8s"] | |
| return X, names, {"off": off_, "state_valid": sv} | |
| def eye_block(xh, Vh, kind, th, prefix=""): | |
| x = np.where(Vh, xh, np.nan).astype(np.float32) | |
| ev = Vh & np.isfinite(x) | |
| closed = ev & ((x > th) if kind == "closed_p" else (x < th)) | |
| open_ = (1.0 - x) if kind == "closed_p" else x | |
| X = np.stack([_share(closed, ev, WIN["3s"]), _longest_run_s(closed, ev), _n_runs(closed, ev), _nan_reduce(open_, WIN["3s"], np.nanmin)], 1).astype(np.float32) | |
| names = [f"{prefix}closed_share_3s", f"{prefix}longest_closure_s_8s", f"{prefix}n_closures_8s", f"{prefix}min_open_3s"] | |
| return X, names, {"closed": closed, "state_valid": ev} | |
| def prob_block(xh, Vh, prefix=""): | |
| x = np.where(Vh, xh, np.nan).astype(np.float32) | |
| X = np.stack([_nan_reduce(x, WIN["3s"], np.nanmean), _nan_reduce(x, WIN["8s"], np.nanmean), _nan_reduce(x, WIN["3s"], np.nanmin)], 1).astype(np.float32) | |
| return X, [f"{prefix}mean_3s", f"{prefix}mean_8s", f"{prefix}min_3s"], {} | |
| def summarise_route(cache: cs.RouteCache, anchors_t, spec=None, hist=HIST, hz=HZ): | |
| """-> dict(S0 f32[n,C], S0valid, X f32[n,K], x_names, valid_share_8s, n_valid_8s, row_ok, Gh [n,hist,C], Vh, Refh {angle ch: [n,hist]}, refs {ch: [M]}, spec, channels, ka).""" | |
| anchors_t = np.asarray(anchors_t, np.float64) | |
| spec = spec_for(cache.meta.get("baseline", ""), cache.channels, spec) | |
| D = dense_series(cache) | |
| G, V, F, MOV, k0, M = D["G"], D["V"], D["F"], D["MOV"], D["k0"], D["M"] | |
| ci = {c: j for j, c in enumerate(cache.channels)} | |
| n = len(anchors_t) | |
| idx = window_index(k0, M, anchors_t, hist, hz) | |
| Gh, Vh = gather(G, V, idx) | |
| i0 = s0_index(k0, M, V, anchors_t, S0_TOL_S, hz) | |
| S0 = np.where((i0 >= 0)[:, None], G[np.clip(i0, 0, max(M - 1, 0))], np.nan).astype(np.float32) if M else np.full((n, cache.C), np.nan, np.float32) | |
| S0valid = i0 >= 0 | |
| ka = anchor_steps(k0, anchors_t, hz) | |
| inside = (ka >= 0) & (ka < M) | |
| row_ok = inside.copy() | |
| row_ok[inside] = F[ka[inside]] | |
| Xs, names, refs, Refh = [], [], {}, {} | |
| for c, th in spec["angle"].items(): | |
| j = ci[c] | |
| X0, n0, _ = angle_block(Gh[:, :, j], Vh, th["on"], th["off"], None, f"{c}_") | |
| refs[c] = causal_expanding_median(G[:, j], V, MOV) | |
| Refh[c] = np.where(idx >= 0, refs[c][np.clip(idx, 0, max(M - 1, 0))], np.nan).astype(np.float32) | |
| X1, n1, _ = angle_block(Gh[:, :, j], Vh, th["on"], th["off"], Refh[c], f"{c}_refm_") | |
| Xs += [X0, X1] | |
| names += n0 + n1 | |
| for c, es in spec["eye_state"].items(): | |
| X2, n2, _ = eye_block(Gh[:, :, ci[c]], Vh, es["kind"], es["th"], f"{c}_") | |
| Xs.append(X2) | |
| names += n2 | |
| for c in spec["prob"]: | |
| X3, n3, _ = prob_block(Gh[:, :, ci[c]], Vh, f"{c}_") | |
| Xs.append(X3) | |
| names += n3 | |
| X = np.concatenate(Xs, 1).astype(np.float32) if Xs else np.zeros((n, 0), np.float32) | |
| assert X.shape == (n, len(names)) | |
| return {"S0": S0, "S0valid": S0valid, "X": X, "x_names": names, "valid_share_8s": Vh.mean(1).astype(np.float32), "n_valid_8s": Vh.sum(1).astype(np.int32), "row_ok": row_ok, | |
| "Gh": Gh, "Vh": Vh, "Refh": Refh, "refs": refs, "spec": spec, "channels": list(cache.channels), "ka": ka, "k0": k0, "M": M} | |
| def summarise_windows(wc: cs.WindowCache, anchors_t, hist_s=HIST_S, anchor_tol_s=ANCHOR_TOL_S): | |
| """-> dict(emb_anchor f32[n,D], emb_mean_8s f32[n,D], n_win_8s i32[n]); windows are causal for T iff win_t_end <= T + 1e-6 (asserted).""" | |
| anchors_t = np.asarray(anchors_t, np.float64) | |
| te = wc.win_t_end.astype(np.float64) | |
| E = wc.emb.astype(np.float32) | |
| valid = wc.win_valid | |
| n, D = len(anchors_t), wc.D | |
| j_hi = np.searchsorted(te, anchors_t + 1e-6, side="right") # windows [0, j_hi) end at or before T | |
| j_lo = np.searchsorted(te, anchors_t - hist_s, side="right") # windows [j_lo, ...) end after T - 8 | |
| j_lo = np.minimum(j_lo, j_hi) | |
| used = j_hi > 0 | |
| assert np.all(te[j_hi[used] - 1] <= anchors_t[used] + 1e-6), "summarise_windows: a window ending after the anchor was selected" | |
| Ez = np.where(valid[:, None], E, 0.0) | |
| csum = np.concatenate([np.zeros((1, D), np.float64), np.cumsum(Ez, 0, dtype=np.float64)]) | |
| ccnt = np.concatenate([[0], np.cumsum(valid.astype(np.int64))]) | |
| cnt = ccnt[j_hi] - ccnt[j_lo] | |
| with np.errstate(invalid="ignore", divide="ignore"): | |
| mean = (csum[j_hi] - csum[j_lo]) / np.maximum(cnt, 1)[:, None] | |
| mean = np.where((cnt > 0)[:, None], mean, np.nan).astype(np.float32) | |
| last_valid = np.maximum.accumulate(np.where(valid, np.arange(len(te)), -1)) if len(te) else np.zeros(0, np.int64) | |
| lv = np.where(used, last_valid[np.clip(j_hi - 1, 0, max(len(te) - 1, 0))] if len(te) else -1, -1) | |
| ok = (lv >= 0) & (lv >= j_lo) | |
| ok[ok] &= (anchors_t[ok] - te[lv[ok]]) <= anchor_tol_s + 1e-6 | |
| emb_anchor = np.where(ok[:, None], E[np.clip(lv, 0, max(len(te) - 1, 0))] if len(te) else np.nan, np.nan).astype(np.float32) | |
| return {"emb_anchor": emb_anchor, "emb_mean_8s": mean, "n_win_8s": cnt.astype(np.int32)} | |
| # ----------------------------------------------------------------------------- pack-level writer | |
| def _route_blocks(uid): | |
| order = np.argsort(uid, kind="stable") | |
| bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| for a, b in zip(bounds[:-1], bounds[1:]): | |
| rows = order[a:b] | |
| yield str(uid[rows[0]]), rows | |
| def write_summary(regime, split, out_path=None, baseline="l2cs", cache_dir=None, window_dir=None, spec=None, limit_routes=0, log=print, with_rules=True): | |
| """-> path of summ_{regime}_{split}.npz (row-aligned with the s12 pack). cache_dir default results/baselines_dms/cache/<baseline>; window_dir optional (embedding baselines).""" | |
| import rules as RL | |
| t0 = time.time() | |
| cache_dir = Path(cache_dir) if cache_dir else cs.cache_dir(baseline) | |
| window_dir = Path(window_dir) if window_dir else None | |
| out_path = Path(out_path) if out_path else cs.summ_dir(baseline) / f"summ_{regime}_{split}.npz" | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| uid, t = z["uid"], z["t"] | |
| N = len(uid) | |
| row_ok = np.zeros(N, bool) | |
| route_dec = np.zeros(N, bool) | |
| S0 = X = None | |
| channels = x_names = None | |
| vs8 = np.full(N, np.nan, np.float32) | |
| nv8 = np.zeros(N, np.int32) | |
| rule_cols = {} | |
| emb = {} | |
| shas = {} | |
| n_routes = n_done = 0 | |
| spec_used = None | |
| chain_absent = [] | |
| for u, rows in _route_blocks(uid): | |
| n_routes += 1 | |
| p = cache_dir / f"{u}.npz" | |
| pw = window_dir / f"{u}.npz" if window_dir else None | |
| if not p.exists() and not (pw and pw.exists()): | |
| continue | |
| if limit_routes and n_done >= limit_routes: | |
| break | |
| n_done += 1 | |
| route_dec[rows] = True | |
| ts = t[rows].astype(np.float64) | |
| if p.exists(): | |
| rc = cs.RouteCache.load(p) | |
| shas[u] = cs.sha256_file(p) | |
| sm = summarise_route(rc, ts, spec) | |
| if S0 is None: | |
| channels, x_names, spec_used = sm["channels"], sm["x_names"], sm["spec"] | |
| S0 = np.full((N, len(channels)), np.nan, np.float32) | |
| X = np.full((N, len(x_names)), np.nan, np.float32) | |
| assert sm["channels"] == channels and sm["x_names"] == x_names, f"{u}: channel / feature set differs from the first route" | |
| S0[rows] = sm["S0"] | |
| X[rows] = sm["X"] | |
| row_ok[rows] = sm["row_ok"] | |
| vs8[rows] = sm["valid_share_8s"] | |
| nv8[rows] = sm["n_valid_8s"] | |
| if with_rules: | |
| for k, v in RL.rule_columns(sm, sm["spec"]).items(): | |
| if k == "chain_absent_channels": | |
| chain_absent = list(v) | |
| continue | |
| rule_cols.setdefault(k, np.full(N, np.nan, np.float32))[rows] = v | |
| if pw and pw.exists(): | |
| wc = cs.WindowCache.load(pw) | |
| shas[f"win:{u}"] = cs.sha256_file(pw) | |
| ew = summarise_windows(wc, ts) | |
| if not emb: | |
| emb = {"emb_anchor": np.full((N, wc.D), np.nan, np.float16), "emb_mean_8s": np.full((N, wc.D), np.nan, np.float16), "n_win_8s": np.zeros(N, np.int32)} | |
| emb["emb_anchor"][rows] = ew["emb_anchor"].astype(np.float16) | |
| emb["emb_mean_8s"][rows] = ew["emb_mean_8s"].astype(np.float16) | |
| emb["n_win_8s"][rows] = ew["n_win_8s"] | |
| if not p.exists(): | |
| row_ok[rows] = ew["n_win_8s"] > 0 | |
| if S0 is None: | |
| S0 = np.zeros((N, 0), np.float32) | |
| X = np.zeros((N, 0), np.float32) | |
| channels, x_names, spec_used = [], [], spec_for(baseline, [], spec) | |
| S0[~row_ok] = np.nan | |
| X[~row_ok] = np.nan | |
| for k in rule_cols: | |
| rule_cols[k][~row_ok] = np.nan | |
| thresholds = {"hz": HZ, "hist_s": HIST_S, "hist_steps": HIST, "s0_tol_s": S0_TOL_S, "gap_reset_steps": GAP_RESET, "dwell_cap_s": DWELL_CAP_S, "ref_n_min": REF_N_MIN, | |
| "anchor_tol_s_embedding": ANCHOR_TOL_S, "angle": spec_used["angle"], "eye_state": spec_used["eye_state"], "rule": RL.RULE, "chain": RL.CHAIN, | |
| "chain_absent_channels": chain_absent} | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert np.array_equal(uid, z["uid"]) and np.array_equal(t, z["t"]) | |
| out_path.parent.mkdir(parents=True, exist_ok=True) | |
| np.savez_compressed(out_path, uid=uid, t=t, row_ok=row_ok, route_decoded=route_dec, S0=S0, S0valid=row_ok & np.isfinite(S0).all(1) if S0.shape[1] else row_ok, channels=np.array(channels), | |
| X=X, x_names=np.array(x_names), valid_share_8s=vs8, n_valid_8s=nv8, rule_names=np.array(sorted(rule_cols)), **{k: rule_cols[k] for k in sorted(rule_cols)}, **emb, | |
| thresholds=np.array(json.dumps(thresholds)), source_cache_sha256=np.array(json.dumps(shas)), hz=np.int64(HZ), hist_s=np.float32(HIST_S), baseline=np.array(baseline), | |
| spec=np.array(json.dumps(spec_used)), cache_dir=np.array(str(cache_dir)), window_dir=np.array(str(window_dir) if window_dir else ""), regime=np.array(regime), split=np.array(split), | |
| n_routes=np.int64(n_routes), n_routes_decoded=np.int64(n_done), written_utc=np.array(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))) | |
| log(f"{baseline} {regime}/{split}: N={N:,} routes {n_routes} decoded {n_done} row_ok {row_ok.mean():.4f} K={len(x_names)} rules {sorted(rule_cols)} {time.time() - t0:.1f}s -> {out_path}") | |
| return out_path | |
| def _check_extractor_specs(): | |
| """the mirrored openface3 spec equals extractors/openface3.SUMMARISE_SPEC (skipped when the extractor module is not importable); the alias is the same object; | |
| window baselines ignore their per-frame channels.""" | |
| assert SPECS["l2cs_gaze"] is SPECS["l2cs"] | |
| for b in ("videomae2", "ov_action"): | |
| sp = spec_for(b, ["vmae2_crop_w", "ova_enc_000", "ova_logit_0"]) | |
| assert sp["prob"] == [] and sp["angle"] == {} and sp["eye_state"] == {} and sp["default_role"] == "ignore", (b, sp) | |
| of3 = spec_for("openface3", [f"of3_emo_{e}" for e in ("neutral", "happy")] + ["of3_head_yaw", "of3_head_pitch", "of3_gaze_yaw", "of3_gaze_pitch", "of3_eye_open_l", "of3_eye_open_r", "of3_face_conf", "of3_au_01", "of3_head_roll"]) | |
| assert not any(c.startswith("of3_emo_") for c in of3["prob"]) and "of3_head_roll" in of3["prob"] and set(of3["angle"]) == {"of3_head_yaw", "of3_head_pitch", "of3_gaze_yaw", "of3_gaze_pitch"}, of3 | |
| try: | |
| sys.path.insert(0, str(PKG / "extractors")) | |
| from openface3 import SUMMARISE_SPEC as X # noqa: F401 (numpy / cv2-level imports only) | |
| except Exception as e: # pragma: no cover | |
| return f"extractor spec not importable ({type(e).__name__}); mirror not cross-checked" | |
| mine = json.loads(json.dumps(SPECS["openface3"])) | |
| theirs = json.loads(json.dumps(X)) | |
| theirs["angle"] = {k: {"on": float(v[0]), "off": float(v[1])} if isinstance(v, list) else v for k, v in theirs["angle"].items()} | |
| theirs["chain"] = {k: (list(v) if isinstance(v, list) else v) for k, v in theirs["chain"].items()} | |
| assert mine == theirs, {"mine": mine, "extractor": theirs} | |
| return "openface3 mirror == extractors/openface3.SUMMARISE_SPEC" | |
| def _self_test(): | |
| import rules as RL | |
| spec_check = _check_extractor_specs() | |
| rc = cs.synthetic_route_cache(channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), t1=120.0) | |
| spec = {"angle": {"yaw_deg": (20, 12), "pitch_deg": (15, 9)}, "eye_state": {"closed_p": {"kind": "closed_p", "th": 0.5}}, "prob": ["conf"], | |
| "rule": {"yaw": "yaw_deg", "pitch": "pitch_deg", "closed": "closed_p"}, "chain": {"head_dev": ("yaw_deg", "pitch_deg"), "closed": "closed_p", "gaze_off": None}} | |
| anchors = np.arange(10.0, 118.0, 0.2) | |
| sm = summarise_route(rc, anchors, spec) | |
| n = len(anchors) | |
| assert sm["S0"].shape == (n, 4) and sm["X"].shape == (n, len(sm["x_names"])) and sm["Gh"].shape == (n, HIST, 4) and len(sm["x_names"]) == 2 * 7 * 2 + 4 + 3 | |
| assert np.isfinite(sm["X"][sm["row_ok"]]).mean() > 0.9 and sm["row_ok"].mean() > 0.9 | |
| rules = RL.rule_columns(sm, sm["spec"]) | |
| # causality: for each of several anchors, a spike at T + 0.1 s (the next 10 Hz step; valid + moving so it would enter every later reference) | |
| # changes NO summary / rule column of THAT anchor (later anchors legitimately see it in their causal route reference) | |
| sparse = anchors[::50] | |
| sm_s = summarise_route(rc, sparse, spec) | |
| rules_s = RL.rule_columns(sm_s, sm_s["spec"]) | |
| k = np.round(rc.t.astype(np.float64) * HZ).astype(np.int64) | |
| pos = {int(kk): i for i, kk in enumerate(k)} | |
| def same(a, b): | |
| a, b = np.asarray(a, np.float64), np.asarray(b, np.float64) | |
| return a.shape == b.shape and np.array_equal(np.isnan(a), np.isnan(b)) and np.array_equal(a[~np.isnan(a)], b[~np.isnan(b)]) | |
| for i, a in enumerate(sparse[:6]): | |
| hit = pos[int(round(a * HZ)) + 1] | |
| feat = rc.feat.astype(np.float32) | |
| valid_sp, fail_sp = rc.valid.copy(), rc.fail.copy() | |
| valid_sp[hit], fail_sp[hit] = True, 0 | |
| feat[hit] = 1e3 | |
| feat[hit, 2] = 1.0 | |
| feat[~valid_sp] = np.nan | |
| mov = rc.moving.copy() | |
| mov[hit] = True | |
| rc_spike = cs.RouteCache(rc.uid, rc.t, rc.frame_idx, rc.box, valid_sp, fail_sp, feat.astype(np.float16), rc.channels, rc.spans, rc.meta, rc.runtime, mov).validate() | |
| sm2 = summarise_route(rc_spike, sparse[i:i + 1], spec) | |
| rules2 = RL.rule_columns(sm2, sm2["spec"]) | |
| assert same(sm_s["X"][i], sm2["X"][0]) and same(sm_s["S0"][i], sm2["S0"][0]) and same(sm_s["valid_share_8s"][i], sm2["valid_share_8s"][0]) and sm_s["row_ok"][i] == sm2["row_ok"][0] | |
| rc_ = [c for c in rules_s if c != "chain_absent_channels"] | |
| assert all(same(rules_s[c][i], rules2[c][0]) for c in rc_), [c for c in rc_ if not same(rules_s[c][i], rules2[c][0])] | |
| sm3 = summarise_route(rc_spike, sparse[i:i + 1] + 0.2, spec) # teeth: 0.2 s later the spike IS in the window | |
| assert not same(sm_s["X"][i], sm3["X"][0]) | |
| # S0 tolerance: invalid frame at T -> frame at T - 0.1 is used | |
| i_t = pos[int(round(anchors[5] * HZ))] | |
| rc_h = cs.RouteCache.load(rc.save(Path("/tmp/summ_selftest_tmp.npz"))) | |
| rc_h.valid[i_t] = False | |
| rc_h.fail[i_t] = 3 | |
| rc_h.feat[i_t] = np.nan | |
| smh = summarise_route(rc_h, anchors[5:6], spec) | |
| prev = rc.feat[i_t - 1].astype(np.float32) | |
| assert smh["S0valid"][0] == rc.valid[i_t - 1] and (not smh["S0valid"][0] or np.array_equal(smh["S0"][0], prev)) | |
| # window caches | |
| M, D = 230, 6 | |
| rng = np.random.default_rng(1) | |
| wv = rng.random(M) < 0.9 | |
| E = rng.normal(0, 1, (M, D)).astype(np.float32) | |
| E[~wv] = np.nan | |
| wc = cs.WindowCache(rc.uid, (np.arange(M) * 0.5 + 1.5).astype(np.float32), E.astype(np.float16), wv, 16, np.array([[0.0, 120.0]], np.float32), rc.meta, rc.runtime) | |
| ew = summarise_windows(wc, anchors) | |
| j = int(np.searchsorted(wc.win_t_end.astype(np.float64), anchors[7] + 1e-6, side="right")) - 1 | |
| assert wc.win_t_end[j] <= anchors[7] and ew["emb_anchor"].shape == (n, D) and ew["n_win_8s"].max() <= 17 and ew["n_win_8s"][40] >= 12 | |
| E2 = wc.emb.astype(np.float32) | |
| E2[j + 1] = 50.0 | |
| wc2 = cs.WindowCache(rc.uid, wc.win_t_end, E2.astype(np.float16), wc.win_valid | (np.arange(M) == j + 1), 16, wc.spans, wc.meta, wc.runtime) | |
| ew2 = summarise_windows(wc2, anchors[7:8]) | |
| assert same(ew2["emb_anchor"][0], ew["emb_anchor"][7]) and same(ew2["emb_mean_8s"][0], ew["emb_mean_8s"][7]) | |
| print(json.dumps({"self_test": "summarise", "ok": True, "spec_check": spec_check, "n_anchors": n, "K": len(sm["x_names"]), "x_names_head": sm["x_names"][:8], "rule_columns": sorted(rules), | |
| "row_ok": float(sm["row_ok"].mean()), "valid_share_8s_mean": float(np.nanmean(sm["valid_share_8s"])), "rule0_defined": float(np.isfinite(rules["rule0"]).mean()), | |
| "ruleref_defined": float(np.isfinite(rules["ruleref"]).mean()), "n_win_8s_median": float(np.median(ew["n_win_8s"]))}, indent=1)) | |
| if __name__ == "__main__": | |
| import argparse | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--self-test", action="store_true") | |
| ap.add_argument("--baseline", default="") | |
| ap.add_argument("--regime", default="manual") | |
| ap.add_argument("--split", default="val") | |
| ap.add_argument("--cache-dir", default="") | |
| ap.add_argument("--window-dir", default="") | |
| ap.add_argument("--out", default="") | |
| ap.add_argument("--limit-routes", type=int, default=0) | |
| a = ap.parse_args() | |
| if a.baseline: | |
| write_summary(a.regime, a.split, a.out or None, a.baseline, a.cache_dir or None, a.window_dir or None, limit_routes=a.limit_routes) | |
| if a.self_test or not a.baseline: | |
| _self_test() | |
Xet Storage Details
- Size:
- 35.9 kB
- Xet hash:
- ffabf086a5829551f0a21d6bcd6b2e26961967cdbe523883971e3063db1cc639
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.