scripts/prepare_rgbd_fresh_qualification.py
#!/usr/bin/env python3
"""Freeze the FRESH-V1 qualification of the sealed V5 RGB-D carriers before any fresh forward."""
from __future__ import annotations
import argparse
import datetime
import shutil
from pathlib import Path
from mcss.mechanism_pilot.geometry_carrier_experiment import read, validate_split
from mcss.mechanism_pilot.small_training import sha, write_json
from mcss.mechanism_pilot.unseen_data import csv_rows
EXPERIMENT = "EXP-3D-RGBD-FRESH-QUALIFICATION-V1"
V5 = Path("outputs/EXP-3D-RGBD-EVIDENCE-CARRIER-V5")
DATA = Path("outputs/EXP-3D-RGBD-FRESH-SCALE-DATA-V1")
PARTITIONS = Path("configs/hypersim_er_partitions.csv")
def independence_audit(records, v5_manifest, partitions, protected):
"""Every FRESH record must pass every check; returns the per-scene audit rows."""
v5_ids = {r["scene_id"] for r in v5_manifest["scenes"]}
v5_volumes = {s.split("_")[1] for s in v5_ids}
v5_physical = {r["physical_scene_id"] for r in v5_manifest["scenes"]}
rows = {r["scene_name"]: r for r in partitions}
audit, volumes, physical = [], set(), set()
for record in records:
sid = record["scene_id"]
volume = sid.split("_")[1]
row = rows.get(sid, {})
checks = {
"fresh_split_label": record["split"] == "FRESH_QUALIFICATION",
"official_val_split": row.get("official_split") == "val"
and row.get("protocol_partition") == "val",
"never_observed": row.get("previously_observed") == "False",
"not_protected": sid not in protected,
"not_in_v5_train_or_dev": sid not in v5_ids,
"volume_disjoint_from_v5": volume not in v5_volumes,
"physical_identity_unique": record["physical_scene_id"] not in v5_physical | physical,
"one_scene_per_volume": volume not in volumes,
"not_marked_exposed": record.get("historically_exposed") is False,
}
volumes.add(volume)
physical.add(record["physical_scene_id"])
audit.append({"scene_id": sid, **checks, "verified": all(checks.values())})
if not audit or not all(a["verified"] for a in audit):
raise PermissionError(f"FRESH independence audit failed: {audit}")
return audit
def prepare(root):
root = Path(root).resolve()
if root.name != EXPERIMENT:
raise PermissionError("Exact fresh-qualification directory required")
if (root / "preregistration.json").exists():
raise FileExistsError("No re-freezing")
if not (root / "PROTOCOL.md").is_file():
raise PermissionError("Frozen protocol text required before freezing")
v5 = V5.resolve()
if read(v5 / "integrity.json")["status"] != "PASS":
raise PermissionError("V5 must be finalized")
gates = read(v5 / "qualification_results.json")
if gates["SURFACE_TRAINING_STATUS"] != "SUPPORTED" or gates["STATIC_DEV_STATUS"] != "SUPPORTED":
raise PermissionError("V5 DEV gates must be SUPPORTED")
v5_manifest = read(v5 / "scene_split.json")
data = DATA.resolve()
fresh = read(data / "manifest_fresh_amended.json")
protected = set(v5_manifest["protected_scene_ids"]) | set(
read(data / "candidate_lock.json")["protected_scene_ids"]
)
audit = independence_audit(fresh["scenes"], v5_manifest, csv_rows(PARTITIONS), protected)
scenes = [{**r, "independence_verified": True} for r in fresh["scenes"]]
manifest = {
"schema": "mcss.rgbd_fresh_qualification.split.v1",
"image_size": fresh["image_size"],
"depth_semantics": "ray_distance_meters",
"scenes": scenes,
"protected_scene_ids": sorted(protected),
"fresh_status": "VERIFIED",
"FRESH_QUALIFICATION_STATUS": "VERIFIED_INDEPENDENCE_AUDITED",
"source_manifest_sha256": sha(data / "manifest_fresh_amended.json"),
}
validate_split(manifest)
write_json(root / "scene_split.json", manifest)
write_json(root / "audit/independence_audit.json", audit)
shutil.copyfile(v5 / "train_depth_prior.json", root / "train_depth_prior.json")
v5_contract = read(v5 / "training_contract.json")
evaluated = read(v5 / "evaluated_variants.json")["variants"]
contract = {
"experiment": EXPERIMENT,
"v5_root": str(v5),
"seeds": v5_contract["seeds"],
"variants": evaluated,
"primary_pair": ["C0", "C1"],
"num_threads": 1,
"device": "cpu",
"fresh_scene_count": len(scenes),
"fresh_scene_ids": [r["scene_id"] for r in scenes],
"training_run": False,
"checkpoints": "the sealed V5 per-seed DEV selections, unchanged",
}
write_json(root / "qualification_contract.json", contract)
v5_rules = read(v5 / "decision_rules.json")
write_json(
root / "decision_rules.json",
{
"bootstrap": {"unit": "physical scene", "draws": 10000, "seed": 20260928},
"roster": "FRESH-V1: every audited scene, one per volume; no model or design "
"decision has used any of them",
"DEPTH_STATUS_FRESH": "frozen V2 primary-contrast gate on C0-C1 over FRESH-V1 "
"(the V5 DEPTH_STATUS definition, unchanged)",
"SCENE_SPECIFICITY_STATUS_FRESH": "C1 wrong-scene damage CIlo>0 and C1 spatial-shuffle "
"damage CIlo>0 over FRESH-V1",
"STATIC_STATUS_FRESH": "frozen V2 static gate for C1 over FRESH-V1",
"FRESH_QUALIFICATION_STATUS": {
"QUALIFIED": "DEPTH, SCENE_SPECIFICITY and STATIC all SUPPORTED on FRESH-V1",
"NOT_QUALIFIED": "otherwise",
},
"power_note": f"{len(scenes)} scenes, below the data lock's minimum of 6 (Amendment 1 "
"found no replacement): QUALIFIED still needs every frozen gate; NOT_QUALIFIED may "
"reflect low power and is reported as such",
"GEOMETRY_FREE_REFERENCE": v5_rules["GEOMETRY_FREE_REFERENCE"],
"consequence": "QUALIFIED -> Core A static carrier qualified on the RGB-D track; "
"DYNAMIC_TTT_NEXT_STAGE_ALLOWED=true under the project rule; the protected official "
"final holdout stays closed either way",
},
)
names = (
"scene_split.json",
"qualification_contract.json",
"decision_rules.json",
"train_depth_prior.json",
"PROTOCOL.md",
"audit/independence_audit.json",
)
selected = read(v5 / "selected_checkpoints.json")
v5_files = [
v5 / n
for n in (
"preregistration.json",
"selected_checkpoints.json",
"evaluated_variants.json",
"qualification_results.json",
"decision_rules.json",
"integrity.json",
"training_contract.json",
"scene_split.json",
)
]
for variant in evaluated:
run_files = ("summary.json", "training.jsonl", "dev_curve.json")
for seed in v5_contract["seeds"]:
v5_files.append(Path(selected[variant][str(seed)]["path"]))
v5_files += [v5 / "checkpoints" / f"{variant}_{seed}" / n for n in run_files]
sources = sorted(Path("src/mcss").rglob("*.py"))
sources += sorted(Path("scripts").glob("*fresh_qualification*.py"))
sources += sorted(Path("scripts").glob("*rgbd_evidence*.py"))
data_files = {}
for record in scenes:
for frame in record["frames"]:
for kind in ("rgb", "depth"):
data_files[str(Path(frame[kind]))] = sha(Path(frame[kind]))
write_json(
root / "preregistration.json",
{
"status": "FROZEN_BEFORE_FRESH_QUALIFICATION",
"experiment": EXPERIMENT,
"created_utc": datetime.datetime.now(datetime.UTC).isoformat(),
"v5_root": str(v5),
"fresh_data_root": str(data),
"input_sha256": {str((root / n).resolve()): sha(root / n) for n in names},
"source_sha256": {str(p.resolve()): sha(p) for p in sources},
"data_sha256": data_files,
"v5_sha256": {str(p.resolve()): sha(p) for p in v5_files},
"model_forward_on_fresh_before_lock": False,
"fresh_media_read_before_lock": "data preparation validity checks only (no model)",
"final_holdout_opened": False,
},
)
print("FROZEN_FRESH_QUALIFICATION", len(scenes), [r["scene_id"] for r in scenes], flush=True)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, default=Path("outputs") / EXPERIMENT)
prepare(parser.parse_args().root)