scripts/run_readout_reanalysis.py
#!/usr/bin/env python3
"""Execute the frozen readout/reference re-analysis: hashes, DEV GT, rows, statistics."""
from __future__ import annotations
import argparse
import json
import shutil
from pathlib import Path
import numpy as np
from mcss.mechanism_pilot.readout_reanalysis import (
EXPERIMENT,
METRICS,
READOUTS,
SOURCES,
VARIANTS,
analyze_with_sensitivity,
score_predictions,
score_references,
validate_matrix,
)
from mcss.mechanism_pilot.small_training import sha, write_json
def verify(lock):
if lock["status"] != "FROZEN_BEFORE_DEV_GT_ANALYSIS" or lock["experiment"] != EXPERIMENT:
raise PermissionError("Frozen re-analysis lock required")
for group in ("source_sha256", "input_sha256", "train_reference_files_sha256"):
for name, digest in lock[group].items():
if sha(Path(name)) != digest:
raise PermissionError(f"Frozen file changed: {name}")
for p in lock["predictions"]:
if sha(Path(p["path"])) != p["sha256"]:
raise PermissionError(f"Sealed prediction changed: {p['path']}")
def dev_gt(manifest, access):
"""DEV primary-query GT depth, read once per frame after every hash is verified."""
gt = {}
for record in manifest["scenes"]:
if record["split"] != "DEV":
continue
frames = {f["frame_id"]: f for f in record["frames"]}
for fid in record["roles"]["primary_query"]:
path = Path(frames[fid]["depth"])
access.append(
{
"scene_id": record["scene_id"],
"frame_id": fid,
"channel": "depth",
"purpose": "POST_SEAL_REANALYSIS_DEV_PRIMARY_QUERY",
"sha256": sha(path),
}
)
data = np.load(path).squeeze().astype(np.float32)
valid = np.isfinite(data) & (data > 0)
gt[record["scene_id"], fid] = np.where(valid, data, 0).astype(np.float32)
return gt
def table(result, readout, metric):
lines = [
"| carrier | mean | vs primary reference | label | vs anchor |",
"|---|---:|---:|---|---|",
]
for name, carrier in result["carriers"].items():
entry = carrier[readout][metric]
ref = entry[f"vs_{result['primary_reference'][metric]}"]
static = result["static"][name][readout][metric]["label"]
lines.append(
f"| {name} | {entry['absolute_mean']:.4f} | {ref['mean']:+.4f} "
f"[{ref['ci95'][0]:+.4f}, {ref['ci95'][1]:+.4f}] | {entry['reference_label']} | "
f"{static} |"
)
return "\n".join(lines)
def readme(result, lock):
refs = lock["references"]
out = [
f"# {EXPERIMENT}",
"",
f"`CARRIER_REFERENCE_STATUS={result['summary']['CARRIER_REFERENCE_STATUS']}`。",
"",
"这是对 V2-V4 已封存 DEV 预测的事后二次分析：不训练、不加载模型、不改变任何冻结状态。"
"参考常数只由 TRAIN primary-query GT 拟合："
f"REF_TRAIN_ABSREL_OPTIMAL={refs['REF_TRAIN_ABSREL_OPTIMAL']:.4f} m（AbsRel 最优常数），"
f"REF_TRAIN_MEDIAN={refs['REF_TRAIN_MEDIAN']:.4f} m。"
"差值为正表示 carrier 更好（AbsRel 更低 / delta1 更高）；"
"CI 为 10000 次 scene-paired bootstrap。",
"",
]
sensitivity = result["sensitivity_excluding_no_hit_scenes"]
out += [
"Amendment 1（见 AMENDMENTS.md，在任何指标计算之前作出）：预测中位数不为正的视角"
"（全部射线落在 bounds 之外、渲染全零的 NO_HIT 视角）在 MEDIAN_SCALED 下保持不缩放；"
f"受影响视角数={result['integrity']['median_scale_undefined_views']}。"
f"排除全 NO_HIT 场景 {sensitivity['excluded_no_hit_scenes']} 的次要敏感性分析："
+ (
f"`CARRIER_REFERENCE_STATUS={sensitivity['summary']['CARRIER_REFERENCE_STATUS']}`，"
f"ABOVE 计数={sensitivity['summary']['carriers_above_reference']}。"
if "summary" in sensitivity
else "无此类场景。"
),
"",
]
for readout in READOUTS:
for metric in METRICS:
out += [f"## {readout} · {metric}", "", table(result, readout, metric), ""]
out += [
"## 冻结主对比在不同 readout 下（仅描述）",
"",
"| source | contrast | readout | metric | gain | 95% CI | label |",
"|---|---|---|---|---:|---:|---|",
]
for source, entry in result["primary_contrasts"].items():
for readout in READOUTS:
for metric in METRICS:
c = entry["by_readout"][readout][metric]
out.append(
f"| {source} | {entry['name']} | {readout} | {metric} | "
f"{c['stats']['mean']:+.4f} | [{c['stats']['ci95'][0]:+.4f}, "
f"{c['stats']['ci95'][1]:+.4f}] | {c['label']} |"
)
out += [
"",
"## Opacity normalization 的作用（正数 = 归一化后更好）",
"",
"| carrier | AbsRel effect | label | delta1 effect | label |",
"|---|---:|---|---:|---|",
]
for name, effect in result["readout_effect"].items():
a, d = effect["depth_absrel"], effect["depth_delta1"]
out.append(
f"| {name} | {a['stats']['mean']:+.4f} | {a['label']} | "
f"{d['stats']['mean']:+.4f} | {d['label']} |"
)
return "\n".join(out) + "\n"
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=f"outputs/{EXPERIMENT}")
root = Path(parser.parse_args().root)
lock = json.loads((root / "preregistration.json").read_text())
verify(lock)
manifest = json.loads(Path(SOURCES["V4"]["root"], "scene_split.json").read_text())
access = []
gt = dev_gt(manifest, access)
write_json(root / "audit" / "gt_access.json", access)
rows, worst = score_predictions(lock["predictions"], gt)
queries = {
r["scene_id"]: list(r["roles"]["primary_query"])
for r in manifest["scenes"]
if r["split"] == "DEV"
}
reference_rows = score_references(lock["references"], gt, queries)
seeds = lock["sources"]["V2"]["seeds"]
validate_matrix(rows, reference_rows, seeds, sorted(queries), queries)
write_json(root / "raw" / "rows.json", rows)
write_json(root / "raw" / "reference_rows.json", reference_rows)
result = analyze_with_sensitivity(rows, reference_rows)
result["primary_reference"] = lock["analysis"]["primary_reference"]
result["integrity"] = {
"max_raw_absrel_deviation": worst["absrel"],
"max_raw_delta1_pixel_deviation": worst["delta1_pixels"],
"median_scale_undefined_views": worst["median_scale_undefined_views"],
"rows": len(rows),
"reference_rows": len(reference_rows),
"variants": list(VARIANTS),
}
write_json(root / "results.json", result)
(root / "README.md").write_text(readme(result, lock))
docs = Path("docs/experiments") / EXPERIMENT
docs.mkdir(parents=True, exist_ok=False)
for name in ("preregistration.json", "results.json", "README.md", "AMENDMENTS.md"):
if (root / name).exists():
shutil.copyfile(root / name, docs / name)
shutil.copyfile(root / "audit" / "gt_access.json", docs / "gt_access.json")
print(
json.dumps(
{
"status": result["summary"]["CARRIER_REFERENCE_STATUS"],
"above": result["summary"]["carriers_above_reference"],
"integrity": result["integrity"],
},
indent=1,
)
)
if __name__ == "__main__":
main()