Buckets:
| """cwm3/tests/test_kill_report.py — eval/kill_report_cwm3.py renders one row per K1–K10 from synthetic CSVs with the expected verdict logic | |
| (kill_checks.beats on paired rows, the K3 −2 sd rule, the K5 twin rule, K10 vs legacy_att), honours the run-name / tag parameters, and never | |
| recomputes anything from predictions. CPU, milliseconds; --skip-pytest so the firewall test is not run recursively. | |
| Run: cd cwm3 && python -m pytest tests/test_kill_report.py -q | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(CWM3 / "eval")) | |
| import kill_report_cwm3 as KR # noqa: E402 | |
| SEEDS = [0, 1, 2] | |
| PCOLS = ["run_stamp", "tag", "regime", "split", "cell", "metric", "a", "b", "value_a", "ci_a_lo", "ci_a_hi", "value_b", "ci_b_lo", "ci_b_hi", | |
| "delta", "delta_ci_lo", "delta_ci_hi", "frac_boot_d_gt0", "n", "n_pos", "n_dev", "n_boot"] | |
| def _prow(tag, reg, cell, a, b, va, vb, lo, hi, stamp="20260911T000000Z", metric="pr_auc"): | |
| d = va - vb | |
| return dict(run_stamp=stamp, tag=tag, regime=reg, split="test", cell=cell, metric=metric, a=a, b=b, value_a=va, ci_a_lo=va - 0.02, ci_a_hi=va + 0.02, value_b=vb, | |
| ci_b_lo=vb - 0.02, ci_b_hi=vb + 0.02, delta=d, delta_ci_lo=d + lo, delta_ci_hi=d + hi, frac_boot_d_gt0=0.9, n=10000, n_pos=500, n_dev=30, n_boot=1000) | |
| def make_inputs(tmp, main="M_s{k}_final", beats_k4=True, veh_manual=0.36): | |
| rows = [] | |
| for k in SEEDS: | |
| m, ind = main.format(k=k), f"IND_s{k}" | |
| # main vs indep rows (vehicle + demand values live in value_a) | |
| for reg in ("manual", "engaged"): | |
| for ev, v in (("brake_on", veh_manual if reg == "manual" else 0.02), ("gas_on", 0.58 if reg == "manual" else 0.04), ("steer_phase_on", 0.78 if reg == "manual" else 0.25)): | |
| rows.append(_prow(f"FIN_s{k}_main_vs_indep", reg, f"veh_onset|{ev}|3s", m, ind, v + 0.001 * k, v - 0.02, -0.01, 0.05)) | |
| for fam, v in zip(KR.FAM, (0.55, 0.15, 0.17, 0.34, 0.12, 0.29)): | |
| rows.append(_prow(f"FIN_s{k}_main_vs_indep", reg, f"d_onset|{fam}|3s", m, ind, v, v - 0.01, -0.02, 0.02)) | |
| # K4 input effect + B1 | |
| for ev in KR.DRV: | |
| d = 0.15 if beats_k4 else -0.15 | |
| rows.append(_prow(f"K4X_s{k}_causal_vs_oldin", reg, f"drv_onset|{ev}|3s", f"U_s{k}", f"O_s{k}", 0.45 + 0.01 * k, 0.45 + 0.01 * k - d, -0.05, 0.05)) | |
| rows.append(_prow(f"B1X_s{k}_u_vs_k0", reg, f"drv_onset|{ev}|3s", f"U_s{k}", "k0", 0.45 + 0.01 * k, 0.30, -0.05, 0.05)) | |
| # P vs margin | |
| for fam, d in zip(KR.PFAM, (-0.2, 0.2, 0.1, 0.0, -0.1, 0.05)): | |
| rows.append(_prow(f"PX_s{k}_p_vs_margin", reg, f"p_onset|{fam}|3s", f"E_s{k}", "margin_rule_S0", 0.3 + d, 0.3, -0.04, 0.04)) | |
| paired = pd.DataFrame(rows, columns=PCOLS) | |
| # a stale earlier stamp for one cell (must be superseded by the latest row) | |
| stale = _prow("FIN_s0_main_vs_indep", "manual", "veh_onset|brake_on|3s", main.format(k=0), "IND_s0", 0.10, 0.30, -0.01, 0.01, stamp="20260910T000000Z") | |
| paired = pd.concat([pd.DataFrame([stale], columns=PCOLS), paired], ignore_index=True) | |
| twin = pd.DataFrame([dict(run_stamp="20260911T000000Z", tag=f"TW_s{k}", regime=reg, event=ev, statistic=st, a=f"A_s{k}", b=f"B_s{k}", value_a=va, value_b=vb, delta=va - vb, | |
| delta_ci_lo=va - vb - 0.01, delta_ci_hi=va - vb + 0.01, n=1000, n_pairs_or_pos=500, n_dev=20, note="") | |
| for k in SEEDS for reg in ("manual", "engaged") for ev in ("hand_leave", "head_turn_onset") | |
| for st, va, vb in (("twin_pair_auroc@3s", 0.70 + 0.01 * k, 0.68), ("random_pair_auroc@3s", 0.85, 0.85))] | |
| + [dict(run_stamp="20260911T000000Z", tag=f"TW_s{k}", regime="manual", event="all", statistic="identity_probe_acc", a=f"A_s{k}", b=f"B_s{k}", value_a=0.48, value_b=0.47, | |
| delta=0.01, delta_ci_lo=np.nan, delta_ci_hi=np.nan, n=1000, n_pairs_or_pos=90, n_dev=90, note="") for k in SEEDS]) | |
| wg = [] | |
| for k in SEEDS: | |
| base = dict(run_stamp="20260911T000000Z", item="itemX", arm="cwm3_W", seed=k, regime=np.nan, split="test", stratum="clean_test", endpoint="gold_W", event_or_family="W_col", tau_s=3.0, | |
| metric="auroc", value=0.71, status="secondary", notes="") | |
| wg.append(dict(base, comparator=np.nan, delta=np.nan, delta_ci_lo=np.nan, delta_ci_hi=np.nan)) | |
| wg.append(dict(base, comparator="legacy_att", delta=0.08, delta_ci_lo=0.01 if k < 2 else -0.02, delta_ci_hi=0.17)) | |
| wg.append(dict(base, arm="legacy_att", event_or_family="att@3s", value=0.63, comparator=np.nan, delta=np.nan, delta_ci_lo=np.nan, delta_ci_hi=np.nan)) | |
| wgold = pd.DataFrame(wg) | |
| ref = pd.DataFrame([dict(item="grid4_v2", arm="can_only_direct", endpoint="veh_onset", event_or_family=ev, tau_s=3.0, metric="pr_auc", regime=reg, seed=k, value=v + 0.005 * k, status="sanity", comparator=np.nan) | |
| for k in SEEDS for reg, vals in (("manual", (0.33, 0.57, 0.75)), ("engaged", (0.01, 0.035, 0.18))) for ev, v in zip(KR.VEH, vals)] | |
| + [dict(item="grid4_v2", arm="current_signal_S0", endpoint="d_onset", event_or_family=fam, tau_s=3.0, metric="pr_auc", regime=reg, seed=0, value=v, status="sanity", comparator=np.nan) | |
| for reg in ("manual", "engaged") for fam, v in zip(KR.FAM, (0.34, 0.31, 0.37, 0.13, 0.29, 0.25))]) | |
| paths = {} | |
| for name, df in (("paired", paired), ("twin", twin), ("wgold", wgold), ("ref", ref)): | |
| p = tmp / f"{name}.csv" | |
| df.to_csv(p, index=False) | |
| paths[name] = str(p) | |
| return paths | |
| def run(tmp, paths, main="M_s{k}_final", extra=()): | |
| out = tmp / "kill_report.md" | |
| argv = ["--label", "unit", "--main-run-prefix", main, "--indep-run-prefix", "IND_s{k}", "--k4-input-tag", "K4X_s{k}_causal_vs_oldin", "--b1-tag", "B1X_s{k}_u_vs_k0", | |
| "--twin-tag", "TW_s{k}", "--p-margin-tag", "PX_s{k}_p_vs_margin", "--warning-item", "itemX", "--warning-column", "W_col", "--paired", paths["paired"], "--twin", paths["twin"], | |
| "--wgold", paths["wgold"], "--ref", paths["ref"], "--out", str(out), "--json-dir", str(tmp / "json"), "--skip-pytest"] + list(extra) | |
| rep = KR.main(argv) | |
| return rep, out.read_text() | |
| def test_renders_ten_rows_and_parametrised_names(tmp_path): | |
| paths = make_inputs(tmp_path) | |
| rep, md = run(tmp_path, paths) | |
| rows = [l for l in md.splitlines() if re.match(r"\| K\d+ \|", l)] | |
| assert len(rows) == 10 and [r.split("|")[1].strip() for r in rows] == [f"K{i}" for i in range(1, 11)] | |
| assert "M_s{k}_final" in md and "K4X_s{k}_causal_vs_oldin" in md and "itemX" in md | |
| assert rep["checks"]["K1"]["verdict"] == "NOT_CLAIMED" and rep["checks"]["K2"]["verdict"] == "NOT_CLAIMED" | |
| assert rep["checks"]["K8"]["verdict"] == "NOT_RUN" | |
| assert Path(rep["json"]).exists() and json.loads(Path(rep["json"]).read_text())["label"] == "unit" | |
| def test_k3_rule_uses_latest_rows_and_two_sd(tmp_path): | |
| paths = make_inputs(tmp_path, veh_manual=0.36) | |
| rep, _ = run(tmp_path, paths) | |
| k3 = rep["checks"]["K3"] | |
| assert k3["verdict"] == "PASS" | |
| c = k3["cells"]["manual|brake_on"] | |
| assert abs(c["main"][0] - 0.36) < 1e-9, "the stale 0.10 row must be superseded by the latest stamp" | |
| assert c["verdict"] == "OK" and abs(c["ref_mean"] - 0.335) < 1e-6 | |
| # a model far below CAN-only fails | |
| paths2 = make_inputs(tmp_path / "b", veh_manual=0.10) if (tmp_path / "b").mkdir() is None else None | |
| rep2, _ = run(tmp_path / "b", paths2) | |
| assert rep2["checks"]["K3"]["verdict"] == "FAIL" and rep2["checks"]["K3"]["cells"]["manual|brake_on"]["verdict"] == "BELOW" | |
| def test_k4_k5_k7_k10_verdicts_via_beats(tmp_path): | |
| paths = make_inputs(tmp_path) | |
| rep, md = run(tmp_path, paths) | |
| C = rep["checks"] | |
| assert all(v["verdict"] == "BEATS" for v in C["K4"]["input_effect"].values()) and C["K4"]["k4_verdict"] == "PASS" and C["K4"]["verdict"] == "PASS" | |
| assert all(v["verdict"] == "BEATS" for v in C["K4"]["b1_vs_k0"].values()) and C["K4"]["b1_verdict"].startswith("6/6") | |
| assert C["K5"]["verdict"] == "PASS" and all(v["verdict"] == "BEATS" for v in C["K5"]["cells"].values()) | |
| assert abs(C["K5"]["cells"]["manual|hand_leave"]["random_minus_twin_a"] - (0.85 - 0.71)) < 1e-6 | |
| assert C["K7"]["p_vs_margin"]["manual|ttc_drop"]["verdict"] == "BEATS" and C["K7"]["p_vs_margin"]["manual|ttc"]["verdict"] == "WORSE" | |
| assert "manual|ttc" in C["K7"]["p_worse"] and C["K7"]["verdict"] == "FLAG" | |
| assert C["K7"]["d_vs_S0"]["manual|ttc"]["verdict"] == "BELOW" and C["K7"]["d_vs_S0"]["manual|steer_rate"]["verdict"] == "ABOVE" | |
| assert C["K10"]["verdict"] == "BEATS" and C["K10"]["n_seeds"] == 3 and "outperforms" in C["K10"]["wording"] | |
| assert "**BEATS**" in md | |
| def test_k4_worse_flags_and_missing_tags_insufficient(tmp_path): | |
| paths = make_inputs(tmp_path, beats_k4=False) | |
| rep, _ = run(tmp_path, paths) | |
| assert rep["checks"]["K4"]["k4_verdict"] == "FLAG" and rep["checks"]["K4"]["verdict"] == "FLAG" | |
| rep2, md2 = run(tmp_path, paths, extra=["--twin-tag", "NOPE_s{k}", "--warning-item", "nope"]) | |
| assert rep2["checks"]["K5"]["verdict"] == "INSUFFICIENT" and rep2["checks"]["K10"]["verdict"] == "INSUFFICIENT" | |
| assert "**INSUFFICIENT**" in md2 | |
| def test_k6_k9_missing_runs_are_insufficient(tmp_path): | |
| paths = make_inputs(tmp_path) | |
| rep, _ = run(tmp_path, paths) | |
| assert rep["checks"]["K6"]["verdict"] == "INSUFFICIENT" and rep["checks"]["K9"]["verdict"] == "INSUFFICIENT" | |
| assert all(v.get("status") == "missing" for v in rep["checks"]["K9"]["runs"].values()) | |
| def test_beats_rows_matches_kill_checks_definition(): | |
| g = pd.DataFrame({"seed": [0, 1, 2], "delta": [0.05, 0.06, 0.04], "delta_ci_lo": [0.01, 0.02, -0.01], "delta_ci_hi": [0.09, 0.10, 0.09], "value_a": [0.5, 0.51, 0.49], "value_b": [0.45, 0.45, 0.45]}) | |
| assert KR.beats_rows(g)["verdict"] == "BEATS" | |
| g2 = g.assign(delta=[0.005, 0.004, -0.002], delta_ci_lo=[-0.01, -0.01, -0.02], delta_ci_hi=[0.02, 0.02, 0.01]) | |
| assert KR.beats_rows(g2)["verdict"] == "NOT_WORSE" | |
| assert KR.beats_rows(g.iloc[:2])["verdict"] == "INSUFFICIENT" | |
Xet Storage Details
- Size:
- 10.4 kB
- Xet hash:
- a752443c7de4d36628b34e6eae66d5fb06a8d59e1679d7dcdcf2d91826e01761
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.