#!/usr/bin/env python3 """Lock and materialize EVAL-V5, the last never-observed Hypersim scenes (no model, no score). `--stage lock` is metadata-only and freezes the ordered candidate list before any media byte is read; `--stage prepare` downloads the locked candidates in order and applies the frozen validity rules and the frozen camera-only frame-role rule of TRAIN-X, EVAL-V3/V4 and TRAIN-EXT. No model is run and no score is read. The rule is fixed by outputs/EXP-3D-RGBD-EVAL-V5-DATA/PLAN_BEFORE_EVAL_V5_DATA.md, written before this script, which was derived from the TRAIN-EXT preparation by count-checked replacements. EVAL-V5: official Hypersim train or val split (same protocol partition), never observed, not protected or excluded, in no earlier cohort (TRAIN72, DEV, FRESH-V1/V2 candidates, EVAL-V3, EVAL-V4, any TRAIN-EXT candidate), never found invalid by an earlier preparation, source asset unique against all of them and within the set, cam_00 >= 16 public frames, non-BAD trajectory, in sha256(seed:scene) order. By the user's decision of 2026-10-03 volumes may be shared with any of them (no volume fence, no volume cap). """ from __future__ import annotations import argparse import hashlib import importlib.util import json import time import zipfile import zlib from pathlib import Path import h5py import numpy as np import requests from PIL import Image from mcss.data.hypersim import convert_hypersim_pose from mcss.mechanism_pilot.calibration import calibrated_intrinsics, published_scene_metadata from mcss.mechanism_pilot.data_prep import BoundedRangeReader, check_geometry, resize_rgb_depth from mcss.mechanism_pilot.optimization_bounds_contracts import FrozenTrainingPrior from mcss.mechanism_pilot.support_redesign_data import safe_source_assets from mcss.mechanism_pilot.unseen_data import asset_identity, csv_rows, frame_quality, select_members EXPERIMENT = "EXP-3D-RGBD-EVAL-V5-DATA" SPLIT = "EVAL_V5" URL = "https://docs-assets.developer.apple.com/ml-research/datasets/hypersim/v1/scenes/{}.zip" PLAN = Path("outputs/EXP-3D-RGBD-EVAL-V5-DATA/PLAN_BEFORE_EVAL_V5_DATA.md") PLAN_SHA256 = "f74b73b395fc338347a32db8ec263315a938b7c875c0b0c042a2306a34ccf7ee" TRAIN_EXT_LOCK = Path("outputs/EXP-3D-RGBD-TRAIN-EXT-DATA/candidate_lock.json") EVAL_V3_SCRIPT = Path("scripts/prepare_rgbd_eval_v3_data.py") EVAL_MANIFESTS = { "EVAL_V3": Path("outputs/EXP-3D-RGBD-EVAL-V3-DATA/manifest_eval_v3.json"), "EVAL_V4": Path("outputs/EXP-3D-RGBD-EVAL-V4-DATA/manifest_eval_v4.json"), "TRAIN_EXT": Path("outputs/EXP-3D-RGBD-TRAIN-EXT-DATA/manifest_train_ext.json"), } FAILURES = { "FRESH_SCALE_V1": Path("outputs/EXP-3D-RGBD-FRESH-SCALE-DATA-V1/scene_failures.json"), "FRESH_V2": Path("outputs/EXP-3D-RGBD-FRESH-V2-DATA/scene_failures.json"), "EVAL_V3": Path("outputs/EXP-3D-RGBD-EVAL-V3-DATA/scene_failures.json"), "EVAL_V4": Path("outputs/EXP-3D-RGBD-EVAL-V4-DATA/scene_failures.json"), "TRAIN_EXT": Path("outputs/EXP-3D-RGBD-TRAIN-EXT-DATA/scene_failures.json"), } SPLITS = ("train", "val") EVAL_V5_MIN = 20 BUDGET = 2 * 1024**3 SELECTION_SEED = "EVAL-V5-20261003" EXPOSURE_BASIS = ( "EVAL-V5 scenes are marked never observed and were never used, considered, read or found " "invalid by any earlier experiment (including every TRAIN-EXT candidate); their source " "assets differ from every used or protected scene. By the user's decision of 2026-10-03 they " "may come from the official val split and may share volumes with TRAIN72, DEV, FRESH, EVAL " "and TRAIN-EXT scenes, so EVAL-V5 is a scene- and asset-disjoint mechanism cohort, weaker " "than EVAL-V3/V4, and never a qualification cohort. The protected official-test final " "holdout is not opened." ) def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def write(path, value): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") def read(path): return json.loads(Path(path).read_text()) def eval_v3_module(repo): """The frozen EVAL-V3 preparation, reused as a library (never modified).""" spec = importlib.util.spec_from_file_location("rgbd_eval_v3_data", repo / EVAL_V3_SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def source_paths(repo): paths = eval_v3_module(repo).source_paths(repo) paths["eval_v3_preparation"] = paths.pop("preparation_source") paths.update({f"{k.lower()}_manifest": repo / p for k, p in EVAL_MANIFESTS.items()}) paths.update({f"{k.lower()}_failures": repo / p for k, p in FAILURES.items()}) paths.update( plan=repo / PLAN, train_ext_lock=repo / TRAIN_EXT_LOCK, preparation_source=Path(__file__).resolve(), ) return paths def selection_key(scene): return hashlib.sha256(f"{SELECTION_SEED}:{scene}".encode()).hexdigest() def select_candidates( partitions, assets, allowed, trajectories, excluded_sets, blocked_assets, fence ): """Metadata-only candidate order; `fence` holds forbidden volumes (none for EVAL-V5).""" blocked = set(blocked_assets) selected = [] for row in sorted(partitions, key=lambda r: selection_key(r["scene_name"])): sid = row["scene_name"] if ( row["official_split"] not in SPLITS or row["protocol_partition"] != row["official_split"] or row["previously_observed"] != "False" or any(sid in s for s in excluded_sets) or sid not in assets or sid.split("_")[1] in fence or asset_identity(assets[sid]) in blocked or len(allowed.get(sid, [])) < 16 or "BAD" in trajectories.get(sid + "_cam_00", {}).get("Scene type", "BAD").upper() ): continue selected.append(sid) blocked.add(asset_identity(assets[sid])) return selected def eval_v5_sets(repo, paths, partitions): """(excluded sets by reason, assets that block, volume fence: none, by user decision).""" current = read(paths["current_manifest"]) v1_lock, fresh_v2_lock = read(paths["fresh_v1_lock"]), read(paths["fresh_v2_lock"]) dev = {r["scene_id"] for r in current["scenes"] if r["split"] == "DEV"} train72 = {r["scene_id"] for r in current["scenes"] if r["split"] == "TRAIN"} if len(dev) != 8 or len(train72) != 72: raise ValueError("TRAIN72 + DEV manifest expected") fresh = set(v1_lock["FRESH_V1_candidates"]) | set(fresh_v2_lock["FRESH_V2_candidates"]) evaluation = { k: {r["scene_id"] for r in read(repo / p)["scenes"]} for k, p in EVAL_MANIFESTS.items() } failed = set().union(*(set(read(repo / p)) for p in FAILURES.values())) train_ext_candidates = set(read(paths["train_ext_lock"])["TRAIN_EXT_candidates"]) protected = ( set(current["protected_scene_ids"]) | set(v1_lock["protected_scene_ids"]) | set(fresh_v2_lock["protected_scene_ids"]) | { r["scene_name"] for r in partitions if r["protocol_partition"] in ("final_holdout", "diagnostic_test") } ) excluded = {next(iter(r.values())) for r in csv_rows(paths["exclusions"])} sets = { "protected": protected, "excluded": excluded, "dev": dev, "train72": train72, "fresh_candidates": fresh, "eval_v3": evaluation["EVAL_V3"], "eval_v4": evaluation["EVAL_V4"], "train_ext": evaluation["TRAIN_EXT"], "train_ext_candidates": train_ext_candidates, "previously_invalid": failed, } blocking = dev | train72 | fresh | evaluation["EVAL_V3"] | evaluation["EVAL_V4"] | protected blocking |= train_ext_candidates return sets, blocking, set() def allowed_frames(v1, path): allowed = {} for split in SPLITS: for sid, ids in v1.allowed_frames(path, split).items(): if sid in allowed: raise ValueError("A scene appears in two official splits") allowed[sid] = ids return allowed def build_lock(repo): if sha(repo / PLAN) != PLAN_SHA256: raise PermissionError("The EVAL-V5 plan changed or is missing") paths = source_paths(repo) ev3 = eval_v3_module(repo) v1 = ev3.v1_module(repo) assets = safe_source_assets(paths["assets"]) partitions = csv_rows(paths["partitions"]) trajectories = {r["Animation"]: r for r in csv_rows(paths["trajectories"])} sets, blocking, fence = eval_v5_sets(repo, paths, partitions) blocked_assets = {asset_identity(assets[s]) for s in blocking if s in assets} allowed = allowed_frames(v1, paths["official_split"]) candidates = select_candidates( partitions, assets, allowed, trajectories, sets.values(), blocked_assets, fence ) observed = {r["scene_name"] for r in partitions if r["previously_observed"] == "True"} return { "schema": "mcss.rgbd_eval_v5.cohort.v1", "experiment": EXPERIMENT, "EVAL_V5_candidates": candidates, "candidate_volumes": sorted({s.split("_")[1] for s in candidates}), "candidate_official_split": { r["scene_name"]: r["official_split"] for r in partitions if r["scene_name"] in candidates }, "previously_observed": sorted(observed & set(candidates)), "minimum_valid": EVAL_V5_MIN, "fence_volumes": sorted(fence), "excluded_scene_ids": {k: sorted(v) for k, v in sets.items()}, "protected_scene_ids": sorted(sets["protected"]), "assets": {s: assets[s] for s in candidates}, "rule": ( "official train or val split with the same protocol partition; previously_observed " "False; not protected or excluded; not DEV, TRAIN72, a FRESH-V1/V2 candidate, " "EVAL-V3, EVAL-V4 or any TRAIN-EXT candidate; not invalid in an earlier preparation " "(FRESH-SCALE-V1, FRESH-V2, EVAL-V3, EVAL-V4, TRAIN-EXT); volumes may be shared with " "any used scene (the user's decision of 2026-10-03); source asset unique vs every " "used, protected and TRAIN-EXT candidate scene and within the set; cam00 >=16 public " f"frames of its split; nonBAD trajectory; sha256('{SELECTION_SEED}:'+scene) order; " "every candidate passing the frozen model-free validity and camera-only role rules " "is kept (no volume cap)" ), "exposure_basis": EXPOSURE_BASIS, "plan_sha256": PLAN_SHA256, "source_sha256": {str(p): sha(p) for p in paths.values()}, "data_budget_bytes": BUDGET, "model_forward_run": False, "protected_media_read": False, } def prepare(repo, root): repo, root = Path(repo).resolve(), Path(root).resolve() lock_path = root / "candidate_lock.json" lock = read(lock_path) if lock != build_lock(repo): raise PermissionError("Candidate lock no longer reproduces from its frozen sources") if (root / "manifest_eval_v5.json").exists(): raise FileExistsError("Completed manifest_eval_v5.json exists; no overwrite") ev3 = eval_v3_module(repo) v1 = ev3.v1_module(repo) paths = source_paths(repo) protected = set(lock["protected_scene_ids"]) observed = set(lock["previously_observed"]) calibration = published_scene_metadata(paths["calibration"]) bounds = read(paths["bounds"]) prior = FrozenTrainingPrior(bounds["near_m"], bounds["far_m"], bounds["source_sha256"]) camera_only_roles = v1.role_rule(repo) allowed = allowed_frames(v1, paths["official_split"]) network = [0] hashes, qualities, geometry, failures, supports = {}, {}, [], {}, {} records = [] budget_stop = False with (root / "access.jsonl").open("a") as log: def access(sid, path, operation): if sid in protected: raise PermissionError("Protected scene access prohibited") entry = { "time": time.time(), "scene_id": sid, "path": str(path), "operation": operation, "purpose": "EVAL_V5_DATA_PREPARATION_NO_FORWARD", } log.write(json.dumps(entry) + "\n") log.flush() for sid in lock["EVAL_V5_candidates"]: if network[0] >= BUDGET: budget_stop = True break session = requests.Session() try: reader = BoundedRangeReader(URL.format(sid), session) reader.shared_budget = network with zipfile.ZipFile(reader) as archive: try: plan = select_members(sid, archive.infolist(), allowed[sid]) except ValueError as exc: failures[sid] = {"status": "INELIGIBLE_MEMBERS", "error": str(exc)} write(root / "scene_failures.json", failures) print("INELIGIBLE", sid, "INELIGIBLE_MEMBERS", flush=True) reader.close() continue plan.update(url=URL.format(sid), etag=reader.etag, split=SPLIT) for member in plan["member_metadata"]: name = member["name"] dest = root / "raw" / name access(sid, name, "download_locked_member") if dest.exists(): content = dest.read_bytes() if ( len(content) != member["bytes"] or zlib.crc32(content) != member["crc32"] ): raise ValueError("Existing download failed CRC; stop") else: dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(archive.read(name)) hashes[str(dest)] = sha(dest) reader.close() finally: session.close() raw = root / "raw" / sid units = { r["parameter_name"]: r["parameter_value"] for r in csv_rows(raw / "_detail/metadata_scene.csv") } scale = float(units["meters_per_asset_unit"]) if not np.isclose( scale, float(calibration[sid]["settings_units_info_meters_scale"]), rtol=1e-6 ): failures[sid] = {"status": "INELIGIBLE_CALIBRATION_SCALE"} write(root / "scene_failures.json", failures) continue def hdf(path, sid=sid): access(sid, path, "decode_hdf5") with h5py.File(path) as handle: return np.array(handle["dataset"]) ids = hdf(raw / "_detail/cam_00/camera_keyframe_frame_indices.hdf5").reshape(-1) positions = hdf(raw / "_detail/cam_00/camera_keyframe_positions.hdf5") rotations = hdf(raw / "_detail/cam_00/camera_keyframe_orientations.hdf5") mapping = {int(fid): j for j, fid in enumerate(ids)} frames, quality = [], {} prepared = root / "prepared" / sid prepared.mkdir(parents=True, exist_ok=True) for fid in plan["frame_ids"]: rgb_path = raw / f"images/scene_cam_00_final_preview/frame.{fid:04d}.color.jpg" access(sid, rgb_path, "decode_RGB") with Image.open(rgb_path) as im: rgb = np.asarray(im.convert("RGB")) depth = hdf( raw / f"images/scene_cam_00_geometry_hdf5/frame.{fid:04d}.depth_meters.hdf5" ) if rgb.shape[:2] != depth.shape: raise ValueError("RGB/depth dimensions differ") quality[fid] = frame_quality(rgb, depth) rotation = rotations[mapping[fid]] orth = float(np.max(np.abs(rotation.T @ rotation - np.eye(3)))) det = float(np.linalg.det(rotation)) quality[fid].update(camera_orthogonality_error=orth, camera_determinant=det) quality[fid]["eligible"] = quality[fid]["eligible"] and orth <= 0.001 and det > 0 pose = convert_hypersim_pose( rotation, positions[mapping[fid]], meters_per_asset_unit=scale ) if fid == plan["frame_ids"][0]: position = hdf( raw / f"images/scene_cam_00_geometry_hdf5/frame.{fid:04d}.position.hdf5" ) try: geometry.append( check_geometry( sid, depth, position, pose, calibrated_intrinsics(calibration[sid], depth.shape), scale, ) ) except ValueError as exc: failures[sid] = {"status": "INELIGIBLE_SOURCE_GEOMETRY", "error": str(exc)} geometry.append({"scene_id": sid, **failures[sid]}) write(root / "geometry_checks.json", geometry) write(root / "scene_failures.json", failures) rgb_small, depth_small = resize_rgb_depth(rgb, depth) rgb_out, depth_out = prepared / f"{fid:04d}.png", prepared / f"{fid:04d}.npy" Image.fromarray(rgb_small).save(rgb_out) np.save(depth_out, depth_small) hashes[str(rgb_out)], hashes[str(depth_out)] = sha(rgb_out), sha(depth_out) frames.append( { "frame_id": fid, "rgb": str(rgb_out), "depth": str(depth_out), "intrinsics": calibrated_intrinsics(calibration[sid], (128, 160)).tolist(), "c2w": pose.tolist(), } ) qualities[sid] = quality write(root / "frame_quality.json", qualities) if sid in failures: print("INELIGIBLE", sid, failures[sid]["status"], flush=True) continue try: roles, support = camera_only_roles(sid, frames, quality, prior) except ValueError as exc: failures[sid] = {"status": "INELIGIBLE_FRAME_ROLES", "error": str(exc)} write(root / "scene_failures.json", failures) print("INELIGIBLE", sid, "INELIGIBLE_FRAME_ROLES", flush=True) continue records.append( { "scene_id": sid, "split": SPLIT, "historically_exposed": sid in observed, "historical_exposure_basis": lock["exposure_basis"], "physical_scene_id": "|".join(asset_identity(lock["assets"][sid])), "official_split": lock["candidate_official_split"][sid], "roles": roles, "frames": frames, } ) supports[sid] = support write(root / "role_support.json", supports) write(root / "progress.json", {SPLIT: [r["scene_id"] for r in records]}) print("PREPARED", sid, flush=True) write(root / "hashes.json", hashes) write( root / "download_cost.json", {"network_bytes": network[0], "full_archives_downloaded": False, "budget_bytes": BUDGET}, ) status = "PASS" if len(records) >= EVAL_V5_MIN else "BLOCKED_INSUFFICIENT_VALID_LOCKED_SCENES" write( root / "manifest_eval_v5.json", { "schema": "mcss.rgbd_eval_v5.data.v1", "image_size": [128, 160], "depth_semantics": "ray_distance_meters", "scenes": records, "candidate_lock_sha256": sha(lock_path), "protected_scene_ids": lock["protected_scene_ids"], }, ) write( root / "preparation_integrity.json", { "status": status, "EVAL_V5": len(records), "valid_volumes": sorted({r["scene_id"].split("_")[1] for r in records}), "budget_stop": budget_stop, "failures": failures, "candidate_replacement": ( "none; every locked candidate is read in order until the budget" ), "protected_media_read": False, "model_forward_run": False, "network_bytes": network[0], "candidate_lock_sha256": sha(lock_path), }, ) print(status, len(records), "failures", len(failures), flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--stage", choices=("lock", "prepare"), required=True) parser.add_argument("--repo", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--root", type=Path, default=Path("outputs") / EXPERIMENT) args = parser.parse_args() root = args.root.resolve() if args.stage == "lock": if (root / "candidate_lock.json").exists(): raise FileExistsError("Candidate lock exists; never re-lock") lock = build_lock(args.repo.resolve()) root.mkdir(parents=True, exist_ok=True) write(root / "candidate_lock.json", lock) print( json.dumps( { "EVAL_V5_candidates": len(lock["EVAL_V5_candidates"]), "candidate_volumes": len(lock["candidate_volumes"]), "previously_observed": len(lock["previously_observed"]), } ) ) else: prepare(args.repo, root) if __name__ == "__main__": main()