scripts/analyze_rgbd_stream_write.py
#!/usr/bin/env python3
"""Core B V1 raw-only statistics: policy contrasts with the frozen V2 bootstrap and gate checks."""
from __future__ import annotations
import argparse
import hashlib
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.geometry_carrier_statistics import _evidence, surface_status
DRAWS, SEED = 10000, 20260928
CONTRASTS = {
"WRITE_GAIN": ("OFF", "ALL"),
"STREAM_WRITE_GAIN": ("NO_STREAM", "ALL"),
"BEYOND_CLAMP_GAIN": ("OFF_CLAMP3", "ALL"),
"WRITE_SPECIFICITY": ("ALL_WRONG_SCENE", "ALL"),
"STREAM_CACHE_EFFECT": ("NO_STREAM", "OFF"),
"CLAMP_EFFECT": ("OFF", "OFF_CLAMP3"),
"CLAMP_VS_STATIC": ("NO_STREAM", "OFF_CLAMP3"),
"FUSE_GAIN": ("OFF", "FUSE"),
"COMPLETE_GAIN": ("OFF", "COMPLETE"),
"TRAINING_GAIN": ("ALL_UNTRAINED", "ALL"),
}
GATED = {
"WRITE_STATUS": "WRITE_GAIN",
"STREAM_STATUS": "STREAM_WRITE_GAIN",
"BEYOND_CLAMP_STATUS": "BEYOND_CLAMP_GAIN",
}
def load(root, name, provenance):
path = Path(root) / name
data = path.read_bytes()
provenance[str(path.resolve())] = hashlib.sha256(data).hexdigest()
return json.loads(data)
def scene_values(rows, policy, metric="depth_absrel", seed=None):
"""Per scene: seeds equally weighted; within a seed, the mean of role means of queries."""
grouped = defaultdict(lambda: defaultdict(list))
for r in rows:
if r["policy"] == policy and (seed is None or r["seed"] == seed):
grouped[r["seed"], r["scene_id"]][r["role"]].append(r[metric])
per_scene = defaultdict(list)
for (_, sid), roles in grouped.items():
per_scene[sid].append(float(np.mean([np.mean(v) for v in roles.values()])))
return {sid: float(np.mean(v)) for sid, v in sorted(per_scene.items())}
def gate(gain, state_ok):
checks = {
"mean_ci_and_scene_consistency": _evidence(gain),
"every_leave_one_scene_out_gain_positive": bool(gain["loso"])
and min(gain["loso"].values()) > 0,
"positive_top1_share_at_most_0_5": gain["positive_top1_contribution"] is not None
and gain["positive_top1_contribution"] <= 0.5,
"state_integrity": state_ok,
}
return surface_status(gain, checks), checks
def analyze(rows, seeds, state_ok):
policies = sorted({r["policy"] for r in rows})
absolute = {p: describe(scene_values(rows, p), draws=DRAWS, seed=SEED) for p in policies}
delta1 = {
p: describe(scene_values(rows, p, "depth_delta1"), draws=DRAWS, seed=SEED) for p in policies
}
contrasts = {
name: paired(scene_values(rows, a), scene_values(rows, b), draws=DRAWS, seed=SEED)
for name, (a, b) in CONTRASTS.items()
}
statuses, checks = {}, {}
for status, name in GATED.items():
statuses[status], checks[status] = gate(contrasts[name], state_ok)
specificity = contrasts["WRITE_SPECIFICITY"]
statuses["WRITE_SPECIFICITY_STATUS"] = (
"SUPPORTED" if specificity["ci95"][0] > 0 else "NOT_ESTABLISHED"
)
if all(statuses[k] == "SUPPORTED" for k in (*GATED, "WRITE_SPECIFICITY_STATUS")):
branch = "A"
elif statuses["WRITE_STATUS"] == "SUPPORTED":
branch = "B"
else:
branch = "C"
per_seed = {
str(seed): {
name: paired(
scene_values(rows, a, seed=seed),
scene_values(rows, b, seed=seed),
draws=DRAWS,
seed=SEED,
)["mean"]
for name, (a, b) in CONTRASTS.items()
}
for seed in seeds
}
direction = all(per_seed[str(s)]["WRITE_GAIN"] > 0 for s in seeds)
return {
"absolute_depth_absrel": absolute,
"absolute_depth_delta1": delta1,
"contrasts": contrasts,
"contrast_definitions": {
k: f"AbsRel({a}) - AbsRel({b})" for k, (a, b) in CONTRASTS.items()
},
"statuses": statuses,
"gate_checks": checks,
"per_seed_mean_gains": per_seed,
"SEED_ROBUSTNESS": "DIRECTION_REPLICATED" if direction else "NOT_REPLICATED",
"INTERPRETATION_BRANCH": branch,
"scene_ids": sorted({r["scene_id"] for r in rows}),
}
def figures(result, curves, directory):
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
directory.mkdir(parents=True, exist_ok=True)
order = [
p
for p in (
"NO_STREAM",
"OFF",
"OFF_CLAMP3",
"FUSE",
"COMPLETE",
"ALL",
"ALL_UNTRAINED",
"ALL_WRONG_SCENE",
)
if p in result["absolute_depth_absrel"]
]
values = [result["absolute_depth_absrel"][p] for p in order]
fig, ax = plt.subplots(figsize=(8, 4))
ax.bar(range(len(order)), [v["mean"] for v in values], color="#0072B2")
for i, v in enumerate(values):
ax.plot([i, i], v["ci95"], color="black")
ax.set(xticks=list(range(len(order))), xticklabels=order, ylabel="DEV query AbsRel")
ax.tick_params(axis="x", rotation=30)
fig.savefig(directory / "01_policy_absrel.png", bbox_inches="tight", dpi=160)
plt.close(fig)
fig, ax = plt.subplots(figsize=(7, 4))
for seed, rows in curves.items():
steps = [r["step"] for r in rows]
window = max(1, min(50, len(rows)))
loss = np.convolve([r["loss_depth"] for r in rows], np.ones(window) / window, mode="same")
ax.plot(steps, loss, label=f"seed {seed}")
ax.set(xlabel="write-rule step", ylabel="TRAIN query depth AbsRel (50-step mean)")
ax.legend()
fig.savefig(directory / "02_write_rule_training.png", bbox_inches="tight", dpi=160)
plt.close(fig)
def run(root, output=None):
root = Path(root)
output = Path(output) if output else root
provenance = {}
rows = load(root, "raw/dev_stream_query_results.json", provenance)
contract = load(root, "training_contract.json", provenance)
audit = load(root, "audit/dev_state_use.json", provenance)
curves = load(root, "training_curves.json", provenance)
result = analyze(rows, contract["seeds"], audit["status"] == "PASS")
output.mkdir(parents=True, exist_ok=True)
(output / "stream_results.json").write_text(
json.dumps(result, indent=2, sort_keys=True, allow_nan=False) + "\n"
)
figures(result, curves, output / "figures")
(output / "statistics_reproduction.json").write_text(
json.dumps(
{
"status": "PASS",
"raw_only": True,
"input_sha256": provenance,
"source_sha256": {
str(p): hashlib.sha256(p.read_bytes()).hexdigest()
for p in (
Path(__file__).resolve(),
Path(__file__).resolve().parents[1]
/ "src/mcss/mechanism_pilot/direct_capacity_statistics.py",
Path(__file__).resolve().parents[1]
/ "src/mcss/mechanism_pilot/geometry_carrier_statistics.py",
)
},
},
indent=2,
sort_keys=True,
)
+ "\n"
)
return result
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, required=True)
parser.add_argument("--output", type=Path)
args = parser.parse_args()
run(args.root, args.output)