scripts/prepare_readout_reanalysis.py
#!/usr/bin/env python3
"""Freeze the post-hoc readout/reference re-analysis before this experiment reads DEV GT."""
from __future__ import annotations
import argparse
import json
import platform
import sys
from datetime import UTC, datetime
from pathlib import Path
import numpy as np
from mcss.mechanism_pilot.readout_reanalysis import (
EXPERIMENT,
METHODS,
OPACITY_FLOOR,
PRIMARY_REFERENCE,
READOUTS,
REFERENCES,
SOURCES,
VARIANTS,
reference_values,
)
from mcss.mechanism_pilot.small_training import sha, write_json
FINAL_MARKERS = ("source_manifest.json", "dirty.patch")
SOURCE_FILES = (
"src/mcss/mechanism_pilot/readout_reanalysis.py",
"scripts/prepare_readout_reanalysis.py",
"scripts/run_readout_reanalysis.py",
"tests/test_readout_reanalysis.py",
"src/mcss/mechanism_pilot/direct_capacity_statistics.py",
"src/mcss/mechanism_pilot/geometry_carrier_statistics.py",
"src/mcss/mechanism_pilot/statistics.py",
)
def dev_roster(manifest):
return {
r["scene_id"]: {
"primary_query": list(r["roles"]["primary_query"]),
"depth": {
str(f["frame_id"]): f["depth"]
for f in r["frames"]
if f["frame_id"] in r["roles"]["primary_query"]
},
}
for r in manifest["scenes"]
if r["split"] == "DEV"
}
def train_roster(manifest):
return sorted(r["scene_id"] for r in manifest["scenes"] if r["split"] == "TRAIN")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=f"outputs/{EXPERIMENT}")
args = parser.parse_args()
root = Path(args.root)
if root.exists():
raise FileExistsError(root)
manifests, inputs, seeds = {}, {}, {}
for key, spec in SOURCES.items():
base = Path(spec["root"])
for marker in FINAL_MARKERS:
if not (base / marker).is_file():
raise PermissionError(f"{key} is not finalized: {marker} missing")
path = base / "scene_split.json"
manifests[key] = json.loads(path.read_text())
inputs[str(path)] = sha(path)
contract = base / "training_contract.json"
inputs[str(contract)] = sha(contract)
seeds[key] = list(json.loads(contract.read_text())["seeds"])
rosters = [dev_roster(m) for m in manifests.values()]
if any(r != rosters[0] for r in rosters) or len(rosters[0]) != 8:
raise ValueError("V2-V4 must share one identical 8-scene DEV primary-query roster")
trains = [train_roster(m) for m in manifests.values()]
if any(t != trains[0] for t in trains) or len(trains[0]) != 24:
raise ValueError("V2-V4 must share one identical 24-scene TRAIN roster")
if any(s != seeds["V2"] for s in seeds.values()):
raise ValueError("V2-V4 must share the frozen seeds")
predictions = []
for key, spec in SOURCES.items():
selected = Path(spec["root"]) / "raw" / "selected_dev"
for variant in VARIANTS:
for seed in seeds[key]:
directory = selected / f"{variant}_{seed}"
results = directory / "query_results.json"
inputs[str(results)] = sha(results)
for row in json.loads(results.read_text()):
if row["method"] not in METHODS:
continue
path = directory / row["prediction_path"]
digest = sha(path)
if digest != row["prediction_file_sha256"]:
raise PermissionError(f"Sealed prediction changed: {path}")
predictions.append(
{
"source": key,
"variant": variant,
"seed": seed,
"scene_id": row["scene_id"],
"role": row["role"],
"query_id": row["query_id"],
"method": row["method"],
"step": row["step"],
"path": str(path),
"sha256": digest,
"sealed_depth_absrel": row["depth_absrel"],
"sealed_depth_delta1": row["depth_delta1"],
"sealed_depth_valid_count": row["depth_valid_count"],
}
)
# Geometry-free constants from TRAIN supervision targets only (primary-query GT).
manifest = manifests["V4"]
train_files, train_depths = {}, []
for record in sorted(
(r for r in manifest["scenes"] if r["split"] == "TRAIN"), key=lambda r: r["scene_id"]
):
frames = {f["frame_id"]: f for f in record["frames"]}
for fid in record["roles"]["primary_query"]:
path = Path(frames[fid]["depth"])
train_files[str(path)] = sha(path)
train_depths.append(np.load(path).squeeze().astype(np.float32))
references = reference_values(train_depths)
source = {name: sha(Path(name)) for name in SOURCE_FILES}
lock = {
"experiment": EXPERIMENT,
"status": "FROZEN_BEFORE_DEV_GT_ANALYSIS",
"created_utc": datetime.now(UTC).isoformat(),
"kind": "post-hoc secondary re-analysis of sealed predictions; no training",
"dev_gt_read_by_this_experiment_before_lock": False,
"frozen_statuses_unchanged": True,
"sources": {k: {**v, "seeds": seeds[k]} for k, v in SOURCES.items()},
"prediction_count": len(predictions),
"predictions": predictions,
"input_sha256": inputs,
"train_reference_files_sha256": train_files,
"references": references,
"analysis": {
"readouts": {
"RAW": "sealed renderer depth sum_i w_i t_i (the frozen metric input)",
"OPACITY_NORMALIZED": f"RAW / max(opacity, {OPACITY_FLOOR})",
"MEDIAN_SCALED": "RAW * median(GT) / median(RAW) per query view on GT-valid "
"pixels; uses GT scale, a shape-only diagnostic",
},
"readout_order": list(READOUTS),
"metrics": "measurement_metrics depth_absrel and depth_delta1 on GT-valid pixels",
"references": {
"REF_TRAIN_ABSREL_OPTIMAL": "1/t-weighted median of pooled TRAIN "
"primary-query GT: the AbsRel-optimal constant",
"REF_TRAIN_MEDIAN": "median of pooled TRAIN primary-query GT",
"under_MEDIAN_SCALED": "both become the per-view GT-median constant",
},
"reference_order": list(REFERENCES),
"primary_reference": PRIMARY_REFERENCE,
"aggregation": "frozen V2 _macro: seeds per scene/role/query, then queries, "
"roles, scenes",
"bootstrap": {"unit": "scene", "draws": 10000, "seed": 20260929, "CI": "pct95"},
"labels": {
"ABOVE": "frozen V2 evidence rule: mean>0, CIlo>0, >=75% scenes not worse",
"BELOW": "mirror rule: mean<0, CIhi<0, >=75% scenes not better",
"NOT_DISTINGUISHABLE": "otherwise",
},
"summary_status": {
"NO_CARRIER_ABOVE_GEOMETRY_FREE_REFERENCE": "no V2-V4 carrier variant ABOVE "
"its primary reference on RAW or OPACITY_NORMALIZED AbsRel/delta1",
"ABOVE_REFERENCE_ONLY_AFTER_OPACITY_NORMALIZATION": "some ABOVE only under "
"OPACITY_NORMALIZED",
"SOME_CARRIER_ABOVE_GEOMETRY_FREE_REFERENCE": "some ABOVE under RAW",
},
"integrity": "RAW recomputation must reproduce every sealed depth_absrel within "
"1e-5 and depth_delta1 within 1/valid_count, else abort",
},
"environment": {"python": sys.version, "platform": platform.platform()},
"source_sha256": source,
}
root.mkdir(parents=True)
for name in ("raw", "audit"):
(root / name).mkdir()
write_json(root / "preregistration.json", lock)
print(json.dumps({k: lock[k] for k in ("status", "prediction_count", "references")}))
if __name__ == "__main__":
main()