Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /data /make_holdout_draw.py

AsianPlayer

16 days ago

download

raw

14.9 kB

"""cwm3/data/make_holdout_draw.py — the BATON-holdout draw (WP5, pre-registration row 2026-09-16T06:46:35Z 'WP5 BATON-holdout').
Population.  The 1,489 canonical leaves (results/baton_leaves.json) minus the 644 benchmark routes (uids of
s12_wm/results/targets_{manual,engaged}_{train,val,test}.npz) = 845 leaves on 208 devices; restricted to the 153 devices with NO
benchmark route (device-disjoint by construction).  Eligibility: cabin video >= 300 s on the frame/20 axis (ffprobe nb_frames / 20,
results/baton_full_corpus_durations.json), and the device's vehicle (carParams.carVin of the leaf's metadata.json; the 17-zero /
empty placeholder counts as unknown) must not be a VIN of any benchmark device (vehicle-disjoint; unknown VINs are kept).
Stratification.  Both control regimes must be represented: stratum `engaged` = canonical engaged time >= ENGAGED_MIN_S (120 s) on the
10 Hz CAN grid of s12_wm/canspec_v2.route_signals (cs_enabled OR cruiseState_enabled, 1 s debounce — the same flag that defines the
benchmark's engaged anchors and the hours table), stratum `manual_only` otherwise.  Covariates recorded per leaf: engaged / manual
seconds, moving-manual seconds, openpilot version (initData.version), car model, dongle, VIN (sha256 prefix only), remote sizes.
Draw.  seed 0 (numpy default_rng); devices in random order; first fill `engaged` with one random eligible leaf per device up to
N_PER_STRATUM (30), then `manual_only` from the remaining devices up to 30; shortfalls are filled from the other stratum with still
unused devices; only if fewer than N_TOTAL devices remain is a second leaf per device allowed (recorded).  One leaf per device.
Stages.  `--stage candidates` writes results/holdout/candidates_v1.json (+ candidates_need_csv.txt: eligible leaves whose
vehicle_dynamics.csv or metadata.json is not local — fetch them with data/fetch_baton_leaves.py --leaves-file ... --csv-only before the
draw, so that the stratification uses the exact CAN flag rather than a proxy).  `--stage draw` writes results/holdout/draw_v1.json
(leaf, uid, dongle, car, duration, stratum, covariates, sha256 of every draw input), holdout_leaves.txt (fetch list) and holdout_uids.txt.
Nothing is fetched, moved or deleted here.  The draw file is final once the decisions row is written; re-running `draw` refuses to
overwrite an existing draw_v1.json.
"""
import argparse, hashlib, json, os, sys, time
from pathlib import Path
import numpy as np

ROOT = Path(__file__).resolve().parents[1]; EXPLORE = ROOT.parent
sys.path.insert(0, str(ROOT / "common")); sys.path.insert(0, str(EXPLORE / "s12_wm"))
from runmeta import write_run_json, sha256_file  # noqa: E402
import canspec_v2 as cs  # noqa: E402

LROOT = Path("/home/USER/BATON"); S12R = EXPLORE / "s12_wm" / "results"
LEAVES = ROOT / "results" / "baton_leaves.json"; DUR = ROOT / "results" / "baton_full_corpus_durations.json"
OUT = ROOT / "results" / "holdout"
REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test")
MIN_DUR_S, ENGAGED_MIN_S, N_PER_STRATUM, N_TOTAL, SEED = 300.0, 120.0, 30, 60, 0
PLACEHOLDER_VINS = {"", "0" * 17, None}


def leaf_of_uid(u): p = u.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:])


def uid_of_leaf(l): return "__".join(l.split("/"))


def vin_of(meta):
    v = (meta.get("carParams") or {}).get("carVin")
    if v in PLACEHOLDER_VINS or not isinstance(v, str) or len(v.strip()) < 11: return None
    return v.strip()


def vin_key(v): return hashlib.sha256(v.encode()).hexdigest()[:12] if v else None


def read_meta(leaf):
    p = LROOT / leaf / "metadata.json"
    if not p.exists(): return None
    try: return json.load(open(p))
    except Exception: return {}


def benchmark_sets():
    uids, dongles = set(), set()
    for r in REGIMES:
        for s in SPLITS:
            z = np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True)
            uids |= set(np.unique(z["uid"]).tolist()); dongles |= set(np.unique(z["dongle"]).tolist())
    return uids, dongles


