Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /data /build_pack_v2_holdout.py

AsianPlayer

16 days ago

download

raw

9.44 kB

"""cwm3/data/build_pack_v2_holdout.py — pack v2 streams (vf256 / vc256) and future-latent targets F for the BATON-holdout split, with the
FROZEN pack-v2 code paths and the FROZEN PCA bases; nothing is re-fitted.
Why a separate entry point: data/build_pack_v2.py is welded to the benchmark (pass 1 fits / reuses the PCA on manual/train, every split is
compared against the v1 pack and the v1 future files, and container_fps.json / alignment.json in results/pack_v2 are shared artefacts).
Here the same functions are imported from it — `S5` (s5 clip library re-indexed to the frame/20 axis by the container fps), `pool_rows`
(the v1 pooling rule: >= 3 clip ENDs in [t-2, t], mean / last / last-first), and the future rule (nearest clip END <= t+tau within 0.6 s) is
copied verbatim from its `future` phase — and applied to s12_wm/results/targets_{regime}_holdout.npz with
    results/pack_v2/pca_Vf_256.npz, pca_Vc_256.npz (streams) and results/pack_v2/future_pca_v2.npz (F)   — frozen, sha256 recorded.
Container fps of the 60 routes: cv2 on the local (ledger-verified) videos, written to results/holdout/pack_v2/container_fps.json (the
benchmark table is not touched).  Outputs results/holdout/pack_v2/{vf256,vc256,future}_{regime}_holdout.npz with the v2 keys
(Z / row_ok / pca_file / pca_sha256 / source / targets / n / stream / roi / axis / version; F / ok / pca_file / pca_sha256 / targets / axis / version).
Requires the s5 features of the 60 routes in s5_features/vjepa2g384/{front,cabin}/<uid>.npz (data/holdout_pipeline.py s5).
"""
import argparse, json, sys, time
from pathlib import Path
import numpy as np

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

H = CWM3 / "results" / "holdout"; OUT = H / "pack_v2"; PCA2 = CWM3 / "results" / "pack_v2"; S12R = bp.S12R; SPLIT = "holdout"
LROOT = Path("/home/USER/BATON"); LEDGER = CWM3 / "results" / "baton_local_fetch_ledger.jsonl"


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


def fps_table(uids, log):
    import cv2
    p = OUT / "container_fps.json"; tab = json.load(open(p)) if p.exists() else {}
    led = {}
    for l in open(LEDGER):
        if l.strip():
            r = json.loads(l)
            if r.get("verified"): led[(r["leaf"], r["file"])] = r["bytes"]
    n = 0
    for u in uids:
        if u in tab and all(c in tab[u] and tab[u][c].get("fps") for c in bp.CAM_FILE): continue
        leaf = leaf_of(u); rec = {}
        for cam, fn in bp.CAM_FILE.items():
            f = LROOT / leaf / fn; ok = f.exists() and (leaf, fn) in led and f.stat().st_size == led[(leaf, fn)]
            if not ok: rec[cam] = {"fps": None, "n_frames": None, "source": "missing_or_unverified"}; continue
            cap = cv2.VideoCapture(str(f)); rec[cam] = {"fps": float(cap.get(cv2.CAP_PROP_FPS)), "n_frames": int(cap.get(cv2.CAP_PROP_FRAME_COUNT)), "source": "cv2_local_ledger_verified"}; cap.release(); n += 1
        tab[u] = rec
    p.write_text(json.dumps(tab, indent=1)); miss = [u for u in uids if any(tab[u][c]["fps"] is None for c in bp.CAM_FILE)]
    log(f"[fps] probed {n} videos; routes without fps {len(miss)}; fps==20 front {sum(tab[u]['front']['fps'] == 20.0 for u in uids if tab[u]['front']['fps'])} cabin {sum(tab[u]['cabin']['fps'] == 20.0 for u in uids if tab[u]['cabin']['fps'])}")
    return tab, miss


