| |
| """Integrity checks for the pendulum-3k dataset. |
| |
| Usage: python verify_dataset.py [dataset_root] |
| """ |
| import csv |
| import json |
| import math |
| import sys |
| from collections import Counter |
| from pathlib import Path |
|
|
| import imageio.v2 as imageio |
|
|
| FRAMES, FPS, SIZE = 160, 12, 196 |
| EXPECTED_VAL = {"val_high_damp": 100, "val_low_damp": 100, "val_large_angle": 100, |
| "val_gravity": 100, "val_quad_damping": 100, "val_driven": 100} |
|
|
|
|
| def main(root="."): |
| root = Path(root) |
| ok = True |
|
|
| def check(cond, msg): |
| nonlocal ok |
| print(("PASS " if cond else "FAIL ") + msg) |
| ok = ok and bool(cond) |
|
|
| idx = list(csv.DictReader(open(root / "index.csv"))) |
| check(len(idx) == 3000, f"index.csv has 3000 rows (got {len(idx)})") |
| by_split = Counter(r["split"] for r in idx) |
| check(by_split["train"] == 2400 and by_split["val"] == 600, |
| f"split counts 2400/600 (got {dict(by_split)})") |
| by_regime = Counter(r["regime"] for r in idx if r["split"] == "val") |
| check(all(by_regime[k] == v for k, v in EXPECTED_VAL.items()), |
| f"100 val videos per OOD regime (got {dict(by_regime)})") |
|
|
| |
| tr = [r for r in idx if r["split"] == "train"] |
| check(all(15 <= abs(float(r["theta0_deg"])) <= 40 for r in tr), |
| "train theta0 within ±(15,40) deg") |
| check(all(0.15 <= float(r["damping_s_inv"]) <= 0.50 for r in tr), |
| "train damping within U(0.15,0.50)") |
| check(all(float(r["gravity_m_s2"]) == 9.81 for r in tr), "train gravity 9.81") |
|
|
| vd = {r["video_id"]: r for r in idx} |
| check(all((root / r["video_path"]).exists() for r in idx), "all 3000 videos exist") |
| check(all((root / r["csv_path"]).exists() for r in idx), "all 3000 CSVs exist") |
|
|
| |
| import random |
| sample = random.Random(0).sample(idx, 40) |
| bad = 0 |
| for r in sample: |
| try: |
| frames = [f for f in imageio.get_reader(root / r["video_path"])] |
| if len(frames) != FRAMES or frames[0].shape[:2] != (SIZE, SIZE): |
| bad += 1 |
| except Exception: |
| bad += 1 |
| check(bad == 0, f"40 sampled videos decode to {FRAMES}x{SIZE}x{SIZE} (bad={bad})") |
|
|
| |
| mis = 0 |
| for r in sample[:10]: |
| rows = list(csv.DictReader(open(root / r["csv_path"]))) |
| if len(rows) != FRAMES: |
| mis += 1 |
| continue |
| if any(abs(float(rows[n]["time_s"]) - n / FPS) > 1e-9 for n in (0, 80, 159)): |
| mis += 1 |
| check(mis == 0, "10 sampled CSVs have 160 frame-aligned rows") |
|
|
| |
| r = sample[0] |
| rows = list(csv.DictReader(open(root / r["csv_path"]))) |
| th = [float(x["theta_rad"]) for x in rows] |
| cross = [n / FPS for n in range(len(th) - 1) if th[n] * th[n + 1] < 0] |
| if float(r["gravity_m_s2"]) == 9.81 and len(cross) >= 4: |
| T_meas = cross[2] - cross[0] |
| T_th = 2 * math.pi * math.sqrt(1.0 / 9.81) |
| check(abs(T_meas - T_th) / T_th < 0.15, |
| f"sample period {T_meas:.2f}s within 15% of small-angle {T_th:.2f}s") |
|
|
| |
| total_q = 0 |
| for split in ("train", "val"): |
| q = json.load(open(root / "questions" / f"questions_{split}.json")) |
| entries = q["scene_annotation"] |
| vids = {e["video_id"] for e in entries} |
| expect = {k for k, v in vd.items() if v["split"] == split} |
| check(vids == expect, f"questions_{split}: every {split} video covered") |
| nbad = 0 |
| types = Counter() |
| for e in entries: |
| if not e["questions"]: |
| nbad += 1 |
| continue |
| for qq in e["questions"]: |
| total_q += 1 |
| types[qq["question_type"]] += 1 |
| if "choices" in qq: |
| if not (0 <= qq["answer"] < len(qq["choices"])): |
| nbad += 1 |
| elif not isinstance(qq["answer"], str): |
| nbad += 1 |
| check(nbad == 0, f"questions_{split}: structure valid ({nbad} bad)") |
| print(f" {split} question type histogram: {dict(types)}", |
| flush=True) |
| check(total_q > 20000, f">20k questions total (got {total_q})") |
|
|
| |
| qv = json.load(open(root / "questions" / "questions_train.json")) |
| pos = Counter(qq["answer"] for e in qv["scene_annotation"] |
| for qq in e["questions"] if "choices" in qq) |
| print(f" train MC answers by choice position: {dict(sorted(pos.items()))}") |
|
|
| print("\n" + ("ALL CHECKS PASSED" if ok else "SOME CHECKS FAILED")) |
| sys.exit(0 if ok else 1) |
|
|
|
|
| if __name__ == "__main__": |
| main(sys.argv[1] if len(sys.argv) > 1 else ".") |
|
|
