Buckets:
| """cwm3/baselines_dms/layout.py — the score-layout writer of the DMS baselines (plan §0 / §1a "同式输出", 2026-09-14). | |
| Contract (identical to the CockpitWM warning head, train/train_warning_head.py:601-602, and the R0 writer baselines_dms/make_r0_layout.py): | |
| results/runs/<arm>_s<seed>/pred_layout_{regime}_{split}.npz | |
| uid, t BYTE-EQUAL to s12_wm/results/targets_{regime}_{split}.npz (asserted here; the pack's own arrays are written) | |
| w_names[W] column names; w f16[N,W] in [0, 1] or NaN (undefined); w_mask bool[N] (False = no score on that row; evaluators drop it) | |
| meta json arm, regime, seed, split, score_source, model_id, dms_fields_used: [] (+ code_sha256, written_utc, baseline details) | |
| arm names start with `abl_dms_` (`abl_` arms need no cwm2/eval/results_schema.py edit; the vocabulary check accepts any abl_* arm). | |
| Missing policy (decisions.md 2026-09-14): NaN + w_mask False where the score is undefined (row_ok False or valid_share_8s < 0.25); the alert-rate | |
| evaluator treats non-finite as "no alert" with unchanged exposure. w_mask = finite(primary column) by default (`mask_col`), or an explicit mask. | |
| Nothing here trains, scores or reads any model / openpilot field; the module only writes and validates layouts. | |
| python baselines_dms/layout.py --self-test # writes a synthetic abl_dms_selftest layout to a temp dir, validates, checks the uid/t assertion | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| RUNS = CWM3 / "results" / "runs" | |
| ARM_PREFIX = "abl_dms_" | |
| META_REQUIRED = ("arm", "regime", "seed", "split", "score_source", "model_id", "dms_fields_used") | |
| SCORE_SOURCES = ("rule", "legacy_chain", "independent_marginals", "") # cwm2/eval/results_schema.SCORE_SOURCES (subset used by the baselines) | |
| def pack_uid_t(regime, split): | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| return z["uid"], z["t"] | |
| def run_dir(arm, seed, root=RUNS): | |
| assert arm.startswith(ARM_PREFIX), f"arm {arm!r} must start with {ARM_PREFIX!r}" | |
| return Path(root) / f"{arm}_s{int(seed)}" | |
| def write_layout(arm, regime, seed, split, uid, t, columns, meta_extra=None, out_root=RUNS, score_source="rule", model_id=None, mask=None, mask_col=None, code_files=None): | |
| """columns: {name: f32/f64[N] in [0,1] or NaN} (insertion order = column order) -> path of the written pred_layout npz. | |
| uid / t must equal the s12 pack (asserted; the pack arrays are written so the file is byte-equal); w is f16; w_mask = finite(columns[mask_col or first]) unless `mask` given.""" | |
| assert arm.startswith(ARM_PREFIX), f"arm {arm!r} must start with {ARM_PREFIX!r}" | |
| assert regime in ("manual", "engaged") and split in ("train", "val", "test"), (regime, split) | |
| assert score_source in SCORE_SOURCES, score_source | |
| puid, pt = pack_uid_t(regime, split) | |
| uid, t = np.asarray(uid), np.asarray(t) | |
| assert uid.shape == puid.shape and np.array_equal(uid.astype(str), puid.astype(str)), f"uid differs from the s12 pack {regime}/{split}" | |
| assert t.shape == pt.shape and np.array_equal(t.astype(np.float64), pt.astype(np.float64)), f"t differs from the s12 pack {regime}/{split}" | |
| N = len(puid) | |
| names = list(columns) | |
| assert names, "no columns" | |
| cols = [] | |
| for k in names: | |
| v = np.asarray(columns[k], np.float32).reshape(-1) | |
| assert v.shape == (N,), (k, v.shape, N) | |
| fin = np.isfinite(v) | |
| assert np.all(v[fin] >= 0) and np.all(v[fin] <= 1), f"column {k}: finite values must lie in [0, 1] (min {v[fin].min() if fin.any() else None}, max {v[fin].max() if fin.any() else None})" | |
| cols.append(v) | |
| W = np.stack(cols, 1).astype(np.float16) | |
| if mask is None: | |
| j = names.index(mask_col) if mask_col else 0 | |
| w_mask = np.isfinite(W[:, j].astype(np.float32)) | |
| else: | |
| w_mask = np.asarray(mask, bool).reshape(-1) | |
| assert w_mask.shape == (N,) | |
| W[~w_mask] = np.nan # masked rows carry no score | |
| files = [Path(__file__)] + [Path(p) for p in (code_files or [])] | |
| h = hashlib.sha256() | |
| for p in files: | |
| h.update(p.read_bytes()) | |
| meta = {"arm": arm, "regime": regime, "seed": int(seed), "split": split, "score_source": score_source, "model_id": model_id or f"{arm}:{names[0]}", | |
| "dms_fields_used": [], "code_sha256": h.hexdigest(), "w_mask_rule": ("explicit" if mask is not None else f"finite({names[names.index(mask_col) if mask_col else 0]})"), | |
| "missing_policy": "NaN + w_mask False where undefined (row_ok False or valid_share_8s < 0.25); alert-rate evaluator: non-finite = no alert, exposure unchanged", | |
| "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "writer": "baselines_dms/layout.py"} | |
| for k, v in (meta_extra or {}).items(): | |
| assert k != "dms_fields_used" or v == [], "dms_fields_used must stay []" | |
| meta[k] = v | |
| for k in META_REQUIRED: | |
| assert k in meta, k | |
| out = run_dir(arm, seed, out_root) | |
| out.mkdir(parents=True, exist_ok=True) | |
| path = out / f"pred_layout_{regime}_{split}.npz" | |
| np.savez_compressed(path, meta=np.array(json.dumps(meta, default=str)), uid=puid, t=pt, w_names=np.array(names), w=W, w_mask=w_mask) | |
| return path | |
| def mark_done(arm, seed, out_root=RUNS, regimes=("manual", "engaged"), splits=("test", "val")): | |
| d = run_dir(arm, seed, out_root) | |
| (d / "PRED_DONE").write_text(json.dumps({"splits": list(splits), "regimes": list(regimes), "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())})) | |
| return d / "PRED_DONE" | |
| def read_layout(path): | |
| with np.load(path, allow_pickle=True) as z: | |
| return {"meta": json.loads(str(z["meta"])), "uid": z["uid"], "t": z["t"], "w_names": [str(c) for c in z["w_names"]], "w": z["w"], "w_mask": z["w_mask"]} | |
| def validate_layout(path, check_pack=True, require_prefix=True): | |
| """asserts the contract on a written file; returns the parsed meta. Used by tests/test_baselines_dms.py (d).""" | |
| L = read_layout(path) | |
| m, W, wm = L["meta"], L["w"], L["w_mask"] | |
| N = len(L["uid"]) | |
| assert W.dtype == np.float16 and W.shape == (N, len(L["w_names"])), (W.dtype, W.shape) | |
| assert wm.dtype == np.bool_ and wm.shape == (N,), (wm.dtype, wm.shape) | |
| f = W.astype(np.float32) | |
| fin = np.isfinite(f) | |
| assert np.all(f[fin] >= 0) and np.all(f[fin] <= 1), "w outside [0, 1]" | |
| assert not fin[~wm].any(), "masked rows must be NaN" | |
| for k in META_REQUIRED: | |
| assert k in m, f"meta lacks {k}" | |
| assert m["dms_fields_used"] == [], m["dms_fields_used"] | |
| if require_prefix: | |
| assert str(m["arm"]).startswith(ARM_PREFIX), m["arm"] | |
| assert m["regime"] in ("manual", "engaged") and m["split"] in ("train", "val", "test") | |
| if check_pack: | |
| puid, pt = pack_uid_t(m["regime"], m["split"]) | |
| assert np.array_equal(L["uid"], puid) and np.array_equal(L["t"], pt), "uid / t differ from the s12 pack" | |
| return m | |
| def schema_row(meta, endpoint="gold_W", metric="auroc", stratum="clean_test"): | |
| """a minimal cwm2/eval/results_schema row for the arm (vocabulary check of the evaluators' table writer).""" | |
| ss = meta.get("score_source", "rule") | |
| return {"arm": meta["arm"], "endpoint": endpoint, "metric": metric, "stratum": stratum, "regime": meta["regime"], "split": meta["split"], | |
| "score_source": ss if ss in SCORE_SOURCES else "rule", # layouts written elsewhere may carry a free-text score_source (R0: fixed_rule_face_pack) -> the table vocabulary value | |
| "roi": "causal_clip_roi", "status": "post_hoc", "calib": "raw", "comparator": ""} | |
| def _self_test(): | |
| import tempfile | |
| puid, pt = pack_uid_t("manual", "val") | |
| rng = np.random.default_rng(0) | |
| N = len(puid) | |
| s = rng.random(N).astype(np.float32) | |
| s[rng.random(N) < 0.2] = np.nan | |
| with tempfile.TemporaryDirectory() as d: | |
| p = write_layout("abl_dms_selftest_rule0", "manual", 0, "val", puid, pt, {"dms_selftest_rule0": s, "dms_selftest_share3s": np.clip(s * 0.5, 0, 1)}, | |
| meta_extra={"baseline": {"id": "selftest"}}, out_root=d, score_source="rule") | |
| m = validate_layout(p) | |
| L = read_layout(p) | |
| assert L["w_names"] == ["dms_selftest_rule0", "dms_selftest_share3s"] and np.array_equal(L["w_mask"], np.isfinite(s)) and m["arm"] == "abl_dms_selftest_rule0" | |
| assert L["uid"].dtype == puid.dtype and L["t"].dtype == pt.dtype and L["uid"].tobytes() == puid.tobytes() and L["t"].tobytes() == pt.tobytes() | |
| try: | |
| sys.path.insert(0, str(EXPLORE / "cwm2" / "eval")) | |
| import results_schema as RS | |
| RS.validate(schema_row(m)) | |
| schema = "results_schema.validate ok" | |
| except ImportError: | |
| schema = "results_schema not importable (skipped)" | |
| for bad_kw in ({"t": pt + 0.2}, {"uid": np.roll(puid, 1)}): | |
| try: | |
| write_layout("abl_dms_selftest_rule0", "manual", 0, "val", bad_kw.get("uid", puid), bad_kw.get("t", pt), {"x": s}, out_root=d) | |
| raise RuntimeError("misaligned uid / t was accepted") | |
| except AssertionError: | |
| pass | |
| try: | |
| write_layout("abl_dms_selftest_rule0", "manual", 0, "val", puid, pt, {"x": s * 2}, out_root=d) | |
| raise RuntimeError("out-of-range score accepted") | |
| except AssertionError: | |
| pass | |
| try: | |
| write_layout("abl_selftest", "manual", 0, "val", puid, pt, {"x": s}, out_root=d) | |
| raise RuntimeError("arm without the abl_dms_ prefix accepted") | |
| except AssertionError: | |
| pass | |
| mark_done("abl_dms_selftest_rule0", 0, d) | |
| print(json.dumps({"self_test": "layout", "ok": True, "N": N, "defined_share": float(np.isfinite(s).mean()), "schema": schema, "meta_keys": sorted(m)}, 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("--validate", default="", help="validate an existing pred_layout npz") | |
| a = ap.parse_args() | |
| if a.validate: | |
| print(json.dumps(validate_layout(a.validate), indent=1)) | |
| if a.self_test or not a.validate: | |
| _self_test() | |
Xet Storage Details
- Size:
- 10.5 kB
- Xet hash:
- 8c7b26eea2f02537157034256e80081b980175d8274b639f6f445f10783e4452
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.