def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--phase", default="all", help="fps|streams|future|all"); a = ap.parse_args(); t0 = time.time()
    OUT.mkdir(parents=True, exist_ok=True); logf = open(OUT / "build_pack_v2_holdout.log", "a")
    def log(m): print(m, flush=True); logf.write(m + "\n"); logf.flush()
    tg = {r: np.load(S12R / f"targets_{r}_{SPLIT}.npz", allow_pickle=True) for r in bp.REGIMES}
    uids = sorted(set().union(*[set(np.unique(tg[r]["uid"]).tolist()) for r in bp.REGIMES]))
    tab, miss = fps_table(uids, log)
    if a.phase == "fps": return
    store = bp.S5(tab); rep = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), routes=len(uids), routes_without_fps=miss, streams={}, future={})
    have_s5 = {u: all((bp.S5_DIR[c] / f"{u}.npz").exists() for c in bp.CAM_FILE) for u in uids}; log(f"s5 front+cabin present for {sum(have_s5.values())}/{len(uids)} routes")
    if a.phase in ("streams", "all"):
        pca2 = {}
        for st in bp.STREAM_CAM:
            pth = PCA2 / f"pca_{st}_256.npz"; z = np.load(pth); pca2[st] = (z["mu"], z["W"], pth); log(f"[pca frozen] {st}: {pth.name} sha256 {sha256_file(pth)[:16]} n_fit {int(z['n_fit'])}")
        for r in bp.REGIMES:
            d = tg[r]; uid, t = d["uid"], d["t"].astype(np.float64); N = len(t); order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1])
            Z = {st: np.zeros((N, bp.K), np.float16) for st in bp.STREAM_CAM}; okc = {c: np.zeros(N, bool) for c in bp.CAM_FILE}; stale = {c: np.full(N, np.nan, np.float32) for c in bp.CAM_FILE}
            for a_, b_ in zip(bounds[:-1], bounds[1:]):
                rows_all = order[a_:b_]; u = uid[rows_all[0]]; gs = {cam: store.get(u, cam) for cam in bp.CAM_FILE}
                if any(g is None for g in gs.values()): continue
                for c0 in range(0, len(rows_all), 8192):
                    rows = rows_all[c0:c0 + 8192]; P = {}
                    for st, cam in bp.STREAM_CAM.items():
                        g = gs[cam]; P[st], ok, last2 = bp.pool_rows(g["tt"], g["F"], t[rows]); okc[cam][rows] = ok; stale[cam][rows[ok]] = (t[rows] - g["tt"][np.clip(last2, 0, None)])[ok]
                    both = okc["front"][rows] & okc["cabin"][rows]
                    for st in P:
                        x = P[st][both].astype(np.float32); Z[st][rows[both]] = ((x - pca2[st][0]) @ pca2[st][1]).astype(np.float16)
                    del P
            row_ok = okc["front"] & okc["cabin"]
            for st in bp.STREAM_CAM:
                pth = pca2[st][2]
                np.savez(OUT / f"{bp.TAG[st]}_{r}_{SPLIT}.npz", Z=Z[st], row_ok=row_ok, pca_file=np.array(pth.name), pca_sha256=np.array(sha256_file(pth)), source=np.array(f"s5 {bp.STREAM_CAM[st]} library re-indexed to frame/20 (container fps table {OUT / 'container_fps.json'}); frozen benchmark PCA {pth}"),
                         targets=np.array(str(S12R / f"targets_{r}_{SPLIT}.npz")), n=np.array(N), stream=np.array(st), roi=np.array("fullframe_causal"), axis=np.array("frame/20 (v2)"), version=np.array(2))
            s2 = {c: stale[c][row_ok] for c in bp.CAM_FILE}
            rep["streams"][f"{r}_{SPLIT}"] = dict(N=int(N), row_ok=int(row_ok.sum()), row_ok_frac=float(row_ok.mean()), routes_ok=int(len(np.unique(uid[row_ok]))), stale_future_content_frac={c: float((v < -1e-6).mean()) if v.size else None for c, v in s2.items()}, stale_max_s={c: float(v.max()) if v.size else None for c, v in s2.items()})
            log(f"[pack v2 holdout] {r}: row_ok {row_ok.sum()}/{N} ({row_ok.mean():.4f}) routes {rep['streams'][f'{r}_{SPLIT}']['routes_ok']}; future-content rows {rep['streams'][f'{r}_{SPLIT}']['stale_future_content_frac']}")
    if a.phase in ("future", "all"):
        pf = PCA2 / "future_pca_v2.npz"; zf = np.load(pf); mu, W = zf["mu"].astype(np.float32), zf["W"].astype(np.float32); log(f"[future pca frozen] {pf.name} sha256 {sha256_file(pf)[:16]} n_fit {int(zf['n_fit'])} train_only {bool(zf['train_only'])}")
        for r in bp.REGIMES:
            d = tg[r]; uid, t = d["uid"], d["t"].astype(np.float64); N = len(t); F2 = np.zeros((N, 5, bp.K), np.float16); ok2 = np.zeros((N, 5), bool)
            order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1])
            for a_, b_ in zip(bounds[:-1], bounds[1:]):
                rows = order[a_:b_]; u = uid[rows[0]]; g = store.get(u, "front")
                if g is None: continue
                proj = None
                for j, tau in enumerate(bp.TAUS):
                    tk = t[rows] + tau; jj = np.searchsorted(g["tt"], tk + 1e-6) - 1; okk = (jj >= 0) & (np.abs(g["tt"][np.clip(jj, 0, None)] - tk) <= 0.6); ok2[rows, j] = okk
                    if okk.any():
                        if proj is None: proj = ((g["F"].astype(np.float32) - mu) @ W).astype(np.float16)
                        F2[rows[okk], j] = proj[jj[okk]]
            np.savez_compressed(OUT / f"future_{r}_{SPLIT}.npz", F=F2, ok=ok2, pca_file=np.array(pf.name), pca_sha256=np.array(sha256_file(pf)), targets=np.array(str(S12R / f"targets_{r}_{SPLIT}.npz")), axis=np.array("frame/20 (v2)"), version=np.array(2))
            rep["future"][f"{r}_{SPLIT}"] = dict(N=int(N), ok_frac=float(ok2.mean()), ok_by_k={f"+{int(k)}s": float(ok2[:, j].mean()) for j, k in enumerate(bp.TAUS)}); log(f"[future v2 holdout] {r}: ok {ok2.mean():.4f} by k {rep['future'][f'{r}_{SPLIT}']['ok_by_k']}")
    (OUT / "alignment_holdout.json").write_text(json.dumps(rep, indent=1))
    write_run_json(OUT, "build_pack_v2_holdout", [Path(__file__), Path(bp.__file__)], [S12R / f"targets_{r}_{SPLIT}.npz" for r in bp.REGIMES] + [PCA2 / "pca_Vf_256.npz", PCA2 / "pca_Vc_256.npz", PCA2 / "future_pca_v2.npz"], 0, t0, extra=dict(summary=rep), input_listing=False)
    log("PACK_V2_HOLDOUT_DONE")


