#!/usr/bin/env python3 """Lock and materialize EVAL-V3, a larger scene-disjoint evaluation cohort (no model, no score). `--stage lock` is metadata-only and freezes the ordered candidate list before any media byte is read; `--stage prepare` downloads locked candidates in order, applies the frozen V2 validity rules and the frozen camera-only V2 frame-role rule, and never runs a model or looks at a score. EVAL-V3: official Hypersim train split, never observed (configs/hypersim_er_partitions.csv), not protected or excluded, never used or considered by an earlier experiment (TRAIN72, DEV, FRESH-V1 and FRESH-V2 candidates, locked TRAIN-X candidates), source asset unique, in sha256(seed:scene) order. Unlike FRESH-V2 it is volume disjoint only from the DEV and FRESH scenes, not from TRAIN72 (no such volume is left): it may share volumes, never scenes or source assets, with the training set. Per volume the first two candidates that pass the model-free validity rules are kept; later candidates of a full volume are never read. """ from __future__ import annotations import argparse import hashlib import importlib.util import json import time import zipfile import zlib from collections import Counter 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-V3-DATA" SPLIT = "EVAL_V3" URL = "https://docs-assets.developer.apple.com/ml-research/datasets/hypersim/v1/scenes/{}.zip" CURRENT_MANIFEST = Path("outputs/EXP-3D-RGBD-TRAIN-SCALE-V6/scene_split.json") V1_LOCK = Path("outputs/EXP-3D-RGBD-FRESH-SCALE-DATA-V1/candidate_lock.json") V1_SCRIPT = Path("scripts/prepare_rgbd_fresh_scale_data.py") FRESH_V2_LOCK = Path("outputs/EXP-3D-RGBD-FRESH-V2-DATA/candidate_lock.json") PER_VOLUME = 2 EVAL_V3_MIN = 30 BUDGET = 2 * 1024**3 SELECTION_SEED = "EVAL-V3-20260929" INDEPENDENCE_BASIS = ( "EVAL-V3 scenes are marked never observed and were never used or considered by any earlier " "experiment; their source assets differ from every used or protected scene; their volumes " "hold no DEV or FRESH scene. They may share volumes with TRAIN72 scenes, so EVAL-V3 is a " "scene- and asset-disjoint mechanism cohort, weaker than the volume-disjoint FRESH cohorts, " "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 v1_module(repo): """The frozen FRESH-V1/TRAIN-X preparation, reused as a library (never modified).""" spec = importlib.util.spec_from_file_location("rgbd_fresh_scale_v1", repo / V1_SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def source_paths(repo): paths = v1_module(repo).source_paths(repo) paths["v1_preparation"] = paths.pop("preparation_source") paths.update( current_manifest=repo / CURRENT_MANIFEST, fresh_v1_lock=repo / V1_LOCK, fresh_v2_lock=repo / FRESH_V2_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, protected, excluded, used, fence): """Metadata-only candidate order; `fence` scenes define the forbidden volumes.""" v1 = v1_module(Path(__file__).resolve().parents[1]) forbidden = {v1.volume(s) for s in fence} blocked = {asset_identity(assets[s]) for s in used | protected if s in assets} selected = [] for row in sorted(partitions, key=lambda r: selection_key(r["scene_name"])): sid = row["scene_name"] if ( row["official_split"] != "train" or row["protocol_partition"] != "train" or row["previously_observed"] != "False" or sid in protected or sid in excluded or sid in used or sid not in assets or v1.volume(sid) in forbidden or asset_identity(assets[sid]) in blocked or len(allowed.get(sid, [])) < 16 or v1.bad_trajectory(trajectories, sid) ): continue selected.append(sid) blocked.add(asset_identity(assets[sid])) return selected def volume_full(counts, scene): """True when the scene's volume already holds PER_VOLUME valid EVAL-V3 scenes.""" return counts[scene.split("_")[1]] >= PER_VOLUME def eval_v3_sets(current, v1_lock, fresh_v2_lock): """(used, fence): every scene used or considered so far; DEV and FRESH scenes fence volumes.""" dev = {r["scene_id"] for r in current["scenes"] if r["split"] == "DEV"} fresh = set(v1_lock["FRESH_V1_candidates"]) | set(fresh_v2_lock["FRESH_V2_candidates"]) used = ( {r["scene_id"] for r in current["scenes"]} | set(v1_lock["current_scene_ids"]) | set(v1_lock["TRAIN_X_candidates"]) | fresh ) return used, dev | fresh def build_lock(repo): paths = source_paths(repo) v1 = 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"])} current = json.loads(paths["current_manifest"].read_text()) v1_lock = json.loads(paths["fresh_v1_lock"].read_text()) fresh_v2_lock = json.loads(paths["fresh_v2_lock"].read_text()) used, fence = eval_v3_sets(current, v1_lock, fresh_v2_lock) excluded = {next(iter(r.values())) for r in csv_rows(paths["exclusions"])} 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") } ) allowed = v1.allowed_frames(paths["official_split"], "train") candidates = select_candidates( partitions, assets, allowed, trajectories, protected, excluded, used, fence ) volumes = sorted({v1.volume(s) for s in candidates}) if len(volumes) * PER_VOLUME < EVAL_V3_MIN: raise ValueError(f"Insufficient metadata candidates: {len(candidates)} in {volumes}") return { "schema": "mcss.rgbd_eval_v3.cohort.v1", "experiment": EXPERIMENT, "EVAL_V3_candidates": candidates, "candidate_volumes": volumes, "per_volume": PER_VOLUME, "minimum_valid": EVAL_V3_MIN, "fence_scene_ids": sorted(fence), "used_scene_ids": sorted(used), "protected_scene_ids": sorted(protected), "excluded_scene_ids": sorted(excluded), "assets": {s: assets[s] for s in candidates}, "eval_rule": ( "official train split & protocol train & previously_observed False; not protected/" "excluded; not used or considered by TRAIN72/DEV/FRESH-V1/FRESH-V2/TRAIN-X; source " "asset unique vs used+protected and within the cohort; volume disjoint from every DEV, " "FRESH-V1 and FRESH-V2 scene (volumes may be shared with TRAIN72); cam00 >=16 public " f"train frames; nonBAD trajectory; sha256('{SELECTION_SEED}:'+scene) order; per volume " f"the first {PER_VOLUME} candidates passing the model-free validity rule are kept; " "later candidates of a full volume are never read" ), "validity_rule": v1_lock["validity_rule"], "role_rule": v1_lock["role_rule"], "independence_basis": INDEPENDENCE_BASIS, "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 = json.loads(lock_path.read_text()) if lock != build_lock(repo): raise PermissionError("Candidate lock no longer reproduces from its frozen sources") if (root / "manifest_eval_v3.json").exists(): raise FileExistsError("Completed manifest_eval_v3.json exists; no overwrite") v1 = v1_module(repo) paths = source_paths(repo) protected = set(lock["protected_scene_ids"]) calibration = published_scene_metadata(paths["calibration"]) bounds = json.loads(paths["bounds"].read_text()) prior = FrozenTrainingPrior(bounds["near_m"], bounds["far_m"], bounds["source_sha256"]) camera_only_roles = v1.role_rule(repo) allowed = v1.allowed_frames(paths["official_split"], "train") network = [0] hashes, qualities, geometry, failures, supports = {}, {}, [], {}, {} records, skipped, counts = [], [], Counter() 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_V3_DATA_PREPARATION_NO_FORWARD", } log.write(json.dumps(entry) + "\n") log.flush() for sid in lock["EVAL_V3_candidates"]: if volume_full(counts, sid): skipped.append(sid) continue 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": False, "historical_exposure_basis": lock["independence_basis"], "physical_scene_id": "|".join(asset_identity(lock["assets"][sid])), "roles": roles, "frames": frames, } ) counts[sid.split("_")[1]] += 1 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_V3_MIN else "BLOCKED_INSUFFICIENT_VALID_LOCKED_SCENES" write( root / "manifest_eval_v3.json", { "schema": "mcss.rgbd_eval_v3.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, SPLIT: len(records), "valid_per_volume": dict(sorted(counts.items())), "never_read_full_volume": skipped, "budget_stop": budget_stop, "failures": failures, "candidate_replacement": "model-free validity only, within volume, in locked order", "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), "skipped", len(skipped), 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_V3_candidates": len(lock["EVAL_V3_candidates"]), "candidate_volumes": lock["candidate_volumes"], } ) ) else: prepare(args.repo, root) if __name__ == "__main__": main()