def regime_seconds(leaf):
    """canonical engaged / manual seconds on the 10 Hz canspec_v2 grid (None when the CSV is absent or unusable)."""
    sig = cs.route_signals(str(LROOT / leaf))
    if sig is None: return None
    eng = sig["engaged"]; v = sig["Y"][:, cs.TARGETS.index("vEgo")]
    return dict(engaged_s=float(eng.sum() / cs.FS), manual_s=float((~eng).sum() / cs.FS), manual_moving_s=float(((~eng) & (v > 0.5)).sum() / cs.FS),
                grid_s=float(len(eng) / cs.FS))


def stage_candidates():
    t0 = time.time(); OUT.mkdir(parents=True, exist_ok=True)
    lv = json.load(open(LEAVES))["leaves"]; dur = json.load(open(DUR)); b_uids, b_dongles = benchmark_sets()
    b_leaves = {leaf_of_uid(u) for u in b_uids}
    # benchmark VINs (local metadata of the 644 benchmark leaves)
    b_vins = {}; n_b_meta = 0
    for l in sorted(b_leaves):
        m = read_meta(l)
        if m is None: continue
        n_b_meta += 1; v = vin_of(m)
        if v: b_vins.setdefault(v, set()).add(l.split("/")[1])
    non_b = sorted(l for l in lv if l not in b_leaves)
    unseen_dev = {l.split("/")[1] for l in non_b} - b_dongles
    rows = []; need_csv = []; n_short = 0
    for l in non_b:
        car, dongle = l.split("/")[0], l.split("/")[1]
        if dongle not in unseen_dev: continue
        d = dur.get(l) or {}; nfr = d.get("nb_frames"); video_s = nfr / 20.0 if nfr else None
        if video_s is None or video_s < MIN_DUR_S: n_short += 1; continue
        m = read_meta(l); csv_ok = (LROOT / l / "vehicle_dynamics.csv").exists()
        if m is None or not csv_ok: need_csv.append(l)
        vin = vin_of(m) if m else None
        rows.append(dict(leaf=l, uid=uid_of_leaf(l), dongle=dongle, car=car, video_s=round(video_s, 2), n_frames=nfr,
                         dcamera_bytes=lv[l].get("dcamera_bytes"), qcamera_bytes=lv[l].get("qcamera_bytes"), meta_local=m is not None, csv_local=csv_ok,
                         vin_key=vin_key(vin), vin_known=vin is not None, vin_in_benchmark=bool(vin and vin in b_vins),
                         op_version=((m or {}).get("initData") or {}).get("version"), car_fingerprint=((m or {}).get("carParams") or {}).get("carFingerprint")))
    rec = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), n_canonical=len(lv), n_benchmark_leaves=len(b_leaves), n_benchmark_devices=len(b_dongles), n_benchmark_meta_local=n_b_meta,
               n_benchmark_vins=len(b_vins), n_non_benchmark_leaves=len(non_b), n_non_benchmark_devices=len({l.split("/")[1] for l in non_b}), n_unseen_devices=len(unseen_dev),
               n_unseen_leaves=sum(1 for l in non_b if l.split("/")[1] in unseen_dev), n_too_short=n_short, min_dur_s=MIN_DUR_S, n_candidates=len(rows),
               n_candidate_devices=len({r["dongle"] for r in rows}), n_need_csv=len(need_csv), n_vin_in_benchmark=sum(r["vin_in_benchmark"] for r in rows), candidates=rows)
    (OUT / "candidates_v1.json").write_text(json.dumps(rec, indent=1))
    (OUT / "candidates_need_csv.txt").write_text("".join(l + "\n" for l in need_csv))
    print(json.dumps({k: v for k, v in rec.items() if k != "candidates"}, indent=1))
    write_run_json(OUT, "holdout_candidates", [Path(__file__)], [LEAVES, DUR] + [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS], SEED, t0, extra=dict(summary={k: v for k, v in rec.items() if k != "candidates"}), input_listing=False)


