scripts/addendum2_rgbd_geometric_baselines.py
#!/usr/bin/env python3
"""Addendum 2 to EXP-3D-RGBD-GEOMETRIC-BASELINES-V1: the sealed V9 carriers vs the sealed baselines.
Descriptive only. Reuses the sealed REPROJ_NN baseline files (hash-checked) and the V7 masks;
scores the V9 sealed DEV carrier predictions (RAW) and their HYBRIDs with the frozen depth
metric. No model is run, no baseline is recomputed.
"""
# ruff: noqa: E501 -- Chinese scientific report prose
import argparse
import hashlib
import importlib.util
import json
from collections import defaultdict
from pathlib import Path
import numpy as np
from mcss.mechanism_pilot.direct_capacity_statistics import describe, paired
from mcss.mechanism_pilot.readout_reanalysis import depth_metrics
ROOT = Path("outputs/EXP-3D-RGBD-GEOMETRIC-BASELINES-V1")
V7 = Path("outputs/EXP-3D-RGBD-DEPTH-BOUNDS-V7")
V9 = Path("outputs/EXP-3D-RGBD-VIEWCOUNT-V9")
CARRIERS = {
"DEV": {
"V9_C0": (V9 / "raw/selected_dev", "C0", "RAW"),
"V9_C1": (V9 / "raw/selected_dev", "C1", "RAW"),
"V9_C2": (V9 / "raw/selected_dev", "C2", "RAW"),
},
}
MANIFESTS = {"DEV": (V7 / "scene_split.json", "DEV")}
MASKS = {"DEV": V7 / "raw/observability"}
def sha(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def read(path):
return json.loads(Path(path).read_text())
def main_module():
spec = importlib.util.spec_from_file_location("geo", "scripts/run_rgbd_geometric_baselines.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def readout(path, kind):
arrays = np.load(path)
depth = arrays["depth"].astype(np.float64)
if kind == "NORM":
return depth / np.maximum(arrays["opacity"].astype(np.float64), 1e-6)
return depth
def run(root=ROOT):
root = Path(root)
if (root / "addendum2_results.json").exists():
raise FileExistsError("Addendum 2 runs once")
if not (root / "ADDENDUM2.md").is_file():
raise PermissionError("Addendum text must be written first")
geo = main_module()
geo.verify(root.resolve())
seal = read(root / "raw/baseline_seal.json")
seeds = read(V9 / "training_contract.json")["seeds"]
rows = []
for cohort, carriers in CARRIERS.items():
manifest_path, split = MANIFESTS[cohort]
manifest = read(manifest_path)
records = {r["scene_id"]: r for r in manifest["scenes"] if r["split"] == split}
sealed_rows = {}
for name, (directory, variant, _) in carriers.items():
for seed in seeds:
run_dir = directory / f"{variant}_{seed}"
for r in read(run_dir / "query_results.json"):
if r["method"] == "direct":
path = run_dir / r["prediction_path"]
if sha(path) != r["prediction_file_sha256"]:
raise PermissionError(f"Sealed prediction changed: {path}")
sealed_rows[name, seed, r["scene_id"], r["role"], r["query_id"]] = path
for sid, record in records.items():
frames = {f["frame_id"]: f for f in record["frames"]}
for role in ("A", "B"):
for qid in record["roles"]["primary_query"]:
entry = seal[f"{cohort}/{sid}/{role}/{qid}"]
if sha(entry["path"]) != entry["sha256"]:
raise AssertionError("Sealed baseline changed")
baseline = dict(np.load(entry["path"]))
gt = np.load(frames[qid]["depth"]).squeeze().astype(np.float64)
obs = np.load(MASKS[cohort] / f"{sid}_{role}_{qid}.npz")["obs_class"]
common = {"cohort": cohort, "scene_id": sid, "role": role, "query_id": qid}
rows.append(
{
**common,
"seed": 0,
"method": "REPROJ_NN",
**depth_metrics(baseline["REPROJ_NN"], gt),
"regions": geo.region_rows(baseline["REPROJ_NN"], gt, obs),
}
)
for name, (_, _, kind) in carriers.items():
for seed in seeds:
prediction = readout(sealed_rows[name, seed, sid, role, qid], kind)
hybrid = np.where(baseline["hit"], baseline["reproj"], prediction)
for method, value in ((name, prediction), (f"HYBRID_{name}", hybrid)):
rows.append(
{
**common,
"seed": seed,
"method": method,
**depth_metrics(value, gt),
"regions": geo.region_rows(value, gt, obs),
}
)
result = {"cohorts": {}, "descriptive_only": True}
for cohort, carriers in CARRIERS.items():
methods = ("REPROJ_NN", *carriers, *(f"HYBRID_{n}" for n in carriers))
values = {m: geo.aggregate(rows, m, cohort) for m in methods}
block = {"absolute_depth_absrel": {}, "vs_reproj_nn": {}, "regions": {}, "delta1": {}}
for method in methods:
block["absolute_depth_absrel"][method] = describe(
values[method], draws=geo.DRAWS, seed=geo.SEED
)
block["delta1"][method] = float(
np.mean(list(geo.aggregate(rows, method, cohort, "depth_delta1").values()))
)
regions = {}
for region in ("OBS0", "OBS1", "OBS2PLUS"):
grouped = defaultdict(list)
for r in rows:
if (
r["method"] == method
and r["cohort"] == cohort
and r["regions"][region] is not None
):
grouped[r["scene_id"]].append(r["regions"][region])
regions[region] = float(np.mean([np.mean(v) for v in grouped.values()]))
block["regions"][method] = regions
if method != "REPROJ_NN":
stats = paired(values["REPROJ_NN"], values[method], draws=geo.DRAWS, seed=geo.SEED)
block["vs_reproj_nn"][method] = {"gain": stats, "label": geo.label(stats)}
result["cohorts"][cohort] = block
dev = result["cohorts"]["DEV"]["absolute_depth_absrel"]
first = read(root / "addendum1_results.json")["cohorts"]["DEV"]["absolute_depth_absrel"]
result["integrity"] = {"v9_c0_equals_v8_c1": abs(dev["V9_C0"]["mean"] - first["V8_C1"]["mean"])}
(root / "addendum2_results.json").write_text(
json.dumps(result, indent=2, sort_keys=True) + "\n"
)
lines = ["# 补充分析 2：训练视角数（V9，描述性）", ""]
for cohort, block in result["cohorts"].items():
lines += [
f"## {cohort}",
"",
"| 方法 | AbsRel | delta1 | OBS0 / OBS1 / OBS2PLUS | 相对 REPROJ_NN（基线−方法） | 标签 |",
"|---|---|---|---|---|---|",
]
for method, v in block["absolute_depth_absrel"].items():
reg = block["regions"][method]
comparison = block["vs_reproj_nn"].get(method)
gain = (
f"{comparison['gain']['mean']:+.4f} [{comparison['gain']['ci95'][0]:+.4f}, {comparison['gain']['ci95'][1]:+.4f}]"
if comparison
else "—"
)
lines.append(
f"| {method} | {v['mean']:.4f} | {block['delta1'][method]:.3f} | {reg['OBS0']:.3f} / {reg['OBS1']:.3f} / {reg['OBS2PLUS']:.3f} | {gain} | {comparison['label'] if comparison else '—'} |"
)
lines.append("")
lines.append(
f"完整性：V9_C0 与补充分析 1 中 V8_C1 的差 {result['integrity']['v9_c0_equals_v8_c1']:.2e}。"
)
(root / "ADDENDUM2_RESULTS.md").write_text("\n".join(lines) + "\n")
print("\n".join(lines), flush=True)
return result
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, default=ROOT)
run(parser.parse_args().root)