scripts/evaluate_rgbd_replication_eval_v4.py
#!/usr/bin/env python3
"""Replication of V11 and INPAINT-V1 on the never-read EVAL-V4 cohort (no training).
The frozen V11 C0/C1 selected checkpoints, V11's state construction and rendering, V11's NEAR/FAR
regions and INPAINT-V1's inpainting are applied unchanged to EVAL-V4, as fixed by
outputs/EXP-3D-RGBD-REPLICATION-EVAL-V4/PLAN_BEFORE_V11_RESULTS.md (written before any V11
result). Stages: `lock` freezes this protocol and code; `seal` renders every prediction and
computes every geometric baseline from context frames and query cameras only, hashes them, and
replays the pipeline on two EVAL-V3 scenes against V11's and INPAINT-V1's sealed files; `score`
reads query depth only after the seal; `analyze` computes the preregistered contrasts.
"""
# ruff: noqa: E501 -- Chinese scientific report prose
from __future__ import annotations
import argparse
import datetime
import functools
import hashlib
import importlib.util
import json
from pathlib import Path
import numpy as np
import torch
from mcss.dynamic.types import hash_scene_state
from mcss.geometry import transform_cameras
from mcss.measurements import FixedMeasurementRenderer
from mcss.mechanism_pilot.readout_reanalysis import depth_metrics
from mcss.mechanism_pilot.rgbd_completion_carrier import PRIMARY_PAIR, build_state, load_checkpoint
from mcss.mechanism_pilot.rgbd_depth_bounds import context_depth_bounds
from mcss.types import Cameras
EXPERIMENT = "EXP-3D-RGBD-REPLICATION-EVAL-V4"
V11 = Path("outputs/EXP-3D-RGBD-COMPLETION-V11")
INPAINT = Path("outputs/EXP-3D-RGBD-INPAINT-BASELINES-V1")
EVAL_V4 = Path("outputs/EXP-3D-RGBD-EVAL-V4-DATA")
PLAN = "PLAN_BEFORE_V11_RESULTS.md"
PLAN_SHA256 = "b726d336a6b6377f49d75c991e228a21380092312d6ee5e9b6f57828f639fb1c"
SCRIPTS = Path(__file__).resolve().parent
REPLAY_SCENES = 2
REGIONS = ("ALL", "NEAR", "FAR")
CONTRASTS = {
"HARMONIC_HYBRID_GAIN": ("REPROJ_HARMONIC", "ALL", "HFILL8_C1", "ALL"),
"HARMONIC_HYBRID_GAIN_C0": ("REPROJ_HARMONIC", "ALL", "HFILL8_C0", "ALL"),
"HARMONIC_FAR_GAIN_C1": ("REPROJ_HARMONIC", "FAR", "C1", "FAR"),
"HARMONIC_FAR_GAIN_C0": ("REPROJ_HARMONIC", "FAR", "C0", "FAR"),
"TELEA_HYBRID_GAIN": ("REPROJ_TELEA", "ALL", "TFILL8_C1", "ALL"),
"TELEA_FAR_GAIN_C1": ("REPROJ_TELEA", "FAR", "C1", "FAR"),
"NN_MINUS_HARMONIC_ALL": ("REPROJ_NN", "ALL", "REPROJ_HARMONIC", "ALL"),
"NN_MINUS_HARMONIC_NEAR": ("REPROJ_NN", "NEAR", "REPROJ_HARMONIC", "NEAR"),
"NN_MINUS_HARMONIC_FAR": ("REPROJ_NN", "FAR", "REPROJ_HARMONIC", "FAR"),
"NN_MINUS_TELEA_ALL": ("REPROJ_NN", "ALL", "REPROJ_TELEA", "ALL"),
"NN_HYBRID_GAIN_V11": ("REPROJ_NN", "ALL", "FILL8_C1", "ALL"),
"HARMONIC_MINUS_NN_HYBRID": ("REPROJ_HARMONIC", "ALL", "FILL8_C1", "ALL"),
}
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, allow_nan=False) + "\n")
def read(path):
return json.loads(Path(path).read_text())
def module(name, filename):
spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename)
loaded = importlib.util.module_from_spec(spec)
spec.loader.exec_module(loaded)
return loaded
@functools.cache
def host():
"""V11's frozen EVAL-V3 module: context loader, aggregation, bootstrap, gates."""
return module("replication_v11_eval", "evaluate_rgbd_completion_eval_v3.py")
@functools.cache
def inpainting():
"""INPAINT-V1's frozen module: ray cosine, harmonic and Telea filling, NEAR mask, label rule."""
return module("replication_inpaint", "run_rgbd_inpaint_baselines.py")
def inputs(root, v11=V11, inpaint=INPAINT, eval_v4=EVAL_V4):
return {
"protocol": root / "PROTOCOL.md",
"plan": root / PLAN,
"script": Path(__file__).resolve(),
"v11_eval_script": SCRIPTS / "evaluate_rgbd_completion_eval_v3.py",
"inpaint_script": SCRIPTS / "run_rgbd_inpaint_baselines.py",
"geometric_script": SCRIPTS / "run_rgbd_geometric_baselines.py",
"v11_contract": v11 / "training_contract.json",
"inpaint_lock": inpaint / "lock.json",
"eval_v4_manifest": eval_v4 / "manifest_eval_v4.json",
"eval_v4_integrity": eval_v4 / "preparation_integrity.json",
"eval_v4_candidate_lock": eval_v4 / "candidate_lock.json",
}
# ---------------------------------------------------------------- stages
def lock(root, v11=V11, inpaint=INPAINT, eval_v4=EVAL_V4):
root = Path(root).resolve()
if root.name != EXPERIMENT:
raise PermissionError("Exact experiment directory required")
if (root / "lock.json").exists():
raise FileExistsError("Never re-lock")
if not (root / "PROTOCOL.md").is_file():
raise PermissionError("Protocol text required before locking")
if sha(root / PLAN) != PLAN_SHA256:
raise PermissionError("The pre-V11 plan changed")
if read(eval_v4 / "preparation_integrity.json")["status"] != "PASS":
raise PermissionError("EVAL-V4 data must be prepared with integrity PASS")
if (root / "raw").exists():
raise PermissionError("No prediction may exist before the lock")
paths = inputs(root, v11, inpaint, eval_v4)
write(
root / "lock.json",
{
"status": "FROZEN_BEFORE_ANY_EVAL_V4_PREDICTION",
"experiment": EXPERIMENT,
"created_utc": datetime.datetime.now(datetime.UTC).isoformat(),
"primary": "AbsRel(REPROJ_HARMONIC) - AbsRel(HFILL8 of V11 C1) on EVAL-V4 (INPAINT-V1 label rule)",
"secondary": "V11 COMPLETION_GAIN and HYBRID_GAIN on EVAL-V4 (frozen V2 gate)",
"input_sha256": {
name: {"path": str(Path(p).resolve()), "sha256": sha(p)}
for name, p in paths.items()
},
"v11_results_existed_at_lock": (v11 / "eval_v3_results.json").exists(),
"inpaint_results_existed_at_lock": (inpaint / "inpaint_results.json").exists(),
"eval_v4_media_scored_before_lock": False,
"final_holdout_opened": False,
},
)
print("LOCKED", EXPERIMENT, flush=True)
def verify(root, v11=V11):
lock_ = read(root / "lock.json")
for name, entry in lock_["input_sha256"].items():
if sha(entry["path"]) != entry["sha256"]:
raise PermissionError(f"Locked input changed: {name}")
if read(v11 / "integrity.json")["status"] != "PASS":
raise PermissionError("V11 must be finalized with integrity PASS")
return lock_
def carriers(v11=V11):
config = read(v11 / "training_contract.json")
selected = read(v11 / "selected_checkpoints.json")
models = {}
for variant in PRIMARY_PAIR:
for seed in config["seeds"]:
item = selected[variant][str(seed)]
if sha(Path(item["path"])) != item["sha256"]:
raise PermissionError("Selected V11 checkpoint changed")
model, _ = load_checkpoint(item["path"], "cpu")
models[variant, seed] = model.eval()
return config, models
@torch.no_grad()
def predict(record, manifest, base, models, constant, access, tag, states=None):
"""Every sealed array of every primary query of one scene (V11's code path plus inpainting).
`tag` only names the construction episode ("eval_v3" reproduces V11's own episode ids).
"""
v11, inp, geo = host(), inpainting(), host().geometry()
renderer = FixedMeasurementRenderer(n_samples=64, ray_chunk_size=2048)
out = {}
sid = record["scene_id"]
for role, context in v11.context_of(record, manifest, base, access).items():
anchor = context[0].camera.c2w
bounds = context_depth_bounds(context, anchor)
points = geo.context_points(context)
built = {}
for (variant, seed), carrier in models.items():
state, state_anchor = build_state(carrier, context, bounds, f"{tag}:{sid}:{role}")
built[variant, seed] = (state, state_anchor)
if states is not None:
states[f"{variant}:{seed}:{sid}/{role}"] = hash_scene_state(state)
for qid in record["roles"]["primary_query"]:
access.append({"scene_id": sid, "frame_id": qid, "purpose": "QUERY_CAMERA_ONLY"})
camera = geo.query_camera(record, qid, manifest["image_size"])
depth, hit = geo.reproject(points, camera)
reproj = geo.fill_nearest(depth, hit, constant) if hit.any() else None
if reproj is None:
reproj = np.full(camera.image_size, constant, dtype=np.float64)
cosine = inp.ray_cosine(camera)
telea, fallback = inp.telea_fill(depth, hit, cosine, constant)
arrays = {
"REPROJ_NN": reproj,
"hit": hit,
"near": inp.near_mask(hit),
"REPROJ_HARMONIC": inp.harmonic_fill(depth, hit, cosine, constant),
"REPROJ_TELEA": telea,
"telea_fallback": np.array(fallback),
}
batched = Cameras(
camera.intrinsics[None, None], camera.c2w[None, None], camera.image_size
)
for (variant, seed), (state, state_anchor) in built.items():
local = transform_cameras(batched, torch.linalg.inv(state_anchor))
pred = renderer(state, local, ("depth",))["depth"][0, 0, 0].numpy()
arrays[f"{variant}_{seed}"] = pred.astype(np.float64)
out[role, qid] = arrays
return out
def replay(models, constant, v11=V11, inpaint=INPAINT):
"""Our code path on the first EVAL-V3 scenes must reproduce V11's and INPAINT-V1's sealed files."""
config = read(v11 / "training_contract.json")
manifest_path = Path(config["eval_v3"]["manifest"])
manifest = read(manifest_path)
seal = read(v11 / "raw" / "eval_v3" / "prediction_seal.json")
inpaint_seal = read(inpaint / "raw" / "inpaint_seal.json")
worst, compared = 0.0, 0
for record in manifest["scenes"][:REPLAY_SCENES]:
sid = record["scene_id"]
mine = predict(record, manifest, manifest_path.parent, models, constant, [], "eval_v3")
for (role, qid), arrays in mine.items():
sealed = dict(np.load(seal["predictions"][f"{sid}/{role}/{qid}"]["path"]))
filled = dict(np.load(inpaint_seal[f"EVAL_V3/{sid}/{role}/{qid}"]["path"]))
if not (
np.array_equal(arrays["hit"], sealed["hit"])
and np.array_equal(arrays["near"], sealed["near"])
):
raise AssertionError(f"Replay hit/near mask differs for {sid}/{role}/{qid}")
pairs = [(arrays["REPROJ_NN"], sealed["REPROJ_NN"])]
pairs += [(arrays[k], sealed[k]) for k in sealed if k[:2] in ("C0", "C1")]
pairs += [(arrays[k], filled[k]) for k in ("REPROJ_HARMONIC", "REPROJ_TELEA")]
for a, b in pairs:
worst = max(worst, float(np.max(np.abs(a - b))))
compared += 1
if worst > 1e-6:
raise AssertionError(f"Replay of the sealed EVAL-V3 pipeline failed: {worst}")
return {"scenes": REPLAY_SCENES, "arrays_compared": compared, "max_abs_deviation": worst}
def seal(root, v11=V11, inpaint=INPAINT, eval_v4=EVAL_V4):
root = Path(root).resolve()
lock_ = verify(root, v11)
out = root / "raw"
if (out / "prediction_seal.json").exists():
raise FileExistsError("EVAL-V4 predictions are sealed once")
config, models = carriers(v11)
constant = config["eval_v3"]["reprojection_fill_constant_m"]
if config["eval_v3"]["near_px"] != inpainting().NEAR_PX:
raise PermissionError("NEAR definition must be V11's")
manifest_path = Path(lock_["input_sha256"]["eval_v4_manifest"]["path"])
manifest = read(manifest_path)
if any(r["split"] != "EVAL_V4" for r in manifest["scenes"]):
raise PermissionError("EVAL-V4 records only")
replayed = replay(models, constant, v11, inpaint)
access, entries, states = [], {}, {}
for record in manifest["scenes"]:
sid = record["scene_id"]
scene = predict(
record, manifest, manifest_path.parent, models, constant, access, "eval_v4", states
)
for (role, qid), arrays in scene.items():
path = out / "predictions" / f"{sid}_{role}_{qid}.npz"
path.parent.mkdir(parents=True, exist_ok=True)
np.savez_compressed(path, **arrays)
entries[f"{sid}/{role}/{qid}"] = {"path": str(path), "sha256": sha(path)}
if any(
"depth" in e.get("channels", []) and e.get("purpose") != "CONTEXT_ONLY_RGBD" for e in access
):
raise AssertionError("Only context depth may be read before the seal")
write(out / "state_hashes.json", states)
write(out / "GT_access_before_seal.json", access)
write(
out / "prediction_seal.json",
{
"predictions": entries,
"state_hashes_sha256": sha(out / "state_hashes.json"),
"query_depth_read": False,
"near_px": config["eval_v3"]["near_px"],
"reprojection_fill_constant_m": constant,
"replay_on_eval_v3": replayed,
},
)
print("EVAL_V4_PREDICTIONS_SEALED", len(entries), json.dumps(replayed), flush=True)
def score(root):
root = Path(root).resolve()
lock_ = read(root / "lock.json")
seal_file = read(root / "raw" / "prediction_seal.json")
manifest = read(lock_["input_sha256"]["eval_v4_manifest"]["path"])
seeds = read(lock_["input_sha256"]["v11_contract"]["path"])["seeds"]
rows, access = [], []
for record in manifest["scenes"]:
sid = record["scene_id"]
frames = {f["frame_id"]: f for f in record["frames"]}
for role in ("A", "B"):
for qid in record["roles"]["primary_query"]:
entry = seal_file["predictions"][f"{sid}/{role}/{qid}"]
if sha(entry["path"]) != entry["sha256"]:
raise PermissionError("A sealed prediction changed")
arrays = dict(np.load(entry["path"]))
access.append(
{"scene_id": sid, "frame_id": qid, "purpose": "POST_SEAL_QUERY_DEPTH"}
)
gt = np.load(frames[qid]["depth"]).squeeze().astype(np.float64)
valid = np.isfinite(gt) & (gt > 0)
near = arrays["near"].astype(bool)
masks = {"ALL": valid, "NEAR": valid & near, "FAR": valid & ~near}
predictions = {
("REPROJ_NN", 0): arrays["REPROJ_NN"],
("REPROJ_HARMONIC", 0): arrays["REPROJ_HARMONIC"],
("REPROJ_TELEA", 0): arrays["REPROJ_TELEA"],
}
for variant in PRIMARY_PAIR:
for seed in seeds:
carrier = arrays[f"{variant}_{seed}"]
predictions[variant, seed] = carrier
predictions[f"FILL8_{variant}", seed] = np.where(
near, arrays["REPROJ_NN"], carrier
)
predictions[f"HFILL8_{variant}", seed] = np.where(
near, arrays["REPROJ_HARMONIC"], carrier
)
predictions[f"TFILL8_{variant}", seed] = np.where(
near, arrays["REPROJ_TELEA"], carrier
)
for (method, seed), value in predictions.items():
row = {
"cohort": "EVAL_V4",
"scene_id": sid,
"role": role,
"query_id": qid,
"method": method,
"seed": seed,
}
for region, mask in masks.items():
row[f"absrel_{region}"] = (
float(np.mean(np.abs(value[mask] - gt[mask]) / gt[mask]))
if mask.any()
else None
)
row["depth_absrel"] = depth_metrics(value, gt)["depth_absrel"]
row["far_fraction"] = float(masks["FAR"].sum() / max(valid.sum(), 1))
rows.append(row)
write(root / "raw" / "query_rows.json", rows)
write(root / "raw" / "GT_access_after_seal.json", access)
print("EVAL_V4_SCORED", len(rows), flush=True)
return rows
def inpaint_block(rows):
"""INPAINT-V1's per-cohort analysis block (same contrasts, aggregation and labels)."""
v11, label = host(), inpainting().geometry().label
methods = sorted({r["method"] for r in rows})
def values(method, region):
return v11.scene_values(rows, method, f"absrel_{region}")
block = {
"absolute_absrel": {
m: {region: v11.summary(values(m, region)) for region in REGIONS} for m in methods
},
"contrasts": {},
"n_scenes": len({r["scene_id"] for r in rows}),
"far_pixel_fraction": float(
np.mean([r["far_fraction"] for r in rows if r["method"] == "REPROJ_NN"])
),
}
for name, (a, ra, b, rb) in CONTRASTS.items():
stats = v11.contrast(values(a, ra), values(b, rb))
entry = {"gain": stats, "definition": f"AbsRel {ra}({a}) - AbsRel {rb}({b})"}
if stats is not None:
entry["label"] = label(stats)
entry["strict_gate"], entry["strict_checks"] = v11.gate(stats)
block["contrasts"][name] = entry
return block
def analyze(root):
root = Path(root).resolve()
rows_path = root / "raw" / "query_rows.json"
rows = read(rows_path)
block = inpaint_block(rows)
v11_rows = [r for r in rows if r["method"] in {"REPROJ_NN", "C0", "C1", "FILL8_C0", "FILL8_C1"}]
v11_gates = host().analyze_rows(v11_rows)
primary = block["contrasts"]["HARMONIC_HYBRID_GAIN"]
result = {
"eval_v4": block,
"v11_gates_on_eval_v4": v11_gates,
"PRIMARY_LABEL": primary.get("label", "UNDEFINED"),
"PRIMARY_STRICT_GATE": primary.get("strict_gate", "UNDEFINED"),
"REPLICATION_BRANCH": {"ABOVE": "A", "NOT_DISTINGUISHABLE": "B", "BELOW": "C"}.get(
primary.get("label"), "UNDEFINED"
),
"V11_BRANCH_ON_EVAL_V4": v11_gates["INTERPRETATION_BRANCH"],
"replay_on_eval_v3": read(root / "raw" / "prediction_seal.json")["replay_on_eval_v3"],
"input_sha256": {str(rows_path): sha(rows_path)},
"final_holdout_opened": False,
}
write(root / "replication_results.json", result)
report(root, result)
return result
def fmt(stats):
if not stats:
return "UNDEFINED"
lo, hi = stats["ci95"]
return f"{stats['mean']:+.4f} [{lo:+.4f}, {hi:+.4f}]，场景 {stats['positive']}/{stats['tied']}/{stats['negative']}"
def report(root, result):
block, gates = result["eval_v4"], result["v11_gates_on_eval_v4"]
lines = [
f"# {EXPERIMENT}",
"",
f"在从未读取过的 EVAL-V4（{block['n_scenes']} 个场景，FAR 像素占 {block['far_pixel_fraction']:.1%}）上复现 V11 与 INPAINT-V1。V11 已选中的 checkpoint、区域定义、补洞方法与统计规则全部冻结，由 V11 出结果之前写定的计划规定。不训练。",
"",
f"- 主对比：AbsRel(REPROJ_HARMONIC) − AbsRel(HFILL8_C1) = {fmt(block['contrasts']['HARMONIC_HYBRID_GAIN']['gain'])}；标签 {result['PRIMARY_LABEL']}，严格 gate {result['PRIMARY_STRICT_GATE']}，复现分支 {result['REPLICATION_BRANCH']}。",
f"- V11 的 COMPLETION_GAIN = {fmt(gates['COMPLETION_GAIN'])}（{gates['statuses']['COMPLETION_STATUS']}）；HYBRID_GAIN = {fmt(gates['HYBRID_GAIN'])}（{gates['statuses']['HYBRID_STATUS']}）；V11 分支 {result['V11_BRANCH_ON_EVAL_V4']}。",
f"- 流程复现：在 EVAL-V3 的前 {result['replay_on_eval_v3']['scenes']} 个场景上重算的 {result['replay_on_eval_v3']['arrays_compared']} 个数组，与 V11 和 INPAINT-V1 的封存文件最大差 {result['replay_on_eval_v3']['max_abs_deviation']:.1e}。",
"",
"| 方法 | ALL | NEAR | FAR |",
"|---|---:|---:|---:|",
]
for method in (
"REPROJ_NN",
"REPROJ_HARMONIC",
"REPROJ_TELEA",
"C0",
"C1",
"FILL8_C1",
"HFILL8_C0",
"HFILL8_C1",
"TFILL8_C1",
):
cells = [block["absolute_absrel"].get(method, {}).get(r) for r in REGIONS]
lines.append(
f"| {method} | "
+ " | ".join("—" if not c else f"{c['mean']:.4f}" for c in cells)
+ " |"
)
lines += ["", "| 对比 | 定义 | 差值 [95% CI]，场景 好/平/差 | 标签 |", "|---|---|---|---|"]
for name, entry in block["contrasts"].items():
lines.append(
f"| {name} | {entry['definition']} | {fmt(entry['gain'])} | {entry.get('label', 'UNDEFINED')} |"
)
lines += [
"",
"说明：EVAL-V4 与 EVAL-V3 一样是机制队列（按场景和源资产与训练集及 EVAL-V3 不相交，可能共享 volume），不是资格验证；受保护的 final holdout 未打开。",
]
(root / "README.md").write_text("\n".join(lines) + "\n")
summary = [
"RGBD REPLICATION EVAL-V4",
f"PRIMARY_LABEL={result['PRIMARY_LABEL']}",
f"PRIMARY_STRICT_GATE={result['PRIMARY_STRICT_GATE']}",
f"REPLICATION_BRANCH={result['REPLICATION_BRANCH']}",
f"V11_BRANCH_ON_EVAL_V4={result['V11_BRANCH_ON_EVAL_V4']}",
f"COMPLETION_STATUS={gates['statuses']['COMPLETION_STATUS']}",
f"HYBRID_STATUS={gates['statuses']['HYBRID_STATUS']}",
f"N_SCENES={block['n_scenes']}",
]
for name, entry in block["contrasts"].items():
if entry["gain"]:
summary.append(
f"{name}={entry['gain']['mean']:.6f} CI={entry['gain']['ci95']} LABEL={entry['label']}"
)
summary += [
f"REPLAY_MAX_DEVIATION={result['replay_on_eval_v3']['max_abs_deviation']}",
"FINAL_HOLDOUT_TOUCHED=false",
]
(root / "terminal_summary.txt").write_text("\n".join(summary) + "\n")
print("\n".join(summary), flush=True)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--stage", choices=("lock", "seal", "score", "analyze"), required=True)
parser.add_argument("--root", type=Path, default=Path("outputs") / EXPERIMENT)
args = parser.parse_args()
torch.set_num_threads(1)
{"lock": lock, "seal": seal, "score": score, "analyze": analyze}[args.stage](args.root)
if __name__ == "__main__":
main()