def stage_draw(allow_proxy=False):
    t0 = time.time(); out = OUT / "draw_v1.json"
    if out.exists(): raise SystemExit(f"{out} exists — the draw is final (delete nothing; a new draw would be draw_v2 with its own decisions row)")
    cand = json.load(open(OUT / "candidates_v1.json")); rows = cand["candidates"]
    # exact regime seconds from the local CSV (fetched by --csv-only for the remote-only ones)
    missing = []
    for r in rows:
        r["csv_local"] = (LROOT / r["leaf"] / "vehicle_dynamics.csv").exists(); m = read_meta(r["leaf"])
        if m is not None:
            v = vin_of(m); r.update(meta_local=True, vin_key=vin_key(v), vin_known=v is not None, op_version=(m.get("initData") or {}).get("version"), car_fingerprint=(m.get("carParams") or {}).get("carFingerprint"))
        rs = regime_seconds(r["leaf"]) if r["csv_local"] else None
        if rs is None: missing.append(r["leaf"]); r.update(engaged_s=None, manual_s=None, manual_moving_s=None, grid_s=None, stratum="unknown")
        else: r.update(**rs, stratum="engaged" if rs["engaged_s"] >= ENGAGED_MIN_S else "manual_only")
    if missing and not allow_proxy: raise SystemExit(f"{len(missing)} candidates without a usable local vehicle_dynamics.csv — fetch them (--csv-only) or pass --allow-proxy: {missing[:5]}")
    # benchmark VINs again (the candidates file carries vin_in_benchmark from stage 1; recompute with the now-complete metadata)
    b_uids, _ = benchmark_sets(); b_vins = set()
    for l in {leaf_of_uid(u) for u in b_uids}:
        m = read_meta(l); v = vin_of(m) if m else None
        if v: b_vins.add(v)
    for r in rows:
        m = read_meta(r["leaf"]); v = vin_of(m) if m else None; r["vin_in_benchmark"] = bool(v and v in b_vins)
    excluded_vin = sorted({r["dongle"] for r in rows if r["vin_in_benchmark"]})
    elig = [r for r in rows if not r["vin_in_benchmark"] and r["stratum"] != "unknown"]
    proxy = [r for r in rows if not r["vin_in_benchmark"] and r["stratum"] == "unknown"]
    rng = np.random.default_rng(SEED)
    by_dev = {}
    for r in elig: by_dev.setdefault(r["dongle"], []).append(r)
    devices = sorted(by_dev); rng.shuffle(devices)
    for d in devices: by_dev[d].sort(key=lambda r: r["leaf"])
    chosen, used = [], set()

    def pick(stratum, n_want):
        for d in devices:
            if len([c for c in chosen if c["stratum"] == stratum]) >= n_want: break
            if d in used: continue
            opts = [r for r in by_dev[d] if r["stratum"] == stratum]
            if not opts: continue
            r = opts[int(rng.integers(len(opts)))]; chosen.append(dict(r, pick_rule=f"stratum_{stratum}")); used.add(d)
    pick("engaged", N_PER_STRATUM); pick("manual_only", N_PER_STRATUM)
    for stratum in ("engaged", "manual_only"):                                  # shortfall: other stratum, unused devices
        if len(chosen) < N_TOTAL: pick(stratum, N_TOTAL - sum(1 for c in chosen if c["stratum"] != stratum))
    second_leaf = 0
    if len(chosen) < N_TOTAL:                                                    # only if fewer than N_TOTAL devices are eligible
        pool = [r for d in devices for r in by_dev[d] if r["leaf"] not in {c["leaf"] for c in chosen}]; rng.shuffle(pool)
        for r in pool[:N_TOTAL - len(chosen)]: chosen.append(dict(r, pick_rule="second_leaf_same_device")); second_leaf += 1
    chosen.sort(key=lambda r: r["leaf"])
    inputs = [LEAVES, DUR, OUT / "candidates_v1.json"] + [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS]
    csv_sha = {r["leaf"]: sha256_file(LROOT / r["leaf"] / "vehicle_dynamics.csv") for r in rows if r["csv_local"]}
    meta_sha = {r["leaf"]: sha256_file(LROOT / r["leaf"] / "metadata.json") for r in rows if (LROOT / r["leaf"] / "metadata.json").exists()}
    h = hashlib.sha256(); [h.update(f"{k}\t{v}\n".encode()) for k, v in sorted(csv_sha.items())]; [h.update(f"{k}\t{v}\n".encode()) for k, v in sorted(meta_sha.items())]
    tot_d = sum((c["dcamera_bytes"] or 0) for c in chosen); tot_q = sum((c["qcamera_bytes"] or 0) for c in chosen)
    rec = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), seed=SEED, n_total=N_TOTAL, n_per_stratum=N_PER_STRATUM, engaged_min_s=ENGAGED_MIN_S, min_dur_s=MIN_DUR_S,
               stratum_rule="engaged: canonical engaged time (canspec_v2.route_signals: cs_enabled OR cruiseState_enabled, 1 s debounce, 10 Hz grid) >= 120 s; manual_only otherwise",
               draw_rule="seed-0 default_rng; devices shuffled; one random eligible leaf per device; fill engaged to 30, then manual_only to 30; shortfall from the other stratum (unused devices); second leaf per device only if < 60 devices eligible",
               vehicle_disjoint_rule="candidate devices whose non-placeholder carVin equals a VIN of any benchmark device are excluded (unknown / placeholder VINs kept)",
               n_candidates=len(rows), n_eligible=len(elig), n_eligible_devices=len(by_dev), n_proxy_unknown_stratum=len(proxy), excluded_devices_vin_in_benchmark=excluded_vin,
               n_candidates_engaged=sum(r["stratum"] == "engaged" for r in elig), n_candidates_manual_only=sum(r["stratum"] == "manual_only" for r in elig),
               n_chosen=len(chosen), n_chosen_devices=len({c["dongle"] for c in chosen}), n_chosen_engaged=sum(c["stratum"] == "engaged" for c in chosen), n_chosen_manual_only=sum(c["stratum"] == "manual_only" for c in chosen),
               second_leaf_same_device=second_leaf, n_cars=len({c["car"] for c in chosen}), video_hours=round(sum(c["video_s"] for c in chosen) / 3600, 2), engaged_hours=round(sum(c["engaged_s"] for c in chosen) / 3600, 2),
               manual_hours=round(sum(c["manual_s"] for c in chosen) / 3600, 2), remote_dcamera_GB=round(tot_d / 1e9, 2), remote_qcamera_GB=round(tot_q / 1e9, 2), remote_video_GB=round((tot_d + tot_q) / 1e9, 2),
               n_dcamera_local=sum((LROOT / c["leaf"] / "dcamera.mp4").exists() for c in chosen), s5_front_present=sum((EXPLORE / "s5_features" / "vjepa2g384" / "front" / f"{c['uid']}.npz").exists() for c in chosen),
               s5_cabin_present=sum((EXPLORE / "s5_features" / "vjepa2g384" / "cabin" / f"{c['uid']}.npz").exists() for c in chosen), frontfeat_present=sum((EXPLORE / "s12_wm" / "cache" / "frontfeat" / f"{c['uid']}.npz").exists() for c in chosen),
               op_versions={v: sum(1 for c in chosen if c["op_version"] == v) for v in sorted({str(c["op_version"]) for c in chosen})},
               inputs_sha256={str(p): sha256_file(p) for p in inputs}, candidate_csv_metadata_digest=h.hexdigest(), n_csv_hashed=len(csv_sha), n_metadata_hashed=len(meta_sha), leaves=chosen)
    out.write_text(json.dumps(rec, indent=1))
    (OUT / "holdout_leaves.txt").write_text("".join(c["leaf"] + "\n" for c in chosen))
    (OUT / "holdout_uids.txt").write_text("".join(c["uid"] + "\n" for c in chosen))
    print(json.dumps({k: v for k, v in rec.items() if k not in ("leaves", "inputs_sha256")}, indent=1))
    write_run_json(OUT, "holdout_draw", [Path(__file__)], inputs, SEED, t0, extra=dict(summary={k: v for k, v in rec.items() if k not in ("leaves", "inputs_sha256")}), input_listing=False)


