Buckets:
| """cwm3/baselines_dms/cache_schema.py — the on-disk contract of the DMS-baseline feature caches (plan §3 "特征缓存", 2026-09-14). | |
| Two cache kinds, both one file per route (uid), written by extractors/base.run_route and read by summarise.py / rules.py / | |
| dataset_adapter.py. Nothing here reads a video, a model or any openpilot field; the module is numpy + json only. | |
| PER-FRAME cache <cache_root>/<baseline>/<uid>.npz (10 Hz, ONE ROW PER DECODED FRAME, causal frame order) | |
| t f32[N] frame time on the frame/20 axis: t = k / 10, strictly increasing | |
| frame_idx i64[N] = 2 k (video frame index; never derived from the container fps) | |
| box f32[N,4] causal driver box in NATIVE pixels (x0, y0, x1, y1) from results/pose_causal_routes[_v2]; NaN when undefined | |
| valid bool[N] the feature row is usable (<=> fail == 0) | |
| fail u8[N] 0 ok | 1 no box | 2 no frame (decode failed / outside the video) | 3 no face (model found nothing) | 4 non-finite output | |
| feat f16[N,C] the baseline's per-frame channels; NaN on every invalid row (and only finite values on valid rows) | |
| channels str[C] channel names (registry vocabulary, e.g. l2cs_yaw / l2cs_pitch / l2cs_face_conf) | |
| spans f32[S,2] the [start, end] time spans (s) that were decoded for this route (union of anchor windows; see video_io.anchor_spans) | |
| moving bool[N] OPTIONAL: vEgo > 2 m/s at the frame (copied from the pose cache); used by rules.py `ruleref` (route reference over MOVING frames) | |
| meta json {baseline, model_ids, weight_sha256, lib_versions, decode_wh, native_wh, code_sha256, video_source, time_base, partial?} | |
| runtime json {decode_ms_p50, decode_ms_p95, infer_ms_p50, infer_ms_p95, n_frames, wall_s} (per-frame milliseconds) | |
| WINDOW cache <cache_root>/<baseline>/<uid>.npz (clip models: O4 VideoMAEv2 / I2 OMZ action; ONE ROW PER 16-FRAME WINDOW) | |
| win_t_end f32[M] END time of the window on the same axis (the window covers [win_t_end - (L-1)/10, win_t_end]; causal for an anchor T iff win_t_end <= T) | |
| emb f16[M,D] window embedding / logits; NaN rows where win_valid is False | |
| win_valid bool[M] all L frames decoded AND a driver box present on every frame (plan §3) | |
| win_len int L (frames per window, 16) | |
| spans, meta, runtime as above | |
| Conventions: NaN wherever a quantity is undefined (missing policy of decisions.md 2026-09-14); dtypes are exactly the ones above (the | |
| loader casts f16 -> f32 for arithmetic); `validate()` enforces the invariants; `covers(spans)` answers the resumability question of | |
| run_route ("does the stored cache already contain every requested span?"). | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import os | |
| import time | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Optional | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| RESULTS_ROOT = CWM3 / "results" / "baselines_dms" | |
| HZ = 10 | |
| FPS = 20 | |
| TIME_BASE = "frame/20: t=k/10, frame_idx=2k" | |
| FAIL_CODES = {0: "ok", 1: "no_box", 2: "no_frame", 3: "no_face", 4: "non_finite"} | |
| META_KEYS = ("baseline", "model_ids", "weight_sha256", "lib_versions", "decode_wh", "native_wh", "code_sha256", "video_source", "time_base") | |
| RUNTIME_KEYS = ("decode_ms_p50", "decode_ms_p95", "infer_ms_p50", "infer_ms_p95", "n_frames") | |
| SPAN_TOL_S = 0.051 # half a 10 Hz step: a requested span is covered when it lies inside a stored span up to this tolerance | |
| def cache_dir(baseline: str, root: Path = RESULTS_ROOT) -> Path: | |
| return Path(root) / "cache" / baseline | |
| def summ_dir(baseline: str, root: Path = RESULTS_ROOT) -> Path: | |
| return Path(root) / "summ" / baseline | |
| def cache_path(baseline: str, uid: str, root: Path = RESULTS_ROOT) -> Path: | |
| return cache_dir(baseline, root) / f"{uid}.npz" | |
| def sha256_file(p, chunk=1 << 22) -> str: | |
| h = hashlib.sha256() | |
| with open(p, "rb") as f: | |
| for b in iter(lambda: f.read(chunk), b""): | |
| h.update(b) | |
| return h.hexdigest() | |
| def code_sha256(files) -> str: | |
| """sha256 over the concatenated bytes of the given source files (provenance of the extractor + this schema).""" | |
| h = hashlib.sha256() | |
| for p in files: | |
| h.update(Path(p).read_bytes()) | |
| return h.hexdigest() | |
| def make_meta(baseline, model_ids, weight_sha256=None, lib_versions=None, decode_wh=None, native_wh=None, code_sha=None, video_source="local", **extra) -> dict: | |
| m = {"baseline": str(baseline), "model_ids": [str(x) for x in (model_ids or [])], "weight_sha256": dict(weight_sha256 or {}), "lib_versions": dict(lib_versions or {}), | |
| "decode_wh": (None if decode_wh is None else [int(decode_wh[0]), int(decode_wh[1])]), "native_wh": (None if native_wh is None else [int(native_wh[0]), int(native_wh[1])]), | |
| "code_sha256": str(code_sha or ""), "video_source": str(video_source), "time_base": TIME_BASE, "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} | |
| m.update(extra) | |
| return m | |
| def make_runtime(decode_ms_per_frame, infer_ms_per_frame, n_frames, wall_s=None) -> dict: | |
| """decode_ms_per_frame / infer_ms_per_frame: sequences of per-frame milliseconds (one entry per decoded block / inferred batch); p50 / p95 over them.""" | |
| def pq(v): | |
| v = np.asarray([x for x in v if np.isfinite(x)], np.float64) | |
| return (float(np.percentile(v, 50)), float(np.percentile(v, 95))) if len(v) else (float("nan"), float("nan")) | |
| d50, d95 = pq(decode_ms_per_frame) | |
| i50, i95 = pq(infer_ms_per_frame) | |
| return {"decode_ms_p50": d50, "decode_ms_p95": d95, "infer_ms_p50": i50, "infer_ms_p95": i95, "n_frames": int(n_frames), "wall_s": (None if wall_s is None else float(wall_s))} | |
| def merge_spans(spans, gap=0.0): | |
| """[S,2] -> sorted, merged (overlapping or gap <= `gap`) spans as f32[S',2]; empty -> shape (0, 2).""" | |
| a = np.asarray(spans, np.float64).reshape(-1, 2) | |
| if len(a) == 0: | |
| return np.zeros((0, 2), np.float32) | |
| a = a[np.argsort(a[:, 0], kind="stable")] | |
| out = [list(a[0])] | |
| for s, e in a[1:]: | |
| if s <= out[-1][1] + gap: | |
| out[-1][1] = max(out[-1][1], e) | |
| else: | |
| out.append([s, e]) | |
| return np.asarray(out, np.float32) | |
| def spans_cover(have, want, tol=SPAN_TOL_S) -> bool: | |
| """every [s, e] in `want` lies inside some merged span of `have` (up to tol).""" | |
| H = merge_spans(have) | |
| for s, e in np.asarray(want, np.float64).reshape(-1, 2): | |
| if not any((hs - tol <= s) and (e <= he + tol) for hs, he in H): | |
| return False | |
| return True | |
| def spans_subtract(want, have, tol=SPAN_TOL_S): | |
| """the parts of `want` not covered by `have` -> f32[S,2] (empty when fully covered); used by run_route to resume a partial cache.""" | |
| H = merge_spans(have) | |
| out = [] | |
| for s, e in merge_spans(want).astype(np.float64): | |
| cur = [(s, e)] | |
| for hs, he in H: | |
| nxt = [] | |
| for a, b in cur: | |
| if he + tol <= a or hs - tol >= b: | |
| nxt.append((a, b)) | |
| continue | |
| if hs - tol > a: | |
| nxt.append((a, hs)) | |
| if he + tol < b: | |
| nxt.append((he, b)) | |
| cur = nxt | |
| out += [(a, b) for a, b in cur if b - a > tol] | |
| return np.asarray(out, np.float32).reshape(-1, 2) | |
| def _json_field(z, key, default): | |
| if key not in z.files: | |
| return default | |
| v = z[key] | |
| try: | |
| return json.loads(str(v.item() if v.shape == () else v)) | |
| except Exception: | |
| return default | |
| def savez_atomic(path, **arrays) -> Path: | |
| """np.savez_compressed into <path>.<pid>.tmp.npz in the same directory, then os.replace onto <path> (atomic on POSIX): a reader never sees a | |
| truncated / half-written cache and two writers of the same route cannot interleave (the last complete file wins). `.npz` is appended to | |
| `path` when missing (np.savez_compressed's own rule); the tmp file is removed when the write fails. -> the final path.""" | |
| path = Path(path) | |
| if path.suffix != ".npz": | |
| path = path.with_name(path.name + ".npz") | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| tmp = path.with_name(f"{path.stem}.{os.getpid()}.tmp.npz") | |
| try: | |
| np.savez_compressed(tmp, **arrays) | |
| os.replace(tmp, path) | |
| except BaseException: | |
| try: | |
| os.unlink(tmp) | |
| except FileNotFoundError: | |
| pass | |
| raise | |
| return path | |
| class RouteCache: | |
| uid: str | |
| t: np.ndarray | |
| frame_idx: np.ndarray | |
| box: np.ndarray | |
| valid: np.ndarray | |
| fail: np.ndarray | |
| feat: np.ndarray | |
| channels: list | |
| spans: np.ndarray | |
| meta: dict = field(default_factory=dict) | |
| runtime: dict = field(default_factory=dict) | |
| moving: Optional[np.ndarray] = None | |
| def n(self): | |
| return int(len(self.t)) | |
| def C(self): | |
| return int(self.feat.shape[1]) if self.feat.ndim == 2 else 0 | |
| def validate(self): | |
| N = self.n | |
| assert self.t.dtype == np.float32 and self.t.shape == (N,), ("t", self.t.dtype, self.t.shape) | |
| assert self.frame_idx.dtype == np.int64 and self.frame_idx.shape == (N,), ("frame_idx", self.frame_idx.dtype, self.frame_idx.shape) | |
| assert self.box.dtype == np.float32 and self.box.shape == (N, 4), ("box", self.box.dtype, self.box.shape) | |
| assert self.valid.dtype == np.bool_ and self.valid.shape == (N,), ("valid", self.valid.dtype, self.valid.shape) | |
| assert self.fail.dtype == np.uint8 and self.fail.shape == (N,), ("fail", self.fail.dtype, self.fail.shape) | |
| assert self.feat.dtype == np.float16 and self.feat.shape == (N, len(self.channels)), ("feat", self.feat.dtype, self.feat.shape, len(self.channels)) | |
| assert self.spans.dtype == np.float32 and self.spans.ndim == 2 and self.spans.shape[1] == 2, ("spans", self.spans.dtype, self.spans.shape) | |
| if N: | |
| assert np.all(np.diff(self.t) > 0), "t must be strictly increasing (causal frame order)" | |
| k = np.round(self.t.astype(np.float64) * HZ) | |
| assert np.allclose(k, self.t.astype(np.float64) * HZ, atol=1e-3), "t is not on the 10 Hz grid k/10" | |
| assert np.array_equal(self.frame_idx, (k * (FPS // HZ)).astype(np.int64)), "frame_idx != 2k (time base frame/20)" | |
| assert np.array_equal(self.valid, self.fail == 0), "valid <=> fail == 0" | |
| assert set(np.unique(self.fail).tolist()) <= set(FAIL_CODES), np.unique(self.fail) | |
| f32 = self.feat.astype(np.float32) | |
| assert np.isnan(f32[~self.valid]).all(), "feat must be NaN on invalid rows" | |
| assert np.isfinite(f32[self.valid]).all(), "feat must be finite on valid rows" | |
| if self.moving is not None: | |
| assert self.moving.dtype == np.bool_ and self.moving.shape == (N,) | |
| for k_ in META_KEYS: | |
| assert k_ in self.meta, f"meta lacks {k_}" | |
| assert self.meta["time_base"] == TIME_BASE, self.meta["time_base"] | |
| return self | |
| def covers(self, spans, tol=SPAN_TOL_S) -> bool: | |
| return spans_cover(self.spans, spans, tol) | |
| def save(self, path) -> Path: | |
| self.validate() | |
| extra = {} if self.moving is None else {"moving": self.moving} | |
| return savez_atomic(path, uid=np.array(self.uid), t=self.t, frame_idx=self.frame_idx, box=self.box, valid=self.valid, fail=self.fail, feat=self.feat, | |
| channels=np.array([str(c) for c in self.channels]), spans=self.spans, meta=np.array(json.dumps(self.meta)), runtime=np.array(json.dumps(self.runtime)), | |
| kind=np.array("route_frames"), fail_codes=np.array(json.dumps(FAIL_CODES)), **extra) | |
| def load(cls, path) -> "RouteCache": | |
| with np.load(path, allow_pickle=False) as z: | |
| assert "feat" in z.files and str(z["kind"]) == "route_frames", f"{path}: not a per-frame route cache" | |
| rc = cls(uid=str(z["uid"]), t=z["t"], frame_idx=z["frame_idx"], box=z["box"], valid=z["valid"], fail=z["fail"], feat=z["feat"], | |
| channels=[str(c) for c in z["channels"]], spans=z["spans"], meta=_json_field(z, "meta", {}), runtime=_json_field(z, "runtime", {}), | |
| moving=(z["moving"] if "moving" in z.files else None)) | |
| return rc.validate() | |
| def empty(cls, uid, channels, meta, spans=None) -> "RouteCache": | |
| C = len(channels) | |
| return cls(uid=uid, t=np.zeros(0, np.float32), frame_idx=np.zeros(0, np.int64), box=np.zeros((0, 4), np.float32), valid=np.zeros(0, bool), fail=np.zeros(0, np.uint8), | |
| feat=np.zeros((0, C), np.float16), channels=list(channels), spans=merge_spans(spans if spans is not None else []), meta=dict(meta), runtime={}) | |
| def merged_with(self, other: "RouteCache") -> "RouteCache": | |
| """union of two caches of the same route / channels (resume of a partial cache): rows of `other` win on duplicate frame_idx.""" | |
| assert self.uid == other.uid and list(self.channels) == list(other.channels), "merge: uid / channels differ" | |
| fi = np.concatenate([self.frame_idx, other.frame_idx]) | |
| order = np.argsort(fi, kind="stable") | |
| fi_s = fi[order] | |
| keep_last = np.r_[fi_s[1:] != fi_s[:-1], True] # duplicates: keep the LAST occurrence = the row from `other` | |
| sel = order[keep_last] | |
| def cat(a, b): | |
| return np.concatenate([a, b])[sel] | |
| mov = None if (self.moving is None or other.moving is None) else cat(self.moving, other.moving) | |
| return RouteCache(self.uid, cat(self.t, other.t), cat(self.frame_idx, other.frame_idx), cat(self.box, other.box), cat(self.valid, other.valid), cat(self.fail, other.fail), | |
| cat(self.feat, other.feat), list(self.channels), merge_spans(np.concatenate([self.spans, other.spans])), dict(other.meta), dict(other.runtime), mov) | |
| class WindowCache: | |
| uid: str | |
| win_t_end: np.ndarray | |
| emb: np.ndarray | |
| win_valid: np.ndarray | |
| win_len: int | |
| spans: np.ndarray | |
| meta: dict = field(default_factory=dict) | |
| runtime: dict = field(default_factory=dict) | |
| def m(self): | |
| return int(len(self.win_t_end)) | |
| def D(self): | |
| return int(self.emb.shape[1]) if self.emb.ndim == 2 else 0 | |
| def validate(self): | |
| M = self.m | |
| assert self.win_t_end.dtype == np.float32 and self.win_t_end.shape == (M,), ("win_t_end", self.win_t_end.dtype, self.win_t_end.shape) | |
| assert self.emb.dtype == np.float16 and self.emb.ndim == 2 and self.emb.shape[0] == M, ("emb", self.emb.dtype, self.emb.shape) | |
| assert self.win_valid.dtype == np.bool_ and self.win_valid.shape == (M,) | |
| assert int(self.win_len) >= 1 | |
| assert self.spans.dtype == np.float32 and self.spans.ndim == 2 and self.spans.shape[1] == 2 | |
| if M: | |
| assert np.all(np.diff(self.win_t_end) > 0), "win_t_end must be strictly increasing" | |
| e32 = self.emb.astype(np.float32) | |
| assert np.isnan(e32[~self.win_valid]).all(), "emb must be NaN on invalid windows" | |
| assert np.isfinite(e32[self.win_valid]).all(), "emb must be finite on valid windows" | |
| for k_ in META_KEYS: | |
| assert k_ in self.meta, f"meta lacks {k_}" | |
| assert self.meta["time_base"] == TIME_BASE | |
| return self | |
| def covers(self, spans, tol=SPAN_TOL_S) -> bool: | |
| return spans_cover(self.spans, spans, tol) | |
| def save(self, path) -> Path: | |
| self.validate() | |
| return savez_atomic(path, uid=np.array(self.uid), win_t_end=self.win_t_end, emb=self.emb, win_valid=self.win_valid, win_len=np.int64(self.win_len), spans=self.spans, | |
| meta=np.array(json.dumps(self.meta)), runtime=np.array(json.dumps(self.runtime)), kind=np.array("route_windows")) | |
| def load(cls, path) -> "WindowCache": | |
| with np.load(path, allow_pickle=False) as z: | |
| assert "emb" in z.files and str(z["kind"]) == "route_windows", f"{path}: not a window cache" | |
| wc = cls(uid=str(z["uid"]), win_t_end=z["win_t_end"], emb=z["emb"], win_valid=z["win_valid"], win_len=int(z["win_len"]), spans=z["spans"], | |
| meta=_json_field(z, "meta", {}), runtime=_json_field(z, "runtime", {})) | |
| return wc.validate() | |
| def cache_kind(path) -> str: | |
| with np.load(path, allow_pickle=False) as z: | |
| return str(z["kind"]) if "kind" in z.files else ("route_windows" if "emb" in z.files else "route_frames") | |
| def load_any(path): | |
| return WindowCache.load(path) if cache_kind(path) == "route_windows" else RouteCache.load(path) | |
| def synthetic_route_cache(uid="synthetic__dev0__seg__route_0-1", t0=0.0, t1=60.0, channels=("yaw_deg", "pitch_deg", "closed_p", "conf"), seed=0, invalid_frac=0.05, meta=None) -> RouteCache: | |
| """a 10 Hz cache with plausible marginals (tests / self-tests only; never data).""" | |
| rng = np.random.default_rng(seed) | |
| k = np.arange(int(round(t0 * HZ)), int(round(t1 * HZ)) + 1) | |
| N, C = len(k), len(channels) | |
| feat = np.zeros((N, C), np.float32) | |
| for j, c in enumerate(channels): | |
| if c.endswith("_deg"): | |
| feat[:, j] = np.cumsum(rng.normal(0, 2.0, N)) * 0.3 + rng.normal(0, 3.0, N) | |
| elif c.endswith("_p"): | |
| feat[:, j] = np.clip(rng.beta(1.2, 6.0, N), 0, 1) | |
| else: | |
| feat[:, j] = np.clip(rng.normal(0.9, 0.05, N), 0, 1) | |
| fail = np.zeros(N, np.uint8) | |
| fail[rng.random(N) < invalid_frac] = 3 | |
| fail[rng.random(N) < invalid_frac / 2] = 1 | |
| valid = fail == 0 | |
| feat[~valid] = np.nan | |
| box = np.tile(np.array([900.0, 300.0, 1500.0, 1150.0], np.float32), (N, 1)) | |
| box[fail == 1] = np.nan | |
| m = make_meta("synthetic", ["synthetic-extractor"], decode_wh=(640, 400), native_wh=(1928, 1208), code_sha="0" * 64, video_source="synthetic") | |
| if meta: | |
| m.update(meta) | |
| return RouteCache(uid=uid, t=(k / HZ).astype(np.float32), frame_idx=(k * 2).astype(np.int64), box=box, valid=valid, fail=fail, feat=feat.astype(np.float16), channels=list(channels), | |
| spans=np.array([[t0, t1]], np.float32), meta=m, runtime=make_runtime([1.0], [1.0], N), moving=rng.random(N) < 0.8).validate() | |
| def _self_test(): | |
| import tempfile | |
| rc = synthetic_route_cache() | |
| assert rc.covers([[10.0, 20.0]]) and not rc.covers([[50.0, 61.0]]) | |
| assert len(spans_subtract([[50.0, 70.0]], rc.spans)) == 1 and abs(spans_subtract([[50.0, 70.0]], rc.spans)[0, 0] - 60.0) < 0.06 | |
| assert len(spans_subtract([[5.0, 6.0]], rc.spans)) == 0 | |
| with tempfile.TemporaryDirectory() as d: | |
| p = rc.save(Path(d) / "rc.npz") | |
| rc2 = RouteCache.load(p) | |
| assert np.array_equal(rc2.frame_idx, rc.frame_idx) and rc2.channels == rc.channels and rc2.meta["baseline"] == "synthetic" and rc2.moving is not None | |
| f1, f2 = rc.feat.astype(np.float32), rc2.feat.astype(np.float32) | |
| assert np.array_equal(np.isnan(f1), np.isnan(f2)) and np.array_equal(f1[rc.valid], f2[rc.valid]) | |
| # merge of a partial cache with a later block | |
| rc_b = synthetic_route_cache(t0=55.0, t1=80.0, seed=1) | |
| rm = rc.merged_with(rc_b) | |
| assert rm.n == len(np.unique(np.concatenate([rc.frame_idx, rc_b.frame_idx]))) and np.all(np.diff(rm.t) > 0) and rm.covers([[0, 80]]) and rm.validate() | |
| # window cache | |
| M, D = 40, 8 | |
| rng = np.random.default_rng(0) | |
| wv = rng.random(M) < 0.9 | |
| emb = rng.normal(0, 1, (M, D)).astype(np.float32) | |
| emb[~wv] = np.nan | |
| wc = WindowCache("synthetic__dev0__seg__route_0-1", (np.arange(M) * 0.5 + 1.5).astype(np.float32), emb.astype(np.float16), wv, 16, np.array([[0.0, 21.0]], np.float32), rc.meta, rc.runtime) | |
| pw = wc.save(Path(d) / "wc.npz") | |
| assert cache_kind(pw) == "route_windows" and cache_kind(p) == "route_frames" and isinstance(load_any(pw), WindowCache) | |
| # invariant violations are caught | |
| bad = synthetic_route_cache() | |
| bad.feat[~bad.valid] = 0 | |
| try: | |
| bad.validate() | |
| raise AssertionError("validate accepted feat != NaN on invalid rows") | |
| except AssertionError as e: | |
| assert "NaN" in str(e) | |
| print(json.dumps({"self_test": "cache_schema", "ok": True, "n_frames": rc.n, "C": rc.C, "fail_codes": FAIL_CODES, "meta_keys": list(META_KEYS), "runtime": rc.runtime}, 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("--inspect", default="", help="print the header of a cache npz") | |
| a = ap.parse_args() | |
| if a.inspect: | |
| c = load_any(a.inspect) | |
| print(json.dumps({"kind": type(c).__name__, "uid": c.uid, "n": getattr(c, "n", getattr(c, "m", None)), "channels": getattr(c, "channels", None), "D": getattr(c, "D", None), | |
| "spans": c.spans.tolist(), "meta": c.meta, "runtime": c.runtime}, indent=1)) | |
| if a.self_test or not a.inspect: | |
| _self_test() | |
Xet Storage Details
- Size:
- 21.1 kB
- Xet hash:
- c051dbe6e68090f504c696a689bfecbae4f3216fb0cdfe248b2e2a059f85ca8b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.