Buckets:
| #!/usr/bin/env python3 | |
| """tools/vlm_handson/rank_candidates.py — VLM-ranked candidate ENRICHMENT of the gold-event page (pre-registration docs/decisions.md | |
| 2026-09-17T01:05:57Z "VLM-ranked candidate ENRICHMENT"). Ordering only; the evaluation and the gold-set definition are untouched. | |
| (1) Pool = ALL v3 ENGAGED TEST first-step hand_return candidates with local video, excluding +-5 s of every previously reviewed / validation zone | |
| (build_gold_engaged.excluded_intervals — the same 200 zones the gold page used), NO per-route cap. For every pool candidate (and every | |
| owner-labelled v3-page / pass-4 item, for the ranking validation) six CLEAN frames at dt = -2.0, -1.0, -0.5, 0.0, +0.5, +1.0 s are read with the | |
| review tool's helpers (build_relabel_v3.read_frames on frame_idx = round(20 t); JPEG q95 round trip like build_validation_frames.py; no | |
| overlay), cropped exactly as the validation run's 'crop' variant (score_vlm_handson.crop_box, long side <= 896) and scored with the | |
| validation scorer (score_vlm_handson.Scorer: Qwen/Qwen3-VL-2B-Instruct, same prompt, P(on) = yes-vs-no first-token softmax). | |
| Frames stay in memory (3 extraction threads -> bounded queue -> GPU batches of 8); nothing but the json outputs is written. | |
| deltaP = mean P(on | dt >= 0) - mean P(on | dt < 0). | |
| (2) Validation of the ranking on owner-labelled items (results/review_tool/labels_latest.json at run time): | |
| gold page G_* hand_return items (at_event / ret_dt=* = true return; never / always = non-return; unsure excluded), v3 page V3_* (ret_dt = true; | |
| always_on / never_on = false; engaged and manual reported separately), pass-4 <orig>__r4 (never_on / always_on = false) -> item-level AUROC of | |
| deltaP (true vs non-return) with a bootstrap CI, yield among the top-K of the pool by deltaP (K = 50, 100, 150) on the labelled subset. | |
| (3) rank_top150.json = the top 150 fresh candidates (not on the page, not within +-5 s of any page item or excluded zone) by deltaP, with the | |
| gold-page item meta, for build_gold_engaged.py --append-ranked. | |
| Outputs (new files only): results/vlm_handson/rank_scores.json, rank_validation.json, rank_top150.json. | |
| python tools/vlm_handson/rank_candidates.py [--cand qwen3vl2b] [--batch 8] [--workers 3] [--top 150] [--limit N] | |
| python tools/vlm_handson/rank_candidates.py --validate-only # re-run (2)+(3) from rank_scores.json (no GPU) | |
| """ | |
| from __future__ import annotations | |
| import os | |
| os.environ.setdefault("OMP_NUM_THREADS", "4"); os.environ.setdefault("MKL_NUM_THREADS", "4"); os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") | |
| import argparse, hashlib, io, json, queue, sys, threading, time | |
| from concurrent.futures import ThreadPoolExecutor | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[2] | |
| sys.path.insert(0, str(CWM3 / "tools/review_tool")); sys.path.insert(0, str(CWM3 / "tools/vlm_handson")) | |
| import build_review_tool as B # noqa: E402 (VIO, FPS, nearest_pose_row) | |
| import build_relabel_v3 as V3B # noqa: E402 (load_cell, read_frames) | |
| import build_gold_engaged as G # noqa: E402 (excluded_intervals, hits, STEP_S, EXCL_S) | |
| import score_vlm_handson as S # noqa: E402 (Scorer, crop_box, LONG_MAX, FULL_W, gpu_gate, CANDS) | |
| OUT = CWM3 / "results/vlm_handson" | |
| RT = CWM3 / "results/review_tool" | |
| ITEMS_GOLD = RT / "items_gold_engaged.json"; ITEMS_V3 = RT / "items_relabel_v3.json"; ITEMS_R4 = RT / "items_relabel_hr_engaged_r4.json"; LABELS = RT / "labels_latest.json" | |
| DTS = [-2.0, -1.0, -0.5, 0.0, 0.5, 1.0] | |
| KS = (50, 100, 150) | |
| TRUE_G = ("at_event",); FALSE_G = ("never", "always"); FALSE_V = ("never_on", "always_on") | |
| # ------------------------------------------------------------------------------------------------------------- items to score | |
| def pool_items(iv, log): | |
| """all engaged test v3 hand_return first-step candidates, local video, outside every excluded zone; no per-route cap.""" | |
| z = V3B.load_cell("engaged"); e = z["event_names"].index("hand_return") | |
| rows = np.where(z["ok"] & z["E1"][:, e] & z["EM1"][:, e])[0] | |
| uids = z["uid"][rows]; te = z["t"][rows] + G.STEP_S | |
| vid = np.array([B.VIO.has_video(u) for u in uids], bool) | |
| excl = np.zeros(len(rows), bool) | |
| for u, lo, hi, _ in iv: | |
| excl |= (uids == u) & (te >= lo) & (te <= hi) | |
| keep = vid & ~excl | |
| st = {"n_rows": int(len(rows)), "n_local_video": int(vid.sum()), "n_excluded": int((vid & excl).sum()), "n_pool": int(keep.sum()), | |
| "n_pool_routes": int(len(np.unique(uids[keep]))), "per_route_cap": None} | |
| log(f"pool: rows {st['n_rows']} local-video {st['n_local_video']} excluded(+-{G.EXCL_S:.0f}s of {len(iv)} reviewed/validation zones) {st['n_excluded']} -> pool {st['n_pool']} on {st['n_pool_routes']} routes (no per-route cap)") | |
| items = [] | |
| for r in rows[keep]: | |
| u, ta = str(z["uid"][r]), float(z["t"][r]); t_ev = round(ta + G.STEP_S, 6) | |
| items.append({"id": f"POOL_{int(r)}", "source": "pool", "uid": u, "car": u.split("__")[0], "regime": "engaged", "event": "hand_return", "t_anchor": ta, "t_event": t_ev, | |
| "row": int(r), "state0": [float(v) for v in z["S0"][r]], "state_names": z["state_names"], "dwheel0_c": float(z["d0"][r]) if np.isfinite(z["d0"][r]) else None, | |
| "wheel_ref_defined": bool(z["wd"][r]), "label_version": "v3"}) | |
| return items, st | |
| def labelled_validation_items(labels, log): | |
| """owner-labelled v3-page (V3_*) and pass-4 (<orig>__r4) items (label not '' / unsure) -> score items with their own ids.""" | |
| items = [] | |
| for src, path, pfx in (("v3", ITEMS_V3, "V3_"), ("pass4", ITEMS_R4, None)): | |
| for it in json.load(open(path, encoding="utf-8"))["items"]: | |
| if pfx and not it["id"].startswith(pfx): continue | |
| lab = (labels.get(it["id"]) or {}).get("label", "") or "" | |
| if lab in ("", "unsure"): continue | |
| m = it["meta"] | |
| items.append({"id": it["id"], "source": src, "uid": m["uid"], "car": m["car"], "regime": m["regime"], "event": m["event"], "t_anchor": float(m["t_anchor"]), "t_event": float(m["t_event"]), | |
| "row": int(m["row"]) if "row" in m else None, "owner_label": lab}) | |
| log(f"labelled validation items: v3 {sum(i['source'] == 'v3' for i in items)} (engaged {sum(i['source'] == 'v3' and i['regime'] == 'engaged' for i in items)}), pass4 {sum(i['source'] == 'pass4' for i in items)}") | |
| return items | |
| def page_index(): | |
| """gold page: {(uid, round(t_event, 3)): id} for event items and [(uid, lo, hi, id)] +-5 s zones of every page item (events and windows).""" | |
| P = json.load(open(ITEMS_GOLD, encoding="utf-8")); key, zones = {}, [] | |
| for it in P["items"]: | |
| m = it["meta"] | |
| if m["event"] == "recall_window": zones.append((m["uid"], m["t0"] - G.EXCL_S, m["t1"] + G.EXCL_S, it["id"])) | |
| else: key[(m["uid"], round(float(m["t_event"]), 3))] = it["id"]; zones.append((m["uid"], m["t_event"] - G.EXCL_S, m["t_event"] + G.EXCL_S, it["id"])) | |
| return P, key, zones | |
| # ------------------------------------------------------------------------------------------------------------- extraction (threads) | |
| def driver_side(boxes): | |
| W = boxes["native_wh"][0]; fin = boxes["box"][np.all(np.isfinite(boxes["box"]), 1)] | |
| cx = float(np.median((fin[:, 0] + fin[:, 2]) / 2) / W) if len(fin) else float("nan") | |
| return ("right" if cx >= 0.5 else "left"), cx | |
| def crop_like_scorer(im, box, side): | |
| """score_vlm_handson.make_input('crop') on an in-memory PIL image: causal box expanded toward the wheel, long side <= 896; full-frame 896 fallback.""" | |
| from PIL import Image | |
| W, H = im.size | |
| if box is None: | |
| return im.resize((S.FULL_W, int(round(H * S.FULL_W / W))), Image.BICUBIC), "full_fallback" | |
| cb = S.crop_box(box, side, W, H) | |
| im2 = im.crop(cb); cw, ch = im2.size | |
| s = min(1.0, S.LONG_MAX / max(cw, ch)) | |
| if s < 1.0: | |
| im2 = im2.resize((max(32, int(round(cw * s))), max(32, int(round(ch * s)))), Image.BICUBIC) | |
| return im2, cb | |
| def extract_route(uid, jobs, q): | |
| """jobs = [(item_id, [times])]; puts (item_id, k, PIL crop | None, crop_box, box_source, side, cx) per frame.""" | |
| import cv2 | |
| from PIL import Image | |
| cv2.setNumThreads(1) | |
| boxes = B.VIO.load_driver_boxes(uid); vp = B.VIO.video_path(uid) | |
| if boxes is None or not vp.exists(): | |
| for iid, times in jobs: | |
| for k in range(len(times)): q.put((iid, k, None, None, "no_video_or_pose", None, None)) | |
| return | |
| side, cx = driver_side(boxes) | |
| cap = cv2.VideoCapture(str(vp)) | |
| try: | |
| for iid, times in sorted(jobs, key=lambda j: j[1][0]): | |
| fis = [int(round(tt * B.FPS)) for tt in times] | |
| frames = V3B.read_frames(cap, fis) | |
| bx, _ = B.VIO.boxes_at(boxes, np.asarray(fis, np.int64)) | |
| for k, (tt, fi) in enumerate(zip(times, fis)): | |
| fr = frames.get(fi) | |
| if fr is None: | |
| q.put((iid, k, None, None, "missing_frame", side, cx)); continue | |
| b, src = bx[k], "pose_exact" | |
| if not np.all(np.isfinite(b)): | |
| j = B.nearest_pose_row(boxes, tt) | |
| if j is not None and np.all(np.isfinite(boxes["box"][j])): b, src = boxes["box"][j], "pose_nearest<=4fr" | |
| else: b, src = None, "none" | |
| ok, buf = cv2.imencode(".jpg", fr, [cv2.IMWRITE_JPEG_QUALITY, 95]) # same q95 JPEG round trip as the validation frames | |
| im = Image.open(io.BytesIO(buf.tobytes())).convert("RGB") | |
| im2, cb = crop_like_scorer(im, None if b is None else [float(v) for v in b], side) | |
| q.put((iid, k, im2, cb, src, side, cx)) | |
| finally: | |
| cap.release() | |
| # ------------------------------------------------------------------------------------------------------------- scoring | |
| def score_all(items, cand, batch, workers, log): | |
| import torch | |
| q = queue.Queue(maxsize=48 * batch) | |
| groups = {} | |
| for it in items: groups.setdefault(it["uid"], []).append((it["id"], [it["t_event"] + dt for dt in DTS])) | |
| routes = sorted(groups.items(), key=lambda g: -len(g[1])) | |
| n_exp = len(items) * len(DTS) | |
| log(f"extraction: {len(items)} items x {len(DTS)} dts = {n_exp} frames on {len(routes)} routes, {workers} threads; scoring {cand} batch {batch}") | |
| free0, ram0 = S.gpu_gate(S.CANDS[cand][2], log) | |
| torch.cuda.reset_peak_memory_stats() | |
| t_load = time.time(); sc = S.Scorer(cand, batch, log); t_load = time.time() - t_load | |
| log(f"load {t_load:.1f}s; allocated after load {torch.cuda.memory_allocated() / 1e9:.2f} GB") | |
| P = {it["id"]: [None] * len(DTS) for it in items}; CB = {it["id"]: [None] * len(DTS) for it in items}; SRC = {it["id"]: [None] * len(DTS) for it in items}; SIDE = {} | |
| t0 = time.time(); t_model = 0.0; n_done = n_scored = 0; buf = []; n_tok_max = n_img_max = 0 | |
| def flush(): | |
| nonlocal t_model, n_scored, n_tok_max, n_img_max | |
| if not buf: return | |
| torch.cuda.synchronize(); tm = time.time() | |
| p, m, top, n_tok, n_img = sc.score([b[2] for b in buf]) | |
| torch.cuda.synchronize(); t_model += time.time() - tm | |
| for (iid, k, _), pv in zip(buf, p): P[iid][k] = round(float(pv), 5) | |
| n_scored += len(buf); n_tok_max, n_img_max = max(n_tok_max, n_tok), max(n_img_max, n_img); buf.clear() | |
| with ThreadPoolExecutor(max_workers=workers) as ex: | |
| futs = [ex.submit(extract_route, uid, jobs, q) for uid, jobs in routes] | |
| while n_done < n_exp: | |
| try: iid, k, im, cb, src, side, cx = q.get(timeout=5.0) | |
| except queue.Empty: | |
| for f in futs: | |
| if f.done() and f.exception() is not None: raise f.exception() | |
| continue | |
| n_done += 1; CB[iid][k] = cb; SRC[iid][k] = src | |
| if side is not None: SIDE[iid] = (side, cx) | |
| if im is not None: | |
| buf.append((iid, k, im)) | |
| if len(buf) >= batch: flush() | |
| if n_done % 400 == 0: | |
| log(f" {n_done}/{n_exp} frames extracted, {n_scored} scored, {time.time() - t0:.0f}s ({n_scored / max(time.time() - t0, 1e-6):.1f} f/s wall)") | |
| flush() | |
| for f in futs: f.result() | |
| wall = time.time() - t0 | |
| peak = torch.cuda.max_memory_allocated() / 1e9; peak_res = torch.cuda.max_memory_reserved() / 1e9 | |
| log(f"scored {n_scored}/{n_exp} frames in {wall:.1f}s wall ({n_scored / wall:.2f} f/s wall; model-only {n_scored / max(t_model, 1e-6):.2f} f/s); peak VRAM alloc {peak:.2f} GB reserved {peak_res:.2f} GB") | |
| for it in items: | |
| it["p_on"] = P[it["id"]]; it["crop_boxes"] = CB[it["id"]]; it["box_source"] = SRC[it["id"]] | |
| s = SIDE.get(it["id"]); it["driver_side"] = s[0] if s else None; it["box_cx_median_frac"] = (None if s is None or s[1] != s[1] else round(s[1], 4)) | |
| it.update(delta_p(it["p_on"])) | |
| timing = {"n_frames_expected": n_exp, "n_frames_scored": n_scored, "wall_s": round(wall, 1), "model_s": round(t_model, 1), "frames_per_s_wall": round(n_scored / wall, 2), | |
| "frames_per_s_model": round(n_scored / max(t_model, 1e-6), 2), "load_s": round(t_load, 1), "peak_vram_alloc_gb": round(peak, 2), "peak_vram_reserved_gb": round(peak_res, 2), | |
| "seq_len_max": n_tok_max, "image_tokens_max": n_img_max, "gate": {"free_vram_gb": round(free0, 2), "mem_available_gb": round(ram0, 2)}, "extraction_threads": workers, "batch": batch, | |
| "nvsmi_used_total_mib_after": S.nvsmi("memory.used")} | |
| return sc.meta, timing | |
| def delta_p(p): | |
| pre = [v for v, dt in zip(p, DTS) if dt < 0 and v is not None]; post = [v for v, dt in zip(p, DTS) if dt >= 0 and v is not None] | |
| if len(pre) < 2 or len(post) < 2: return {"deltaP": None, "mean_pre": None, "mean_post": None} | |
| return {"deltaP": round(float(np.mean(post) - np.mean(pre)), 5), "mean_pre": round(float(np.mean(pre)), 5), "mean_post": round(float(np.mean(post)), 5)} | |
| # ------------------------------------------------------------------------------------------------------------- validation | |
| def auroc(y, s): | |
| y = np.asarray(y, int); s = np.asarray(s, float); n1, n0 = int(y.sum()), int((1 - y).sum()) | |
| if n1 == 0 or n0 == 0: return None | |
| from scipy.stats import rankdata | |
| r = rankdata(s) | |
| return float((r[y == 1].sum() - n1 * (n1 + 1) / 2) / (n1 * n0)) | |
| def auroc_ci(y, s, n_boot=2000, seed=0): | |
| rng = np.random.default_rng(seed); y = np.asarray(y, int); s = np.asarray(s, float); vals = [] | |
| for _ in range(n_boot): | |
| i = rng.integers(0, len(y), len(y)); a = auroc(y[i], s[i]) | |
| if a is not None: vals.append(a) | |
| return [round(float(np.percentile(vals, 2.5)), 3), round(float(np.percentile(vals, 97.5)), 3)] if vals else None | |
| def wilson(k, n, z=1.96): | |
| if n == 0: return [None, None] | |
| p = k / n; d = 1 + z * z / n; c = (p + z * z / (2 * n)) / d; h = z * np.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / d | |
| return [round(float(c - h), 3), round(float(c + h), 3)] | |
| def label_truth(source, lab): | |
| """-> 1 true return, 0 non-return, None excluded.""" | |
| if not lab or lab == "unsure": return None | |
| if source == "gold": return 1 if (lab in TRUE_G or lab.startswith("ret_dt=")) else 0 if lab in FALSE_G else None | |
| return 1 if lab.startswith("ret_dt=") else 0 if lab in FALSE_V else None | |
| def validate(items, labels, page_key, out_top, top_n, log): | |
| by_key = {(it["uid"], round(it["t_event"], 3)): it for it in items if it["source"] == "pool"} | |
| L = [] # labelled rows | |
| for it in items: | |
| if it["source"] == "pool": | |
| gid = page_key.get((it["uid"], round(it["t_event"], 3))); it["page_id"] = gid | |
| if gid is None: continue | |
| lab = (labels.get(gid) or {}).get("label", "") or ""; y = label_truth("gold", lab); grp = "gold" | |
| else: | |
| lab = it["owner_label"]; y = label_truth(it["source"], lab); grp = f"{it['source']}_{it['regime']}" | |
| if y is None or it["deltaP"] is None: continue | |
| L.append({"id": it.get("page_id") or it["id"], "group": grp, "regime": it["regime"], "label": lab, "y": y, "deltaP": it["deltaP"], "mean_pre": it["mean_pre"], "mean_post": it["mean_post"], "p_on": it["p_on"]}) | |
| sets = {"gold_page_G": lambda r: r["group"] == "gold", "v3_engaged": lambda r: r["group"] == "v3_engaged", "v3_manual": lambda r: r["group"] == "v3_manual", "pass4_engaged": lambda r: r["group"] == "pass4_engaged", | |
| "v3+pass4_engaged": lambda r: r["group"] in ("v3_engaged", "pass4_engaged"), "engaged_all (G + v3 eng + pass4)": lambda r: r["regime"] == "engaged", | |
| "all (engaged + v3 manual)": lambda r: True} | |
| res = {} | |
| for name, f in sets.items(): | |
| rows = [r for r in L if f(r)]; y = [r["y"] for r in rows]; s = [r["deltaP"] for r in rows] | |
| a = auroc(y, s) if rows else None | |
| res[name] = {"n": len(rows), "n_true": int(sum(y)), "n_false": int(len(y) - sum(y)), "auroc": None if a is None else round(a, 3), "auroc_ci95_boot": auroc_ci(y, s) if a is not None else None, | |
| "deltaP_true": [r["deltaP"] for r in rows if r["y"] == 1], "deltaP_false_median": (round(float(np.median([r["deltaP"] for r in rows if r["y"] == 0])), 4) if any(r["y"] == 0 for r in rows) else None), | |
| "deltaP_false_p90": (round(float(np.percentile([r["deltaP"] for r in rows if r["y"] == 0], 90)), 4) if any(r["y"] == 0 for r in rows) else None)} | |
| log(f" AUROC {name:40s} n={len(rows):3d} (true {int(sum(y))}) -> {res[name]['auroc']} CI {res[name]['auroc_ci95_boot']}") | |
| # ---- top-K yields: threshold = K-th largest deltaP over the whole pool; yield on the labelled items above it ------------------- | |
| pool = [it for it in items if it["source"] == "pool" and it["deltaP"] is not None] | |
| dp_pool = np.sort([it["deltaP"] for it in pool])[::-1] | |
| yields = {} | |
| for K in KS: | |
| thr = float(dp_pool[min(K, len(dp_pool)) - 1]) | |
| for name, f in (("engaged_all", lambda r: r["regime"] == "engaged"), ("gold_page_G", lambda r: r["group"] == "gold"), ("all", lambda r: True)): | |
| rows = [r for r in L if f(r)]; above = [r for r in rows if r["deltaP"] >= thr]; k = sum(r["y"] for r in above) | |
| base = sum(r["y"] for r in rows) / len(rows) if rows else None | |
| yields[f"K{K}_{name}"] = {"K": K, "deltaP_threshold": round(thr, 4), "labelled_above": len(above), "true_above": int(k), "yield": (round(k / len(above), 3) if above else None), "yield_ci95": wilson(k, len(above)) if above else None, | |
| "labelled_total": len(rows), "true_total": int(sum(r["y"] for r in rows)), "base_rate": (round(base, 3) if base is not None else None), | |
| "true_missed_below": int(sum(r["y"] for r in rows if r["deltaP"] < thr))} | |
| # ---- labelled-subset ranking: cumulative precision along the deltaP order -------------------------------------------------------- | |
| cum = {} | |
| for name, f in (("engaged_all", lambda r: r["regime"] == "engaged"), ("all", lambda r: True)): | |
| rows = sorted([r for r in L if f(r)], key=lambda r: -r["deltaP"]); ys = np.cumsum([r["y"] for r in rows]) | |
| cum[name] = {"n": len(rows), "order": [r["id"] for r in rows], "y_in_order": [r["y"] for r in rows], "deltaP_in_order": [r["deltaP"] for r in rows], | |
| "precision_at": {str(k): round(float(ys[k - 1] / k), 3) for k in (5, 10, 20, 30, 40, 50) if k <= len(rows)}, "rank_of_true": [i + 1 for i, r in enumerate(rows) if r["y"] == 1]} | |
| # ---- fresh candidates for block RE -------------------------------------------------------------------------------------------------- | |
| P, key, zones = page_index() | |
| fresh = [] | |
| for it in pool: | |
| if it.get("page_id"): continue | |
| if any(z[0] == it["uid"] and z[1] <= it["t_event"] <= z[2] for z in zones): it["near_page_item"] = True; continue | |
| fresh.append(it) | |
| fresh.sort(key=lambda it: -it["deltaP"]) | |
| top = fresh[:top_n] | |
| n_nan = sum(1 for it in items if it["source"] == "pool" and it["deltaP"] is None) | |
| log(f"pool {len(pool)} scored (+{n_nan} without deltaP); on page {sum(1 for it in items if it.get('page_id'))}; near a page item {sum(1 for it in items if it.get('near_page_item'))}; fresh {len(fresh)} -> top {len(top)} (deltaP >= {top[-1]['deltaP'] if top else None})") | |
| thr_re = top[-1]["deltaP"] if top else None | |
| re_yield = {} | |
| if thr_re is not None: | |
| for name, f in (("engaged_all", lambda r: r["regime"] == "engaged"), ("gold_page_G", lambda r: r["group"] == "gold")): | |
| rows = [r for r in L if f(r)]; above = [r for r in rows if r["deltaP"] >= thr_re]; k = sum(r["y"] for r in above) | |
| re_yield[name] = {"deltaP_threshold": round(thr_re, 4), "labelled_above": len(above), "true_above": int(k), "yield": (round(k / len(above), 3) if above else None), "yield_ci95": wilson(k, len(above)) if above else None} | |
| dist = {"pool": _dist([it["deltaP"] for it in pool]), "fresh": _dist([it["deltaP"] for it in fresh]), "top": _dist([it["deltaP"] for it in top]), | |
| "on_page_labelled_true": [r["deltaP"] for r in L if r["group"] == "gold" and r["y"] == 1], "on_page_labelled_false": _dist([r["deltaP"] for r in L if r["group"] == "gold" and r["y"] == 0])} | |
| val = {"built_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "labels_file": str(LABELS.relative_to(CWM3)), "labels_sha256": hashlib.sha256(LABELS.read_bytes()).hexdigest(), | |
| "n_labels_G": sum(1 for k, v in labels.items() if k.startswith("G_") and v.get("label")), "label_rules": {"gold G_*": "at_event | ret_dt=* -> true; never | always -> false; unsure -> excluded; only hand_return items (blocks R200/R100) are in the pool", | |
| "v3 V3_*": "ret_dt=* -> true; never_on | always_on -> false; unsure -> excluded", "pass4 <orig>__r4": "never_on | always_on -> false (no true returns); unsure -> excluded"}, | |
| "deltaP": "mean P(on | dt in {0, +0.5, +1}) - mean P(on | dt in {-2, -1, -0.5})", "auroc": res, "topK_yield_pool_threshold": yields, "labelled_ranking": cum, | |
| "block_RE": {"n_pool_scored": len(pool), "n_pool_without_deltaP": n_nan, "n_on_page": sum(1 for it in items if it.get("page_id")), "n_near_page_item": sum(1 for it in items if it.get("near_page_item")), | |
| "n_fresh": len(fresh), "n_top": len(top), "deltaP_threshold": thr_re, "yield_estimate_on_labelled": re_yield, "n_routes_top": len({it["uid"] for it in top}), | |
| "per_route_top_max": max([sum(1 for it in top if it["uid"] == u) for u in {it["uid"] for it in top}] or [0])}, | |
| "deltaP_distribution": dist, "labelled_rows": L} | |
| json.dump(val, open(OUT / "rank_validation.json", "w", encoding="utf-8"), indent=1, ensure_ascii=False) | |
| doc = {"built_utc": val["built_utc"], "source": "results/vlm_handson/rank_scores.json", "selection": f"pool candidates not on the gold page and not within +-{G.EXCL_S:.0f} s of any page item (events and windows) or excluded zone, sorted by deltaP desc, top {top_n}", | |
| "n_fresh": len(fresh), "n_items": len(top), "deltaP_min": thr_re, "page_items_json_sha256": hashlib.sha256(ITEMS_GOLD.read_bytes()).hexdigest(), "n_page_items_at_selection": len(P["items"]), | |
| "items": [{"rank": i + 1, "uid": it["uid"], "car": it["car"], "regime": "engaged", "event": "hand_return", "row": it["row"], "t_anchor": it["t_anchor"], "t_event": it["t_event"], "state0": it["state0"], "state_names": it["state_names"], | |
| "dwheel0_c": it["dwheel0_c"], "wheel_ref_defined": it["wheel_ref_defined"], "deltaP": it["deltaP"], "mean_pre": it["mean_pre"], "mean_post": it["mean_post"], "p_on": it["p_on"], "driver_side": it["driver_side"]} for i, it in enumerate(top)]} | |
| json.dump(doc, open(out_top, "w", encoding="utf-8"), indent=1, ensure_ascii=False) | |
| return val | |
| def _dist(v): | |
| v = np.asarray([x for x in v if x is not None], float) | |
| if not len(v): return None | |
| return {"n": int(len(v)), "mean": round(float(v.mean()), 4), "sd": round(float(v.std()), 4), "min": round(float(v.min()), 4), "p10": round(float(np.percentile(v, 10)), 4), "p25": round(float(np.percentile(v, 25)), 4), | |
| "median": round(float(np.median(v)), 4), "p75": round(float(np.percentile(v, 75)), 4), "p90": round(float(np.percentile(v, 90)), 4), "max": round(float(v.max()), 4), | |
| "share_gt_0.05": round(float((v > 0.05).mean()), 3), "share_gt_0.1": round(float((v > 0.1).mean()), 3), "share_gt_0.2": round(float((v > 0.2).mean()), 3)} | |
| def cross_check_validation_scores(items, log): | |
| """same frames were scored in the validation run (dt in {-1,-0.5,0,+0.5,+1} of V3_* / __r4 items, crop variant): agreement of P(on).""" | |
| try: V = json.load(open(OUT / "validation_frames.json")); Sc = json.load(open(OUT / "scores_qwen3vl2b.json")) | |
| except FileNotFoundError: return None | |
| pv = Sc["variants"]["crop"]["p_yes"]; ref = {} | |
| for f, p in zip(Sc["frames"], pv): ref[(f["item"], round(f["dt"], 1))] = p | |
| pairs = [] | |
| for it in items: | |
| if it["source"] == "pool": continue | |
| for dt, p in zip(DTS, it["p_on"]): | |
| r = ref.get((it["id"], round(dt, 1))) | |
| if r is not None and p is not None: pairs.append((r, p)) | |
| if not pairs: return None | |
| a = np.array(pairs) | |
| out = {"n_pairs": len(pairs), "mean_abs_diff": round(float(np.abs(a[:, 0] - a[:, 1]).mean()), 4), "max_abs_diff": round(float(np.abs(a[:, 0] - a[:, 1]).max()), 4), "corr": round(float(np.corrcoef(a[:, 0], a[:, 1])[0, 1]), 4), | |
| "note": "validation run read the clean q95 jpg from disk in batches of 8 by item order; here the same q95 round trip happens in memory and batch composition differs (bf16) -> tiny differences expected"} | |
| log(f"cross-check vs validation crop scores: {out}") | |
| return out | |
| # ------------------------------------------------------------------------------------------------------------- main | |
| def main(argv=None) -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--cand", default="qwen3vl2b", choices=list(S.CANDS)); ap.add_argument("--batch", type=int, default=8); ap.add_argument("--workers", type=int, default=3) | |
| ap.add_argument("--top", type=int, default=150); ap.add_argument("--limit", type=int, default=0, help="debug: only the first N pool items"); ap.add_argument("--validate-only", action="store_true") | |
| a = ap.parse_args(argv) | |
| OUT.mkdir(parents=True, exist_ok=True); logl = [] | |
| def log(s): | |
| s = f"[{time.strftime('%H:%M:%S')}] {s}"; print(s, flush=True); logl.append(s) | |
| t_start = time.time() | |
| labels = json.load(open(LABELS, encoding="utf-8")) | |
| P, page_key, _ = page_index() | |
| if a.validate_only: | |
| R = json.load(open(OUT / "rank_scores.json", encoding="utf-8")); items = R["items"] | |
| for it in items: | |
| if it["source"] != "pool": it["owner_label"] = (labels.get(it["id"]) or {}).get("label", "") or it.get("owner_label", "") | |
| log(f"validate-only from rank_scores.json ({len(items)} items, built {R['built_utc']})") | |
| validate(items, labels, page_key, OUT / f"rank_top{a.top}.json", a.top, log) | |
| return 0 | |
| iv = G.excluded_intervals() | |
| log(f"exclusion intervals: {len(iv)} on {len({x[0] for x in iv})} routes (build_gold_engaged.excluded_intervals)") | |
| pool, pst = pool_items(iv, log) | |
| if a.limit: pool = pool[:a.limit] | |
| val_items = labelled_validation_items(labels, log) | |
| items = pool + val_items | |
| n_on_page = sum(1 for it in pool if (it["uid"], round(it["t_event"], 3)) in page_key) | |
| log(f"pool items on the gold page already: {n_on_page}; not on page: {len(pool) - n_on_page}") | |
| meta, timing = score_all(items, a.cand, a.batch, a.workers, log) | |
| xc = cross_check_validation_scores(items, log) | |
| code = [Path(__file__), CWM3 / "tools/vlm_handson/score_vlm_handson.py", CWM3 / "tools/review_tool/build_gold_engaged.py", CWM3 / "tools/review_tool/build_relabel_v3.py", CWM3 / "tools/review_tool/build_review_tool.py"] | |
| h = hashlib.sha256() | |
| for p in code: h.update(p.name.encode()); h.update(p.read_bytes()) | |
| R = {"built_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "purpose": "VLM-ranked candidate enrichment of the gold-event page (decisions 2026-09-17T01:05:57Z); ordering only", | |
| "cand": a.cand, "scorer_meta": meta, "variant": "crop", "dts": DTS, "deltaP": "mean P(on | dt >= 0) - mean P(on | dt < 0)", "pool_stats": {**pst, "n_on_page": n_on_page, "n_not_on_page": len(pool) - n_on_page, "n_exclusion_intervals": len(iv)}, | |
| "n_validation_items": len(val_items), "timing": timing, "cross_check_vs_validation_run": xc, "frame_convention": "build_relabel_v3.read_frames on frame_idx = round(20 t) (same as the gold page and the validation frames); JPEG q95 round trip in memory; crop = score_vlm_handson.crop_box on the causal box (nearest pose row <= 4 frames), long side <= 896; full-frame 896 fallback when no box", | |
| "command": " ".join([sys.executable] + sys.argv), "code_sha256": h.hexdigest(), "code_files": [str(p.relative_to(CWM3)) for p in code], "wall_s_total": round(time.time() - t_start, 1), "log": logl, "items": items} | |
| json.dump(R, open(OUT / "rank_scores.json", "w", encoding="utf-8"), indent=0, ensure_ascii=False) | |
| log(f"rank_scores.json written ({len(items)} items)") | |
| validate(items, labels, page_key, OUT / f"rank_top{a.top}.json", a.top, log) | |
| R["log"] = logl; json.dump(R, open(OUT / "rank_scores.json", "w", encoding="utf-8"), indent=0, ensure_ascii=False) | |
| log(f"RANK_DONE total {time.time() - t_start:.0f}s") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 29.7 kB
- Xet hash:
- 016c1885a4fdfea898b4a85e8b99c5b6875181df90a93c4186c917d19c7783e2
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.