File size: 4,845 Bytes
9b2853a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | #!/usr/bin/env python3
"""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)})")
# global parameter windows
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")
# decode a sample of videos: frame count + size
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})")
# CSV alignment + physics spot-check on decoded sample
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")
# period consistency for one Earth-gravity sample against theory (<10% with damping)
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")
# question integrity
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})")
# MC answer-position balance
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 ".")
|