if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 9.44 kB
Xet hash:: 9e1e477fcc95438ae1e931612ab9bd3b5c8cd3007c4e4ea78d194985f5fc2f45

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

	"""cwm3/data/build_pack_v2_holdout.py — pack v2 streams (vf256 / vc256) and future-latent targets F for the BATON-holdout split, with the
	FROZEN pack-v2 code paths and the FROZEN PCA bases; nothing is re-fitted.

	Why a separate entry point: data/build_pack_v2.py is welded to the benchmark (pass 1 fits / reuses the PCA on manual/train, every split is
	compared against the v1 pack and the v1 future files, and container_fps.json / alignment.json in results/pack_v2 are shared artefacts).
	Here the same functions are imported from it — `S5` (s5 clip library re-indexed to the frame/20 axis by the container fps), `pool_rows`
	(the v1 pooling rule: >= 3 clip ENDs in [t-2, t], mean / last / last-first), and the future rule (nearest clip END <= t+tau within 0.6 s) is
	copied verbatim from its `future` phase — and applied to s12_wm/results/targets_{regime}_holdout.npz with
	results/pack_v2/pca_Vf_256.npz, pca_Vc_256.npz (streams) and results/pack_v2/future_pca_v2.npz (F) — frozen, sha256 recorded.
	Container fps of the 60 routes: cv2 on the local (ledger-verified) videos, written to results/holdout/pack_v2/container_fps.json (the
	benchmark table is not touched). Outputs results/holdout/pack_v2/{vf256,vc256,future}_{regime}_holdout.npz with the v2 keys
	(Z / row_ok / pca_file / pca_sha256 / source / targets / n / stream / roi / axis / version; F / ok / pca_file / pca_sha256 / targets / axis / version).
	Requires the s5 features of the 60 routes in s5_features/vjepa2g384/{front,cabin}/<uid>.npz (data/holdout_pipeline.py s5).
	"""
	import argparse, json, sys, time
	from pathlib import Path
	import numpy as np

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

	H = CWM3 / "results" / "holdout"; OUT = H / "pack_v2"; PCA2 = CWM3 / "results" / "pack_v2"; S12R = bp.S12R; SPLIT = "holdout"
	LROOT = Path("/home/USER/BATON"); LEDGER = CWM3 / "results" / "baton_local_fetch_ledger.jsonl"


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


	def fps_table(uids, log):
	import cv2
	p = OUT / "container_fps.json"; tab = json.load(open(p)) if p.exists() else {}
	led = {}
	for l in open(LEDGER):
	if l.strip():
	r = json.loads(l)
	if r.get("verified"): led[(r["leaf"], r["file"])] = r["bytes"]
	n = 0
	for u in uids:
	if u in tab and all(c in tab[u] and tab[u][c].get("fps") for c in bp.CAM_FILE): continue
	leaf = leaf_of(u); rec = {}
	for cam, fn in bp.CAM_FILE.items():
	f = LROOT / leaf / fn; ok = f.exists() and (leaf, fn) in led and f.stat().st_size == led[(leaf, fn)]
	if not ok: rec[cam] = {"fps": None, "n_frames": None, "source": "missing_or_unverified"}; continue
	cap = cv2.VideoCapture(str(f)); rec[cam] = {"fps": float(cap.get(cv2.CAP_PROP_FPS)), "n_frames": int(cap.get(cv2.CAP_PROP_FRAME_COUNT)), "source": "cv2_local_ledger_verified"}; cap.release(); n += 1
	tab[u] = rec
	p.write_text(json.dumps(tab, indent=1)); miss = [u for u in uids if any(tab[u][c]["fps"] is None for c in bp.CAM_FILE)]
	log(f"[fps] probed {n} videos; routes without fps {len(miss)}; fps==20 front {sum(tab[u]['front']['fps'] == 20.0 for u in uids if tab[u]['front']['fps'])} cabin {sum(tab[u]['cabin']['fps'] == 20.0 for u in uids if tab[u]['cabin']['fps'])}")
	return tab, miss


	def main():
	ap = argparse.ArgumentParser(); ap.add_argument("--phase", default="all", help="fps\|streams\|future\|all"); a = ap.parse_args(); t0 = time.time()
	OUT.mkdir(parents=True, exist_ok=True); logf = open(OUT / "build_pack_v2_holdout.log", "a")
	def log(m): print(m, flush=True); logf.write(m + "\n"); logf.flush()
	tg = {r: np.load(S12R / f"targets_{r}_{SPLIT}.npz", allow_pickle=True) for r in bp.REGIMES}
	uids = sorted(set().union(*[set(np.unique(tg[r]["uid"]).tolist()) for r in bp.REGIMES]))
	tab, miss = fps_table(uids, log)
	if a.phase == "fps": return
	store = bp.S5(tab); rep = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), routes=len(uids), routes_without_fps=miss, streams={}, future={})
	have_s5 = {u: all((bp.S5_DIR[c] / f"{u}.npz").exists() for c in bp.CAM_FILE) for u in uids}; log(f"s5 front+cabin present for {sum(have_s5.values())}/{len(uids)} routes")
	if a.phase in ("streams", "all"):
	pca2 = {}
	for st in bp.STREAM_CAM:
	pth = PCA2 / f"pca_{st}_256.npz"; z = np.load(pth); pca2[st] = (z["mu"], z["W"], pth); log(f"[pca frozen] {st}: {pth.name} sha256 {sha256_file(pth)[:16]} n_fit {int(z['n_fit'])}")
	for r in bp.REGIMES:
	d = tg[r]; uid, t = d["uid"], d["t"].astype(np.float64); N = len(t); order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1])
	Z = {st: np.zeros((N, bp.K), np.float16) for st in bp.STREAM_CAM}; okc = {c: np.zeros(N, bool) for c in bp.CAM_FILE}; stale = {c: np.full(N, np.nan, np.float32) for c in bp.CAM_FILE}
	for a_, b_ in zip(bounds[:-1], bounds[1:]):
	rows_all = order[a_:b_]; u = uid[rows_all[0]]; gs = {cam: store.get(u, cam) for cam in bp.CAM_FILE}
	if any(g is None for g in gs.values()): continue
	for c0 in range(0, len(rows_all), 8192):
	rows = rows_all[c0:c0 + 8192]; P = {}
	for st, cam in bp.STREAM_CAM.items():
	g = gs[cam]; P[st], ok, last2 = bp.pool_rows(g["tt"], g["F"], t[rows]); okc[cam][rows] = ok; stale[cam][rows[ok]] = (t[rows] - g["tt"][np.clip(last2, 0, None)])[ok]
	both = okc["front"][rows] & okc["cabin"][rows]
	for st in P:
	x = P[st][both].astype(np.float32); Z[st][rows[both]] = ((x - pca2[st][0]) @ pca2[st][1]).astype(np.float16)
	del P
	row_ok = okc["front"] & okc["cabin"]
	for st in bp.STREAM_CAM:
	pth = pca2[st][2]
	np.savez(OUT / f"{bp.TAG[st]}_{r}_{SPLIT}.npz", Z=Z[st], row_ok=row_ok, pca_file=np.array(pth.name), pca_sha256=np.array(sha256_file(pth)), source=np.array(f"s5 {bp.STREAM_CAM[st]} library re-indexed to frame/20 (container fps table {OUT / 'container_fps.json'}); frozen benchmark PCA {pth}"),
	targets=np.array(str(S12R / f"targets_{r}_{SPLIT}.npz")), n=np.array(N), stream=np.array(st), roi=np.array("fullframe_causal"), axis=np.array("frame/20 (v2)"), version=np.array(2))
	s2 = {c: stale[c][row_ok] for c in bp.CAM_FILE}
	rep["streams"][f"{r}_{SPLIT}"] = dict(N=int(N), row_ok=int(row_ok.sum()), row_ok_frac=float(row_ok.mean()), routes_ok=int(len(np.unique(uid[row_ok]))), stale_future_content_frac={c: float((v < -1e-6).mean()) if v.size else None for c, v in s2.items()}, stale_max_s={c: float(v.max()) if v.size else None for c, v in s2.items()})
	log(f"[pack v2 holdout] {r}: row_ok {row_ok.sum()}/{N} ({row_ok.mean():.4f}) routes {rep['streams'][f'{r}_{SPLIT}']['routes_ok']}; future-content rows {rep['streams'][f'{r}_{SPLIT}']['stale_future_content_frac']}")
	if a.phase in ("future", "all"):
	pf = PCA2 / "future_pca_v2.npz"; zf = np.load(pf); mu, W = zf["mu"].astype(np.float32), zf["W"].astype(np.float32); log(f"[future pca frozen] {pf.name} sha256 {sha256_file(pf)[:16]} n_fit {int(zf['n_fit'])} train_only {bool(zf['train_only'])}")
	for r in bp.REGIMES:
	d = tg[r]; uid, t = d["uid"], d["t"].astype(np.float64); N = len(t); F2 = np.zeros((N, 5, bp.K), np.float16); ok2 = np.zeros((N, 5), bool)
	order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1])
	for a_, b_ in zip(bounds[:-1], bounds[1:]):
	rows = order[a_:b_]; u = uid[rows[0]]; g = store.get(u, "front")
	if g is None: continue
	proj = None
	for j, tau in enumerate(bp.TAUS):
	tk = t[rows] + tau; jj = np.searchsorted(g["tt"], tk + 1e-6) - 1; okk = (jj >= 0) & (np.abs(g["tt"][np.clip(jj, 0, None)] - tk) <= 0.6); ok2[rows, j] = okk
	if okk.any():
	if proj is None: proj = ((g["F"].astype(np.float32) - mu) @ W).astype(np.float16)
	F2[rows[okk], j] = proj[jj[okk]]
	np.savez_compressed(OUT / f"future_{r}_{SPLIT}.npz", F=F2, ok=ok2, pca_file=np.array(pf.name), pca_sha256=np.array(sha256_file(pf)), targets=np.array(str(S12R / f"targets_{r}_{SPLIT}.npz")), axis=np.array("frame/20 (v2)"), version=np.array(2))
	rep["future"][f"{r}_{SPLIT}"] = dict(N=int(N), ok_frac=float(ok2.mean()), ok_by_k={f"+{int(k)}s": float(ok2[:, j].mean()) for j, k in enumerate(bp.TAUS)}); log(f"[future v2 holdout] {r}: ok {ok2.mean():.4f} by k {rep['future'][f'{r}_{SPLIT}']['ok_by_k']}")
	(OUT / "alignment_holdout.json").write_text(json.dumps(rep, indent=1))
	write_run_json(OUT, "build_pack_v2_holdout", [Path(__file__), Path(bp.__file__)], [S12R / f"targets_{r}_{SPLIT}.npz" for r in bp.REGIMES] + [PCA2 / "pca_Vf_256.npz", PCA2 / "pca_Vc_256.npz", PCA2 / "future_pca_v2.npz"], 0, t0, extra=dict(summary=rep), input_listing=False)
	log("PACK_V2_HOLDOUT_DONE")


	if __name__ == "__main__":
	main()