Buckets:
| """cwm3/baselines_dms/extractors/ov_action.py — I2: Intel OMZ driver-action-recognition-adas-0002 (encoder + decoder) as a DMS baseline (plan §3, 2026-09-14). | |
| Model (registry ids omz_driver_action_recognition_adas_0002_encoder / _decoder; OMZ 2023.0 FP16 IR, Apache-2.0; CPU openvino) | |
| encoder input '0' [B,3,224,224] f32 -> output '513' [B,512,1,1] per-frame MobileNetV2 embedding. The IR carries its own preprocessing | |
| (rt_info conversion_parameters: mean_values [123.675,116.28,103.53], scale_values [58.395,57.12,57.375], reverse_input_channels=True), | |
| i.e. the network expects the RAW 0..255 BGR frame (cv2 order) in NCHW and does BGR->RGB + ImageNet normalisation internally. | |
| decoder input '0' [1,16,512] = the 16 most recent encoder embeddings -> output '804' [1,9] logits over the 9 driver actions (ACTIONS below; batch 1 only; | |
| order as recorded from the OMZ README / demos driver_actions.txt in the registry notes). | |
| Per-frame RouteCache (10 Hz) channels ova_enc_000..511 (role embedding) = the encoder embedding of the driver crop (video_io.driver_crop, +10 %), | |
| letterboxed to 224x224 (aspect preserved, centred, padding = ImageNet mean colour so that it is ~0 after the IR's own mean subtraction). | |
| valid <=> frame decoded AND driver box present AND the 512 values finite (fail codes of extractors/base.run_route). | |
| WindowCache (5 Hz) for every 5 Hz grid time T (k5 / 5) inside the requested spans: the 16 frames k10 - 15 .. k10 (k10 = 2 k5; causal 1.6 s ending AT T) | |
| -> emb f16[M,9] = decoder logits (channels ova_logit_0..8, role logit); win_valid = all 16 frames present with a box and a valid embedding; NaN otherwise. | |
| Summary channel ova_p_safe (role prob) = softmax(logits)[0] (Safe driving) — derived from emb, not stored. | |
| Nothing here reads an openpilot field; the video / time axis / driver box come from video_io only; no network; no GPU. | |
| python baselines_dms/extractors/ov_action.py --self-test [--uid UID] # 1 local TRAIN route, 30 frames: encoder, RouteCache, decoder windows | |
| python baselines_dms/extractors/ov_action.py --phase1 [--workers 3] [--max-minutes 60] [--force] | |
| -> results/baselines_dms/phase1/ov_action/{cache/<uid>.npz, cache_win/<uid>.npz, measurements.json, sign_convention_contact_sheet.png, logs/} | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import importlib.util | |
| import json | |
| import math | |
| import resource | |
| import subprocess | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| HERE = Path(__file__).resolve().parent | |
| PKG = HERE.parent | |
| CWM3 = PKG.parent | |
| for p in (str(PKG), str(HERE)): | |
| if p not in sys.path: | |
| sys.path.insert(0, p) | |
| import cache_schema as cs # noqa: E402 | |
| import video_io as vio # noqa: E402 | |
| from base import EXIT_BUDGET, EXIT_DONE, EXIT_FAIL, Extractor, run_route # noqa: E402 | |
| _spec = importlib.util.spec_from_file_location("bdms_registry", PKG / "registry.py") | |
| REG = importlib.util.module_from_spec(_spec) | |
| _spec.loader.exec_module(REG) | |
| BASELINE = "ov_action" | |
| ENC_ID = "omz_driver_action_recognition_adas_0002_encoder" | |
| DEC_ID = "omz_driver_action_recognition_adas_0002_decoder" | |
| ENC_XML = "driver-action-recognition-adas-0002-encoder.xml" | |
| DEC_XML = "driver-action-recognition-adas-0002-decoder.xml" | |
| SIZE = 224 | |
| D_EMB = 512 | |
| WIN_LEN = 16 | |
| WIN_HZ = 5 | |
| N_CLASSES = 9 | |
| PAD_BGR = (104, 116, 124) # ImageNet mean (103.53, 116.28, 123.675) in BGR order -> ~0 after the IR's baked-in mean subtraction | |
| # class order of driver-action-recognition-adas-0002 (OMZ README / demos/action_recognition_demo/python/driver_actions.txt, as recorded in registry.json notes) | |
| ACTIONS = ["safe_driving", "texting_left", "talking_phone_left", "texting_right", "talking_phone_right", "operating_radio", "drinking_eating", "reaching_behind", "hair_makeup"] | |
| ACTIONS_OMZ = ["Safe driving", "Texting left", "Talking on the phone left", "Texting right", "Talking on the phone right", "Operating the radio", "Drinking eating", | |
| "Reaching behind", "Hair and makeup"] | |
| ENC_CHANNELS = [f"ova_enc_{i:03d}" for i in range(D_EMB)] | |
| LOGIT_CHANNELS = [f"ova_logit_{i}" for i in range(N_CLASSES)] | |
| P_SAFE = "ova_p_safe" | |
| LEFT_IDX, RIGHT_IDX = (1, 2), (3, 4) # hand-side classes used for the descriptive left/right asymmetry proxy (measurements only) | |
| PHASE1_DIR = cs.RESULTS_ROOT / "phase1" / BASELINE | |
| MANIFEST = cs.RESULTS_ROOT / "phase1" / "manifest_v1.csv" | |
| FACE_PACK = CWM3 / "results" / "targets_face_causal_routes_v2" | |
| LM_YAW_DEV_CH = 1 # facehist_causal_*_train.npz Fh8_c[N,80,9] channel 1 = lm_yaw_dev (10 Hz, step 79 = t_anchor) | |
| # ----------------------------------------------------------------------------- preprocessing | |
| def letterbox(crop, size=SIZE, pad=PAD_BGR): | |
| """u8 BGR crop [h,w,3] -> u8 BGR [size,size,3]: longest side scaled to `size` (INTER_AREA down / INTER_LINEAR up), centred, padded with `pad`.""" | |
| import cv2 | |
| h, w = crop.shape[:2] | |
| s = size / float(max(h, w)) | |
| nw, nh = max(1, int(round(w * s))), max(1, int(round(h * s))) | |
| r = cv2.resize(crop, (nw, nh), interpolation=cv2.INTER_AREA if s < 1.0 else cv2.INTER_LINEAR) | |
| out = np.empty((size, size, 3), np.uint8) | |
| out[:] = pad | |
| x0, y0 = (size - nw) // 2, (size - nh) // 2 | |
| out[y0:y0 + nh, x0:x0 + nw] = r | |
| return out | |
| def softmax(z, axis=-1): | |
| z = np.asarray(z, np.float64) | |
| z = z - np.nanmax(z, axis=axis, keepdims=True) | |
| e = np.exp(z) | |
| return (e / e.sum(axis=axis, keepdims=True)).astype(np.float32) | |
| def grid_steps(a, b, hz): | |
| """requested span [a, b] (s) -> (k_lo, k_hi): the hz-grid steps inside it, rounding an endpoint that is within 0.02 step of the grid TO the grid | |
| (else floor / ceil as video_io.frames_in_spans does). f32(232.4) * 5 = 1161.99997 would otherwise floor to 1161 and drop the last window.""" | |
| lo, hi = float(a) * hz, float(b) * hz | |
| k_lo = int(round(lo)) if abs(lo - round(lo)) <= 0.02 else int(math.ceil(lo)) | |
| k_hi = int(round(hi)) if abs(hi - round(hi)) <= 0.02 else int(math.floor(hi)) | |
| return k_lo, k_hi | |
| def grid_span(span, hz=cs.HZ): | |
| """[a, b] (s) -> f32 [a', b'] equal to the requested span to within 1 f32 ulp but with a' >= a and b' <= b in exact arithmetic for endpoints on the | |
| 10 Hz grid: video_io.frames_in_spans applies floor(a * hz) / ceil(b * hz) to the f32 values, and e.g. f32(834.2) = 834.2000122 would make ceil() | |
| read the frame at 834.3, ONE 10 Hz STEP PAST THE REQUESTED SPAN END (f32(376.8) = 376.79998779 one step before the start). Verified 2026-09-14: | |
| 22 / 50 manifest windows were affected at the end, 21 at the start. Endpoints off the grid (> 0.02 step) are passed through unchanged.""" | |
| out = [] | |
| for x, is_end in ((float(span[0]), False), (float(span[1]), True)): | |
| k = x * hz | |
| r = round(k) | |
| if abs(k - r) > 0.02: | |
| out.append(np.float32(x)) | |
| continue | |
| f = np.float32(r / hz) | |
| if is_end: | |
| while float(f) * hz > r: | |
| f = np.nextafter(f, np.float32(-np.inf)) | |
| else: | |
| while float(f) * hz < r: | |
| f = np.nextafter(f, np.float32(np.inf)) | |
| out.append(f) | |
| return np.asarray(out, np.float32) | |
| def _rt(model, path): | |
| try: | |
| return str(model.get_rt_info(path).astype(str)) | |
| except Exception: | |
| return None | |
| # ----------------------------------------------------------------------------- extractor | |
| class OvActionExtractor(Extractor): | |
| baseline = BASELINE | |
| channels = ENC_CHANNELS | |
| model_ids = [ENC_ID, DEC_ID] | |
| needs_crop = True | |
| decode_wh = (960, 600) # 1.6 aspect like the default (640, 400); the driver crop keeps >= 224 px on both sides before the letterbox | |
| batch = 32 | |
| win_len = WIN_LEN | |
| win_hz = WIN_HZ | |
| win_channels = LOGIT_CHANNELS | |
| summary_channels = LOGIT_CHANNELS + [P_SAFE] | |
| def __init__(self, n_threads=4): | |
| self.n_threads = int(n_threads) | |
| self.infer_ms = [] # encoder ms per crop, one entry per crop (pooled p50 / p95 for measurements) | |
| self.dec_ms = [] # decoder ms per window | |
| self._enc = self._dec = None | |
| self.registry_entry = None | |
| self.preproc = {} | |
| def spec(): | |
| """channel roles: per-frame embedding channels, window logits, and the derived softmax probability of Safe driving.""" | |
| d = {c: "embedding" for c in ENC_CHANNELS} | |
| d.update({c: "logit" for c in LOGIT_CHANNELS}) | |
| d[P_SAFE] = "prob" | |
| return d | |
| def setup(self, registry_entry=None): | |
| import cv2 | |
| import openvino as ov | |
| ents = registry_entry if isinstance(registry_entry, dict) and "encoder" in registry_entry else {"encoder": REG.get(ENC_ID), "decoder": REG.get(DEC_ID)} | |
| enc_dir, dec_dir = REG.resolve(ENC_ID), REG.resolve(DEC_ID) # asserts sha256 of xml + bin | |
| core = ov.Core() | |
| cfg = {"INFERENCE_NUM_THREADS": self.n_threads} | |
| m_enc = core.read_model(str(enc_dir / ENC_XML)) | |
| self.preproc = {"mean_values": _rt(m_enc, ["conversion_parameters", "mean_values"]), "scale_values": _rt(m_enc, ["conversion_parameters", "scale_values"]), | |
| "reverse_input_channels": _rt(m_enc, ["conversion_parameters", "reverse_input_channels"]), "layout": _rt(m_enc, ["conversion_parameters", "layout"]), | |
| "ir_version": _rt(m_enc, "version"), "mo_runtime": _rt(m_enc, "Runtime_version"), "input_fed": "u8 BGR 0..255 as f32, NCHW, letterbox 224x224, pad BGR %s" % (PAD_BGR,)} | |
| assert self.preproc["reverse_input_channels"] in ("True", None), self.preproc | |
| m_enc.reshape([self.batch, 3, SIZE, SIZE]) | |
| self._enc = core.compile_model(m_enc, "CPU", cfg) | |
| m_dec = core.read_model(str(dec_dir / DEC_XML)) # static [1,16,512]: the IR's attention VariadicSplit bakes batch 1 (dynamic batch fails at infer) | |
| self._dec = core.compile_model(m_dec, "CPU", cfg) | |
| self._enc_out, self._dec_out = self._enc.output(0), self._dec.output(0) | |
| self.registry_entry = {k: {kk: v.get(kk) for kk in ("id", "version", "sha256", "local_path", "license", "url")} for k, v in ents.items()} | |
| self.weight_sha256 = {f["name"]: f["sha256"] for e in ents.values() for f in e.get("files", [])} | |
| self.lib_versions = {"openvino": ov.__version__, "numpy": np.__version__, "opencv": cv2.__version__, "device": "CPU", "inference_num_threads": self.n_threads} | |
| return self | |
| def teardown(self): | |
| self._enc = self._dec = None | |
| # per-frame encoder ------------------------------------------------------------------------------------------------------------------------ | |
| def encode(self, X): | |
| """X f32[m,3,224,224] BGR 0..255 -> f32[m,512]; static batch of `self.batch` (the remainder is zero-padded and dropped).""" | |
| B, m = self.batch, len(X) | |
| out = np.empty((m, D_EMB), np.float32) | |
| for s in range(0, m, B): | |
| xb = X[s:s + B] | |
| if len(xb) < B: | |
| xb = np.concatenate([xb, np.zeros((B - len(xb),) + xb.shape[1:], np.float32)], 0) | |
| y = self._enc(np.ascontiguousarray(xb))[self._enc_out] | |
| out[s:s + B] = np.asarray(y, np.float32).reshape(B, D_EMB)[:min(B, m - s)] | |
| return out | |
| def infer(self, crops): | |
| out = np.full((len(crops), D_EMB), np.nan, np.float32) | |
| idx = [i for i, c in enumerate(crops) if c is not None and c.size and c.shape[0] >= 2 and c.shape[1] >= 2] | |
| if not idx: | |
| return out | |
| t0 = time.time() | |
| X = np.stack([letterbox(crops[i]) for i in idx]).transpose(0, 3, 1, 2).astype(np.float32) | |
| out[idx] = self.encode(X) | |
| self.infer_ms += [1000.0 * (time.time() - t0) / len(idx)] * len(idx) | |
| return out | |
| # 16-frame decoder --------------------------------------------------------------------------------------------------------------------------- | |
| def decode(self, W): | |
| """W f32[m,16,512] -> logits f32[m,9]; one static [1,16,512] forward per window (the decoder IR only accepts batch 1).""" | |
| out = np.empty((len(W), N_CLASSES), np.float32) | |
| for i in range(len(W)): | |
| out[i] = np.asarray(self._dec(np.ascontiguousarray(W[i:i + 1], dtype=np.float32))[self._dec_out], np.float32).reshape(N_CLASSES) | |
| return out | |
| def decode_windows(self, rc: cs.RouteCache, spans=None) -> cs.WindowCache: | |
| """RouteCache (10 Hz embeddings) -> WindowCache over every 5 Hz grid time inside `spans` (default: the cache's spans); window = frames k10-15..k10.""" | |
| t0 = time.time() | |
| spans = cs.merge_spans(rc.spans if spans is None else spans) | |
| step = cs.HZ // self.win_hz | |
| k = np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64) | |
| k0 = int(k.min()) if len(k) else 0 | |
| M = int(k.max() - k0 + 1) if len(k) else 0 | |
| E = np.full((M, D_EMB), np.nan, np.float32) | |
| V = np.zeros(M, bool) | |
| if M: | |
| E[k - k0] = rc.feat.astype(np.float32) | |
| V[k - k0] = rc.valid | |
| k5 = np.unique(np.concatenate([np.arange(*np.add(grid_steps(a, b, self.win_hz), (0, 1))) for a, b in spans.astype(np.float64)] | |
| + [np.zeros(0, np.int64)])).astype(np.int64) | |
| k10 = k5 * step | |
| idx = k10[:, None] + np.arange(-(self.win_len - 1), 1)[None, :] - k0 | |
| inside = (idx >= 0) & (idx < M) | |
| wv = inside.all(1) | |
| if wv.any(): | |
| wv[wv] = V[idx[wv]].all(1) | |
| emb = np.full((len(k10), N_CLASSES), np.nan, np.float32) | |
| dec_ms = [] | |
| if wv.any(): | |
| t1 = time.time() | |
| emb[wv] = self.decode(E[idx[wv]]) | |
| dec_ms = [1000.0 * (time.time() - t1) / int(wv.sum())] * int(wv.sum()) | |
| self.dec_ms += dec_ms | |
| P = np.zeros(M, bool) # a cache row exists for the step (any fail code) | |
| if M: | |
| P[k - k0] = True | |
| eligible = inside.all(1) # the 16-frame history consists of decoded steps (else the window is invalid by construction, not by the model) | |
| if eligible.any(): | |
| eligible[eligible] = P[idx[eligible]].all(1) | |
| assert not np.any(wv & ~eligible) | |
| meta = dict(rc.meta) | |
| meta.update(kind="route_windows", win_len=self.win_len, win_hz=self.win_hz, win_channels=list(self.win_channels), class_names=list(ACTIONS), class_names_omz=list(ACTIONS_OMZ), | |
| summary_channels={P_SAFE: "softmax(emb)[0] (Safe driving)"}, window_rule="frames k10-15..k10 at 10 Hz for every 5 Hz step k5 (k10 = 2 k5) inside the spans; valid = all 16 valid", | |
| n_windows_eligible=int(eligible.sum()), decoder_input="[M,16,512]", decoder_output="[M,9] logits", written_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) | |
| rt = dict(rc.runtime) | |
| p50, p95 = (float(np.percentile(dec_ms, 50)), float(np.percentile(dec_ms, 95))) if dec_ms else (float("nan"), float("nan")) | |
| rt.update(encoder_infer_ms_p50=rc.runtime.get("infer_ms_p50"), encoder_infer_ms_p95=rc.runtime.get("infer_ms_p95"), infer_ms_p50=p50, infer_ms_p95=p95, infer_unit="ms per 16-frame window (decoder)", | |
| n_windows=int(len(k10)), n_windows_valid=int(wv.sum()), n_windows_eligible=int(eligible.sum()), wall_s_windows=float(time.time() - t0)) | |
| return cs.WindowCache(rc.uid, (k10 / cs.HZ).astype(np.float32), emb.astype(np.float16), wv, self.win_len, spans.astype(np.float32), meta, rt) | |
| def window_summary(emb): | |
| """emb [M,9] logits -> dict of summary channels {ova_logit_i: [M], ova_p_safe: [M]} (NaN propagates).""" | |
| e = np.asarray(emb, np.float32) | |
| out = {c: e[:, i] for i, c in enumerate(LOGIT_CHANNELS)} | |
| ps = np.full(len(e), np.nan, np.float32) | |
| ok = np.isfinite(e).all(1) | |
| if ok.any(): | |
| ps[ok] = softmax(e[ok])[:, 0] | |
| out[P_SAFE] = ps | |
| return out | |
| # ----------------------------------------------------------------------------- measurements helpers | |
| def _spearman(x, y): | |
| x, y = np.asarray(x, np.float64), np.asarray(y, np.float64) | |
| ok = np.isfinite(x) & np.isfinite(y) | |
| if ok.sum() < 3: | |
| return float("nan"), int(ok.sum()) | |
| try: | |
| from scipy.stats import spearmanr | |
| return float(spearmanr(x[ok], y[ok]).correlation), int(ok.sum()) | |
| except Exception: | |
| def rank(v): | |
| o = np.argsort(v, kind="stable") | |
| r = np.empty(len(v), np.float64) | |
| r[o] = np.arange(len(v)) | |
| return r | |
| rx, ry = rank(x[ok]), rank(y[ok]) | |
| return float(np.corrcoef(rx, ry)[0, 1]), int(ok.sum()) | |
| def _peak_rss_mb(): | |
| return {"self_mb": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0, "children_max_mb": resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss / 1024.0} | |
| def _vram_mb_this_process(): | |
| """VRAM of THIS pid per nvidia-smi (CPU-only extractor -> 0); None when nvidia-smi is unavailable.""" | |
| try: | |
| import os | |
| out = subprocess.run(["nvidia-smi", "--query-compute-apps=pid,used_memory", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=10).stdout | |
| tot = 0.0 | |
| for line in out.splitlines(): | |
| parts = [p.strip() for p in line.split(",")] | |
| if len(parts) >= 2 and parts[0].isdigit() and int(parts[0]) == os.getpid(): | |
| tot += float(parts[1]) | |
| return tot | |
| except Exception: | |
| return None | |
| def read_manifest(path=MANIFEST): | |
| rows = list(csv.DictReader(open(path))) | |
| for r in rows: | |
| for k in ("t_anchor", "win_start", "win_end", "t0"): | |
| r[k] = float(r[k]) | |
| r["row"], r["idx"], r["y"] = int(r["row"]), int(r["idx"]), int(r["y"]) | |
| return rows | |
| def _lm_yawdev_history(rows): | |
| """manifest rows -> {idx: (t f64[80], yawdev f32[80] NaN invalid)} from the face pack Fh8_c[row, :, 1] (10 Hz history ending at t_anchor).""" | |
| out = {} | |
| for regime in sorted({r["regime"] for r in rows}): | |
| p = FACE_PACK / f"facehist_causal_{regime}_train.npz" | |
| if not p.exists(): | |
| continue | |
| sub = [r for r in rows if r["regime"] == regime] | |
| with np.load(p, allow_pickle=True) as z: | |
| uid, t = z["uid"], z["t"] | |
| ridx = np.array([r["row"] for r in sub]) | |
| assert all(str(uid[r["row"]]) == r["uid"] and abs(float(t[r["row"]]) - r["t_anchor"]) < 1e-6 for r in sub), "manifest row / face pack mismatch" | |
| ch = [str(c) for c in z["channels"]] | |
| assert ch[LM_YAW_DEV_CH] == "lm_yaw_dev", ch | |
| Fh = z["Fh8_c"][ridx][:, :, LM_YAW_DEV_CH].astype(np.float32) | |
| Fv = z["Fh8valid"][ridx] | |
| for r, f, v in zip(sub, Fh, Fv): | |
| th = r["t_anchor"] - (79 - np.arange(80)) / 10.0 | |
| out[r["idx"]] = (th, np.where(v, f, np.nan).astype(np.float32)) | |
| return out | |
| def _contact_sheet(ex, picks, out_png, decode_wh, workers=1): | |
| """picks: list of dict(uid, t, yawdev, p_safe, top, lr) (12) -> 3 x 4 sheet of the letterboxed driver crops with the values printed.""" | |
| import cv2 | |
| tiles = [] | |
| by_uid = {} | |
| for p in picks: | |
| by_uid.setdefault(p["uid"], []).append(p) | |
| for uid, ps in by_uid.items(): | |
| want = {int(round(p["t"] * cs.HZ)): p for p in ps} | |
| spans = cs.merge_spans([[p["t"], p["t"]] for p in ps]) | |
| fs = vio.FrameStream(uid, spans, hz=cs.HZ, workers=workers, decode_wh=decode_wh, block=8) | |
| for t_, fi_, fr_, bx_ in fs: | |
| kk = int(round(t_ * cs.HZ)) | |
| if kk in want and fr_ is not None: | |
| crop = vio.driver_crop(fr_, bx_) | |
| want[kk]["tile"] = letterbox(crop) if crop is not None else np.full((SIZE, SIZE, 3), 40, np.uint8) | |
| for p in picks: | |
| tile = p.get("tile", np.full((SIZE, SIZE, 3), 40, np.uint8)).copy() | |
| canvas = np.full((SIZE + 78, SIZE, 3), 255, np.uint8) | |
| canvas[:SIZE] = tile | |
| lines = [f"{p['uid'].split('__')[0][:16]} t={p['t']:.1f}", f"lm_yaw_dev={p['yawdev']:+.3f}", f"p_safe={p['p_safe']:.2f} lr={p['lr']:+.2f}", f"top={p['top']}"] | |
| for i, s in enumerate(lines): | |
| cv2.putText(canvas, s, (3, SIZE + 15 + 17 * i), cv2.FONT_HERSHEY_SIMPLEX, 0.42, (0, 0, 0), 1, cv2.LINE_AA) | |
| tiles.append(canvas) | |
| while len(tiles) < 12: | |
| tiles.append(np.full((SIZE + 78, SIZE, 3), 255, np.uint8)) | |
| rows_img = [np.concatenate(tiles[i:i + 4], 1) for i in range(0, 12, 4)] | |
| sheet = np.concatenate(rows_img, 0) | |
| head = np.full((44, sheet.shape[1], 3), 255, np.uint8) | |
| cv2.putText(head, "ov_action (OMZ driver-action-recognition-adas-0002): no yaw output. Row 1-2 = 6 most negative lm_yaw_dev frames, row 2-3 = 6 most positive (face pack, IOD units).", | |
| (6, 17), cv2.FONT_HERSHEY_SIMPLEX, 0.42, (0, 0, 0), 1, cv2.LINE_AA) | |
| cv2.putText(head, "Printed: softmax p_safe, lr = (texting_left+phone_left) - (texting_right+phone_right) logits, top class. Crops as fed to the encoder (letterbox 224, BGR).", | |
| (6, 36), cv2.FONT_HERSHEY_SIMPLEX, 0.42, (0, 0, 0), 1, cv2.LINE_AA) | |
| out_png = Path(out_png) | |
| out_png.parent.mkdir(parents=True, exist_ok=True) | |
| cv2.imwrite(str(out_png), np.concatenate([head, sheet], 0)) | |
| return out_png | |
| # ----------------------------------------------------------------------------- Phase 1 | |
| def phase1(workers=3, max_minutes=60.0, force=False, out_root=PHASE1_DIR, n_threads=4, log=print): | |
| t_all = time.time() | |
| rows = read_manifest() | |
| spans_by_uid = {} | |
| for r in rows: | |
| spans_by_uid.setdefault(r["uid"], []).append([r["win_start"], r["win_end"]]) | |
| uids = sorted(spans_by_uid) | |
| spans_by_uid = {u: cs.merge_spans(v) for u, v in spans_by_uid.items()} | |
| out_root = Path(out_root) | |
| cache_dir, win_dir, log_dir = out_root / "cache", out_root / "cache_win", out_root / "logs" | |
| log_dir.mkdir(parents=True, exist_ok=True) | |
| ex = OvActionExtractor(n_threads=n_threads).setup(None) | |
| t0 = time.time() | |
| infos, code = [], EXIT_DONE | |
| for u in uids: | |
| p = cache_dir / f"{u}.npz" | |
| if force and p.exists(): | |
| p.unlink() | |
| for sp in spans_by_uid[u]: # ONE run_route call PER SPAN (the cache is extended via the resume path): a multi-span FrameStream block | |
| if time.time() - t0 > max_minutes * 60: # would make video_io.decode_block grab every video frame across the gap between two anchors' spans | |
| code = EXIT_BUDGET | |
| break | |
| c, _, info = run_route(ex, u, grid_span(sp)[None], cache_dir, max_minutes, workers, log=log, block=100, t_start=t0) # grid_span: exactly the frames of [win_start, win_end] | |
| info["span"] = [float(sp[0]), float(sp[1])] | |
| log(f"{u[:60]} [{sp[0]:.1f},{sp[1]:.1f}] {info['status']} frames {info.get('n_frames')} valid {info.get('valid_rate')} wall {info.get('wall_s')}s") | |
| infos.append(info) | |
| if c == EXIT_FAIL: | |
| code = EXIT_FAIL | |
| elif c == EXIT_BUDGET: | |
| code = EXIT_BUDGET | |
| break | |
| if code == EXIT_BUDGET: | |
| break | |
| wall_extract = time.time() - t0 | |
| n_spans = int(sum(len(v) for v in spans_by_uid.values())) | |
| summ = {"n_routes": len(uids), "n_spans": n_spans, "n_span_runs": len(infos), "n_exists": sum(1 for i in infos if i["status"] == "exists"), "n_failed": sum(1 for i in infos if i["status"] == "no_video"), | |
| "n_remaining": n_spans - sum(1 for i in infos if i["status"] in ("done", "exists")), "budget_hit": code == EXIT_BUDGET, "wall_s": round(wall_extract, 1), "baseline": BASELINE, | |
| "per_span": infos} | |
| rss_after_extract = _peak_rss_mb() | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| (log_dir / f"phase1_extract_{stamp}.json").write_text(json.dumps({"summary": summ, "code": code, "wall_extract_s": wall_extract, "describe": ex.describe()}, indent=1, default=str)) | |
| if code == EXIT_BUDGET: | |
| log(f"BUDGET_EXIT after {wall_extract:.0f}s — rerun --phase1 to resume ({summ['n_remaining']} spans remaining)") | |
| return code | |
| # window (decoder) pass | |
| t1 = time.time() | |
| wcs = {} | |
| for u in uids: | |
| p = cache_dir / f"{u}.npz" | |
| if not p.exists(): | |
| continue | |
| rc = cs.RouteCache.load(p) | |
| pw = win_dir / f"{u}.npz" | |
| if pw.exists() and not force: | |
| try: | |
| wc = cs.WindowCache.load(pw) | |
| if wc.covers(spans_by_uid[u]) and wc.meta.get("code_sha256") == rc.meta.get("code_sha256"): | |
| wcs[u] = wc | |
| continue | |
| except Exception: | |
| pass | |
| wc = ex.decode_windows(rc, spans_by_uid[u]) | |
| wc.save(pw) | |
| wcs[u] = wc | |
| wall_win = time.time() - t1 | |
| # ---- measurements | |
| rcs = {u: cs.RouteCache.load(cache_dir / f"{u}.npz") for u in uids if (cache_dir / f"{u}.npz").exists()} | |
| n_frames = int(sum(rc.n for rc in rcs.values())) | |
| valid = np.concatenate([rc.valid for rc in rcs.values()]) if rcs else np.zeros(0, bool) | |
| had_box = np.concatenate([np.isfinite(rc.box).all(1) for rc in rcs.values()]) if rcs else np.zeros(0, bool) | |
| fail = np.concatenate([rc.fail for rc in rcs.values()]) if rcs else np.zeros(0, np.uint8) | |
| fail_hist = {cs.FAIL_CODES[int(k)]: int(v) for k, v in zip(*np.unique(fail, return_counts=True))} | |
| span_walls = [i.get("wall_s") for i in summ["per_span"] if i.get("status") == "done"] | |
| span_fps = [i.get("decode_worker_fps") for i in summ["per_span"] if i.get("status") == "done" and i.get("decode_worker_fps") is not None] | |
| dec_p50 = [rc.runtime.get("decode_ms_p50") for rc in rcs.values()] | |
| dec_p95 = [rc.runtime.get("decode_ms_p95") for rc in rcs.values()] | |
| enc_ms = np.asarray(ex.infer_ms, np.float64) | |
| if len(enc_ms) == 0: # resumed run: fall back to the cached per-route runtimes | |
| enc_ms = np.asarray([rc.runtime.get("infer_ms_p50") for rc in rcs.values() if rc.runtime.get("infer_ms_p50") is not None], np.float64) | |
| win_valid = np.concatenate([wc.win_valid for wc in wcs.values()]) if wcs else np.zeros(0, bool) | |
| n_win_elig = int(sum(wc.meta.get("n_windows_eligible", 0) for wc in wcs.values())) | |
| dec_ms = np.asarray(ex.dec_ms, np.float64) | |
| # per anchor | |
| yaw_hist = _lm_yawdev_history(rows) | |
| per_anchor, pairs = [], {"lr_asym": [], "yawdev": [], "one_minus_p_safe": [], "abs_yawdev": [], "logits": []} | |
| for r in rows: | |
| u = r["uid"] | |
| rec = {k: r[k] for k in ("idx", "uid", "dongle", "regime", "y", "t_anchor", "win_start", "win_end")} | |
| rc, wc = rcs.get(u), wcs.get(u) | |
| if rc is not None: | |
| inw = (rc.t >= r["win_start"] - 1e-6) & (rc.t <= r["win_end"] + 1e-6) | |
| rec.update(n_frames_window=int(inw.sum()), frame_valid_share_window=float(rc.valid[inw].mean()) if inw.any() else None, | |
| frame_box_share_window=float(np.isfinite(rc.box[inw]).all(1).mean()) if inw.any() else None) | |
| if wc is not None: | |
| j = np.flatnonzero(np.abs(wc.win_t_end.astype(np.float64) - r["t_anchor"]) < 1e-3) | |
| if len(j): | |
| j = int(j[0]) | |
| S = window_summary(wc.emb[j:j + 1].astype(np.float32)) | |
| lg = [float(S[c][0]) for c in LOGIT_CHANNELS] | |
| ps = float(S[P_SAFE][0]) | |
| rec.update(win_valid_at_anchor=bool(wc.win_valid[j]), win_t_end=float(wc.win_t_end[j]), **{c: (None if not np.isfinite(v) else v) for c, v in zip(LOGIT_CHANNELS, lg)}, | |
| ova_p_safe=(None if not np.isfinite(ps) else ps), top_class=(ACTIONS[int(np.argmax(lg))] if np.isfinite(lg).all() else None)) | |
| inw8 = (wc.win_t_end > r["t_anchor"] - 8.0 - 1e-6) & (wc.win_t_end <= r["t_anchor"] + 1e-6) | |
| rec.update(n_windows_8s=int(inw8.sum()), win_valid_share_8s=float(wc.win_valid[inw8].mean()) if inw8.any() else None) | |
| # correlation pairs on the anchor's 10 Hz face history (5 Hz sub-grid j odd -> t on the 0.2 s grid) | |
| if r["idx"] in yaw_hist: | |
| th, yd = yaw_hist[r["idx"]] | |
| kw = np.round(wc.win_t_end.astype(np.float64) * cs.HZ).astype(np.int64) | |
| pos = {int(kk): i for i, kk in enumerate(kw)} | |
| for tt, v in zip(th, yd): | |
| i = pos.get(int(round(tt * cs.HZ))) | |
| if i is None or not wc.win_valid[i] or not np.isfinite(v): | |
| continue | |
| e = wc.emb[i].astype(np.float32) | |
| pr = softmax(e) | |
| pairs["lr_asym"].append(float(e[list(LEFT_IDX)].sum() - e[list(RIGHT_IDX)].sum())) | |
| pairs["yawdev"].append(float(v)) | |
| pairs["one_minus_p_safe"].append(float(1.0 - pr[0])) | |
| pairs["abs_yawdev"].append(abs(float(v))) | |
| pairs["logits"].append(e) | |
| per_anchor.append(rec) | |
| rho_lr, n_pairs = _spearman(pairs["lr_asym"], pairs["yawdev"]) | |
| rho_dist, _ = _spearman(pairs["one_minus_p_safe"], pairs["abs_yawdev"]) | |
| rho_dist_signed, _ = _spearman(pairs["one_minus_p_safe"], pairs["yawdev"]) | |
| L = np.asarray(pairs["logits"], np.float32).reshape(-1, N_CLASSES) if pairs["logits"] else np.zeros((0, N_CLASSES), np.float32) | |
| rho_per_logit = {ACTIONS[i]: {"vs_yawdev": _spearman(L[:, i], pairs["yawdev"])[0], "vs_abs_yawdev": _spearman(L[:, i], pairs["abs_yawdev"])[0]} for i in range(N_CLASSES)} if len(L) else {} | |
| # descriptive score distribution (y only describes; nothing is tuned) | |
| desc = {} | |
| for regime in ("manual", "engaged", "pooled"): | |
| for y in (0, 1): | |
| v = np.asarray([1.0 - a["ova_p_safe"] for a in per_anchor if a.get("ova_p_safe") is not None and a["y"] == y and (regime == "pooled" or a["regime"] == regime)], np.float64) | |
| desc[f"{regime}_y{y}"] = {"n": int(len(v)), "mean_1_minus_p_safe": float(v.mean()) if len(v) else None, "median": float(np.median(v)) if len(v) else None} | |
| tops = [a.get("top_class") for a in per_anchor if a.get("top_class")] | |
| top_hist = {c: int(sum(1 for t in tops if t == c)) for c in ACTIONS} | |
| allE = np.concatenate([wc.emb[wc.win_valid].astype(np.float32) for wc in wcs.values() if wc.win_valid.any()]) if wcs else np.zeros((0, N_CLASSES), np.float32) | |
| all_top = {c: int((np.argmax(allE, 1) == i).sum()) for i, c in enumerate(ACTIONS)} if len(allE) else {} | |
| p_safe_all = softmax(allE)[:, 0] if len(allE) else np.zeros(0, np.float32) | |
| # contact sheet: 6 most negative / 6 most positive lm_yaw_dev among valid windows (<= 2 per anchor) | |
| picks = [] | |
| cand = [] | |
| for r in rows: | |
| wc = wcs.get(r["uid"]) | |
| if wc is None or r["idx"] not in yaw_hist: | |
| continue | |
| th, yd = yaw_hist[r["idx"]] | |
| kw = np.round(wc.win_t_end.astype(np.float64) * cs.HZ).astype(np.int64) | |
| pos = {int(kk): i for i, kk in enumerate(kw)} | |
| for tt, v in zip(th, yd): | |
| i = pos.get(int(round(tt * cs.HZ))) | |
| if i is None or not wc.win_valid[i] or not np.isfinite(v): | |
| continue | |
| e = wc.emb[i].astype(np.float32) | |
| cand.append({"uid": r["uid"], "idx": r["idx"], "t": float(wc.win_t_end[i]), "yawdev": float(v), "p_safe": float(softmax(e)[0]), "top": ACTIONS[int(np.argmax(e))], | |
| "lr": float(e[list(LEFT_IDX)].sum() - e[list(RIGHT_IDX)].sum())}) | |
| def take(sorted_c, n): | |
| got, per = [], {} | |
| for c in sorted_c: | |
| if per.get(c["idx"], 0) >= 1: | |
| continue | |
| per[c["idx"]] = 1 | |
| got.append(c) | |
| if len(got) >= n: | |
| break | |
| return got | |
| neg = take(sorted(cand, key=lambda c: c["yawdev"]), 6) | |
| posi = take(sorted(cand, key=lambda c: -c["yawdev"]), 6) | |
| picks = neg + posi | |
| png = None | |
| if picks: | |
| png = _contact_sheet(ex, picks, out_root / "sign_convention_contact_sheet.png", ex.decode_wh, workers=1) | |
| wall_total = time.time() - t_all | |
| meas = { | |
| "baseline": BASELINE, "manifest": str(MANIFEST), "n_anchors": len(rows), "n_routes": len(uids), "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), | |
| "extractor": ex.describe(), "spec_roles": {"ova_enc_000..511": "embedding", "ova_logit_0..8": "logit", P_SAFE: "prob"}, "class_order": ACTIONS, "class_order_omz": ACTIONS_OMZ, | |
| "preprocessing": ex.preproc, "registry": ex.registry_entry, "weight_sha256": ex.weight_sha256, "lib_versions": ex.lib_versions, | |
| "run": {"exit_code": code, "workers_decode": workers, "block": 100, "one_run_route_call_per_span": True, "summary": summ, "wall_extract_s": wall_extract, "wall_windows_s": wall_win, "wall_total_s": wall_total, | |
| "span_wall_s": {"median": float(np.median(span_walls)) if span_walls else None, "max": float(np.max(span_walls)) if span_walls else None, "n": len(span_walls)}, | |
| "decode_worker_fps": {"median": float(np.median(span_fps)) if span_fps else None, "min": float(np.min(span_fps)) if span_fps else None, "max": float(np.max(span_fps)) if span_fps else None}, | |
| "frames_per_s_end_to_end_extract": (n_frames / wall_extract if wall_extract > 0 else None), "frames_per_s_end_to_end_total": (n_frames / wall_total if wall_total > 0 else None), | |
| "n_crops_inferred_this_run": int(len(ex.infer_ms)), "resumed_from_existing_caches": bool(len(ex.infer_ms) < int(had_box.sum())), | |
| "peak_rss_mb_after_extraction": rss_after_extract, "peak_rss_mb_end": _peak_rss_mb(), "vram_mb_this_process": _vram_mb_this_process(), | |
| "device": "CPU (openvino)", "vram_mb_peak": 0.0}, | |
| "frames": {"n_frames": n_frames, "valid_share": float(valid.mean()) if n_frames else None, "n_with_box": int(had_box.sum()), "valid_share_with_box": float(valid[had_box].mean()) if had_box.any() else None, | |
| "fail_hist": fail_hist, "decode_ms_per_frame": {"route_p50_median": float(np.nanmedian(dec_p50)) if dec_p50 else None, "route_p50_min": float(np.nanmin(dec_p50)) if dec_p50 else None, | |
| "route_p50_max": float(np.nanmax(dec_p50)) if dec_p50 else None, "route_p95_median": float(np.nanmedian(dec_p95)) if dec_p95 else None, | |
| "route_p95_max": float(np.nanmax(dec_p95)) if dec_p95 else None, "note": "per-frame ms inside the decode workers (seek+grab+retrieve+resize), per route"}, | |
| "encoder_infer_ms_per_frame": {"p50": float(np.percentile(enc_ms, 50)) if len(enc_ms) else None, "p95": float(np.percentile(enc_ms, 95)) if len(enc_ms) else None, | |
| "n": int(len(enc_ms)), "note": "letterbox + encoder forward, static batch 32, per crop"}}, | |
| "windows": {"n_windows": int(len(win_valid)), "n_valid": int(win_valid.sum()), "valid_share": float(win_valid.mean()) if len(win_valid) else None, "n_eligible": n_win_elig, | |
| "valid_share_eligible": (float(win_valid.sum() / n_win_elig) if n_win_elig else None), | |
| "decoder_infer_ms_per_window": {"p50": float(np.percentile(dec_ms, 50)) if len(dec_ms) else None, "p95": float(np.percentile(dec_ms, 95)) if len(dec_ms) else None, "n": int(len(dec_ms))}, | |
| "top_class_hist_all_valid_windows": all_top, "p_safe_all_valid_windows": {"mean": float(p_safe_all.mean()) if len(p_safe_all) else None, "p10": float(np.percentile(p_safe_all, 10)) if len(p_safe_all) else None, | |
| "p50": float(np.percentile(p_safe_all, 50)) if len(p_safe_all) else None, "p90": float(np.percentile(p_safe_all, 90)) if len(p_safe_all) else None}}, | |
| "per_anchor": per_anchor, "anchor_top_class_hist": top_hist, "descriptive_by_y": desc, | |
| "spearman_vs_face_pack": {"note": "ov_action has no head/gaze yaw channel; proxies on the 5 Hz sub-grid of each anchor's 10 Hz face history (t_anchor-7.9..t_anchor), valid windows only", | |
| "n_pairs": n_pairs, "rho_lr_asym_vs_lm_yaw_dev": rho_lr, "rho_1_minus_p_safe_vs_abs_lm_yaw_dev": rho_dist, "rho_1_minus_p_safe_vs_lm_yaw_dev": rho_dist_signed, | |
| "rho_per_logit": rho_per_logit, "lr_asym_def": "(logit texting_left + logit talking_phone_left) - (logit texting_right + logit talking_phone_right)", | |
| "face_pack": str(FACE_PACK), "lm_yaw_dev_units": "dimensionless (IOD-normalised), face pack channel 1"}, | |
| "contact_sheet": {"png": (str(png) if png else None), "picks": [{k: v for k, v in p.items() if k != "tile"} for p in picks]}, | |
| } | |
| (out_root / "measurements.json").write_text(json.dumps(meas, indent=1, default=str)) | |
| log(json.dumps({"n_frames": n_frames, "valid_share": meas["frames"]["valid_share"], "valid_share_with_box": meas["frames"]["valid_share_with_box"], "fail_hist": fail_hist, | |
| "windows_valid_share": meas["windows"]["valid_share"], "fps_extract": meas["run"]["frames_per_s_end_to_end_extract"], "enc_ms_p50": meas["frames"]["encoder_infer_ms_per_frame"]["p50"], | |
| "rho_lr": rho_lr, "rho_dist_abs": rho_dist, "n_pairs": n_pairs, "wall_total_s": wall_total}, indent=1)) | |
| ex.teardown() | |
| return code | |
| # ----------------------------------------------------------------------------- self-test | |
| def _self_test(uid=None, workers=2, n_threads=4): | |
| import tempfile | |
| ex = OvActionExtractor(n_threads=n_threads).setup(None) | |
| # letterbox: aspect preserved, centred, padded with the mean colour | |
| c = np.zeros((300, 100, 3), np.uint8) | |
| c[:] = (10, 20, 30) | |
| L = letterbox(c) | |
| assert L.shape == (SIZE, SIZE, 3) and tuple(L[112, 112]) == (10, 20, 30) and tuple(L[112, 5]) == PAD_BGR and tuple(L[112, 218]) == PAD_BGR, "letterbox geometry" | |
| assert (L[:, 74:149] == (10, 20, 30)).all() and (L[:, :74] == PAD_BGR).all(), "letterbox content columns" | |
| # grid_span / grid_steps: floor/ceil of the f32 endpoints recover exactly the requested 10 Hz steps (f32(834.2) * 10 = 8342.000122 -> ceil 8343 before) | |
| for a, b in ((824.0, 834.2), (376.8, 387.0), (222.2, 232.4), (30.8, 41.0), (5541.6, 5551.8)): | |
| g = grid_span([a, b]) | |
| assert math.floor(float(g[0]) * 10) == round(a * 10) and math.ceil(float(g[1]) * 10) == round(b * 10), (a, b, g.tolist()) | |
| assert abs(float(g[0]) - a) < 1e-3 and abs(float(g[1]) - b) < 1e-3 and grid_steps(g[0], g[1], 5) == (round(a * 5), round(b * 5)), (a, b, g.tolist()) | |
| assert grid_steps(1.03, 2.07, 10) == (11, 20), grid_steps(1.03, 2.07, 10) | |
| # infer contract: None -> NaN, finite output, deterministic, batch remainder handled | |
| rng = np.random.default_rng(0) | |
| crops = [None] + [rng.integers(0, 255, (h, w, 3), dtype=np.uint8) for h, w in ((300, 200), (150, 150), (400, 260))] * 12 # 37 crops -> 32 + 5 remainder | |
| Y = ex.infer(crops) | |
| assert Y.shape == (37, D_EMB) and np.isnan(Y[0]).all() and np.isfinite(Y[1:]).all(), Y.shape | |
| Y2 = ex.infer(crops[1:4]) | |
| assert np.allclose(Y[1:4], Y2, atol=1e-4) and np.allclose(Y[1:4], Y[4:7], atol=1e-4), "determinism / batch-position independence" | |
| assert Y[1:4].std(0).mean() > 1e-3, "embeddings must differ across crops" | |
| rep = {"self_test": "extractors.ov_action", "unit_checks": "ok", "preproc": ex.preproc, "describe": ex.describe(), "n_spec_roles": len(ex.spec())} | |
| if uid is None: | |
| cands = [(vio.video_path(u).stat().st_size, u) for u in vio.anchors_by_uid(("manual_train", "engaged_train")) if vio.has_video(u) and vio.pose_path(u) is not None] | |
| if not cands: | |
| rep["route"] = "none local; skipped" | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return rep | |
| uid = min(cands)[1] | |
| ts = vio.anchors_of(uid) | |
| W, H, n = vio.video_info(vio.video_path(uid)) | |
| t_a = float(ts[ts >= 3.0][0]) if (ts >= 3.0).any() else float(ts[0]) | |
| span = grid_span([round(t_a - 2.9, 1), t_a])[None] # 30 frames at 10 Hz (grid_span: no frame outside the span) | |
| n_exp = len(vio.frames_in_spans(span, cs.HZ, n)[0]) | |
| with tempfile.TemporaryDirectory() as d: | |
| t0 = time.time() | |
| code, p, info = run_route(ex, uid, span, d, max_minutes=10, workers=workers, block=10) | |
| rc = cs.RouteCache.load(p) | |
| assert code == EXIT_DONE and rc.n == n_exp and rc.C == D_EMB and rc.channels == ENC_CHANNELS, (code, rc.n, n_exp) | |
| assert rc.valid.any() and np.isfinite(rc.feat.astype(np.float32)[rc.valid]).all() | |
| wc = ex.decode_windows(rc) | |
| wc2 = cs.WindowCache.load(wc.save(Path(d) / "win.npz")) | |
| k0 = int(round(float(rc.t[0]) * cs.HZ)) | |
| k5 = np.arange(*np.add(grid_steps(span[0, 0], span[0, 1], 5), (0, 1))) | |
| assert rc.t.min() >= span[0, 0] - 1e-4 and rc.t.max() <= span[0, 1] + 1e-4 and rc.n == 30, "grid_span: a frame outside the requested span was decoded" | |
| n_elig = int(((2 * k5 - 15) >= k0).sum()) | |
| assert wc2.m == len(k5) and wc2.D == N_CLASSES and wc2.win_len == WIN_LEN and wc2.meta["n_windows_eligible"] == n_elig, (wc2.m, len(k5), n_elig) | |
| assert wc2.win_valid.sum() <= n_elig and (not rc.valid.all() or wc2.win_valid.sum() == n_elig), (wc2.win_valid.sum(), n_elig, rc.valid.mean()) | |
| # causality: the window ending at T uses frames <= T only — perturbing the frame AFTER T changes no valid window ending at T | |
| if wc2.win_valid.any(): | |
| j = int(np.flatnonzero(wc2.win_valid)[0]) | |
| kT = int(round(float(wc2.win_t_end[j]) * cs.HZ)) | |
| i_after = np.flatnonzero(np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64) == kT + 1) | |
| if len(i_after): | |
| feat2 = rc.feat.astype(np.float32).copy() | |
| feat2[i_after[0]] = 5.0 | |
| v2, f2 = rc.valid.copy(), rc.fail.copy() | |
| v2[i_after[0]], f2[i_after[0]] = True, 0 | |
| feat2[~v2] = np.nan | |
| rc2 = cs.RouteCache(rc.uid, rc.t, rc.frame_idx, rc.box, v2, f2, feat2.astype(np.float16), rc.channels, rc.spans, rc.meta, rc.runtime, rc.moving) | |
| wc3 = ex.decode_windows(rc2) | |
| assert np.array_equal(wc3.emb[j], wc2.emb[j]), "a frame after T changed the window ending at T" | |
| rep["causality_check"] = "ok" | |
| S = window_summary(wc2.emb.astype(np.float32)) | |
| ps = S[P_SAFE][wc2.win_valid] | |
| assert np.all((ps >= 0) & (ps <= 1)) and np.allclose(softmax(wc2.emb[wc2.win_valid].astype(np.float32)).sum(1), 1.0, atol=1e-4) | |
| rep.update(route=uid, native_wh=[W, H], span=span.tolist(), n_frames=rc.n, valid_share=float(rc.valid.mean()), fail_hist=info["fail_hist"], runtime=rc.runtime, | |
| n_windows=wc2.m, n_windows_eligible=n_elig, n_windows_valid=int(wc2.win_valid.sum()), window_runtime=wc2.runtime, | |
| top_classes=[ACTIONS[int(np.argmax(e))] for e in wc2.emb[wc2.win_valid].astype(np.float32)], p_safe=[round(float(v), 3) for v in ps], | |
| wall_s=round(time.time() - t0, 2), rss=_peak_rss_mb()) | |
| 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("--uid", default="") | |
| ap.add_argument("--workers", type=int, default=3, help="decode worker processes (<= 3 on the shared box)") | |
| ap.add_argument("--threads", type=int, default=4, help="openvino CPU inference threads") | |
| ap.add_argument("--max-minutes", type=float, default=60.0) | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--out-root", default=str(PHASE1_DIR)) | |
| a = ap.parse_args() | |
| if a.phase1: | |
| return phase1(workers=min(a.workers, 3), max_minutes=a.max_minutes, force=a.force, out_root=Path(a.out_root), n_threads=a.threads) | |
| _self_test(a.uid or None, min(a.workers, 3), a.threads) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 44.4 kB
- Xet hash:
- 5f6d3f5ce8c32190567620f7187234667f777de8e4cdb676a9306cef5451d2c3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.