Buckets:
| #!/usr/bin/env python3 | |
| """cwm3/baselines_dms/fit_readout.py — LAYER 2b of the DMS-baseline comparison (plan §2 第二层 (b) / §2b B, C; 2026-09-14): the COMMON learned warning read-out, | |
| fitted on a baseline's 8 s causal summaries with EXACTLY the protocol of the CockpitWM head W_learned_* (train/train_warning_head.py): the same label table, | |
| model family, grid, cross-validation, selection rule, temperature fit and layout contract; nothing in train_warning_head.py is modified — its functions are imported. | |
| PROTOCOL (every element is the W_learned one; file:line of train/train_warning_head.py as of 2026-09-14) | |
| labels LB.train_label_table() (data/labels.py:79; consensus, TRAIN devices, lenient groups excluded; 979 rows / 182 positives / 91 dongles); the clean gold | |
| (test / val) never enters a fit — asserted as in :459 (no clean_test idx in the training table) | |
| models TW.make_model(kind, hp, seed) :255-274 (L2 logistic regression with 0-imputation + standardisation; HistGradientBoosting depth 3, l2 1.0, leaf 20, no early | |
| stopping, monotone constraints = TW.monotone_constraints(names) (all 0 for summary features: no prefix matches); MLP 64 hidden) over TW.GRID :277-279 | |
| selection TW.cv_auroc :282-291 = GroupKFold(5) by dongle, mean fold AUROC; the best (kind, hp) by mean CV AUROC is refit on all labels (:486-505) | |
| seeds 0 / 1 / 2 -> make_model(..., seed) random_state (HGB / MLP); LR is seed-invariant (the CockpitWM head's seeds are the joint model's seeds; for a | |
| summaries-only read-out the seed only moves the model's own randomness — stated in meta) | |
| temperature WG.temperature_fit on the clean-val gold rows (cwm2/eval/warning_gold.py:94; T_GRID logspace 0.1..20) -> meta (the layout carries RAW probabilities, as | |
| W_learned; warning_gold.py refits the temperature itself) | |
| layout results/runs/abl_dms_<b>_<variant>_s<k>/pred_layout_{regime}_{split}.npz through baselines_dms/layout.write_layout (uid / t byte-equal to the s12 pack, | |
| w f16 in [0, 1] / NaN, w_mask, meta arm / regime / seed / split / score_source / model_id / roi fullframe_causal / dms_fields_used []), the contract of | |
| train_warning_head.py:601-602. Columns: dms_<b>_<variant>_{lr,hgb,mlp}_t15_w10_n979 (+ _sel_ = the CV-selected kind). | |
| FEATURES (label-free, causal, <= t) | |
| learned X[N,K] of summ_{regime}_{split}.npz (summarise.py 8 s window blocks: share_off / dwell / max_abs / mean / std per angle channel, eye-state blocks, | |
| probability blocks) + the fixed-rule columns (rule0 / ruleref ...) + valid_share_8s; window baselines (videomae2 / ov_action) have an empty X -> | |
| features = emb_anchor[D] ‖ emb_mean_8s[D] ‖ n_win_8s (stated in meta feature_source) | |
| --with-u0 + the layer-1 base U0_dms_<b>_s<k> anchor outputs: onset_p of hand_return / hand_leave / head_turn_onset / head_return at 1 / 2 / 3 / 5 s (16 dims) from | |
| its pred_layout (test / val) and from pred_u0_{regime}_train_labelrows.npz (fit_u_base.predict_label_rows) for the label rows — symmetric to W reading U | |
| --ctx arm abl_dms_<b>_ctx (plan §2b C (ii)): the features above ‖ CockpitWM's CAUSAL CURRENT-STATE block = TW.features_from_pred(pred, ...) :180 masked with | |
| TW.feature_mask(names, "current") :117 (anchor states z0, current kinematics K0 + validity, current margins + validity, P0 / R0 states, vf_ok / tok_all16 / | |
| V_valid bits, mode, vEgo — NO predicted quantity: pi_*, onset3s_*, K_hat are `is_future` :112 and masked out) read from the FROZEN | |
| results/runs/I1_all_minus_road2drv_s<k>_final pred packs (pred_{regime}_{split}.npz; label rows from pred_{regime}_train_labelrows.npz as :430-436) | |
| MISSING POLICY (decisions.md 2026-09-14): a row is undefined when row_ok is False or valid_share_8s < 0.25 (window baselines: no valid anchor window) -> score NaN, | |
| w_mask False; undefined label rows are dropped from the fit (count in meta). `--impute prior` (secondary arm suffix `imp`): undefined rows get the training base | |
| rate (a constant) and w_mask True, so the alert-rate evaluator counts them as scored rows. | |
| Nothing here reads a video, a model or any openpilot field (the legacy chain enters only through warning_gold.py as the labelled comparator). | |
| python baselines_dms/fit_readout.py --baseline l2cs_gaze --seed 0 [--ctx] [--with-u0] [--impute prior] [--models lr,hgb,mlp] [--summ-root D --out-root D --res-root D] | |
| python baselines_dms/fit_readout.py --dry-run # CPU: synthetic summaries + synthetic label table -> layout in a temp dir, validated | |
| python baselines_dms/fit_readout.py --baseline videomae2 --seed 0 [--ctx] --apply-only --summ-root results/baselines_dms/summ_full/videomae2 \ | |
| [--model-dir results/baselines_dms/readout/abl_dms_videomae2_learned_s0_<stamp>] [--ref-summ-root results/baselines_dms/summ/videomae2] [--arm-suffix _full] | |
| APPLY-ONLY (layer 3, 2026-09-15; additive): the ALREADY-FITTED read-out of a tier-A run is re-applied to NEW full-exposure summaries (tiers A+B+C) -> layouts under | |
| results/runs/<arm><suffix>_s<k>/ (default suffix `_full`); NOTHING is refitted, re-selected or re-tempered: the same 979 label rows, the same selected (kind, hp), | |
| the same seed; the clean-val temperature of the tier-A layout meta is copied verbatim (the layout carries raw probabilities; the evaluators refit T themselves). | |
| Model source: <model-dir>/models.joblib (fit() persists the fitted estimators there since 2026-09-15). Tier-A runs of 2026-09-15T00:41-02:01Z predate that file: | |
| for them the SAME estimators are re-materialised deterministically from <model-dir>/selected.json (per_model hp) -> TW.make_model(kind, hp, seed).fit(X, y) on the | |
| identical label rows / identical tier-A train summaries (--ref-summ-root; LR = lbfgs deterministic, HGB / MLP random_state = seed), without CV or selection; the | |
| identity is PROVEN by the bit-equality check below, and the materialised estimators are persisted (models.joblib) so any later application loads them. | |
| Equality check (the "same inputs -> same outputs" assertion, equality_check): on the tier-A rows (w_mask of the reference layout results/runs/<arm>_s<k>): | |
| (i) routes whose caches are UNCHANGED (sha256 recorded in `source_cache_sha256` of both summaries; window cache too) must have identical summary features; | |
| (ii) every row whose summary feature vector is identical must score bit-equal (f16, every non-MLP column; MLP columns within 2 f16 ulps — a re-materialised MLP | |
| reproduces the tier-A fit only to BLAS rounding, measured 1 ulp on 0.02 % of rows; MLP is the selected kind only for abl_dms_mp_face_learned_s0); | |
| (iii) rows whose features changed are reported, not asserted: tier-A routes that reappear in tiers B/C are EXTENDED by the extractor (RouteCache.merged_with keeps | |
| the tier-A frames and adds the rest of the route), and tier-A caches held a TRUNCATED 8 s history for the exposure rows inside the anchor spans (valid_share_8s / | |
| n_win_8s grow; for face models the route-causal `*_refm_*` reference also moves) — the full-exposure scores of those rows are the correct ones. The clean gold | |
| anchors (own spans in tier A) are reported separately (share identical features / equal selected score). Tier-A defined rows must stay defined (asserted). | |
| --equality report only logs. Everything is written into apply.json of the new readout dir and into the layout meta (baseline.apply_only.tierA_equality). | |
| """ | |
| from __future__ import annotations | |
| import os | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| os.environ.setdefault("MKL_NUM_THREADS", "4") | |
| import argparse # noqa: E402 | |
| import json # noqa: E402 | |
| import sys # noqa: E402 | |
| import time # noqa: E402 | |
| from pathlib import Path # noqa: E402 | |
| import numpy as np # noqa: E402 | |
| PKG = Path(__file__).resolve().parent | |
| CWM3 = PKG.parent | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| for _p in (PKG, CWM3 / "train", CWM3 / "data", CWM3 / "eval", CWM3 / "common", CWM2 / "eval", CWM2 / "data", CWM2, CWM2 / "baselines" / "legacy_cockpitwm"): | |
| if str(_p) not in sys.path: | |
| sys.path.insert(0, str(_p)) | |
| import cache_schema as cs # noqa: E402 | |
| import layout as LY # noqa: E402 | |
| import fit_u_base as FU # noqa: E402 | |
| REGIMES = ("manual", "engaged") | |
| TAU, WIN, SRC_TAG = 15, 10, "n979" | |
| MIN_VALID_SHARE = 0.25 | |
| U0_EVENTS = ("hand_return", "hand_leave", "head_turn_onset", "head_return") | |
| U0_TAUS = (5, 10, 15, 25) | |
| FROZEN_JOINT = "I1_all_minus_road2drv_s{seed}_final" | |
| READOUT_ROOT = cs.RESULTS_ROOT / "readout" | |
| _TW = {} | |
| def TW(): | |
| """train/train_warning_head.py imported lazily (heavy imports; unchanged code).""" | |
| if "m" not in _TW: | |
| import train_warning_head as m | |
| _TW["m"] = m | |
| return _TW["m"] | |
| def arm_name(baseline, ctx=False, with_u0=False, impute=None): | |
| return f"abl_dms_{baseline}_" + ("ctx" if ctx else "learned") + ("u0" if with_u0 else "") + ("imp" if impute == "prior" else "") | |
| # ----------------------------------------------------------------------------- features | |
| def summary_features(baseline, regime, split, summ_root=None): | |
| """summ npz -> (X f32[N,K] (NaN undefined), names[K], defined bool[N], uid, t, info).""" | |
| p = FU.summ_path(baseline, regime, split, summ_root) | |
| with np.load(p, allow_pickle=True) as z: | |
| uid, t, row_ok = z["uid"], z["t"], z["row_ok"].astype(bool) | |
| N = len(uid) | |
| X = z["X"].astype(np.float32) if "X" in z.files else np.zeros((N, 0), np.float32) | |
| names = [str(c) for c in z["x_names"]] if "x_names" in z.files else [] | |
| vs8 = z["valid_share_8s"].astype(np.float32) if "valid_share_8s" in z.files else np.full(N, np.nan, np.float32) | |
| src = "summarise.X (8 s causal window blocks)" | |
| if "emb_anchor" in z.files and (X.shape[1] == 0 or baseline in FU.WINDOW_BASELINES): | |
| # window baselines: the anchor quantity is the window embedding; any per-frame X (crop geometry / encoder units, summarise `default_role: ignore` | |
| # makes it empty) is NOT a feature of the read-out (2026-09-14 layer-2 launch; stated in feature_source) | |
| dropped = int(X.shape[1]) | |
| E, Em = z["emb_anchor"].astype(np.float32), z["emb_mean_8s"].astype(np.float32) | |
| nw = z["n_win_8s"].astype(np.float32) if "n_win_8s" in z.files else np.full(N, np.nan, np.float32) | |
| X = np.concatenate([E, Em, nw[:, None]], 1) | |
| names = [f"emb_anchor_{j}" for j in range(E.shape[1])] + [f"emb_mean8s_{j}" for j in range(Em.shape[1])] + ["n_win_8s"] | |
| src = "summarise_windows emb_anchor ‖ emb_mean_8s ‖ n_win_8s (window baseline)" + (f"; {dropped} per-frame summary columns dropped" if dropped else "") | |
| rules = [str(c) for c in z["rule_names"]] if "rule_names" in z.files else [] | |
| for r in rules: | |
| if r in z.files and r not in names and not r.startswith("chain_absent"): | |
| X = np.concatenate([X, z[r].astype(np.float32).reshape(N, 1)], 1) | |
| names.append(r) | |
| if np.isfinite(vs8).any(): | |
| X = np.concatenate([X, vs8.reshape(N, 1)], 1) | |
| names.append("valid_share_8s") | |
| defined = row_ok & np.isfinite(X).any(1) & np.where(np.isfinite(vs8), vs8 >= MIN_VALID_SHARE, True) | |
| X = np.where(defined[:, None], X, np.nan).astype(np.float32) | |
| return X, names, defined, uid, t, {"feature_source": src, "summary": str(p), "summary_sha256_16": cs.sha256_file(p)[:16], "K": int(X.shape[1])} | |
| def u0_features(baseline, seed, regime, split, rows=None, run_dir=None): | |
| """layer-1 base outputs at the anchors: onset_p of the 4 U events at 1 / 2 / 3 / 5 s (16 dims; NaN where the base did not predict).""" | |
| rd = Path(run_dir) if run_dir else FU.RUNS / FU.U0_DMS.format(b=baseline, seed=seed) | |
| names = [f"u0_onset_{ev}_{tau}" for ev in U0_EVENTS for tau in U0_TAUS] | |
| if split == "train": | |
| p = rd / f"pred_u0_{regime}_train_labelrows.npz" | |
| assert p.exists(), f"{p} missing: run fit_u_base.py --baseline {baseline} --predict-label-rows --seed {seed} first" | |
| with np.load(p, allow_pickle=True) as z: | |
| pos = {int(r): i for i, r in enumerate(z["rows"])} | |
| O = z["onset_p"].astype(np.float32) # [n,4 events,4 taus] | |
| ev = [str(e) for e in z["event_names"]] | |
| sel = np.array([pos[int(r)] for r in rows]) | |
| X = np.stack([O[sel, ev.index(e), U0_TAUS.index(tau)] for e in U0_EVENTS for tau in U0_TAUS], 1) | |
| return X.astype(np.float32), names | |
| with np.load(rd / f"pred_layout_{regime}_{split}.npz", allow_pickle=True) as z: | |
| ev = [str(e) for e in z["event_names"]] | |
| taus = [int(x) for x in z["onset_taus"]] | |
| O = z["onset_p"] | |
| rows = np.arange(len(z["uid"])) if rows is None else rows | |
| X = np.stack([O[rows, taus.index(tau), ev.index(e)].astype(np.float32) for e in U0_EVENTS for tau in U0_TAUS], 1) | |
| return X.astype(np.float32), names | |
| def ctx_features(seed, regime, split, rows): | |
| """CockpitWM's causal CURRENT-STATE block of the frozen main model (TW.features_from_pred + feature_mask 'current'); no predicted quantity.""" | |
| T = TW() | |
| import cwm3_dataset as ds | |
| rd = FU.RUNS / FROZEN_JOINT.format(seed=seed) | |
| cfg = json.loads((rd / "config.json").read_text())["cfg"] | |
| ds.set_kin_source(cfg.get("kin_source", "cwm2") or "cwm2", cfg.get("kin_version", "v1") or "v1") | |
| ds.set_pack_version(cfg.get("pack_version", "v1") or "v1") | |
| names_all = T.feature_names("P", extra=()) | |
| mask = T.feature_mask(names_all, "current") | |
| rows = np.asarray(rows, np.int64) | |
| if split == "train": | |
| pr = np.load(rd / f"pred_{regime}_train_labelrows.npz", allow_pickle=True) | |
| posmap = {int(r): i for i, r in enumerate(pr["rows"].astype(np.int64))} | |
| pos = np.array([posmap[int(r)] for r in rows], np.int64) | |
| X = T.features_from_pred(pr, regime, "train", rows, pos, "P") | |
| else: | |
| pr = np.load(rd / f"pred_{regime}_{split}.npz", allow_pickle=True) | |
| X = T.features_from_pred(pr, regime, split, rows, None, "P") | |
| assert X.shape[1] == len(names_all) | |
| return X[:, mask].astype(np.float32), [n for n, m in zip(names_all, mask) if m] | |
| def features(baseline, seed, regime, split, rows=None, ctx=False, with_u0=False, summ_root=None, u0_dir=None): | |
| X, names, defined, uid, t, info = summary_features(baseline, regime, split, summ_root) | |
| rows = np.arange(len(uid)) if rows is None else np.asarray(rows, np.int64) | |
| X, defined = X[rows], defined[rows] | |
| blocks = [info["feature_source"]] | |
| if with_u0: | |
| Xu, nu = u0_features(baseline, seed, regime, split, rows, u0_dir) | |
| X, names = np.concatenate([X, Xu], 1), names + nu | |
| blocks.append(f"U0_dms_{baseline}_s{seed} onset_p (16)") | |
| if ctx: | |
| Xc, nc = ctx_features(seed, regime, split, rows) | |
| X, names = np.concatenate([X, Xc], 1), names + nc | |
| blocks.append(f"CockpitWM current-state block ({len(nc)} dims, {FROZEN_JOINT.format(seed=seed)}; feature_mask current)") | |
| X = np.where(defined[:, None], X, np.nan).astype(np.float32) | |
| return X, names, defined, uid[rows], t[rows], dict(info, blocks=blocks) | |
| # ----------------------------------------------------------------------------- fit | |
| def fit(baseline, seed=0, ctx=False, with_u0=False, impute=None, models=("lr", "hgb", "mlp"), label_table=None, splits=("test", "val"), regimes=REGIMES, summ_root=None, | |
| out_root=LY.RUNS, res_root=READOUT_ROOT, gold=True, u0_dir=None, log=print): | |
| """-> dict(arm, layouts, selected, cv rows); label_table None -> LB.train_label_table() (979); a synthetic table is accepted for the dry-run (gold=False).""" | |
| T = TW() | |
| LB, WG = T.LB, T.WG | |
| arm = arm_name(baseline, ctx, with_u0, impute) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| res = Path(res_root) / f"{arm}_s{seed}_{stamp}" | |
| res.mkdir(parents=True, exist_ok=True) | |
| tab = LB.train_label_table() if label_table is None else label_table | |
| Xs, ys, gs, ds_ = [], [], [], [] | |
| names = None | |
| for regime in regimes: | |
| sub = tab[tab.regime == regime] | |
| if not len(sub): | |
| continue | |
| X, names, defined, _, _, _ = features(baseline, seed, regime, "train", sub.row.to_numpy().astype(np.int64), ctx, with_u0, summ_root, u0_dir) | |
| Xs.append(X) | |
| ys.append(sub.y.to_numpy().astype(int)) | |
| gs.append(sub.dongle.to_numpy()) | |
| ds_.append(defined) | |
| X_all, y_all, g_all, d_all = np.concatenate(Xs), np.concatenate(ys), np.concatenate(gs), np.concatenate(ds_) | |
| n_lab = len(y_all) | |
| X, y, groups = X_all[d_all], y_all[d_all], g_all[d_all] | |
| prior = float(y.mean()) if len(y) else float("nan") | |
| log(f"{arm} s{seed}: labels {n_lab} rows / {int(y_all.sum())} pos; defined {int(d_all.sum())} rows / {int(y.sum())} pos / {len(np.unique(groups))} devices; features {X.shape[1]} ({names[:6]}...)") | |
| assert d_all.sum() >= 40 and y.min() != y.max(), "too few defined labelled rows for the read-out" | |
| cv_rows, best = [], {} | |
| for kind in models: | |
| for hp0 in T.GRID[kind]: | |
| hp = dict(hp0, demand="P", names=names) | |
| m, folds = T.cv_auroc(kind, hp, X, y, groups, 5, seed) | |
| cv_rows.append({"model": kind, "hp": json.dumps({k: v for k, v in hp.items() if k != "names"}), "cv_auroc": m, "folds": json.dumps([round(f, 4) for f in folds])}) | |
| log(f" cv {kind} {hp0}: AUROC {m:.4f}") | |
| if kind not in best or m > best[kind][0]: | |
| best[kind] = (m, hp) | |
| sel = max(best, key=lambda k: best[k][0]) | |
| fits = {kind: T.make_model(kind, hp, seed).fit(X, y) for kind, (cvm, hp) in best.items()} | |
| persist_models(res, fits, sel, names, prior, seed, arm, log) # 2026-09-15 (additive): models.joblib for fit_readout --apply-only | |
| selected = {"selected": sel, "cv_auroc": best[sel][0], "hp": {k: v for k, v in best[sel][1].items() if k != "names"}, "per_model": {k: {"cv_auroc": v[0], "hp": {kk: vv for kk, vv in v[1].items() if kk != "names"}} for k, v in best.items()}, | |
| "n_features": len(names), "n_labels": int(n_lab), "n_labels_defined": int(d_all.sum()), "n_pos_defined": int(y.sum()), "prior": prior} | |
| log(f" selected {sel} (cv AUROC {best[sel][0]:.4f})") | |
| # gold anchors (temperature on clean-val only; the clean gold never enters a fit) | |
| G = None | |
| if gold: | |
| EP = LB.episode_labels() | |
| lab = EP[EP.label != "unsure"].reset_index(drop=True) | |
| packs = {(r, s): WG.load_pack(r, s) for r in REGIMES for s in ("test", "val")} | |
| G = WG.build_gold(lab, packs, WG.merge_legacy(seed), "legacy_first") | |
| assert not set(G[G.clean_test].idx) & set(tab.idx), "gold test episode inside the training table" | |
| col = lambda kind: f"dms_{baseline}_{'ctx' if ctx else 'learned'}{'u0' if with_u0 else ''}_{kind}_t{TAU}_w{WIN}_{SRC_TAG}" # noqa: E731 | |
| layouts, scores = [], {} | |
| for regime in regimes: | |
| for split in splits: | |
| Xrs, _, defined, uid, t, info = features(baseline, seed, regime, split, None, ctx, with_u0, summ_root, u0_dir) | |
| cols = {} | |
| for kind, m in fits.items(): | |
| p = np.full(len(uid), np.nan, np.float32) | |
| if defined.any(): | |
| p[defined] = m.predict_proba(Xrs[defined])[:, 1].astype(np.float32) | |
| if impute == "prior": | |
| p[~defined] = prior | |
| cols[col(kind)] = np.clip(p, 0, 1) | |
| cols[col("sel")] = cols[col(sel)] | |
| scores[(regime, split)] = (cols, defined) | |
| mask = np.ones(len(uid), bool) if impute == "prior" else defined | |
| temps = {} | |
| if G is not None: | |
| gv = G[(G.regime == regime) & (G.split == split) & G.clean_val] | |
| if len(gv) >= 10: | |
| r_ = gv.row.to_numpy() | |
| for k, v in cols.items(): | |
| ok = np.isfinite(v[r_]) | |
| yy = gv.y.to_numpy().astype(float)[ok] | |
| temps[k] = WG.temperature_fit(v[r_][ok].astype(np.float64), yy) if ok.sum() >= 10 and yy.min() != yy.max() else (None, None) | |
| meta = {"baseline": {"id": baseline, "kind": "common learned warning read-out (train_warning_head protocol) on the baseline's causal 8 s summaries", "features": names, "feature_blocks": info["blocks"], | |
| "n_features": len(names), "labels": "LB.train_label_table consensus 979 (TRAIN devices)", "n_labels": int(n_lab), "n_labels_defined": int(d_all.sum()), "selected": selected, | |
| "grid": {k: T.GRID[k] for k in models}, "cv": "GroupKFold(5) by dongle, mean AUROC (TW.cv_auroc)", "impute": impute, "prior": prior, | |
| "missing_policy": "row_ok False or valid_share_8s < 0.25 -> NaN, w_mask False" + (" (imputed with the training base rate, w_mask True)" if impute == "prior" else ""), | |
| "temperature_clean_val_by_column": {k: {"T": v[0], "pinned": v[1]} for k, v in temps.items()}, "ctx_source": (FROZEN_JOINT.format(seed=seed) if ctx else None), | |
| "u0_source": (FU.U0_DMS.format(b=baseline, seed=seed) if with_u0 else None), "seed_note": "seed moves the model's own randomness only (HGB / MLP random_state); LR is seed-invariant"}, | |
| "roi": "fullframe_causal", "roi_streams": ["face"] + (["pose", "can", "road"] if ctx else []), "driver_input_shift_s": 0.0, | |
| "notes": "DMS-baseline learned read-out; comparator only, never an input to any model"} | |
| p_ = LY.write_layout(arm, regime, seed, split, uid, t, cols, meta_extra=meta, out_root=out_root, score_source="", model_id=f"{arm}:{sel}", mask=mask, code_files=[Path(__file__)]) | |
| layouts.append(str(p_)) | |
| log(f" wrote {p_} defined {defined.mean():.4f} sel mean {np.nanmean(cols[col('sel')]):.4f}") | |
| LY.mark_done(arm, seed, out_root, regimes, splits) | |
| import pandas as pd | |
| pd.DataFrame(cv_rows).to_csv(res / "cv.csv", index=False) | |
| (res / "selected.json").write_text(json.dumps({"arm": arm, "seed": seed, "baseline": baseline, "ctx": ctx, "with_u0": with_u0, "impute": impute, **selected, "layouts": layouts}, indent=1, default=str)) | |
| return {"arm": arm, "layouts": layouts, "selected": selected, "cv": cv_rows, "res_dir": str(res)} | |
| def persist_models(res_dir, fits, sel, names, prior, seed, arm, log=print): | |
| """<res_dir>/models.joblib = the fitted estimators of fit() (all kinds) + the selection; loaded by apply_only (never refitted).""" | |
| import joblib | |
| import sklearn | |
| p = Path(res_dir) / "models.joblib" | |
| joblib.dump({"fits": fits, "selected": sel, "names": list(names), "prior": float(prior), "seed": int(seed), "arm": arm, "sklearn": sklearn.__version__, | |
| "written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}, p) | |
| log(f" persisted fitted models -> {p}") | |
| return p | |
| # ----------------------------------------------------------------------------- apply-only (layer 3, 2026-09-15): tier-A read-outs on full-exposure summaries | |
| def newest_model_dir(arm, seed, res_root=READOUT_ROOT, ref_root=LY.RUNS): | |
| """newest results/baselines_dms/readout/<arm>_s<seed>_<stamp>/ whose selected.json layouts point at <ref_root>/<arm>_s<seed>/ (the tier-A fit).""" | |
| cands = sorted(Path(res_root).glob(f"{arm}_s{seed}_*")) | |
| ok = [] | |
| for d in cands: | |
| sj = d / "selected.json" | |
| if not sj.exists(): | |
| continue | |
| lay = json.loads(sj.read_text()).get("layouts", []) | |
| if lay and all(Path(l).parent == Path(ref_root) / f"{arm}_s{seed}" for l in lay): | |
| ok.append(d) | |
| assert ok, f"no readout dir of {arm}_s{seed} under {res_root} pointing at {ref_root}" | |
| return ok[-1] | |
| def label_features(baseline, seed, ctx, with_u0, summ_root, label_table=None, regimes=REGIMES, u0_dir=None): | |
| """the fit() training matrix: label rows of LB.train_label_table (979) in REGIMES order -> (X[defined], y[defined], names, n_lab, n_defined).""" | |
| T = TW() | |
| tab = T.LB.train_label_table() if label_table is None else label_table | |
| Xs, ys, ds_, names = [], [], [], None | |
| for regime in regimes: | |
| sub = tab[tab.regime == regime] | |
| if not len(sub): | |
| continue | |
| X, names, defined, _, _, _ = features(baseline, seed, regime, "train", sub.row.to_numpy().astype(np.int64), ctx, with_u0, summ_root, u0_dir) | |
| Xs.append(X) | |
| ys.append(sub.y.to_numpy().astype(int)) | |
| ds_.append(defined) | |
| X_all, y_all, d_all = np.concatenate(Xs), np.concatenate(ys), np.concatenate(ds_) | |
| return X_all[d_all], y_all[d_all], names, int(len(y_all)), int(d_all.sum()) | |
| def materialise_models(baseline, seed, ctx, with_u0, model_dir, ref_summ_root, label_table=None, log=print): | |
| """-> (fits {kind: fitted estimator}, sel, names, prior, info). models.joblib when present; else the deterministic re-materialisation from selected.json (see module doc).""" | |
| T = TW() | |
| model_dir = Path(model_dir) | |
| sj = json.loads((model_dir / "selected.json").read_text()) | |
| pj = model_dir / "models.joblib" | |
| if pj.exists(): | |
| import joblib | |
| M = joblib.load(pj) | |
| assert M["selected"] == sj["selected"] and int(M["seed"]) == int(seed), (M["selected"], sj["selected"], M["seed"], seed) | |
| log(f" loaded fitted models {pj} (kinds {sorted(M['fits'])}, selected {M['selected']}, K {len(M['names'])})") | |
| return M["fits"], M["selected"], list(M["names"]), float(M["prior"]), {"model_source": str(pj), "materialisation": "joblib (persisted by fit())"} | |
| X, y, names, n_lab, n_def = label_features(baseline, seed, ctx, with_u0, ref_summ_root, label_table) | |
| assert n_lab == int(sj["n_labels"]) and n_def == int(sj["n_labels_defined"]) and len(names) == int(sj["n_features"]), \ | |
| f"label / feature table differs from the fit: labels {n_lab} vs {sj['n_labels']}, defined {n_def} vs {sj['n_labels_defined']}, K {len(names)} vs {sj['n_features']}" | |
| assert int(y.sum()) == int(sj["n_pos_defined"]), (int(y.sum()), sj["n_pos_defined"]) | |
| fits = {} | |
| for kind, rec in sj["per_model"].items(): | |
| hp = dict(rec["hp"], names=names) | |
| fits[kind] = T.make_model(kind, hp, seed).fit(X, y) | |
| prior = float(y.mean()) | |
| assert abs(prior - float(sj["prior"])) < 1e-12, (prior, sj["prior"]) | |
| log(f" re-materialised {sorted(fits)} from {model_dir / 'selected.json'} on the identical {n_def} defined label rows ({int(y.sum())} pos, K {len(names)}); selected {sj['selected']} " | |
| f"(no CV / no re-selection; identity proven by the tier-A bit-equality below)") | |
| return fits, sj["selected"], names, prior, {"model_source": str(model_dir / "selected.json"), "materialisation": "re-materialised: TW.make_model(kind, selected.json per_model hp, seed).fit on the identical label rows / tier-A train summaries"} | |
| def _cache_shas(summ_path): | |
| with np.load(summ_path, allow_pickle=True) as z: | |
| return json.loads(str(z["source_cache_sha256"])) if "source_cache_sha256" in z.files else {} | |
| def equality_check(ref_path, new_path, baseline, regime, split, ref_summ_root, new_summ_root, gold_rows=None, log=print): | |
| """the "same inputs -> same outputs" check on the tier-A rows (w_mask of the reference layout), at the FEATURE level: | |
| (i) route-level: rows of routes whose caches are unchanged (sha256 in both summaries' `source_cache_sha256`) -> their summary features must be identical; | |
| (ii) rows whose summary feature vector is identical (equal or both NaN, every column; ctx / u0 blocks are frozen files, identical by construction) -> scores must be | |
| bit-equal (f16) for every non-MLP column; MLP columns within 2 f16 ulps (re-materialised MLPs reproduce the tier-A fit only to BLAS rounding: 1 ulp on ~0.02 % rows); | |
| (iii) rows whose features changed (tier-A caches held a TRUNCATED 8 s history for exposure rows inside the anchor spans: valid_share_8s / n_win_8s grow once the route | |
| is fully decoded; face models: the route-causal `*_refm_*` reference also moves) -> reported, not asserted. | |
| gold_rows: pack rows of the clean gold anchors of this (regime, split) -> their identity / equality shares are reported separately. -> dict (asserted by the caller).""" | |
| R, N_ = LY.read_layout(ref_path), LY.read_layout(new_path) | |
| assert R["w_names"] == N_["w_names"], (R["w_names"], N_["w_names"]) | |
| assert R["uid"].tobytes() == N_["uid"].tobytes() and R["t"].tobytes() == N_["t"].tobytes() and R["uid"].dtype == N_["uid"].dtype and R["t"].dtype == N_["t"].dtype, "uid / t not byte-equal" | |
| ref_def, new_def = R["w_mask"], N_["w_mask"] | |
| lost = ref_def & ~new_def | |
| rows = np.flatnonzero(ref_def) | |
| sr, sn = _cache_shas(FU.summ_path(baseline, regime, split, ref_summ_root)), _cache_shas(FU.summ_path(baseline, regime, split, new_summ_root)) | |
| win = baseline in FU.WINDOW_BASELINES | |
| routes_ref = {u.replace("win:", "") for u in sr} | |
| unchanged_uids = {u for u in routes_ref if sn.get(u) == sr.get(u) and (not win or sn.get(f"win:{u}") == sr.get(f"win:{u}"))} | |
| in_unch = np.isin(R["uid"].astype(str)[rows], sorted(unchanged_uids)) | |
| Xr, names_r, _, _, _, _ = summary_features(baseline, regime, split, ref_summ_root) | |
| Xn, names_n, _, _, _, _ = summary_features(baseline, regime, split, new_summ_root) | |
| assert names_r == names_n, "summary feature names differ between the tier-A and the full summaries" | |
| Xr, Xn = Xr[rows], Xn[rows] | |
| ident = np.all((Xr == Xn) | (np.isnan(Xr) & np.isnan(Xn)), 1) | |
| del Xr, Xn | |
| Wr, Wn = R["w"][rows], N_["w"][rows] | |
| fr, fn = Wr.astype(np.float32), Wn.astype(np.float32) | |
| same = (Wr == Wn) | (np.isnan(fr) & np.isnan(fn)) | |
| ulp = np.spacing(np.maximum(np.abs(Wr), np.abs(Wn)).astype(np.float16)).astype(np.float32) | |
| within = same | (np.abs(fr - fn) <= 2 * ulp) | |
| is_mlp = np.array(["_mlp_" in c for c in R["w_names"]]) | |
| nonmlp = ~is_mlp | |
| per_col = {c: {"share_equal_identical_features": float(same[ident, j].mean()) if ident.any() else 1.0, "max_abs_diff_identical_features": float(np.nanmax(np.abs(fr[ident, j] - fn[ident, j]))) if ident.any() else 0.0, | |
| "share_equal_changed_features": float(same[~ident, j].mean()) if (~ident).any() else 1.0, "max_abs_diff_changed_features": float(np.nanmax(np.abs(fr[~ident, j] - fn[~ident, j]))) if (~ident).any() else 0.0} | |
| for j, c in enumerate(R["w_names"])} | |
| with np.load(FU.summ_path(baseline, regime, split, ref_summ_root), allow_pickle=True) as zr, np.load(FU.summ_path(baseline, regime, split, new_summ_root), allow_pickle=True) as zn: | |
| cov = {} | |
| for k in ("valid_share_8s", "n_valid_8s", "n_win_8s"): | |
| if k in zr.files and k in zn.files: | |
| cov[k] = {"tierA_mean_changed_rows": float(np.nanmean(zr[k][rows][~ident])) if (~ident).any() else None, "full_mean_changed_rows": float(np.nanmean(zn[k][rows][~ident])) if (~ident).any() else None} | |
| out = {"reference": str(ref_path), "new": str(new_path), "columns": R["w_names"], "n_rows": int(len(ref_def)), "tierA_defined_rows": int(len(rows)), "new_defined_rows": int(new_def.sum()), | |
| "tierA_defined_share": float(ref_def.mean()), "new_defined_share": float(new_def.mean()), "lost_rows": int(lost.sum()), | |
| "routes_in_tierA_summary": int(len(routes_ref)), "routes_cache_unchanged": int(len(unchanged_uids)), "routes_cache_changed": int(len(routes_ref - unchanged_uids)), | |
| "rows_unchanged_routes": int(in_unch.sum()), "unchanged_routes_features_identical_share": float(ident[in_unch].mean()) if in_unch.any() else 1.0, | |
| "rows_features_identical": int(ident.sum()), "rows_features_changed": int((~ident).sum()), | |
| "equal_identical_nonmlp": bool(same[ident][:, nonmlp].all()) if ident.any() else True, | |
| "within_2ulp_identical_mlp": bool(within[ident][:, is_mlp].all()) if (ident.any() and is_mlp.any()) else True, | |
| "share_rows_equal_all_columns_identical": float(same[ident].all(1).mean()) if ident.any() else 1.0, | |
| "share_rows_equal_all_columns_changed": float(same[~ident].all(1).mean()) if (~ident).any() else 1.0, | |
| "per_column": per_col, "history_coverage_changed_rows": cov} | |
| if gold_rows is not None: | |
| g = np.asarray(gold_rows, np.int64) | |
| gd = ref_def[g] | |
| pos = {r: i for i, r in enumerate(rows)} | |
| gi = np.array([pos[r] for r in g[gd]], np.int64) | |
| jsel = [j for j, c in enumerate(R["w_names"]) if c.endswith("_sel_t15_w10_n979")] or [len(R["w_names"]) - 1] | |
| out["gold"] = {"n_gold_rows": int(len(g)), "n_gold_defined_tierA": int(gd.sum()), "n_gold_defined_full": int(new_def[g].sum()), | |
| "share_features_identical": float(ident[gi].mean()) if len(gi) else None, "share_sel_equal": float(same[gi][:, jsel[0]].mean()) if len(gi) else None, | |
| "max_abs_diff_sel": float(np.nanmax(np.abs(fr[gi, jsel[0]] - fn[gi, jsel[0]]))) if len(gi) else 0.0} | |
| gtxt = (f"; gold rows {out['gold']['n_gold_defined_tierA']}/{out['gold']['n_gold_rows']} defined: features identical {out['gold']['share_features_identical']}, sel equal {out['gold']['share_sel_equal']}" | |
| if gold_rows is not None else "") | |
| log(f" equality {Path(new_path).name}: tier-A defined {len(rows):,} ({out['tierA_defined_share']:.4f}) -> full defined {out['new_defined_rows']:,} ({out['new_defined_share']:.4f}), lost {out['lost_rows']}; " | |
| f"routes unchanged {out['routes_cache_unchanged']} / changed {out['routes_cache_changed']} (features identical on unchanged routes: {out['unchanged_routes_features_identical_share']:.4f}); " | |
| f"features identical rows {out['rows_features_identical']:,}: bit-equal non-MLP {out['equal_identical_nonmlp']}, MLP within 2 ulp {out['within_2ulp_identical_mlp']} (all-column equal share {out['share_rows_equal_all_columns_identical']:.5f}); " | |
| f"features changed rows {out['rows_features_changed']:,} (truncated tier-A history: {json.dumps(cov, default=str)}) all-column equal share {out['share_rows_equal_all_columns_changed']:.4f}{gtxt}") | |
| return out | |
| def apply_only(baseline, seed=0, ctx=False, with_u0=False, model_dir=None, summ_root=None, ref_summ_root=None, out_root=LY.RUNS, res_root=READOUT_ROOT, ref_root=LY.RUNS, | |
| model_root=READOUT_ROOT, arm_suffix="_full", splits=("test", "val"), regimes=REGIMES, equality="strict", gold=True, u0_dir=None, label_table=None, log=print): | |
| """the tier-A read-out (model_dir) applied to the summaries under summ_root -> layouts results/runs/<arm><suffix>_s<seed>/ (+ apply.json, models.joblib in a new readout dir). | |
| Nothing is refitted (see the module doc); the bit-equality on unchanged tier-A rows is asserted (equality='strict').""" | |
| T = TW() | |
| LB, WG = T.LB, T.WG | |
| arm0 = arm_name(baseline, ctx, with_u0, None) | |
| arm = arm0 + arm_suffix | |
| assert arm_suffix and arm != arm0, "apply-only must write a NEW arm (suffix)" | |
| ref_summ_root = Path(ref_summ_root) if ref_summ_root else cs.summ_dir(baseline) | |
| summ_root = Path(summ_root) if summ_root else None | |
| assert summ_root is not None and Path(summ_root).resolve() != Path(ref_summ_root).resolve(), "--summ-root must be the NEW full-exposure summary dir (not the tier-A dir)" | |
| model_dir = Path(model_dir) if model_dir else newest_model_dir(arm0, seed, model_root, ref_root) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| res = Path(res_root) / f"{arm}_s{seed}_{stamp}" | |
| res.mkdir(parents=True, exist_ok=True) | |
| t0 = time.time() | |
| log(f"{arm} s{seed}: APPLY-ONLY of {model_dir.name} on {summ_root} (reference layouts {Path(ref_root) / f'{arm0}_s{seed}'}; tier-A summaries {ref_summ_root})") | |
| fits, sel, names, prior, minfo = materialise_models(baseline, seed, ctx, with_u0, model_dir, ref_summ_root, label_table, log) | |
| persist_models(res, fits, sel, names, prior, seed, arm, log) | |
| ref_dir = Path(ref_root) / f"{arm0}_s{seed}" | |
| G = None | |
| if gold: | |
| EP = LB.episode_labels() | |
| lab = EP[EP.label != "unsure"].reset_index(drop=True) | |
| packs = {(r, s): WG.load_pack(r, s) for r in REGIMES for s in ("test", "val")} | |
| G = WG.build_gold(lab, packs, WG.merge_legacy(seed), "legacy_first") | |
| col = lambda kind: f"dms_{baseline}_{'ctx' if ctx else 'learned'}{'u0' if with_u0 else ''}_{kind}_t{TAU}_w{WIN}_{SRC_TAG}" # noqa: E731 (the tier-A column names) | |
| layouts, eq_all, defined_share = [], {}, {} | |
| for regime in regimes: | |
| for split in splits: | |
| ref_p = ref_dir / f"pred_layout_{regime}_{split}.npz" | |
| ref_meta = LY.read_layout(ref_p)["meta"] | |
| Xrs, names2, defined, uid, t, info = features(baseline, seed, regime, split, None, ctx, with_u0, summ_root, u0_dir) | |
| assert names2 == names, f"feature names differ from the fit ({len(names2)} vs {len(names)})" | |
| cols = {} | |
| idx = np.flatnonzero(defined) | |
| for kind, m in fits.items(): | |
| p = np.full(len(uid), np.nan, np.float32) | |
| for ch in np.array_split(idx, max(1, len(idx) // 200_000 + 1)) if len(idx) else []: | |
| p[ch] = m.predict_proba(Xrs[ch])[:, 1].astype(np.float32) | |
| cols[col(kind)] = np.clip(p, 0, 1) | |
| cols[col("sel")] = cols[col(sel)] # column names = the tier-A layout's (checked in equality_check) | |
| temps_new = {} | |
| if G is not None: | |
| gv = G[(G.regime == regime) & (G.split == split) & G.clean_val] | |
| if len(gv) >= 10: | |
| r_ = gv.row.to_numpy() | |
| for k, v in cols.items(): | |
| ok = np.isfinite(v[r_]) | |
| yy = gv.y.to_numpy().astype(float)[ok] | |
| temps_new[k] = WG.temperature_fit(v[r_][ok].astype(np.float64), yy) if ok.sum() >= 10 and yy.min() != yy.max() else (None, None) | |
| base = dict(ref_meta.get("baseline", {})) | |
| base["temperature_clean_val_by_column"] = ref_meta.get("baseline", {}).get("temperature_clean_val_by_column", {}) # copied verbatim from the tier-A fit | |
| base["temperature_clean_val_by_column_recomputed_on_full"] = {k: {"T": v[0], "pinned": v[1]} for k, v in temps_new.items()} | |
| base["apply_only"] = {"reference_arm": arm0, "reference_layout": str(ref_p), "reference_readout_dir": str(model_dir), **minfo, "summary": info["summary"], "summary_sha256_16": info["summary_sha256_16"], | |
| "tierA_summary_dir": str(ref_summ_root), "rule": "no refit / no re-selection / no re-tempering: the tier-A estimators (same 979 label rows, same seed) applied to the tiers A+B+C summaries"} | |
| meta = {"baseline": base, "roi": ref_meta.get("roi", "fullframe_causal"), "roi_streams": ref_meta.get("roi_streams", ["face"]), "driver_input_shift_s": ref_meta.get("driver_input_shift_s", 0.0), | |
| "notes": ref_meta.get("notes", "DMS-baseline learned read-out; comparator only, never an input to any model") + "; layer-3 full-exposure layout (apply-only)"} | |
| p_ = LY.write_layout(arm, regime, seed, split, uid, t, cols, meta_extra=meta, out_root=out_root, score_source="", model_id=f"{arm}:{sel}", mask=defined, code_files=[Path(__file__)]) | |
| gold_rows = G[(G.regime == regime) & (G.split == split) & (G.clean_test | G.clean_val)].row.to_numpy() if G is not None else None | |
| eq = equality_check(ref_p, p_, baseline, regime, split, ref_summ_root, summ_root, gold_rows, log) | |
| assert eq["lost_rows"] == 0, f"{regime}/{split}: {eq['lost_rows']} tier-A defined rows became undefined" | |
| if equality == "strict": | |
| assert eq["unchanged_routes_features_identical_share"] == 1.0, f"{regime}/{split}: features differ on routes whose caches are unchanged (share identical {eq['unchanged_routes_features_identical_share']:.4f})" | |
| assert eq["equal_identical_nonmlp"], f"{regime}/{split}: identical features -> different scores (non-MLP columns): {json.dumps(eq['per_column'])}" | |
| assert eq["within_2ulp_identical_mlp"], f"{regime}/{split}: identical features -> MLP scores differ by > 2 f16 ulps: {json.dumps(eq['per_column'])}" | |
| eq_all[f"{regime}_{split}"] = eq | |
| defined_share[f"{regime}_{split}"] = eq["new_defined_share"] | |
| layouts.append(str(p_)) | |
| log(f" wrote {p_} defined {defined.mean():.4f} sel mean {np.nanmean(cols[col('sel')]):.4f}") | |
| # the equality result is also stamped into every layout meta (re-open, add, rewrite: the files are new and belong to this run) | |
| for p_ in layouts: | |
| with np.load(p_, allow_pickle=True) as z: | |
| d = {k: z[k] for k in z.files} | |
| m = json.loads(str(d["meta"])) | |
| m["baseline"]["apply_only"]["tierA_equality"] = eq_all[f"{m['regime']}_{m['split']}"] | |
| d["meta"] = np.array(json.dumps(m, default=str)) | |
| np.savez_compressed(p_, **d) | |
| LY.mark_done(arm, seed, out_root, regimes, splits) | |
| rep = {"arm": arm, "seed": seed, "baseline": baseline, "ctx": ctx, "with_u0": with_u0, "reference_arm": arm0, "model_dir": str(model_dir), **minfo, "selected": sel, "n_features": len(names), "prior": prior, | |
| "summ_root": str(summ_root), "ref_summ_root": str(ref_summ_root), "equality_mode": equality, "layouts": layouts, "defined_share": defined_share, "equality": eq_all, "wall_s": round(time.time() - t0, 1)} | |
| (res / "apply.json").write_text(json.dumps(rep, indent=1, default=str)) | |
| log(f" APPLY-ONLY done {arm} s{seed}: defined share {json.dumps({k: round(v, 4) for k, v in defined_share.items()})}; identical features -> bit-equal scores (non-MLP): " | |
| f"{all(e['equal_identical_nonmlp'] for e in eq_all.values())}; MLP within 2 ulp: {all(e['within_2ulp_identical_mlp'] for e in eq_all.values())}; -> {res / 'apply.json'}") | |
| return rep | |
| # ----------------------------------------------------------------------------- dry run | |
| def synthetic_label_table(uid, t, defined, dongle, n=240, seed=0): | |
| """a TRAIN-like label table on the defined rows of a synthetic summary (dry-run only): idx, y, regime, row, dongle, uid, t0.""" | |
| import pandas as pd | |
| rng = np.random.default_rng(seed) | |
| rows = np.flatnonzero(defined) | |
| rows = rng.choice(rows, min(n, len(rows)), replace=False) | |
| groups = np.array([f"synthdev{i % 6}" for i in range(len(rows))]) # >= 5 pseudo-devices so GroupKFold(5) is defined (the real table has 91 dongles) | |
| return pd.DataFrame({"idx": 10_000_000 + np.arange(len(rows)), "y": (rng.random(len(rows)) < 0.25).astype(int), "regime": "manual", "row": rows, "dongle": groups, "uid": uid[rows], "t0": t[rows]}) | |
| def dry_run(keep_dir=None, log=print): | |
| import tempfile | |
| root = Path(keep_dir) if keep_dir else Path(tempfile.mkdtemp(prefix="dms_readout_dryrun_")) | |
| b = "synthetic" | |
| summ_root = root / "summ" / b | |
| for split, sd in (("train", 0), ("val", 1)): | |
| FU.synthetic_summary(b, "manual", split, summ_root / f"summ_manual_{split}.npz", n_routes=3, seed=sd) | |
| X, names, defined, uid, t, info = summary_features(b, "manual", "train", summ_root) | |
| with np.load(S12R / "targets_manual_train.npz", allow_pickle=True) as z: | |
| dongle = z["dongle"] | |
| tab = synthetic_label_table(uid, t, defined, dongle, n=240) | |
| t0 = time.time() | |
| out = fit(b, 0, models=("lr", "hgb"), label_table=tab, splits=("val",), regimes=("manual",), summ_root=summ_root, out_root=root / "runs", res_root=root / "readout", gold=False, log=log) | |
| m = LY.validate_layout(out["layouts"][0]) | |
| rep = {"root": str(root), "arm": out["arm"], "layout": out["layouts"][0], "selected": out["selected"]["selected"], "cv_auroc": out["selected"]["cv_auroc"], "n_features": out["selected"]["n_features"], | |
| "meta_keys": sorted(m), "wall_s": round(time.time() - t0, 1)} | |
| try: | |
| import results_schema as RS | |
| RS.validate(LY.schema_row(m)) | |
| rep["results_schema"] = "validate ok" | |
| except ImportError: | |
| rep["results_schema"] = "not importable" | |
| print(json.dumps(rep, indent=1, default=str)) | |
| return rep | |
| def main(argv=None): | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--baseline", default="") | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--ctx", action="store_true") | |
| ap.add_argument("--with-u0", action="store_true") | |
| ap.add_argument("--impute", default="", choices=["", "prior"]) | |
| ap.add_argument("--models", default="lr,hgb,mlp") | |
| ap.add_argument("--dry-run", action="store_true") | |
| ap.add_argument("--keep-dir", default="") | |
| ap.add_argument("--summ-root", default="", help="directory holding summ_{regime}_{split}.npz (default results/baselines_dms/summ/<b>); dry runs on partial caches") | |
| ap.add_argument("--out-root", default="", help="layout root (default results/runs); dry runs must point elsewhere") | |
| ap.add_argument("--res-root", default="", help="cv.csv / selected.json root (default results/baselines_dms/readout)") | |
| ap.add_argument("--apply-only", action="store_true", help="layer 3: re-apply the tier-A fitted read-out (no refit) to the summaries under --summ-root -> new arm <arm>--arm-suffix (see module doc)") | |
| ap.add_argument("--model-dir", default="", help="apply-only: readout dir of the tier-A fit (models.joblib / selected.json); default = newest results/baselines_dms/readout/<arm>_s<k>_* pointing at results/runs/<arm>_s<k>") | |
| ap.add_argument("--ref-summ-root", default="", help="apply-only: the tier-A summary dir (label-row features for the re-materialisation + reference cache sha256); default results/baselines_dms/summ/<b>") | |
| ap.add_argument("--ref-runs-root", default="", help="apply-only: root of the reference (tier-A) layouts (default results/runs)") | |
| ap.add_argument("--arm-suffix", default="_full", help="apply-only: suffix of the new arm / run dir (default _full -> abl_dms_<b>_<arm>_full_s<k>)") | |
| ap.add_argument("--equality", default="strict", choices=["strict", "report"], help="apply-only: assert (strict) or only log the bit-equality on unchanged-cache tier-A rows") | |
| ap.add_argument("--no-gold", action="store_true", help="apply-only: skip the informational clean-val temperature recomputation") | |
| a = ap.parse_args(argv) | |
| if a.dry_run: | |
| dry_run(a.keep_dir or None) | |
| return 0 | |
| assert a.baseline | |
| if a.apply_only: | |
| assert a.summ_root, "--apply-only needs --summ-root (the NEW full-exposure summary dir)" | |
| apply_only(a.baseline, a.seed, a.ctx, a.with_u0, a.model_dir or None, a.summ_root, a.ref_summ_root or None, out_root=(Path(a.out_root) if a.out_root else LY.RUNS), | |
| res_root=(Path(a.res_root) if a.res_root else READOUT_ROOT), ref_root=(Path(a.ref_runs_root) if a.ref_runs_root else LY.RUNS), arm_suffix=a.arm_suffix, | |
| equality=a.equality, gold=not a.no_gold) | |
| return 0 | |
| fit(a.baseline, a.seed, a.ctx, a.with_u0, a.impute or None, tuple(a.models.split(",")), summ_root=(a.summ_root or None), | |
| out_root=(Path(a.out_root) if a.out_root else LY.RUNS), res_root=(Path(a.res_root) if a.res_root else READOUT_ROOT)) | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 47.8 kB
- Xet hash:
- 6226033e47e3a38a7c770b0cdc9e70f5b95015e16c65598dbd3599abd341576a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.