scripts/evaluate_rgbd_volume_abstention.py
#!/usr/bin/env python3
"""V13: abstain outside the state's volume (no training).
The frozen V11 C1 selected checkpoints render depth and opacity for queries far from the context
frames (EVAL-FAR) and for the near queries of EVAL-V3/V4. AHFILL8 uses REPROJ_HARMONIC on NEAR
pixels and on FAR pixels whose rendered opacity is below 0.5, the carrier elsewhere, as fixed by
PROTOCOL.md. Stages: `lock` freezes the protocol, code and inputs before any far query is rendered;
`seal --cohort` replays two sealed EVAL-V3 scenes, then renders and hashes every prediction and
geometric baseline of one cohort before any query depth is read; `score` needs every cohort's 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 generate_rays, intersect_aabb, transform_cameras
from mcss.measurements import FixedMeasurementRenderer
from mcss.mechanism_pilot import rgbd_completion_carrier as v11_carrier
from mcss.mechanism_pilot.readout_reanalysis import depth_metrics
from mcss.mechanism_pilot.rgbd_depth_bounds import context_depth_bounds
from mcss.types import Cameras
EXPERIMENT = "EXP-3D-RGBD-VOLUME-V13"
V11 = Path("outputs/EXP-3D-RGBD-COMPLETION-V11")
INPAINT = Path("outputs/EXP-3D-RGBD-INPAINT-BASELINES-V1")
EVAL_FAR = Path("outputs/EXP-3D-RGBD-EVAL-FAR-DATA")
EXPLORATORY = "audit/exploratory_opacity_fallback"
COHORTS = {
"EVAL_FAR": (EVAL_FAR / "manifest_eval_far.json", "far_query", "EVAL_FAR"),
"EVAL_V3_NEAR": (
Path("outputs/EXP-3D-RGBD-EVAL-V3-DATA/manifest_eval_v3.json"),
"primary_query",
"EVAL_V3",
),
"EVAL_V4_NEAR": (
Path("outputs/EXP-3D-RGBD-EVAL-V4-DATA/manifest_eval_v4.json"),
"primary_query",
"EVAL_V4",
),
}
PRIMARY = "EVAL_FAR"
THRESHOLD = 0.5
DESCRIPTIVE_THRESHOLDS = (0.3, 0.7, 0.9)
REGIONS = ("ALL", "NEAR", "FAR")
REPLAY_SCENES = 2
METHODS = (
"CONSTANT",
"REPROJ_NN",
"REPROJ_HARMONIC",
"C1",
"HFILL8",
"AHFILL8",
*(f"AHFILL8_t{t}" for t in DESCRIPTIVE_THRESHOLDS),
)
CONTRASTS = {
"ABSTAIN_GAIN": ("HFILL8", "ALL", "AHFILL8", "ALL"),
"HARMONIC_VS_AHFILL8": ("REPROJ_HARMONIC", "ALL", "AHFILL8", "ALL"),
"HARMONIC_VS_HFILL8": ("REPROJ_HARMONIC", "ALL", "HFILL8", "ALL"),
"NN_VS_AHFILL8": ("REPROJ_NN", "ALL", "AHFILL8", "ALL"),
"CONSTANT_VS_C1": ("CONSTANT", "ALL", "C1", "ALL"),
"NN_VS_C1": ("REPROJ_NN", "ALL", "C1", "ALL"),
"HARMONIC_VS_C1_FAR": ("REPROJ_HARMONIC", "FAR", "C1", "FAR"),
"HARMONIC_VS_AHFILL8_FAR": ("REPROJ_HARMONIC", "FAR", "AHFILL8", "FAR"),
**{
f"HARMONIC_VS_AHFILL8_t{t}": ("REPROJ_HARMONIC", "ALL", f"AHFILL8_t{t}", "ALL")
for t in DESCRIPTIVE_THRESHOLDS
},
}
SCRIPTS = Path(__file__).resolve().parent
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 and V2 gate."""
return module("v13_v11_eval", "evaluate_rgbd_completion_eval_v3.py")
@functools.cache
def inpainting():
"""INPAINT-V1's frozen module: ray cosine, harmonic filling, NEAR mask, label rule."""
return module("v13_inpaint", "run_rgbd_inpaint_baselines.py")
def abstain(near, opacity, harmonic, carrier, threshold):
"""AHFILL8: harmonic on NEAR pixels and on FAR pixels the carrier renders with low opacity."""
return np.where(near | (opacity < threshold), harmonic, carrier)
def inputs(root, v11=V11, inpaint=INPAINT):
paths = {
"protocol": root / "PROTOCOL.md",
"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",
"v11_selected": v11 / "selected_checkpoints.json",
"v11_eval_v3_seal": v11 / "raw/eval_v3/prediction_seal.json",
"inpaint_lock": inpaint / "lock.json",
"eval_far_candidate_lock": EVAL_FAR / "candidate_lock.json",
"eval_far_integrity": EVAL_FAR / "preparation_integrity.json",
}
for name, (manifest, _, _) in COHORTS.items():
paths[f"{name}_manifest"] = manifest
for cohort in ("EVAL_V3", "EVAL_V4", "REPLICA"):
paths[f"exploratory_{cohort}"] = root / EXPLORATORY / f"fallback_{cohort}.json"
return paths
# ---------------------------------------------------------------- stages
def lock(root, v11=V11, inpaint=INPAINT):
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 (root / "raw").exists():
raise PermissionError("No prediction may exist before the lock")
if read(EVAL_FAR / "preparation_integrity.json")["status"] != "PASS":
raise PermissionError("EVAL-FAR data must be prepared with integrity PASS")
paths = inputs(root, v11, inpaint)
write(
root / "lock.json",
{
"status": "FROZEN_BEFORE_ANY_FAR_QUERY_PREDICTION",
"experiment": EXPERIMENT,
"created_utc": datetime.datetime.now(datetime.UTC).isoformat(),
"threshold": THRESHOLD,
"primary": "EVAL_FAR: P1 AbsRel(HFILL8) - AbsRel(AHFILL8) (frozen V2 gate); P2 AbsRel(REPROJ_HARMONIC) - AbsRel(AHFILL8) (INPAINT-V1 label rule)",
"input_sha256": {
name: {"path": str(Path(p).resolve()), "sha256": sha(p)}
for name, p in paths.items()
},
"far_query_rendered_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")["C1"]
models = {}
for seed in config["seeds"]:
item = selected[str(seed)]
if sha(Path(item["path"])) != item["sha256"]:
raise PermissionError("Selected V11 C1 checkpoint changed")
models[seed] = v11_carrier.load_checkpoint(item["path"], "cpu")[0].eval()
return config, models
@torch.no_grad()
def predict(record, manifest, base, models, constant, role_name, access, tag, states=None):
"""Every sealed array of every query of one scene: carrier depth and opacity, box, geometry.
`tag` only names the construction episode ("eval_v3" reproduces V11's own episode ids).
"""
stage, inp = host(), inpainting()
geo = stage.geometry()
renderer = FixedMeasurementRenderer(n_samples=64, ray_chunk_size=2048)
out, sid = {}, record["scene_id"]
for role, context in stage.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 seed, model in models.items():
state, state_anchor = v11_carrier.build_state(
model, context, bounds, f"{tag}:{sid}:{role}"
)
built[seed] = (state, state_anchor)
if states is not None:
states[f"C1:{seed}:{sid}/{role}"] = hash_scene_state(state)
box_state, box_anchor = next(iter(built.values()))
if not all(
torch.equal(s.bounds, box_state.bounds) and torch.equal(a, box_anchor)
for s, a in built.values()
):
raise AssertionError("All seeds of one context must share the volume and the anchor")
for qid in record["roles"][role_name]:
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)
arrays = {
"REPROJ_NN": reproj,
"hit": hit,
"near": inp.near_mask(hit),
"REPROJ_HARMONIC": inp.harmonic_fill(depth, hit, inp.ray_cosine(camera), constant),
}
batched = Cameras(
camera.intrinsics[None, None], camera.c2w[None, None], camera.image_size
)
box_local = transform_cameras(batched, torch.linalg.inv(box_anchor))
origins, directions = generate_rays(box_local.to(dtype=torch.float64))
t_near, t_exit, box_hit = intersect_aabb(
origins[0, 0], directions[0, 0], box_state.bounds[0].double()
)
arrays |= {
"BOX_NEAR": t_near.numpy(),
"BOX_EXIT": t_exit.numpy(),
"BOX_HIT": box_hit.numpy(),
}
for seed, (state, state_anchor) in built.items():
local = transform_cameras(batched, torch.linalg.inv(state_anchor))
pred = renderer(state, local, ("depth", "visibility"))
arrays[f"C1_{seed}"] = pred["depth"][0, 0, 0].numpy().astype(np.float64)
arrays[f"OPACITY_{seed}"] = pred["visibility"][0, 0, 0].numpy()
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."""
manifest_path, role_name, _ = COHORTS["EVAL_V3_NEAR"]
manifest = read(manifest_path)
seal_ = read(v11 / "raw/eval_v3/prediction_seal.json")["predictions"]
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, role_name, [], "eval_v3"
)
for (role, qid), arrays in mine.items():
sealed = dict(np.load(seal_[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"]),
(arrays["REPROJ_HARMONIC"], filled["REPROJ_HARMONIC"]),
]
pairs += [(arrays[f"C1_{seed}"], sealed[f"C1_{seed}"]) for seed in models]
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, cohort, v11=V11, inpaint=INPAINT):
root = Path(root).resolve()
verify(root, v11)
out = root / "raw" / cohort
if (out / "prediction_seal.json").exists():
raise FileExistsError("Each cohort is sealed once")
if (root / "raw" / "query_rows.json").exists():
raise PermissionError("No cohort may be sealed after scoring")
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, role_name, split = COHORTS[cohort]
manifest = read(manifest_path)
if any(r["split"] != split for r in manifest["scenes"]):
raise PermissionError(f"{cohort}: {split} records only")
replayed = replay(models, constant, v11, inpaint)
print("REPLAY_OK", cohort, json.dumps(replayed), flush=True)
access, entries, states = [], {}, {}
for number, record in enumerate(manifest["scenes"], 1):
sid = record["scene_id"]
scene = predict(
record,
manifest,
manifest_path.parent,
models,
constant,
role_name,
access,
cohort.lower(),
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)}
print("SEALED_SCENE", cohort, number, len(manifest["scenes"]), sid, flush=True)
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",
{
"cohort": cohort,
"role": role_name,
"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("PREDICTIONS_SEALED", cohort, len(entries), flush=True)
def absrel(pred, gt, mask):
return float(np.mean(np.abs(pred[mask] - gt[mask]) / gt[mask])) if mask.any() else None
def score(root):
root = Path(root).resolve()
lock_ = verify(root)
seals = {c: read(root / "raw" / c / "prediction_seal.json") for c in COHORTS}
seeds = read(lock_["input_sha256"]["v11_contract"]["path"])["seeds"]
rows, flags, access = [], [], []
for cohort, (manifest_path, role_name, _) in COHORTS.items():
seal_ = seals[cohort]
constant = seal_["reprojection_fill_constant_m"]
for record in read(manifest_path)["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"][role_name]:
entry = seal_["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(
{
"cohort": cohort,
"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)
far = valid & ~near
harmonic = arrays["REPROJ_HARMONIC"]
inside = (
arrays["BOX_HIT"].astype(bool)
& (gt >= arrays["BOX_NEAR"])
& (gt <= arrays["BOX_EXIT"])
)
masks = {"ALL": valid, "NEAR": valid & near, "FAR": far}
predictions = {
("CONSTANT", 0): np.full(gt.shape, constant, dtype=np.float64),
("REPROJ_NN", 0): arrays["REPROJ_NN"],
("REPROJ_HARMONIC", 0): harmonic,
}
for seed in seeds:
carrier = arrays[f"C1_{seed}"]
opacity = arrays[f"OPACITY_{seed}"].astype(np.float64)
predictions["C1", seed] = carrier
predictions["HFILL8", seed] = np.where(near, harmonic, carrier)
predictions["AHFILL8", seed] = abstain(
near, opacity, harmonic, carrier, THRESHOLD
)
for t in DESCRIPTIVE_THRESHOLDS:
predictions[f"AHFILL8_t{t}", seed] = abstain(
near, opacity, harmonic, carrier, t
)
flagged = far & (opacity < THRESHOLD)
flags.append(
{
"cohort": cohort,
"scene_id": sid,
"role": role,
"query_id": qid,
"seed": seed,
"surface_in_volume": float(inside[valid].mean()),
"far_fraction": float(far.sum() / max(valid.sum(), 1)),
"flagged_share_of_far": float(flagged.sum() / max(far.sum(), 1)),
"flagged": float(flagged.sum()),
"flagged_outside": float((flagged & ~inside).sum()),
"far_outside": float((far & ~inside).sum()),
}
)
for (method, seed), value in predictions.items():
row = {
"cohort": cohort,
"scene_id": sid,
"role": role,
"query_id": qid,
"method": method,
"seed": seed,
}
for region, mask in masks.items():
row[f"absrel_{region}"] = absrel(value, gt, mask)
row["depth_absrel"] = depth_metrics(value, gt)["depth_absrel"]
row["far_fraction"] = float(far.sum() / max(valid.sum(), 1))
rows.append(row)
write(root / "raw" / "query_rows.json", rows)
write(root / "raw" / "flag_rows.json", flags)
write(root / "raw" / "GT_access_after_seal.json", access)
print("SCORED", len(rows), flush=True)
return rows
def cohort_block(rows, flags):
stage, label = host(), inpainting().geometry().label
def values(method, region):
return stage.scene_values(rows, method, f"absrel_{region}")
methods = sorted({r["method"] for r in rows})
block = {
"absolute_absrel": {
m: {region: stage.summary(values(m, region)) for region in REGIONS} for m in methods
},
"n_scenes": len({r["scene_id"] for r in rows}),
"contrasts": {},
}
for name, (a, ra, b, rb) in CONTRASTS.items():
stats = stage.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["gate"], entry["gate_checks"] = stage.gate(stats)
block["contrasts"][name] = entry
per_scene = {}
for f in flags:
per_scene.setdefault(f["scene_id"], []).append(f)
def scene_mean(key):
return float(np.mean([np.mean([f[key] for f in v]) for v in per_scene.values()]))
pooled = {
key: sum(f[key] for f in flags) for key in ("flagged", "flagged_outside", "far_outside")
}
block["coverage"] = {
"surface_in_volume": scene_mean("surface_in_volume"),
"far_fraction": scene_mean("far_fraction"),
"flagged_share_of_far": scene_mean("flagged_share_of_far"),
"precision_outside_volume": pooled["flagged_outside"] / max(pooled["flagged"], 1),
"recall_outside_volume": pooled["flagged_outside"] / max(pooled["far_outside"], 1),
}
return block
def analyze_rows(rows, flags):
result = {"cohorts": {}}
for cohort in COHORTS:
subset = [r for r in rows if r["cohort"] == cohort]
if subset:
result["cohorts"][cohort] = cohort_block(
subset, [f for f in flags if f["cohort"] == cohort]
)
primary = result["cohorts"][PRIMARY]["contrasts"]
p1 = primary["ABSTAIN_GAIN"].get("gate", "UNDEFINED")
p2 = primary["HARMONIC_VS_AHFILL8"].get("label", "UNDEFINED")
first, second = p1 == "SUPPORTED", p2 == "ABOVE"
result |= {
"ABSTAIN_STATUS": p1,
"HARMONIC_LABEL": p2,
"INTERPRETATION_BRANCH": "A" if first and second else "B" if first or second else "C",
}
return result
def analyze(root):
root = Path(root).resolve()
rows_path, flags_path = root / "raw" / "query_rows.json", root / "raw" / "flag_rows.json"
result = analyze_rows(read(rows_path), read(flags_path))
result["replay_on_eval_v3"] = {
c: read(root / "raw" / c / "prediction_seal.json")["replay_on_eval_v3"] for c in COHORTS
}
result["input_sha256"] = {str(p): sha(p) for p in (rows_path, flags_path)}
result["final_holdout_opened"] = False
write(root / "volume_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):
primary = result["cohorts"][PRIMARY]["contrasts"]
worst = max(r["max_abs_deviation"] for r in result["replay_on_eval_v3"].values())
lines = [
f"# {EXPERIMENT}",
"",
"状态体积之外的弃权回退（不训练）：冻结的 V11 C1 在渲染不透明度 < 0.5 的 FAR 像素上弃权，改用调和插值补洞（AHFILL8）。主队列 EVAL-FAR 的 query 远离上下文帧，此前从未被任何模型读取。协议见 [PROTOCOL.md](PROTOCOL.md)。",
"",
f"- **P1 ABSTAIN_GAIN（HFILL8 − AHFILL8）：** {fmt(primary['ABSTAIN_GAIN']['gain'])}，{result['ABSTAIN_STATUS']}",
f"- **P2 调和插值补洞 − AHFILL8：** {fmt(primary['HARMONIC_VS_AHFILL8']['gain'])}，{result['HARMONIC_LABEL']}",
f"- **分支 {result['INTERPRETATION_BRANCH']}**；复现 V11 封存预测的最大差 {worst}",
]
for cohort, block in result["cohorts"].items():
cov = block["coverage"]
lines += [
"",
f"## {cohort}（{block['n_scenes']} 个场景）",
"",
f"GT 表面落在体积内 {cov['surface_in_volume']:.1%}；FAR 像素占 {cov['far_fraction']:.1%}，其中被标记 {cov['flagged_share_of_far']:.1%}；标记对体积外像素的精度 {cov['precision_outside_volume']:.2f}、召回 {cov['recall_outside_volume']:.2f}。",
"",
"| 方法 | ALL | NEAR | FAR |",
"|---|---:|---:|---:|",
]
for method in METHODS:
cells = [block["absolute_absrel"].get(method, {}).get(region) for region in REGIONS]
lines.append(
f"| {method} | "
+ " | ".join("—" if not c else f"{c['mean']:.4f}" for c in cells)
+ " |"
)
lines += [
"",
"| 对比 | 定义 | 差值 [95% CI]，场景 好/平/差 | 标签 | gate |",
"|---|---|---|---|---|",
]
for name, entry in block["contrasts"].items():
lines.append(
f"| {name} | {entry['definition']} | {fmt(entry['gain'])} | {entry.get('label', 'UNDEFINED')} | {entry.get('gate', 'UNDEFINED')} |"
)
lines += [
"",
"**限制：** EVAL-FAR 的场景已被用于近处 query，它是机制队列，不作资格验证；阈值 0.5 来自已用队列上的探索性分析；受保护的 final holdout 未打开。",
]
(root / "README.md").write_text("\n".join(lines) + "\n")
summary = [
"RGBD VOLUME V13",
f"ABSTAIN_STATUS={result['ABSTAIN_STATUS']}",
f"HARMONIC_LABEL={result['HARMONIC_LABEL']}",
f"INTERPRETATION_BRANCH={result['INTERPRETATION_BRANCH']}",
f"REPLAY_MAX_ABS_DEVIATION={worst}",
]
for cohort, block in result["cohorts"].items():
for method in ("CONSTANT", "REPROJ_HARMONIC", "C1", "HFILL8", "AHFILL8"):
value = block["absolute_absrel"].get(method, {}).get("ALL")
if value:
summary.append(f"{cohort}_{method}_ALL={value['mean']:.6f}")
for name in ("ABSTAIN_GAIN", "HARMONIC_VS_AHFILL8", "CONSTANT_VS_C1"):
entry = block["contrasts"][name]
if entry["gain"]:
summary.append(
f"{cohort}_{name}={entry['gain']['mean']:.6f} CI={entry['gain']['ci95']} LABEL={entry['label']} GATE={entry['gate']}"
)
summary.append(f"{cohort}_SURFACE_IN_VOLUME={block['coverage']['surface_in_volume']:.4f}")
summary.append("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("--cohort", choices=tuple(COHORTS))
parser.add_argument("--root", type=Path, default=Path("outputs") / EXPERIMENT)
args = parser.parse_args()
torch.set_num_threads(1)
if args.stage == "seal":
if args.cohort is None:
parser.error("--stage seal needs --cohort")
seal(args.root, args.cohort)
else:
{"lock": lock, "score": score, "analyze": analyze}[args.stage](args.root)
if __name__ == "__main__":
main()