Buckets:
| """cwm3/tests/test_pred_layout_contract.py — the joint layout writer obeys the cwm2 pred_layout contract and its closed-form composition | |
| agrees with a Monte-Carlo path composition (the independence composition cwm2's pred_adapter used). | |
| (a) chain algebra (numpy, synthetic hazards, 1,000 rows x 8,000 sampled paths): forward_marginal, first_onset_law, no_event_prob (on / off, | |
| from state 0 / 1), onset_within, the C1 composition and the C2 away-and-return composition each agree with path frequencies within the Monte-Carlo noise (mean |diff| < 0.008-0.012). | |
| (b) if a joint run with a pred file exists on disk, write its layout to a temp dir and check: every EVENT_NAMES / EP_NAMES / w column present, | |
| shapes, dtypes f16, uid/t witness, onset_p monotone in tau, values in [0, 1], V blocks NaN exactly where valid_V is False, meta keys the | |
| evaluators read (arm, regime, seed, split, score_source, roi, dms_fields_used == []); pred_layout.load_pred round-trips. | |
| Run: cd cwm3 && python -m pytest tests/test_pred_layout_contract.py -q | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import numpy as np # noqa: E402 | |
| import pytest # noqa: E402 | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| CWM2 = CWM3.parent / "cwm2" | |
| for p in (CWM3 / "eval", CWM2 / "eval"): | |
| sys.path.insert(0, str(p)) | |
| import write_pred_layout_joint as W # noqa: E402 | |
| from pred_layout import EVENT_NAMES, EP_NAMES, load_pred # noqa: E402 | |
| def _sample_paths(rng, h_on, h_off, z0, S): | |
| N, T = h_on.shape | |
| z = np.zeros((S, N, T), bool) | |
| prev = np.broadcast_to(z0.astype(bool), (S, N)).copy() | |
| for k in range(T): | |
| u = rng.random((S, N)) | |
| nxt = np.where(prev, u >= h_off[:, k], u < h_on[:, k]) | |
| z[:, :, k] = nxt | |
| prev = nxt | |
| return z | |
| def test_chain_algebra_matches_paths(): | |
| rng = np.random.default_rng(0) | |
| N, T, S = 1000, 25, 8000 | |
| h_on = rng.uniform(0.01, 0.3, (N, T)) | |
| h_off = rng.uniform(0.01, 0.4, (N, T)) | |
| z0 = (rng.random(N) < 0.4).astype(np.float64) | |
| z = _sample_paths(rng, h_on, h_off, z0, S) | |
| pi = W.forward_marginal(h_on, h_off, z0) | |
| assert np.abs(pi - z.mean(0)).max() < 0.05 and np.abs(pi - z.mean(0)).mean() < 0.008 # S = 8000 paths: sd ≈ 0.0056 per cell | |
| prev = np.concatenate([np.broadcast_to(z0.astype(bool), (S, N))[:, :, None], z[:, :, :-1]], 2) | |
| onset, offset = z & ~prev, ~z & prev | |
| # first-onset law from z0 = 0 | |
| rows0 = z0 == 0 | |
| first = np.where(onset.any(2), onset.argmax(2), -1) | |
| q = W.first_onset_law(h_on) | |
| for k in (0, 3, 10): | |
| emp = (first[:, rows0] == k).mean(0) | |
| assert np.abs(q[rows0, k] - emp).mean() < 0.008 | |
| # no_event_prob from the marginal state at k (on / off) | |
| for kind, ev in (("on", onset), ("off", offset)): | |
| for k in (0, 7, 14): | |
| emp = 1 - ev[:, :, k + 1:k + 11].any(2).mean(0) | |
| model = W.no_event_prob(h_on, h_off, k, 10, 1 - pi[:, k], pi[:, k], kind) | |
| assert np.abs(model - emp).mean() < 0.008, (kind, k) | |
| # onset_within from z0 | |
| emp = onset[:, :, :15].any(2).mean(0) | |
| assert np.abs(W.onset_within(h_on, h_off, z0, 15) - emp).mean() < 0.008 | |
| # C1-style composition: independent hands chain (chain A) vs vehicle chain (chain B) | |
| hA_on, hA_off = rng.uniform(0.01, 0.2, (N, T)), rng.uniform(0.01, 0.3, (N, T)) | |
| zA = _sample_paths(rng, hA_on, hA_off, np.zeros(N), S) | |
| onA = zA & ~np.concatenate([np.zeros((S, N, 1), bool), zA[:, :, :-1]], 2) | |
| K1 = np.where(onA.any(2), onA.argmax(2), -1) | |
| cond = (K1 >= 0) & (K1 <= 14) | |
| Kc = np.clip(K1, 0, 24) | |
| cs = np.concatenate([np.zeros((S, N, 1), np.int32), np.cumsum(onset, 2, dtype=np.int32)], 2) | |
| hit = (np.take_along_axis(cs, np.minimum(Kc + 11, 25)[..., None], 2) - np.take_along_axis(cs, (Kc + 1)[..., None], 2))[..., 0] > 0 | |
| emp = (cond & hit).sum(0) / np.maximum(cond.sum(0), 1) | |
| qA = W.first_onset_law(hA_on)[:, :15] | |
| win = np.stack([1 - W.no_event_prob(h_on, h_off, k, 10, 1 - pi[:, k], pi[:, k], "on") for k in range(15)], 1) | |
| c1 = (qA * win).sum(1) / qA.sum(1) | |
| ok = cond.sum(0) > 200 | |
| assert np.abs(c1[ok] - emp[ok]).mean() < 0.012 | |
| # C2-style away-and-return with two independent driver chains at a fixed k | |
| k = 5 | |
| hands, head = z, zA # hands_on := chain z (state 1 = on); head_turned := chain A | |
| a_h_emp = (~hands[:, :, k]).mean(0) | |
| ret_h = onset[:, :, k + 1:k + 11].any(2) # hand_return = hands 0->1 | |
| off_A = ~zA & np.concatenate([np.zeros((S, N, 1), bool), zA[:, :, :-1]], 2) | |
| ret_d = off_A[:, :, k + 1:k + 11].any(2) # head_return = head 1->0 | |
| away = (~hands[:, :, k]) | zA[:, :, k] | |
| emp_ar = (away & (ret_h | ret_d)).mean(0) | |
| piA = W.forward_marginal(hA_on, hA_off, np.zeros(N)) | |
| one, zero = np.ones(N), np.zeros(N) | |
| ah, ad = 1 - pi[:, k], piA[:, k] | |
| r_h0 = 1 - W.no_event_prob(h_on, h_off, k, 10, one, zero, "on") | |
| r_h1 = 1 - W.no_event_prob(h_on, h_off, k, 10, zero, one, "on") | |
| r_d0 = 1 - W.no_event_prob(hA_on, hA_off, k, 10, zero, one, "off") | |
| r_d1 = 1 - W.no_event_prob(hA_on, hA_off, k, 10, one, zero, "off") | |
| model_ar = ah * ad * (1 - (1 - r_h0) * (1 - r_d0)) + ah * (1 - ad) * (1 - (1 - r_h0) * (1 - r_d1)) + (1 - ah) * ad * (1 - (1 - r_h1) * (1 - r_d0)) | |
| assert np.abs(ah - a_h_emp).mean() < 0.008 | |
| assert np.abs(model_ar - emp_ar).mean() < 0.01 | |
| import re as _re | |
| def _find_pred(): | |
| for rd in sorted((CWM3 / "results" / "runs").glob("I1_*_s0")) + sorted((CWM3 / "results" / "runs").glob("F1_*_s0")): | |
| for p in sorted(rd.glob("pred_*_val.npz")): | |
| if _re.fullmatch(r"pred_(manual|engaged)_val\.npz", p.name) and (rd / "PRED_DONE").exists(): | |
| return rd, p | |
| return None, None | |
| def test_layout_contract_on_disk(tmp_path): | |
| rd, pf = _find_pred() | |
| regime, split = pf.stem.split("_")[1], pf.stem.split("_")[2] | |
| import shutil | |
| shutil.copy(pf, tmp_path / pf.name) | |
| outp = W.write_layout(tmp_path, regime, split, {"stage": "I1"}, [Path(W.__file__)], log=lambda *_: None) | |
| z = np.load(outp, allow_pickle=True) | |
| meta = json.loads(str(z["meta"])) | |
| for k in ("arm", "regime", "seed", "split", "score_source", "roi", "dms_fields_used", "model_id", "code_sha256", "inputs_digest"): | |
| assert k in meta | |
| assert meta["dms_fields_used"] == [] and meta["arm"] in ("cwm3_I", "cwm3_F") and meta["score_source"] == "independent_marginals" | |
| assert list(map(str, z["event_names"])) == list(EVENT_NAMES) and list(map(str, z["ep_names"])) == list(EP_NAMES) | |
| N = len(z["uid"]) | |
| assert z["onset_p"].shape == (N, 4, len(EVENT_NAMES)) and z["h_on"].shape == (N, 25, len(EVENT_NAMES)) and z["onset_p"].dtype == np.float16 | |
| assert z["ep_p"].shape == (N, len(EP_NAMES)) and z["w"].shape[0] == N and len(z["w_names"]) == z["w"].shape[1] | |
| op = z["onset_p"].astype(np.float32) | |
| fin = np.isfinite(op) | |
| assert (op[fin] >= 0).all() and (op[fin] <= 1).all() | |
| assert np.all(np.diff(np.nan_to_num(op, nan=0.0), axis=1) >= -1e-3), "onset_p must be monotone in tau" | |
| pv = np.load(pf, allow_pickle=True) | |
| vV = pv["valid_V"].astype(bool) | |
| e = list(EVENT_NAMES).index("brake_on") | |
| assert np.isnan(op[~vV, :, e]).all() and np.isfinite(op[vV, :, e]).all() | |
| e = list(EVENT_NAMES).index("hand_leave") | |
| assert np.isfinite(op[:, :, e]).all() | |
| wn = list(map(str, z["w_names"])) | |
| for req in ("W_direct_t15_w10_RV_6fam", "driver_only_t15", "demand_only_t15_6fam"): | |
| assert req in wn | |
| ep = z["ep_p"].astype(np.float32) | |
| j = list(EP_NAMES).index("C2_ttc") | |
| assert np.isfinite(ep[:, j]).all() and z["ep_mask"][:, j].all() | |
| j = list(EP_NAMES).index("C1") | |
| assert np.isfinite(ep[vV, j]).all() and np.isnan(ep[~vV, j]).all() | |
| tg = np.load(CWM3.parent / "s12_wm" / "results" / f"targets_{regime}_{split}.npz", allow_pickle=True) | |
| pr = load_pred(outp, tg) | |
| assert pr["meta"]["regime"] == regime | |
Xet Storage Details
- Size:
- 8.26 kB
- Xet hash:
- 1f52a44192404f79e755a4ad213295b1d219c8e3c90549357d69bee1058802d1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.