Buckets:
| """cwm3/eval/newtest_render.py — clip packages for the human-evaluation programme (docs/newtest_protocol_draft.md §2; 2026-09-12): the physically cut | |
| prefix [t-5, t], post [t, t+10] (frame t frozen for 1 s) and full 15 s clips, the whitelist CAN evidence, the gaze windows [t-3, t+1], the server | |
| manifests and the remote-pull list. Test clips never leave the machine; nothing here calls any model or VLM. | |
| Inputs: results/newtest/manifest_<v>.csv (eval/newtest_sample.py) and, with --gaze, results/gaze_validation/manifest_<v>.csv. | |
| Recipes (all ffmpeg, CPU, nice 19): | |
| full from the route video: [max(0, t-5), +15 s], scale 640:-2, libx264 veryfast crf 27, -an, +faststart (= s15_cockpit/prep_clips_v2.cut) | |
| prefix from the full clip: -t anchor_s, libx264 veryfast crf 23 (= annotation_human/active_select.cut_prefix) | |
| post from the full clip: -ss anchor_s, 10 s, tpad start_duration=1 start_mode=clone (frame t frozen 1 s) -> 11 s, libx264 crf 23 | |
| gaze from the route video: [max(0, t-3), +4 s], scale 640:-2, libx264 crf 23 | |
| S3 (round-2 test clips): the prefix cuts clips_prefix_r2/<idx>_{cab,road}.mp4 and the s15 full clips already exist and are REUSED (never re-cut); only | |
| the post clip is added; CAN = can_r2/<idx>.*.json when present, else built. S1 / S2 / S3_hard: everything is cut from the route video | |
| (/home/USER/BATON/<leaf>/{dcamera,qcamera}.mp4); a clip whose route video is not on disk is listed in results/newtest/routes_to_fetch.txt | |
| (leaf, remote path, phases, strata, n clips) for the data agent — this script never fetches anything. | |
| CAN evidence: cwm2/data/build_evidence.build (canspec whitelist; 1 Hz; no driver-monitoring field exists in the reader) -> can_newtest/<idx>.prefix.json | |
| (t <= anchor) and .outcome.json (t > anchor) in clip time (clip starts at max(0, t-5), anchor at anchor_s), exactly as active_select.materialise_can. | |
| Manifests (server-side; the server sends only a whitelisted subset to raters): cwm2/annotation_human/clips_newtest.json (warning clips with every | |
| rendered file present: idx, clip_id, uid, t0, dongle, regime, anchor_s, split, stratum, phase, calibration, qpos, pattern, cell, pi, fired_*, t_alert_*, | |
| score_*, clips{prefix_*, post_*, full_*}, can{prefix, outcome}) and clips_gaze.json (gaze windows). The manifests are REGENERATED from the set of | |
| rendered packages on every run (a clip is never removed once rendered); round-1 files (clips_h5.json, clips_prefix/, can/) are never touched. | |
| Phases: --phase 1 = S3 (100) + the first 60 S1 clips + the S2 calibration clips (+ gaze windows with --gaze); --phase 2 = the rest; --phase all. | |
| Every duration is checked with ffprobe (prefix <= anchor + 0.1 s; post 10.5–11.3 s; gaze <= 4.2 s); a violation aborts the manifest write. | |
| python eval/newtest_render.py --phase 1 --gaze [--version v1] [--workers 6] [--dry-run] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import collections | |
| import json | |
| import subprocess | |
| import sys | |
| import time | |
| from concurrent.futures import ThreadPoolExecutor | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| ANN = CWM2 / "annotation_human" | |
| for p in (CWM3 / "common", CWM2 / "data", ANN): | |
| sys.path.insert(0, str(p)) | |
| from runmeta import write_run_json # noqa: E402 | |
| import h5_common as H # noqa: E402 | |
| FFMPEG, FFPROBE = "/usr/bin/ffmpeg", "/usr/bin/ffprobe" | |
| BATON_LOCAL = Path("/home/USER/BATON") | |
| REMOTE_ROOT = "USER@REMOTE_HOST:/data/backup/BATON" | |
| S15_CLIPS = EXPLORE / "s15_cockpit" / "annotate_clips_v2" | |
| OUT_CLIPS = ANN / "clips_newtest" | |
| OUT_CAN = ANN / "can_newtest" | |
| OUT_GAZE = ANN / "clips_gaze" | |
| MAN_WARN = ANN / "clips_newtest.json" | |
| MAN_GAZE = ANN / "clips_gaze.json" | |
| NEWTEST = CWM3 / "results" / "newtest" | |
| GAZEV = CWM3 / "results" / "gaze_validation" | |
| FULL_S, PRE_S, POST_S = 15.0, 5.0, 10.0 | |
| FREEZE_S = H.EVENT_FREEZE_S | |
| MIN_POST_CONTENT_S = 1.0 # a post clip needs >= 1 s of content after the anchor (route end); shorter -> excluded with reason | |
| GAZE_PRE, GAZE_POST = H.GAZE_PRE_S, H.GAZE_POST_S | |
| SERVER_FIELDS = ["stratum", "layout_regime", "cell", "pattern", "pattern_cluster", "dev_group", "pi", "N_cell", "n_cell", "N_stratum", "n_stratum", | |
| "s3_stratum", "s3_draw_rank", "s15_idx", "speed_anchor_mps", "leaf", "dcam", "qcam"] | |
| def run(cmd): | |
| r = subprocess.run(["nice", "-n", "19"] + cmd, capture_output=True, text=True) | |
| return r.returncode == 0, r.stderr[-300:] | |
| def probe_dur(p): | |
| r = subprocess.run([FFPROBE, "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(p)], capture_output=True, text=True) | |
| try: | |
| return float(r.stdout.strip()) | |
| except ValueError: | |
| return float("nan") | |
| def cut_full(src, dst, ss): | |
| if dst.exists(): | |
| return "cached" | |
| tmp = dst.with_suffix(".part.mp4") | |
| ok, err = run([FFMPEG, "-v", "error", "-y", "-ss", f"{ss:.2f}", "-to", f"{ss + FULL_S:.2f}", "-i", str(src), "-vf", "scale=640:-2", "-c:v", "libx264", "-preset", "veryfast", | |
| "-crf", "27", "-an", "-movflags", "+faststart", str(tmp)]) | |
| if not ok: | |
| return "fail:" + err | |
| tmp.rename(dst) | |
| return "ok" | |
| def cut_prefix(src, dst, dur): | |
| if dst.exists(): | |
| return "cached" | |
| tmp = dst.with_suffix(".part.mp4") | |
| ok, err = run([FFMPEG, "-v", "error", "-y", "-i", str(src), "-t", f"{dur:.3f}", "-an", "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-movflags", "+faststart", str(tmp)]) | |
| if not ok: | |
| return "fail:" + err | |
| tmp.rename(dst) | |
| return "ok" | |
| def cut_post(src, dst, anchor): | |
| if dst.exists(): | |
| return "cached" | |
| tmp = dst.with_suffix(".part.mp4") | |
| # -ss / -t as INPUT options: read exactly [anchor, anchor + 10 s] of the full clip, then tpad clones frame t for 1 s -> an 11 s output | |
| ok, err = run([FFMPEG, "-v", "error", "-y", "-ss", f"{anchor:.3f}", "-t", f"{POST_S:.3f}", "-i", str(src), "-vf", f"tpad=start_duration={FREEZE_S:.1f}:start_mode=clone", | |
| "-an", "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-movflags", "+faststart", str(tmp)]) | |
| if not ok: | |
| return "fail:" + err | |
| tmp.rename(dst) | |
| return "ok" | |
| def cut_gaze(src, dst, ss): | |
| if dst.exists(): | |
| return "cached" | |
| tmp = dst.with_suffix(".part.mp4") | |
| ok, err = run([FFMPEG, "-v", "error", "-y", "-ss", f"{ss:.2f}", "-to", f"{ss + GAZE_PRE + GAZE_POST:.2f}", "-i", str(src), "-vf", "scale=640:-2", "-c:v", "libx264", "-preset", "veryfast", | |
| "-crf", "23", "-an", "-movflags", "+faststart", str(tmp)]) | |
| if not ok: | |
| return "fail:" + err | |
| tmp.rename(dst) | |
| return "ok" | |
| def shift_rows(rows, shift, keep): | |
| out = [] | |
| for r in rows: | |
| r = dict(r) | |
| r["t_s"] = round(r["t_s"] + shift, 3) | |
| if r["t_s"] < -1e-6: | |
| continue | |
| if keep(r["t_s"]): | |
| out.append(r) | |
| return out | |
| def materialise_can(clip, can_dir, build_fn, reuse_dir=None): | |
| """can_newtest/<idx>.prefix.json (t <= anchor) and .outcome.json (t > anchor), clip time — active_select.materialise_can verbatim in spirit.""" | |
| idx, a = clip["idx"], clip["anchor_s"] | |
| pp, po = can_dir / f"{idx}.prefix.json", can_dir / f"{idx}.outcome.json" | |
| if pp.exists() and po.exists(): | |
| return "cached" | |
| if reuse_dir is not None and (reuse_dir / f"{idx}.prefix.json").exists() and (reuse_dir / f"{idx}.outcome.json").exists(): | |
| return f"reuse:{reuse_dir.name}" | |
| r = build_fn(clip["clip_id"], dict(dcam=clip["dcam"], t0=clip["t0"], ctrl=clip.get("regime"))) | |
| if r is None: | |
| return "missing" | |
| pre, out = r | |
| ws = clip["t0"] - PRE_S | |
| ss = max(0.0, ws) | |
| shift = ws - ss | |
| rows = pre["rows"] + out["rows"] | |
| base = {k: v for k, v in pre.items() if k not in ("rows", "window")} | |
| base["clip_time_note"] = "t_s is clip-relative seconds (clip starts at max(0, t0-5)); anchor at anchor_s" | |
| base["anchor_s"] = a | |
| json.dump(dict(base, window="prefix", rows=shift_rows(rows, shift, lambda t: t <= a + 1e-6)), open(pp, "w"), indent=1) | |
| json.dump(dict(base, window="outcome", rows=shift_rows(rows, shift, lambda t: t > a + 1e-6)), open(po, "w"), indent=1) | |
| return "built" | |
| def write_deviations(path, new, unit, log): | |
| """Append-only pre-analysis exclusion log (machine-readable; the analysis scripts drop these units and re-weight their strata). One entry per idx; an | |
| idx already present is not duplicated; entries are never removed here (a unit that later renders is removed by the analysis check against the manifest).""" | |
| old = json.load(open(path)) if path.exists() else {"exclusions": []} | |
| have = {e["idx"] for e in old["exclusions"]} | |
| added = 0 | |
| for e in new: | |
| if e["idx"] not in have: | |
| old["exclusions"].append(dict(e, logged_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))) | |
| added += 1 | |
| old["note"] = f"pre-analysis exclusions of {unit}s that cannot be rendered / rated (reason per entry); written by eval/newtest_render.py; consumed by the analysis (exclude + re-weight the stratum)" | |
| json.dump(old, open(path, "w"), indent=1, default=lambda o: None if (isinstance(o, float) and np.isnan(o)) else str(o)) | |
| log(f"deviations {path.name}: {len(old['exclusions'])} entries ({added} new)") | |
| def select_phase(M, phase): | |
| if phase == "all": | |
| return M | |
| if phase == "1": | |
| return M[(M.phase == 1) | (M.calibration == True)] # noqa: E712 | |
| return M[(M.phase == 2) & (M.calibration == False)] # noqa: E712 | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--version", default="v1") | |
| ap.add_argument("--phase", default="1", choices=["1", "2", "all"]) | |
| ap.add_argument("--gaze", action="store_true", help="also render the gaze windows of results/gaze_validation/manifest_<v>.csv") | |
| ap.add_argument("--gaze-supp", default="", help="also render a supplementary gaze manifest (e.g. supp_v1): results/gaze_validation/manifest_<supp>.csv, deviations_<supp>.json; entries merged into clips_gaze.json with stratum_group") | |
| ap.add_argument("--workers", type=int, default=6) | |
| ap.add_argument("--dry-run", action="store_true", help="only write routes_to_fetch.txt and report; no ffmpeg") | |
| ap.add_argument("--include-optional", action="store_true", help="also render the optional S3_hard clips") | |
| a = ap.parse_args() | |
| t_start = time.time() | |
| for d in (OUT_CLIPS / "full", OUT_CLIPS / "prefix", OUT_CLIPS / "post", OUT_CAN, OUT_GAZE): | |
| d.mkdir(parents=True, exist_ok=True) | |
| lines = [] | |
| def log(m): | |
| line = f"[{time.strftime('%H:%M:%S')}] {m}" | |
| print(line, flush=True) | |
| lines.append(line) | |
| M_all = pd.read_csv(NEWTEST / f"manifest_{a.version}.csv") | |
| M = M_all if a.include_optional else M_all[M_all.optional == False] # noqa: E712 | |
| sel = select_phase(M, a.phase) | |
| G = pd.read_csv(GAZEV / f"manifest_{a.version}.csv") if a.gaze else pd.DataFrame() | |
| if a.gaze and a.gaze_supp: | |
| Gs = pd.read_csv(GAZEV / f"manifest_{a.gaze_supp}.csv") | |
| Gs["supp_src"] = a.gaze_supp | |
| G["supp_src"] = "" | |
| G = pd.concat([G, Gs], ignore_index=True) | |
| log(f"manifest {a.version}: {len(M)} clips; phase {a.phase} -> {len(sel)} warning clips ({dict(collections.Counter(sel.stratum))}); gaze windows {len(G)}") | |
| # ---- remote-pull list: every clip (all phases) + gaze window whose route video is not on disk | |
| need = collections.defaultdict(lambda: dict(n=0, phases=set(), strata=set())) | |
| for r in M_all.itertuples(): # the fetch list always covers every clip, optional ones included | |
| if r.stratum == "S3": | |
| continue # prefix + full exist; the post clip is cut from the s15 full clip | |
| if not (Path(r.dcam).exists() and Path(r.qcam).exists()): | |
| e = need[r.leaf] | |
| e["n"] += 1 | |
| e["phases"].add(int(r.phase)) | |
| e["strata"].add(r.stratum) | |
| for r in G.itertuples(): | |
| if not (Path(r.dcam).exists() and Path(r.qcam).exists()): | |
| e = need[r.leaf] | |
| e["n"] += 1 | |
| e["phases"].add(1) | |
| e["strata"].add("gaze") | |
| fetch_p = NEWTEST / "routes_to_fetch.txt" | |
| with open(fetch_p, "w") as f: | |
| f.write(f"# leaves without local dcamera.mp4 + qcamera.mp4 needed by the human-evaluation clip packages ({time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}); " | |
| f"pull {REMOTE_ROOT}/<leaf>/{{dcamera,qcamera}}.mp4 -> {BATON_LOCAL}/<leaf>/ (data agent); delete after rendering\n") | |
| f.write("# BATON/<vehicle>/<dongle>/<segment>/<route_leaf>\tn_clips\tphases\tstrata (first field = the path under the BATON root, local and remote)\n") | |
| for leaf in sorted(need, key=lambda k: (min(need[k]["phases"]), -need[k]["n"])): | |
| e = need[leaf] | |
| f.write(f"BATON/{leaf}\t{e['n']}\t{','.join(map(str, sorted(e['phases'])))}\t{','.join(sorted(e['strata']))}\n") | |
| log(f"routes_to_fetch.txt: {len(need)} leaves ({sum(e['n'] for e in need.values())} clips/windows) without local video; phase-1 leaves {sum(1 for e in need.values() if 1 in e['phases'])}") | |
| if a.dry_run: | |
| return | |
| import build_evidence as BE # whitelist-only canspec reader | |
| BE.cs.load_can = BE.load_can_cached | |
| # ---- warning clips | |
| jobs_full, jobs_pre, jobs_post = [], [], [] | |
| clips = [] | |
| skipped = collections.Counter() | |
| for r in sel.itertuples(): | |
| idx = int(r.idx) | |
| c = dict(idx=idx, clip_id=r.clip_id, uid=r.uid, t0=float(r.t_anchor), t1=float(r.t_anchor) + POST_S, dongle=str(r.dongle), regime=r.regime, anchor_s=float(r.anchor_s), split="test", | |
| double_rate=True, qpos=int(r.qpos), **{k: (None if (isinstance(getattr(r, k), float) and np.isnan(getattr(r, k))) else getattr(r, k)) for k in SERVER_FIELDS if hasattr(r, k)}) | |
| for X in ("A", "B", "C"): | |
| for p in ("val10", "val20", "test10", "test20"): | |
| c[f"fired_{X}_{p}"] = bool(getattr(r, f"fired_{X}_{p}")) | |
| for p in ("val10", "val20", "test10", "test20", "frame"): | |
| v = getattr(r, f"t_alert_{X}_{p}") | |
| c[f"t_alert_{X}_{p}"] = None if (isinstance(v, float) and np.isnan(v)) else float(v) | |
| c[f"score_{X}_anchor"] = float(getattr(r, f"score_{X}_anchor")) | |
| c["calibration"], c["optional"], c["phase"] = bool(r.calibration), bool(r.optional), int(r.phase) | |
| n5 = f"{idx:05d}" | |
| if r.stratum == "S3": | |
| full = {cam: Path(getattr(r, f"full_{cam}")) for cam in ("cab", "road")} | |
| pre = {cam: Path(getattr(r, f"prefix_{cam}")) for cam in ("cab", "road")} | |
| if not all(p.exists() for p in list(full.values()) + list(pre.values())): | |
| skipped["S3_missing_existing_cuts"] += 1 | |
| continue | |
| c["can_reuse"] = ANN / "can_r2" | |
| else: | |
| if not (Path(r.dcam).exists() and Path(r.qcam).exists()): | |
| skipped[f"{r.stratum}_no_local_video"] += 1 | |
| continue | |
| full = {cam: OUT_CLIPS / "full" / f"{n5}_{cam}.mp4" for cam in ("cab", "road")} | |
| pre = {cam: OUT_CLIPS / "prefix" / f"{n5}_{cam}.mp4" for cam in ("cab", "road")} | |
| ss = max(0.0, c["t0"] - PRE_S) | |
| for cam, src in (("cab", r.dcam), ("road", r.qcam)): | |
| jobs_full.append((Path(src), full[cam], ss)) | |
| for cam in ("cab", "road"): | |
| jobs_pre.append((full[cam], pre[cam], c["anchor_s"])) | |
| post = {cam: OUT_CLIPS / "post" / f"{n5}_{cam}.mp4" for cam in ("cab", "road")} | |
| for cam in ("cab", "road"): | |
| jobs_post.append((full[cam], post[cam], c["anchor_s"])) | |
| c["clips"] = dict(prefix_cab=str(pre["cab"]), prefix_road=str(pre["road"]), post_cab=str(post["cab"]), post_road=str(post["road"]), full_cab=str(full["cab"]), full_road=str(full["road"])) | |
| c["dcam"], c["qcam"] = r.dcam, r.qcam | |
| clips.append(c) | |
| log(f"render plan: {len(clips)} clips ({dict(collections.Counter(c['stratum'] for c in clips))}); skipped {dict(skipped)}; ffmpeg jobs full {len(jobs_full)} prefix {len(jobs_pre)} post {len(jobs_post)}") | |
| stats = collections.Counter() | |
| with ThreadPoolExecutor(a.workers) as ex: | |
| for st in ex.map(lambda j: cut_full(*j), jobs_full): | |
| stats["full:" + st.split(":")[0]] += 1 | |
| with ThreadPoolExecutor(a.workers) as ex: | |
| for st in ex.map(lambda j: cut_prefix(*j), jobs_pre): | |
| stats["prefix:" + st.split(":")[0]] += 1 | |
| for st in ex.map(lambda j: cut_post(*j), jobs_post): | |
| stats["post:" + st.split(":")[0]] += 1 | |
| log(f"ffmpeg: {dict(stats)}") | |
| # ---- CAN evidence | |
| can_how = collections.Counter() | |
| for c in clips: | |
| how = materialise_can(c, OUT_CAN, BE.build, c.pop("can_reuse", None)) | |
| can_how[how.split(":")[0]] += 1 | |
| if how.startswith("reuse"): | |
| c["can"] = dict(prefix=str(ANN / "can_r2" / f"{c['idx']}.prefix.json"), outcome=str(ANN / "can_r2" / f"{c['idx']}.outcome.json")) | |
| else: | |
| c["can"] = dict(prefix=str(OUT_CAN / f"{c['idx']}.prefix.json"), outcome=str(OUT_CAN / f"{c['idx']}.outcome.json")) | |
| c["can_missing"] = how == "missing" | |
| log(f"CAN evidence: {dict(can_how)}") | |
| # ---- duration checks; keep only complete packages | |
| bad, complete, truncated, excluded = [], [], [], [] | |
| for c in clips: | |
| ok = True | |
| for cam in ("cab", "road"): | |
| for kind, lo, hi in (("prefix", 0.0, c["anchor_s"] + 0.1), ("post", FREEZE_S + MIN_POST_CONTENT_S, POST_S + FREEZE_S + 0.3), ("full", 0.0, FULL_S + 0.3)): | |
| p = Path(c["clips"][f"{kind}_{cam}"]) | |
| if not p.exists(): | |
| ok = False | |
| bad.append((c["idx"], kind, cam, "missing")) | |
| continue | |
| d = probe_dur(p) | |
| if not (lo <= d <= hi): | |
| ok = False | |
| bad.append((c["idx"], kind, cam, d)) | |
| elif kind == "post" and d < POST_S + FREEZE_S - 0.3: # route ends < 10 s after the anchor: kept, flagged (Task 2 sees post_content_s) | |
| c["post_truncated"], c["post_content_s"] = True, round(d - FREEZE_S, 2) | |
| truncated.append((c["idx"], round(d - FREEZE_S, 2))) | |
| if ok: | |
| complete.append(c) | |
| else: | |
| src_d = {cam: probe_dur(Path(c[k])) if Path(c[k]).exists() else float("nan") for cam, k in (("cab", "dcam"), ("road", "qcam"))} | |
| reasons = sorted({(f"{b[1]}_{b[2]}_missing" if b[3] == "missing" else f"{b[1]}_{b[2]}_{b[3]:.2f}s") for b in bad if b[0] == c["idx"]}) | |
| why = "no_road_video" if not Path(c["qcam"]).exists() else ("no_cabin_video" if not Path(c["dcam"]).exists() else | |
| ("post_content_below_minimum" if any(b[1] == "post" and b[3] != "missing" for b in bad if b[0] == c["idx"]) else "render_failure")) | |
| excluded.append(dict(idx=c["idx"], stratum=c["stratum"], cell=c.get("cell"), regime=c["regime"], t_anchor=c["t0"], reason=why, details=reasons, | |
| source_duration_s=src_d, leaf=c.get("leaf"))) | |
| log(f"complete packages {len(complete)} / {len(clips)}; violations {len(bad)}: {bad[:6]}; post clips shorter than 10 s (route end; kept + flagged) {sorted(set(truncated))}; " | |
| f"excluded {[(e['idx'], e['reason']) for e in excluded]}") | |
| # clips whose route video is absent on the remote too (data agent's verdict) are logged from the fetch list: qcamera missing -> no_road_video | |
| for r in M_all.itertuples(): | |
| if r.idx in {c["idx"] for c in clips} or r.stratum == "S3": | |
| continue | |
| if Path(r.dcam).exists() and not Path(r.qcam).exists(): | |
| excluded.append(dict(idx=int(r.idx), stratum=r.stratum, cell=r.cell, regime=r.regime, t_anchor=float(r.t_anchor), reason="no_road_video", details=["qcamera.mp4 absent locally and on the remote (dcamera only)"], | |
| source_duration_s={"cab": probe_dur(Path(r.dcam)), "road": None}, leaf=r.leaf)) | |
| write_deviations(NEWTEST / f"deviations_{a.version}.json", excluded, "clip", log) | |
| too_long = [b for b in bad if b[3] != "missing" and ((b[1] == "post" and b[3] > POST_S + FREEZE_S + 0.3) or (b[1] == "prefix" and b[3] > 5.1) or (b[1] == "full" and b[3] > FULL_S + 0.3))] | |
| assert not too_long, f"clips LONGER than the recipe allows (recipe error): {too_long[:10]}" | |
| # ---- gaze windows | |
| gaze = [] | |
| if len(G): | |
| jobs_g, gskip = [], 0 | |
| for r in G.itertuples(): | |
| if not (Path(r.dcam).exists() and Path(r.qcam).exists()): | |
| gskip += 1 | |
| continue | |
| n5 = f"{int(r.idx):05d}" | |
| ss = max(0.0, float(r.t) - GAZE_PRE) | |
| g = dict(idx=int(r.idx), clip_id=r.clip_id, uid=r.uid, t=float(r.t), t0=float(r.t), dongle=str(r.dongle), regime=r.regime, anchor_s=float(r.anchor_s), split="val", | |
| calibration=bool(r.calibration), qpos=int(r.qpos) + (100000 if getattr(r, "supp_src", "") else 0), cell=r.cell, stratum=r.stratum, asym_half=r.asym_half, pi=float(r.pi), N_stratum=int(r.N_stratum), n_stratum=int(r.n_stratum), | |
| stratum_group=(getattr(r, "stratum_group", "") if isinstance(getattr(r, "stratum_group", ""), str) else "") or ("gaze_supp" if getattr(r, "supp_src", "") else "v1"), supp_version=getattr(r, "supp_src", "") or None, | |
| gaze_off=int(r.gaze_off), dwell_s=None if np.isnan(r.dwell_s) else float(r.dwell_s), head_turned=None if np.isnan(r.head_turned) else float(r.head_turned), | |
| face_valid=bool(r.face_valid), kin_valid=bool(r.kin_valid), ear_conf_asym=None if np.isnan(r.ear_conf_asym) else float(r.ear_conf_asym), leaf=r.leaf, dcam=r.dcam, qcam=r.qcam, | |
| clips=dict(gaze_cab=str(OUT_GAZE / f"{n5}_cab.mp4"), gaze_road=str(OUT_GAZE / f"{n5}_road.mp4"))) | |
| jobs_g += [(Path(r.dcam), OUT_GAZE / f"{n5}_cab.mp4", ss), (Path(r.qcam), OUT_GAZE / f"{n5}_road.mp4", ss)] | |
| gaze.append(g) | |
| gst = collections.Counter() | |
| with ThreadPoolExecutor(a.workers) as ex: | |
| for st in ex.map(lambda j: cut_gaze(*j), jobs_g): | |
| gst[st.split(":")[0]] += 1 | |
| gbad, gexcl = [], [] | |
| keep = [] | |
| src_dur_cache = {} | |
| for g in gaze: | |
| ok = True | |
| for cam in ("cab", "road"): | |
| p = Path(g["clips"][f"gaze_{cam}"]) | |
| d = probe_dur(p) if p.exists() else float("nan") | |
| if not (np.isfinite(d) and GAZE_PRE + GAZE_POST - 1.0 <= d <= GAZE_PRE + GAZE_POST + 0.2): | |
| ok = False | |
| gbad.append((g["idx"], cam, d)) | |
| if ok: | |
| keep.append(g) | |
| else: | |
| dd = {} | |
| for cam, k in (("cab", "dcam"), ("road", "qcam")): | |
| if g[k] not in src_dur_cache: | |
| src_dur_cache[g[k]] = probe_dur(Path(g[k])) if Path(g[k]).exists() else float("nan") | |
| dd[cam] = src_dur_cache[g[k]] | |
| # the rater judges frame t: a window whose cabin (road) video ends before t (or within the +1 s tail) has no frame to rate | |
| why = ("no_cabin_video_at_t" if np.isfinite(dd["cab"]) and float(g["t"]) >= dd["cab"] - 0.25 else | |
| "no_road_video_at_t" if np.isfinite(dd["road"]) and float(g["t"]) >= dd["road"] - 0.25 else "render_failure") | |
| gexcl.append(dict(idx=g["idx"], cell=g["cell"], regime=g["regime"], stratum=g["stratum"], t=float(g["t"]), reason=why, source_duration_s=dd, leaf=g["leaf"], | |
| details=[f"{b[1]}_{'nan' if not np.isfinite(b[2]) else f'{b[2]:.2f}s'}" for b in gbad if b[0] == g["idx"]])) | |
| for cam in ("cab", "road"): # never leave an empty mp4 behind | |
| p = Path(g["clips"][f"gaze_{cam}"]) | |
| if p.exists() and p.stat().st_size < 1000: | |
| p.unlink() | |
| gaze = keep | |
| log(f"gaze windows: local video {len(jobs_g) // 2} / {len(G)} (skipped {gskip}); ffmpeg {dict(gst)}; complete {len(gaze)}; excluded {dict(collections.Counter(e['reason'] for e in gexcl))} " | |
| f"{[e['idx'] for e in gexcl]}") | |
| assert not [e for e in gexcl if e["reason"] == "render_failure"], [e for e in gexcl if e["reason"] == "render_failure"][:5] | |
| write_deviations(GAZEV / f"deviations_{a.version}.json", [e for e in gexcl if e["idx"] < 91000], "gaze window", log) | |
| if a.gaze_supp: | |
| write_deviations(GAZEV / f"deviations_{a.gaze_supp}.json", [e for e in gexcl if e["idx"] >= 91000], "supplementary gaze window", log) | |
| # remote-pull list for supplementary windows without local video (data agent) | |
| need_s = sorted({r.leaf for r in Gs.itertuples() if not (Path(r.dcam).exists() and Path(r.qcam).exists())}) | |
| fs = GAZEV / f"routes_to_fetch_{a.gaze_supp}.txt" | |
| fs.write_text("# BATON/<vehicle>/<dongle>/<segment>/<route_leaf> needed by the supplementary gaze windows (no local dcamera+qcamera); pull from " + REMOTE_ROOT + "\n" + "".join(f"BATON/{l}\n" for l in need_s)) | |
| log(f"supplementary windows without local video: {sum(1 for r in Gs.itertuples() if not (Path(r.dcam).exists() and Path(r.qcam).exists()))} on {len(need_s)} leaves -> {fs.name}") | |
| # ---- manifests: regenerate from every rendered package (previous entries kept; rendered files must still exist) | |
| def excluded_idx(dev_path): | |
| return {int(e["idx"]) for e in json.load(open(dev_path))["exclusions"]} if dev_path.exists() else set() | |
| def merge_manifest(path, new, dev_path, key="idx"): | |
| """previous entries are kept (a rendered package is never dropped by a later run) EXCEPT units in the deviations log: an excluded unit must never be | |
| servable, its stale clip files are deleted.""" | |
| old = json.load(open(path))["clips"] if path.exists() else [] | |
| by = {c[key]: c for c in old} | |
| for c in new: | |
| by[c[key]] = c | |
| excl = excluded_idx(dev_path) | |
| out = [] | |
| for c in by.values(): | |
| if c[key] in excl: | |
| for f in c["clips"].values(): | |
| if Path(f).exists() and Path(f).parent in (OUT_GAZE, OUT_CLIPS / "post", OUT_CLIPS / "prefix", OUT_CLIPS / "full"): | |
| Path(f).unlink() | |
| continue | |
| if all(Path(f).exists() for f in c["clips"].values()): | |
| out.append(c) | |
| out.sort(key=lambda c: c["qpos"]) | |
| return out | |
| def promote_calibration(entries, dev_path, planned_csv, log_): | |
| """an excluded calibration unit is replaced (display flag only) by the first ratable unit of the same cell, same regime if possible, lowest qpos.""" | |
| excl = excluded_idx(dev_path) | |
| plan = pd.read_csv(planned_csv) | |
| lost = plan[(plan.calibration == True) & plan.idx.isin(excl)] # noqa: E712 | |
| have = {c["idx"] for c in entries if c.get("calibration")} | |
| for r in lost.itertuples(): | |
| cand = sorted([c for c in entries if c.get("cell") == r.cell and not c.get("calibration") and c["idx"] not in have], key=lambda c: (c.get("regime") != r.regime, c["qpos"])) | |
| if cand: | |
| cand[0]["calibration"] = True | |
| cand[0]["calibration_promoted_for"] = int(r.idx) | |
| have.add(cand[0]["idx"]) | |
| log_(f"calibration: excluded {r.idx} ({r.cell}, {r.regime}) replaced by {cand[0]['idx']} ({cand[0].get('regime')})") | |
| return entries | |
| warn_all = merge_manifest(MAN_WARN, complete, NEWTEST / f"deviations_{a.version}.json") | |
| warn_all = promote_calibration(warn_all, NEWTEST / f"deviations_{a.version}.json", NEWTEST / f"manifest_{a.version}.csv", log) | |
| json.dump(dict(generated=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), source=str(NEWTEST / f"manifest_{a.version}.csv"), schema_version=H.SCHEMA_VERSION, n=len(warn_all), | |
| note="server-side manifest of the human-evaluation test set; stratum / cell / pattern / pi / fired_* / t_alert_* / score_* / uid / dongle / split are never sent to a rater " | |
| "(annotate_h5.py whitelists the client queue); prefix [t-5, t] physically truncated; post [t, t+10] with frame t frozen 1 s; full 15 s; CAN 1 Hz whitelist", | |
| clips=warn_all), open(MAN_WARN, "w"), indent=1) | |
| if a.gaze: | |
| gaze_all = merge_manifest(MAN_GAZE, gaze, GAZEV / f"deviations_{a.version}.json") | |
| if a.gaze_supp: | |
| gaze_all = merge_manifest(MAN_GAZE, gaze_all, GAZEV / f"deviations_{a.gaze_supp}.json") | |
| gaze_all = promote_calibration(gaze_all, GAZEV / f"deviations_{a.version}.json", GAZEV / f"manifest_{a.version}.csv", log) | |
| json.dump(dict(generated=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), source=str(GAZEV / f"manifest_{a.version}.csv"), schema_version=H.SCHEMA_VERSION, n=len(gaze_all), | |
| note="server-side manifest of the gaze_off validation windows (VAL split); cell / gaze_off / dwell / head_turned / pi never sent to a rater; clip [t-3, t+1]", clips=gaze_all), | |
| open(MAN_GAZE, "w"), indent=1) | |
| log(f"manifests: {MAN_WARN.name} {len(warn_all)} clips ({dict(collections.Counter(c['stratum'] for c in warn_all))}; calibration {sum(c['calibration'] for c in warn_all)}); " | |
| + (f"{MAN_GAZE.name} {len(gaze_all)} windows total ({len(gaze)} complete this run; calibration {sum(c.get('calibration', False) for c in gaze_all)})" if a.gaze else "gaze not requested")) | |
| write_run_json(NEWTEST, "newtest_render", [Path(__file__), Path(BE.__file__), Path(H.__file__)], [NEWTEST / f"manifest_{a.version}.csv"] + ([GAZEV / f"manifest_{a.version}.csv"] if a.gaze else []), | |
| seed=0, t_start=t_start, input_listing=False, | |
| extra=dict(version=a.version, phase=a.phase, gaze=a.gaze, n_rendered=len(complete), n_manifest=len(warn_all), n_gaze_rendered=len(gaze), skipped=dict(skipped), ffmpeg=dict(stats), | |
| can=dict(can_how), routes_to_fetch=len(need), manifest_sha256={MAN_WARN.name: H.sha256_file(MAN_WARN), **({MAN_GAZE.name: H.sha256_file(MAN_GAZE)} if a.gaze and MAN_GAZE.exists() else {})}, | |
| log=lines, dms_fields_used=[])) | |
| log(f"DONE in {time.time() - t_start:.0f}s") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 30.9 kB
- Xet hash:
- 6143a564e740a3e6620cd544f61445509e38a75bf705c62a7c2ba2b6c518867d
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.