scripts/supplement_rgbd_fresh_data.py
#!/usr/bin/env python3
"""FRESH-V1 Amendment 1: one *valid* scene per volume (metadata-only rule, model-free trigger).
The frozen lock kept the lexicographically first metadata-eligible val scene of each volume.
When that scene failed a frozen, model-free validity check, its volume had no FRESH scene.
Amendment 1 lets such a volume use its next metadata-eligible val scene (same rules, still one
scene per volume). No model has been run on any FRESH scene; no score is involved.
"""
from __future__ import annotations
import argparse
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
def base_module(repo):
spec = importlib.util.spec_from_file_location(
"fresh_scale_data", repo / "scripts/prepare_rgbd_fresh_scale_data.py"
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def supplement_candidates(base, repo, lock, failures):
"""Next metadata-eligible val scene, per volume whose locked FRESH candidate failed."""
paths = base.source_paths(repo)
assets = safe_source_assets(paths["assets"])
partitions = csv_rows(paths["partitions"])
trajectories = {r["Animation"]: r for r in csv_rows(paths["trajectories"])}
allowed = base.allowed_frames(paths["official_split"], "val")
current = set(lock["current_scene_ids"])
protected, excluded = set(lock["protected_scene_ids"]), set(lock["excluded_scene_ids"])
failed_volumes = {base.volume(s) for s in lock["FRESH_V1_candidates"] if s in failures}
valid_volumes = {base.volume(s) for s in lock["FRESH_V1_candidates"] if s not in failures}
used_volumes = {base.volume(s) for s in current}
blocked = {asset_identity(assets[s]) for s in current | protected if s in assets}
blocked |= {asset_identity(assets[s]) for s in lock["FRESH_V1_candidates"]}
chosen, filled = [], set()
for row in sorted(partitions, key=lambda r: r["scene_name"]):
sid, vol = row["scene_name"], base.volume(row["scene_name"])
if (
vol not in failed_volumes
or vol in filled | valid_volumes | used_volumes
or sid in lock["FRESH_V1_candidates"]
or row["official_split"] != "val"
or row["protocol_partition"] != "val"
or row["previously_observed"] != "False"
or sid in protected
or sid in excluded
or sid in current
or sid not in assets
or asset_identity(assets[sid]) in blocked
or len(allowed.get(sid, [])) < 16
or base.bad_trajectory(trajectories, sid)
):
continue
chosen.append(sid)
filled.add(vol)
blocked.add(asset_identity(assets[sid]))
return chosen, assets
def run(repo, root):
repo, root = Path(repo).resolve(), Path(root).resolve()
base = base_module(repo)
lock = json.loads((root / "candidate_lock.json").read_text())
integrity = json.loads((root / "preparation_integrity.json").read_text())
failures = json.loads((root / "scene_failures.json").read_text())
fresh = json.loads((root / "manifest_fresh.json").read_text())
if (root / "amendment1_lock.json").exists():
raise FileExistsError("Amendment 1 already locked")
candidates, assets = supplement_candidates(base, repo, lock, failures)
amendment = {
"schema": "mcss.rgbd_fresh_scale.fresh_amendment1.v1",
"rule": "one VALID scene per volume: a volume whose locked FRESH candidate failed a "
"frozen model-free validity check may use its next metadata-eligible val scene "
"(lexicographic, same rules, still one scene per volume)",
"trigger": {s: failures[s]["status"] for s in lock["FRESH_V1_candidates"] if s in failures},
"candidate_lock_sha256": base.sha(root / "candidate_lock.json"),
"preparation_integrity_before": integrity,
"supplement_candidates": candidates,
"model_forward_run_on_fresh": False,
"created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
base.write(root / "amendment1_lock.json", amendment)
paths = base.source_paths(repo)
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 = base.role_rule(repo)
allowed = base.allowed_frames(paths["official_split"], "val")
records, supplement_failures, network = [], {}, [0]
with (root / "access.jsonl").open("a") as log:
def access(sid, path, operation):
if sid in set(lock["protected_scene_ids"]):
raise PermissionError("Protected scene access prohibited")
log.write(
json.dumps(
{
"time": time.time(),
"scene_id": sid,
"path": str(path),
"operation": operation,
"purpose": "FRESH_AMENDMENT1_DATA_PREPARATION_NO_FORWARD",
}
)
+ "\n"
)
log.flush()
for sid in candidates:
session = requests.Session()
try:
reader = BoundedRangeReader(base.URL.format(sid), session)
reader.shared_budget = network
with zipfile.ZipFile(reader) as archive:
plan = select_members(sid, archive.infolist(), allowed[sid])
for member in plan["member_metadata"]:
dest = root / "raw" / member["name"]
access(sid, member["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(member["name"]))
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"])
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, failed = [], {}, None
if not np.isclose(
scale, float(calibration[sid]["settings_units_info_meters_scale"]), rtol=1e-6
):
failed = {"status": "INELIGIBLE_CALIBRATION_SCALE"}
prepared = root / "prepared" / sid
prepared.mkdir(parents=True, exist_ok=True)
for fid in plan["frame_ids"]:
if failed:
break
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"
)
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]["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:
check_geometry(
sid,
depth,
position,
pose,
calibrated_intrinsics(calibration[sid], depth.shape),
scale,
)
except ValueError as exc:
failed = {"status": "INELIGIBLE_SOURCE_GEOMETRY", "error": str(exc)}
break
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)
frames.append(
{
"frame_id": fid,
"rgb": str(rgb_out),
"depth": str(depth_out),
"intrinsics": calibrated_intrinsics(calibration[sid], (128, 160)).tolist(),
"c2w": pose.tolist(),
}
)
if failed is None:
try:
roles, _ = camera_only_roles(sid, frames, quality, prior)
except ValueError as exc:
failed = {"status": "INELIGIBLE_FRAME_ROLES", "error": str(exc)}
if failed is not None:
supplement_failures[sid] = failed
print("INELIGIBLE", sid, failed["status"], flush=True)
continue
records.append(
{
"scene_id": sid,
"split": "FRESH_QUALIFICATION",
"historically_exposed": False,
"historical_exposure_basis": lock["independence_basis"],
"physical_scene_id": "|".join(asset_identity(assets[sid])),
"roles": roles,
"frames": frames,
}
)
print("PREPARED", sid, flush=True)
amended = dict(fresh)
amended["scenes"] = fresh["scenes"] + records
amended["amendment1_lock_sha256"] = base.sha(root / "amendment1_lock.json")
base.write(root / "manifest_fresh_amended.json", amended)
base.write(
root / "amendment1_result.json",
{
"supplement_prepared": [r["scene_id"] for r in records],
"supplement_failures": supplement_failures,
"fresh_scene_count": len(amended["scenes"]),
"network_bytes": network[0],
},
)
print("FRESH_AMENDED", len(amended["scenes"]), [r["scene_id"] for r in amended["scenes"]])
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repo", type=Path, default=Path(__file__).resolve().parents[1])
parser.add_argument(
"--root", type=Path, default=Path("outputs/EXP-3D-RGBD-FRESH-SCALE-DATA-V1")
)
args = parser.parse_args()
run(args.repo, args.root)