if __name__ == "__main__":
    ap = argparse.ArgumentParser(); ap.add_argument("--stage", choices=["candidates", "draw"], required=True); ap.add_argument("--allow-proxy", action="store_true", help="draw even if some candidates have no local CAN csv (their stratum is 'unknown' and they are excluded; documented)")
    a = ap.parse_args()
    stage_candidates() if a.stage == "candidates" else stage_draw(a.allow_proxy)

Xet Storage Details

Size:: 14.9 kB
Xet hash:: 75c5e094dc4732b4e038546d1f56e91fa9789880edc589d401d5a6b39211e2bf

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""cwm3/data/make_holdout_draw.py — the BATON-holdout draw (WP5, pre-registration row 2026-09-16T06:46:35Z 'WP5 BATON-holdout').

	Population. The 1,489 canonical leaves (results/baton_leaves.json) minus the 644 benchmark routes (uids of
	s12_wm/results/targets_{manual,engaged}_{train,val,test}.npz) = 845 leaves on 208 devices; restricted to the 153 devices with NO
	benchmark route (device-disjoint by construction). Eligibility: cabin video >= 300 s on the frame/20 axis (ffprobe nb_frames / 20,
	results/baton_full_corpus_durations.json), and the device's vehicle (carParams.carVin of the leaf's metadata.json; the 17-zero /
	empty placeholder counts as unknown) must not be a VIN of any benchmark device (vehicle-disjoint; unknown VINs are kept).
	Stratification. Both control regimes must be represented: stratum `engaged` = canonical engaged time >= ENGAGED_MIN_S (120 s) on the
	10 Hz CAN grid of s12_wm/canspec_v2.route_signals (cs_enabled OR cruiseState_enabled, 1 s debounce — the same flag that defines the
	benchmark's engaged anchors and the hours table), stratum `manual_only` otherwise. Covariates recorded per leaf: engaged / manual
	seconds, moving-manual seconds, openpilot version (initData.version), car model, dongle, VIN (sha256 prefix only), remote sizes.
	Draw. seed 0 (numpy default_rng); devices in random order; first fill `engaged` with one random eligible leaf per device up to
	N_PER_STRATUM (30), then `manual_only` from the remaining devices up to 30; shortfalls are filled from the other stratum with still
	unused devices; only if fewer than N_TOTAL devices remain is a second leaf per device allowed (recorded). One leaf per device.
	Stages. `--stage candidates` writes results/holdout/candidates_v1.json (+ candidates_need_csv.txt: eligible leaves whose
	vehicle_dynamics.csv or metadata.json is not local — fetch them with data/fetch_baton_leaves.py --leaves-file ... --csv-only before the
	draw, so that the stratification uses the exact CAN flag rather than a proxy). `--stage draw` writes results/holdout/draw_v1.json
	(leaf, uid, dongle, car, duration, stratum, covariates, sha256 of every draw input), holdout_leaves.txt (fetch list) and holdout_uids.txt.
	Nothing is fetched, moved or deleted here. The draw file is final once the decisions row is written; re-running `draw` refuses to
	overwrite an existing draw_v1.json.
	"""
	import argparse, hashlib, json, os, sys, time
	from pathlib import Path
	import numpy as np

	ROOT = Path(__file__).resolve().parents[1]; EXPLORE = ROOT.parent
	sys.path.insert(0, str(ROOT / "common")); sys.path.insert(0, str(EXPLORE / "s12_wm"))
	from runmeta import write_run_json, sha256_file # noqa: E402
	import canspec_v2 as cs # noqa: E402

	LROOT = Path("/home/USER/BATON"); S12R = EXPLORE / "s12_wm" / "results"
	LEAVES = ROOT / "results" / "baton_leaves.json"; DUR = ROOT / "results" / "baton_full_corpus_durations.json"
	OUT = ROOT / "results" / "holdout"
	REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test")
	MIN_DUR_S, ENGAGED_MIN_S, N_PER_STRATUM, N_TOTAL, SEED = 300.0, 120.0, 30, 60, 0
	PLACEHOLDER_VINS = {"", "0" * 17, None}


	def leaf_of_uid(u): p = u.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:])


	def uid_of_leaf(l): return "__".join(l.split("/"))


	def vin_of(meta):
	v = (meta.get("carParams") or {}).get("carVin")
	if v in PLACEHOLDER_VINS or not isinstance(v, str) or len(v.strip()) < 11: return None
	return v.strip()


	def vin_key(v): return hashlib.sha256(v.encode()).hexdigest()[:12] if v else None


	def read_meta(leaf):
	p = LROOT / leaf / "metadata.json"
	if not p.exists(): return None
	try: return json.load(open(p))
	except Exception: return {}


	def benchmark_sets():
	uids, dongles = set(), set()
	for r in REGIMES:
	for s in SPLITS:
	z = np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True)
	uids \|= set(np.unique(z["uid"]).tolist()); dongles \|= set(np.unique(z["dongle"]).tolist())
	return uids, dongles


	def regime_seconds(leaf):
	"""canonical engaged / manual seconds on the 10 Hz canspec_v2 grid (None when the CSV is absent or unusable)."""
	sig = cs.route_signals(str(LROOT / leaf))
	if sig is None: return None
	eng = sig["engaged"]; v = sig["Y"][:, cs.TARGETS.index("vEgo")]
	return dict(engaged_s=float(eng.sum() / cs.FS), manual_s=float((~eng).sum() / cs.FS), manual_moving_s=float(((~eng) & (v > 0.5)).sum() / cs.FS),
	grid_s=float(len(eng) / cs.FS))


	def stage_candidates():
	t0 = time.time(); OUT.mkdir(parents=True, exist_ok=True)
	lv = json.load(open(LEAVES))["leaves"]; dur = json.load(open(DUR)); b_uids, b_dongles = benchmark_sets()
	b_leaves = {leaf_of_uid(u) for u in b_uids}
	# benchmark VINs (local metadata of the 644 benchmark leaves)
	b_vins = {}; n_b_meta = 0
	for l in sorted(b_leaves):
	m = read_meta(l)
	if m is None: continue
	n_b_meta += 1; v = vin_of(m)
	if v: b_vins.setdefault(v, set()).add(l.split("/")[1])
	non_b = sorted(l for l in lv if l not in b_leaves)
	unseen_dev = {l.split("/")[1] for l in non_b} - b_dongles
	rows = []; need_csv = []; n_short = 0
	for l in non_b:
	car, dongle = l.split("/")[0], l.split("/")[1]
	if dongle not in unseen_dev: continue
	d = dur.get(l) or {}; nfr = d.get("nb_frames"); video_s = nfr / 20.0 if nfr else None
	if video_s is None or video_s < MIN_DUR_S: n_short += 1; continue
	m = read_meta(l); csv_ok = (LROOT / l / "vehicle_dynamics.csv").exists()
	if m is None or not csv_ok: need_csv.append(l)
	vin = vin_of(m) if m else None
	rows.append(dict(leaf=l, uid=uid_of_leaf(l), dongle=dongle, car=car, video_s=round(video_s, 2), n_frames=nfr,
	dcamera_bytes=lv[l].get("dcamera_bytes"), qcamera_bytes=lv[l].get("qcamera_bytes"), meta_local=m is not None, csv_local=csv_ok,
	vin_key=vin_key(vin), vin_known=vin is not None, vin_in_benchmark=bool(vin and vin in b_vins),
	op_version=((m or {}).get("initData") or {}).get("version"), car_fingerprint=((m or {}).get("carParams") or {}).get("carFingerprint")))
	rec = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), n_canonical=len(lv), n_benchmark_leaves=len(b_leaves), n_benchmark_devices=len(b_dongles), n_benchmark_meta_local=n_b_meta,
	n_benchmark_vins=len(b_vins), n_non_benchmark_leaves=len(non_b), n_non_benchmark_devices=len({l.split("/")[1] for l in non_b}), n_unseen_devices=len(unseen_dev),
	n_unseen_leaves=sum(1 for l in non_b if l.split("/")[1] in unseen_dev), n_too_short=n_short, min_dur_s=MIN_DUR_S, n_candidates=len(rows),
	n_candidate_devices=len({r["dongle"] for r in rows}), n_need_csv=len(need_csv), n_vin_in_benchmark=sum(r["vin_in_benchmark"] for r in rows), candidates=rows)
	(OUT / "candidates_v1.json").write_text(json.dumps(rec, indent=1))
	(OUT / "candidates_need_csv.txt").write_text("".join(l + "\n" for l in need_csv))
	print(json.dumps({k: v for k, v in rec.items() if k != "candidates"}, indent=1))
	write_run_json(OUT, "holdout_candidates", [Path(__file__)], [LEAVES, DUR] + [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS], SEED, t0, extra=dict(summary={k: v for k, v in rec.items() if k != "candidates"}), input_listing=False)


	def stage_draw(allow_proxy=False):
	t0 = time.time(); out = OUT / "draw_v1.json"
	if out.exists(): raise SystemExit(f"{out} exists — the draw is final (delete nothing; a new draw would be draw_v2 with its own decisions row)")
	cand = json.load(open(OUT / "candidates_v1.json")); rows = cand["candidates"]
	# exact regime seconds from the local CSV (fetched by --csv-only for the remote-only ones)
	missing = []
	for r in rows:
	r["csv_local"] = (LROOT / r["leaf"] / "vehicle_dynamics.csv").exists(); m = read_meta(r["leaf"])
	if m is not None:
	v = vin_of(m); r.update(meta_local=True, vin_key=vin_key(v), vin_known=v is not None, op_version=(m.get("initData") or {}).get("version"), car_fingerprint=(m.get("carParams") or {}).get("carFingerprint"))
	rs = regime_seconds(r["leaf"]) if r["csv_local"] else None
	if rs is None: missing.append(r["leaf"]); r.update(engaged_s=None, manual_s=None, manual_moving_s=None, grid_s=None, stratum="unknown")
	else: r.update(**rs, stratum="engaged" if rs["engaged_s"] >= ENGAGED_MIN_S else "manual_only")
	if missing and not allow_proxy: raise SystemExit(f"{len(missing)} candidates without a usable local vehicle_dynamics.csv — fetch them (--csv-only) or pass --allow-proxy: {missing[:5]}")
	# benchmark VINs again (the candidates file carries vin_in_benchmark from stage 1; recompute with the now-complete metadata)
	b_uids, _ = benchmark_sets(); b_vins = set()
	for l in {leaf_of_uid(u) for u in b_uids}:
	m = read_meta(l); v = vin_of(m) if m else None
	if v: b_vins.add(v)
	for r in rows:
	m = read_meta(r["leaf"]); v = vin_of(m) if m else None; r["vin_in_benchmark"] = bool(v and v in b_vins)
	excluded_vin = sorted({r["dongle"] for r in rows if r["vin_in_benchmark"]})
	elig = [r for r in rows if not r["vin_in_benchmark"] and r["stratum"] != "unknown"]
	proxy = [r for r in rows if not r["vin_in_benchmark"] and r["stratum"] == "unknown"]
	rng = np.random.default_rng(SEED)
	by_dev = {}
	for r in elig: by_dev.setdefault(r["dongle"], []).append(r)
	devices = sorted(by_dev); rng.shuffle(devices)
	for d in devices: by_dev[d].sort(key=lambda r: r["leaf"])
	chosen, used = [], set()

	def pick(stratum, n_want):
	for d in devices:
	if len([c for c in chosen if c["stratum"] == stratum]) >= n_want: break
	if d in used: continue
	opts = [r for r in by_dev[d] if r["stratum"] == stratum]
	if not opts: continue
	r = opts[int(rng.integers(len(opts)))]; chosen.append(dict(r, pick_rule=f"stratum_{stratum}")); used.add(d)
	pick("engaged", N_PER_STRATUM); pick("manual_only", N_PER_STRATUM)
	for stratum in ("engaged", "manual_only"): # shortfall: other stratum, unused devices
	if len(chosen) < N_TOTAL: pick(stratum, N_TOTAL - sum(1 for c in chosen if c["stratum"] != stratum))
	second_leaf = 0
	if len(chosen) < N_TOTAL: # only if fewer than N_TOTAL devices are eligible
	pool = [r for d in devices for r in by_dev[d] if r["leaf"] not in {c["leaf"] for c in chosen}]; rng.shuffle(pool)
	for r in pool[:N_TOTAL - len(chosen)]: chosen.append(dict(r, pick_rule="second_leaf_same_device")); second_leaf += 1
	chosen.sort(key=lambda r: r["leaf"])
	inputs = [LEAVES, DUR, OUT / "candidates_v1.json"] + [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS]
	csv_sha = {r["leaf"]: sha256_file(LROOT / r["leaf"] / "vehicle_dynamics.csv") for r in rows if r["csv_local"]}
	meta_sha = {r["leaf"]: sha256_file(LROOT / r["leaf"] / "metadata.json") for r in rows if (LROOT / r["leaf"] / "metadata.json").exists()}
	h = hashlib.sha256(); [h.update(f"{k}\t{v}\n".encode()) for k, v in sorted(csv_sha.items())]; [h.update(f"{k}\t{v}\n".encode()) for k, v in sorted(meta_sha.items())]
	tot_d = sum((c["dcamera_bytes"] or 0) for c in chosen); tot_q = sum((c["qcamera_bytes"] or 0) for c in chosen)
	rec = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), seed=SEED, n_total=N_TOTAL, n_per_stratum=N_PER_STRATUM, engaged_min_s=ENGAGED_MIN_S, min_dur_s=MIN_DUR_S,
	stratum_rule="engaged: canonical engaged time (canspec_v2.route_signals: cs_enabled OR cruiseState_enabled, 1 s debounce, 10 Hz grid) >= 120 s; manual_only otherwise",
	draw_rule="seed-0 default_rng; devices shuffled; one random eligible leaf per device; fill engaged to 30, then manual_only to 30; shortfall from the other stratum (unused devices); second leaf per device only if < 60 devices eligible",
	vehicle_disjoint_rule="candidate devices whose non-placeholder carVin equals a VIN of any benchmark device are excluded (unknown / placeholder VINs kept)",
	n_candidates=len(rows), n_eligible=len(elig), n_eligible_devices=len(by_dev), n_proxy_unknown_stratum=len(proxy), excluded_devices_vin_in_benchmark=excluded_vin,
	n_candidates_engaged=sum(r["stratum"] == "engaged" for r in elig), n_candidates_manual_only=sum(r["stratum"] == "manual_only" for r in elig),
	n_chosen=len(chosen), n_chosen_devices=len({c["dongle"] for c in chosen}), n_chosen_engaged=sum(c["stratum"] == "engaged" for c in chosen), n_chosen_manual_only=sum(c["stratum"] == "manual_only" for c in chosen),
	second_leaf_same_device=second_leaf, n_cars=len({c["car"] for c in chosen}), video_hours=round(sum(c["video_s"] for c in chosen) / 3600, 2), engaged_hours=round(sum(c["engaged_s"] for c in chosen) / 3600, 2),
	manual_hours=round(sum(c["manual_s"] for c in chosen) / 3600, 2), remote_dcamera_GB=round(tot_d / 1e9, 2), remote_qcamera_GB=round(tot_q / 1e9, 2), remote_video_GB=round((tot_d + tot_q) / 1e9, 2),
	n_dcamera_local=sum((LROOT / c["leaf"] / "dcamera.mp4").exists() for c in chosen), s5_front_present=sum((EXPLORE / "s5_features" / "vjepa2g384" / "front" / f"{c['uid']}.npz").exists() for c in chosen),
	s5_cabin_present=sum((EXPLORE / "s5_features" / "vjepa2g384" / "cabin" / f"{c['uid']}.npz").exists() for c in chosen), frontfeat_present=sum((EXPLORE / "s12_wm" / "cache" / "frontfeat" / f"{c['uid']}.npz").exists() for c in chosen),
	op_versions={v: sum(1 for c in chosen if c["op_version"] == v) for v in sorted({str(c["op_version"]) for c in chosen})},
	inputs_sha256={str(p): sha256_file(p) for p in inputs}, candidate_csv_metadata_digest=h.hexdigest(), n_csv_hashed=len(csv_sha), n_metadata_hashed=len(meta_sha), leaves=chosen)
	out.write_text(json.dumps(rec, indent=1))
	(OUT / "holdout_leaves.txt").write_text("".join(c["leaf"] + "\n" for c in chosen))
	(OUT / "holdout_uids.txt").write_text("".join(c["uid"] + "\n" for c in chosen))
	print(json.dumps({k: v for k, v in rec.items() if k not in ("leaves", "inputs_sha256")}, indent=1))
	write_run_json(OUT, "holdout_draw", [Path(__file__)], inputs, SEED, t0, extra=dict(summary={k: v for k, v in rec.items() if k not in ("leaves", "inputs_sha256")}), input_listing=False)


	if __name__ == "__main__":
	ap = argparse.ArgumentParser(); ap.add_argument("--stage", choices=["candidates", "draw"], required=True); ap.add_argument("--allow-proxy", action="store_true", help="draw even if some candidates have no local CAN csv (their stratum is 'unknown' and they are excluded; documented)")
	a = ap.parse_args()
	stage_candidates() if a.stage == "candidates" else stage_draw(a.allow_proxy)