#!/usr/bin/env python3 """EXP-MEASUREMENT-PLANNING-ORACLE-V1: oracle headroom audit of budgeted depth acquisition (no training). Task (MEASUREMENT_PLANNING_TASK_V1, preregistration.md): an episode is one record and one context role. The role's three RGB-D frames are measured; the record's frames that no role uses form the candidate pool, whose RGB and cameras are free and whose depth is revealed only on request; the primary queries are scored. A subset S of requested candidates is reconstructed by the frozen model-free backend (z-buffer of all measured frames + harmonic fill, i.e. REPROJ_HARMONIC over context + S). Every subset of the pool is evaluated (teacher path), which gives the exact best-K oracle, the cross-query oracle and the exact random expectation; deployable heuristics act through the MeasurementEnv only. Stages (repository root, project environment, through cv_host_run.py): lock freeze contracts, scene manifest and hashes (refused once results exist) tests run and record this experiment's tests compute --cohort NAME per-episode raw tables (EVAL34, EVAL34_GAP3, EVAL_FAR, REPLICA, DEV) analyze statistics, gate, figures, final summary """ from __future__ import annotations import argparse import hashlib import json import os import subprocess import sys import time from collections import defaultdict from multiprocessing import Pool from pathlib import Path import numpy as np from mcss.measurement_planning import geometry as g from mcss.measurement_planning import heuristics as h from mcss.measurement_planning.information import ( CANDIDATE_DEPTH_PURPOSE, CANDIDATE_RGB_PURPOSE, CONTEXT_PURPOSE, ROLE_KEYS, candidate_pool, query_ids, ) from mcss.measurement_planning.oracle import FIELDS, run_episode from mcss.mechanism_pilot.direct_capacity_statistics import describe, paired from mcss.mechanism_pilot.geometry_carrier_statistics import _evidence, surface_status EXPERIMENT = "EXP-MEASUREMENT-PLANNING-ORACLE-V1" REPO = g.REPO # A rehearsal redirects every output (MP_ORACLE_REHEARSAL_OUT) and treats DEV as its primary cohort # (code-path check only; it prints no numbers). The formal run never sets it. REHEARSAL = "MP_ORACLE_REHEARSAL_OUT" in os.environ OUT = Path(os.environ.get("MP_ORACLE_REHEARSAL_OUT", REPO / "outputs" / EXPERIMENT)) PRIMARY = "DEV" if REHEARSAL else "EVAL34" DOCS = REPO / "docs" / "experiments" / EXPERIMENT DRAWS, SEED = 10000, 20260928 PRIMARY_BUDGETS = (1, 2, 4) HEADROOM_FRACTION_MIN = 0.20 REPLAY_TOLERANCE_M = 1e-9 # K = 0 against the sealed REPROJ_HARMONIC (INPAINT-V1 precedent) GAP = 3 V17_CONTRACT = REPO / "outputs/EXP-3D-RGBD-LONG-TRAIN-V17/training_contract.json" MANIFESTS = { "EVAL_V3": REPO / "outputs/EXP-3D-RGBD-EVAL-V3-DATA/manifest_eval_v3.json", "EVAL_V4": REPO / "outputs/EXP-3D-RGBD-EVAL-V4-DATA/manifest_eval_v4.json", "EVAL_FAR": REPO / "outputs/EXP-3D-RGBD-EVAL-FAR-DATA/manifest_eval_far.json", "REPLICA": REPO / "outputs/EXP-3D-RGBD-REPLICA-DATA-V1/manifest_replica.json", "V7_SPLIT": REPO / "outputs/EXP-3D-RGBD-DEPTH-BOUNDS-V7/scene_split.json", } SEALS = { "EVAL34": Path.home() / "cv_work/depthsplat/probe_eval34/raw/eval/EVAL34/prediction_seal.json", "DEV": Path.home() / "cv_work/depthsplat/probe_dev/raw/eval/DEV/prediction_seal.json", } COHORTS = { # name: (sources, query role, minimum frame gap to queries, use) "EVAL34": (("EVAL_V3", "EVAL_V4"), "primary_query", 0, "PRIMARY"), "EVAL34_GAP3": (("EVAL_V3", "EVAL_V4"), "primary_query", GAP, "SECONDARY_GAP_POOL"), "EVAL_FAR": (("EVAL_FAR",), "far_query", 0, "SECONDARY_FAR_QUERIES"), "REPLICA": (("REPLICA",), "primary_query", 0, "DESCRIPTIVE_CROSS_DATASET"), "DEV": (("V7_SPLIT",), "primary_query", 0, "REHEARSAL_ONLY"), } DIAGNOSTIC = ("ORACLE", "XORACLE", "RANDOM_EXPECTED", "WORST", "QUERY_NEAREST") CODE = ( "src/mcss/measurement_planning/__init__.py", "src/mcss/measurement_planning/information.py", "src/mcss/measurement_planning/geometry.py", "src/mcss/measurement_planning/heuristics.py", "src/mcss/measurement_planning/oracle.py", "scripts/run_measurement_planning_oracle_v1.py", "tests/test_measurement_planning_oracle_v1.py", ) def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def read(path): return json.loads(Path(path).read_text()) 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 fill_constant(): return float(read(V17_CONTRACT)["eval"]["reprojection_fill_constant_m"]) # ---------------------------------------------------------------- cohorts and the frame-role lock def records(name): sources, query_role, gap, _ = COHORTS[name] out, protected = [], set() for source in sources: manifest = read(MANIFESTS[source]) protected |= set(manifest.get("protected_scene_ids", [])) rows = manifest["scenes"] if source == "V7_SPLIT": rows = [r for r in rows if r["split"] == "DEV"] out.extend(rows) ids = [r["scene_id"] for r in out] if len(ids) != len(set(ids)) or set(ids) & protected: raise PermissionError("Duplicate or protected scene in a cohort") image_size = tuple(read(MANIFESTS[sources[0]])["image_size"]) return out, query_role, gap, image_size def episodes(name): rows, query_role, gap, image_size = records(name) jobs = [] for record in rows: pool = candidate_pool(record, gap) queries = list(record["roles"][query_role]) if set(pool) & set(queries) or set(pool) & set(query_ids(record)): raise PermissionError("A candidate is a query") for role in ("A", "B"): jobs.append( { "cohort": name, "scene_id": record["scene_id"], "role": role, "context": list(record["roles"][ROLE_KEYS[role]]), "pool": pool, "queries": queries, "image_size": list(image_size), } ) return rows, jobs # ---------------------------------------------------------------- lock def lock(): if (OUT / "raw").exists() and any((OUT / "raw").rglob("*.npz")): raise PermissionError( "Results exist: the contracts are frozen once, before any computation" ) for path in ("PROTOCOL.md", "preregistration.md"): if not (OUT / path).exists(): raise FileNotFoundError(f"{path} must be written before the lock") constant = fill_constant() scene_manifest, frame_roles = {}, {} for name in COHORTS: rows, jobs = episodes(name) scene_manifest[name] = { "use": COHORTS[name][3], "scenes": sorted({j["scene_id"] for j in jobs}), "episodes": len(jobs), "query_role": COHORTS[name][1], "min_query_gap": COHORTS[name][2], "pool_sizes": sorted({len(j["pool"]) for j in jobs}), } frame_roles[name] = { r["scene_id"]: { "roles": {k: v for k, v in r["roles"].items() if v}, "pool": candidate_pool(r, COHORTS[name][2]), } for r in rows } train = {r["scene_id"] for r in read(MANIFESTS["V7_SPLIT"])["scenes"] if r["split"] == "TRAIN"} for name, entry in scene_manifest.items(): if set(entry["scenes"]) & train: raise PermissionError(f"{name} overlaps TRAIN72") write(OUT / "scene_manifest.json", scene_manifest) write(OUT / "frame_role_lock.json", frame_roles) write( OUT / "task_contract.json", { "task": "MEASUREMENT_PLANNING_TASK_V1", "episode": "one record x one context role (A or B)", "initial_context": "the role's three RGB-D frames (context_a or context_b)", "candidate_pool": ( "frames no role uses (not context of either role, query, stream, " "ignored, far_query)" ), "gap_pool": ( f"EVAL34_GAP3: candidates at least {GAP} frame indices from every primary query" ), "queries": {k: v[1] for k, v in COHORTS.items()}, "primary_budgets": list(PRIMARY_BUDGETS), "budget_curve": "K = 0 .. pool size", "backend": "frozen REPROJ_HARMONIC over initial context + requested candidates", "regions": ( "NEAR / FAR from the initial context's hits (frozen 8 px rule), " "fixed for every subset" ), "metric": "AbsRel on GT-valid query pixels (arms-evaluator rule); " "episode = mean over its queries; " "scene = mean over roles", "primary_metric": "ALL AbsRel", "fill_constant_m": constant, "cohorts": {k: {"use": v[3], "sources": list(v[0])} for k, v in COHORTS.items()}, }, ) write( OUT / "information_contract.json", { "deployable_may_see": [ "initial context RGB, depth and cameras", "candidate RGB and cameras (all candidates, before any request)", "depth of candidates it has requested, within the budget", ], "deployable_never_sees": [ "candidate depth before its request", "query RGB, depth or camera", "the other role's context frames", "oracle utilities or any teacher output", ], "enforcement": "MeasurementEnv holds a sanitized record " "(role context + candidates only); " "request_depth enforces membership, uniqueness and budget; " "every read is logged by purpose", "teacher_only": ["ORACLE", "XORACLE", "RANDOM_EXPECTED", "WORST", "QUERY_NEAREST"], "purposes": { "context": CONTEXT_PURPOSE, "candidate_rgb": CANDIDATE_RGB_PURPOSE, "candidate_depth": CANDIDATE_DEPTH_PURPOSE, }, }, ) write( OUT / "baseline_contract.json", { "deployable": {name: (fn.__doc__ or "").strip() for name, fn in h.DEPLOYABLE.items()}, "diagnostic": { "ORACLE": ( "best subset of size K per episode by the episode's mean ALL AbsRel (in-sample)" ), "XORACLE": ( "cross-query oracle: subset chosen on the other query of the " "episode, scored on this one" ), "RANDOM_EXPECTED": ( "exact mean over all subsets of size K (uniform random policy expectation)" ), "WORST": "worst subset of size K (range only)", "QUERY_NEAREST": ( "K candidates nearest to any query camera (uses query cameras; not deployable)" ), }, "constants": { "fill_cap_px": h.FILL_CAP_PX, "pose_metres_per_radian": h.POSE_METRES_PER_RADIAN, "novelty_relative": h.NOVELTY_RELATIVE, "random_seed": h.RANDOM_SEED, }, "best_heuristic_rule": "per K, the deployable selector with the lowest mean " "scene ALL AbsRel " "(chosen on the same data: conservative for headroom)", }, ) hashes = { "code": {p: sha(REPO / p) for p in CODE}, "frozen_functions": g.frozen_hashes(), "manifests": {k: sha(v) for k, v in MANIFESTS.items()}, "seals": {k: sha(v) for k, v in SEALS.items() if v.exists()}, "v17_contract": sha(V17_CONTRACT), "documents": {p: sha(OUT / p) for p in ("PROTOCOL.md", "preregistration.md")}, "contracts": { p: sha(OUT / p) for p in ( "task_contract.json", "information_contract.json", "baseline_contract.json", "scene_manifest.json", "frame_role_lock.json", ) }, } gate = { "primary_cohort": "EVAL34", "budgets": list(PRIMARY_BUDGETS), "headroom_fraction_min": HEADROOM_FRACTION_MIN, "rule": "HEADROOM_SUPPORTED at K iff the cross-query contrast BEST_H_K - XORACLE_K " "passes the V2 gate " "(SUPPORTED) and (BEST_H_K - ORACLE_K) / (NO_DEPTH - ORACLE_K) >= 0.20 in scene means; " "HEADROOM_STATUS = SUPPORTED if any K in (1, 2, 4) is supported, else NOT_ESTABLISHED; " "POLICY_TRAINING_ALLOWED iff SUPPORTED", "statistics": {"unit": "scene", "draws": DRAWS, "seed": SEED}, } write( OUT / "preregistration.json", { "experiment": EXPERIMENT, "gate": gate, "hashes": hashes, "rehearsal": REHEARSAL, "locked_at": time.strftime("%Y-%m-%d %H:%M:%S %z"), }, ) print("LOCKED", sha(OUT / "preregistration.json"), flush=True) def verify_lock(): pre = read(OUT / "preregistration.json") for group, mapping in pre["hashes"].items(): if group == "frozen_functions": current = g.frozen_hashes() if current != mapping: raise PermissionError("A frozen function changed") continue if isinstance(mapping, str): if sha(V17_CONTRACT) != mapping: raise PermissionError("V17 contract changed") continue for key, digest in mapping.items(): path = { "code": lambda k: REPO / k, "manifests": lambda k: MANIFESTS[k], "seals": lambda k: SEALS[k], "documents": lambda k: OUT / k, "contracts": lambda k: OUT / k, }[group](key) if sha(path) != digest: raise PermissionError(f"{group}/{key} changed after the lock") return pre def compute(cohort, workers): verify_lock() rows, jobs = episodes(cohort) by_id = {r["scene_id"]: r for r in rows} constant = fill_constant() out = OUT / "raw" / cohort if out.exists() and any(out.glob("*.npz")): raise FileExistsError(f"{cohort} was computed once") seal_prefix = "EVAL34" if cohort in ("EVAL34", "EVAL34_GAP3") else cohort seal_path = SEALS.get(seal_prefix) for job in jobs: job["record"] = by_id[job["scene_id"]] job["constant"] = constant job["repo"] = str(REPO) job["out_dir"] = str(out) job["seal_prefix"] = seal_prefix job["seal_path"] = str(seal_path) if seal_path is not None and seal_path.exists() else None started = time.time() with Pool(workers) as pool: done = pool.map(run_episode, jobs, chunksize=1) write( out / "_done.json", {"episodes": sorted(done), "seconds": round(time.time() - started, 1)} ) print( "COMPUTED", cohort, len(done), "episodes", round(time.time() - started, 1), "s", flush=True ) # ---------------------------------------------------------------- analysis def load(cohort): folder = OUT / "raw" / cohort episodes_ = {} for path in sorted(folder.glob("*.json")): if path.name.startswith("_"): continue meta = read(path) npz = folder / f"{path.stem}.npz" if sha(npz) != meta["table_sha256"]: raise PermissionError(f"{npz} changed") meta["table"] = dict(np.load(npz)) episodes_[path.stem] = meta return episodes_ def popcount(bits): return bin(bits).count("1") def mean_over_queries(values): """Episode value per subset: mean over the episode's queries, skipping empty queries.""" with np.errstate(all="ignore"): out = np.nanmean(values, axis=1) return out def episode_methods(meta): """Per method and K: the chosen subset(s) and the episode's value of every field.""" table = {f: meta["table"][f] for f in FIELDS} pool = meta["pool"] n = len(pool) sizes = np.array([popcount(b) for b in range(1 << n)]) score = mean_over_queries(table["all"]) per_query_all = table["all"] index = {f: i for i, f in enumerate(pool)} def bits_of(frames): return int(sum(1 << index[f] for f in frames)) def values(bits): return {f: float(mean_over_queries(table[f][[bits]])[0]) for f in FIELDS} result = {0: {"NO_DEPTH": values(0)}} for k in range(1, n + 1): members = np.flatnonzero(sizes == k) best = int(members[np.argmin(score[members])]) worst = int(members[np.argmax(score[members])]) methods = { "ORACLE": values(best), "WORST": values(worst), "RANDOM_EXPECTED": { f: float(np.nanmean(mean_over_queries(table[f][members]))) for f in FIELDS }, } nq = per_query_all.shape[1] cross = defaultdict(list) for q in range(nq): others = [o for o in range(nq) if o != q] if not others: continue chooser = np.nanmean(per_query_all[members][:, others], axis=1) pick = int(members[np.argmin(chooser)]) for f in FIELDS: cross[f].append(table[f][pick, q]) methods["XORACLE"] = { f: float(np.nanmean(v)) if np.isfinite(v).any() else float("nan") for f, v in cross.items() } for name, chosen in meta["selections"].items(): methods[name] = values(bits_of(chosen[str(k)])) methods["QUERY_NEAREST"] = values( bits_of(meta["diagnostic_selections"]["QUERY_NEAREST"][str(k)]) ) methods["_oracle_frames"] = [pool[i] for i in range(n) if best >> i & 1] result[k] = methods return result def scene_table(episodes_): """{K: {method: {field: {scene: value}}}}; scene = mean over its roles (NaN roles skipped).""" grouped = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: defaultdict(list)))) for meta in episodes_.values(): for k, methods in episode_methods(meta).items(): for method, fields in methods.items(): if method.startswith("_"): continue for f, v in fields.items(): if v is not None and np.isfinite(v): grouped[k][method][f][meta["scene_id"]].append(v) return { k: { m: { f: {s: float(np.mean(v)) for s, v in scenes.items()} for f, scenes in fields.items() } for m, fields in methods.items() } for k, methods in grouped.items() } def gate(stats): checks = { "mean_ci_and_scene_consistency": _evidence(stats), "every_leave_one_scene_out_gain_positive": bool(stats["loso"]) and min(stats["loso"].values()) > 0, "positive_top1_share_at_most_0_5": stats["positive_top1_contribution"] is not None and stats["positive_top1_contribution"] <= 0.5, } return surface_status(stats, checks), checks def contrast(a, b): common = sorted(set(a) & set(b)) stats = paired({s: a[s] for s in common}, {s: b[s] for s in common}, draws=DRAWS, seed=SEED) stats.pop("per_scene", None) return stats def headroom(table, budgets, deployable): """Per K: best deployable heuristic, in-sample and cross-query headroom, oracle gain, gate.""" out = {} nodepth = table[0]["NO_DEPTH"]["all"] for k in budgets: if k not in table: continue rows = table[k] means = {m: float(np.mean(list(rows[m]["all"].values()))) for m in deployable if m in rows} best = min(means, key=lambda m: (means[m], m)) best_values = rows[best]["all"] d = contrast(best_values, rows["ORACLE"]["all"]) x = contrast(best_values, rows["XORACLE"]["all"]) gain = contrast(nodepth, rows["ORACLE"]["all"]) fraction = d["mean"] / gain["mean"] if gain["mean"] > 0 else None status, checks = gate(x) supported = ( status == "SUPPORTED" and fraction is not None and fraction >= HEADROOM_FRACTION_MIN ) out[str(k)] = { "best_heuristic": best, "heuristic_means_all": means, "oracle_minus_best": d, "xoracle_minus_best": x, "oracle_gain_over_no_depth": gain, "best_gain_over_no_depth": contrast(nodepth, best_values), "best_vs_random_expected": contrast(rows["RANDOM_EXPECTED"]["all"], best_values), "headroom_fraction": fraction, "xoracle_gate_status": status, "xoracle_gate_checks": checks, "headroom_supported": bool(supported), } return out def budget_curve(table): curve = {} for k in sorted(table): curve[str(k)] = { m: {f: float(np.mean(list(v.values()))) if v else None for f, v in fields.items()} for m, fields in table[k].items() } return curve def spearman(x, y): from scipy.stats import spearmanr if len(set(x)) < 2 or len(set(y)) < 2: return None return float(spearmanr(x, y).statistic) def utility_structure(episodes_): """Single-step utilities U_i = L(no depth) - L({i}) (ALL) against deployable features, within episodes.""" names = ( "translation_m", "rotation_deg", "pose_distance", "uncovered", "fill_distance_px", "photometric", "novelty3d", "pool_index", ) per_scene = defaultdict(lambda: defaultdict(list)) top1 = defaultdict(list) points = [] oracle_positions = defaultdict(int) marginal = defaultdict(list) for meta in episodes_.values(): pool = meta["pool"] table = mean_over_queries(meta["table"]["all"]) far_cov = mean_over_queries(meta["table"]["coverage_far"]) far_err = mean_over_queries(meta["table"]["far"]) utility = {f: float(table[0] - table[1 << i]) for i, f in enumerate(pool)} best = max(pool, key=lambda f: (utility[f], -f)) oracle_positions[pool.index(best)] += 1 feats = meta["features"] for name in names: xs = [feats[str(f)][name] for f in pool] ys = [utility[f] for f in pool] rho = spearman(xs, ys) if rho is not None: per_scene[name][meta["scene_id"]].append(rho) pick = max(pool, key=lambda f: (feats[str(f)][name], -f)) top1[name].append(pick == best) for i, f in enumerate(pool): points.append( { "utility": utility[f], **{name: feats[str(f)][name] for name in names}, "far_coverage_gain": float(far_cov[1 << i] - far_cov[0]) if np.isfinite(far_cov[0]) else None, "far_improvement": float(far_err[0] - far_err[1 << i]) if np.isfinite(far_err[0]) else None, "scene_id": meta["scene_id"], } ) n = len(pool) for bits in range(1 << n): m = popcount(bits) if m > 3: continue for i in range(n): if not bits >> i & 1: marginal[m].append(float(table[bits] - table[bits | 1 << i])) correlations = {} for name, scenes in per_scene.items(): values = {s: float(np.mean(v)) for s, v in scenes.items()} stats = describe(values, draws=DRAWS, seed=SEED) stats.pop("per_scene", None) correlations[name] = { "mean_within_episode_spearman": stats, "top1_agreement": float(np.mean(top1[name])), } best = max( correlations, key=lambda k: abs(correlations[k]["mean_within_episode_spearman"]["mean"]) ) rho = abs(correlations[best]["mean_within_episode_spearman"]["mean"]) label = "STRONG" if rho > 0.6 else "MODERATE" if rho >= 0.3 else "WEAK" total = sum(oracle_positions.values()) modal = max(oracle_positions.values()) / total tercile = [] by_scene = defaultdict(list) for p in points: by_scene[p["scene_id"]].append(p) for group in by_scene.values(): u = np.array([p["utility"] for p in group]) c = np.array([p["uncovered"] for p in group]) hi_c, lo_c = c >= np.quantile(c, 2 / 3), c <= np.quantile(c, 1 / 3) hi_u, lo_u = u >= np.quantile(u, 2 / 3), u <= np.quantile(u, 1 / 3) tercile.append( { "high_cov_low_util": int((hi_c & lo_u).sum()), "low_cov_high_util": int((lo_c & hi_u).sum()), "n": len(group), } ) return { "correlations": correlations, "best_feature": best, "predictability_label": label, "oracle_best1_positions": {str(k): v for k, v in sorted(oracle_positions.items())}, "oracle_best1_modal_share": modal, "utility_scene_dependent": bool(modal < 0.5), "mismatch_counts": { "high_uncovered_low_utility": sum(t["high_cov_low_util"] for t in tercile), "low_uncovered_high_utility": sum(t["low_cov_high_util"] for t in tercile), "candidates": sum(t["n"] for t in tercile), }, "marginal_utility_by_history": { str(m): {"mean": float(np.mean(v)), "median": float(np.median(v)), "n": len(v)} for m, v in sorted(marginal.items()) }, "_points": points, "_marginal": {str(m): v for m, v in marginal.items()}, } def figures(curve, scenes, util, gates, folder): import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt folder.mkdir(parents=True, exist_ok=True) ks = sorted(int(k) for k in curve) fig, ax = plt.subplots(figsize=(7, 4.5)) for method, style in ( ("RANDOM_EXPECTED", ":"), ("UNIFORM", "--"), ("ORACLE", "-"), ("XORACLE", "-."), ): xs = [k for k in ks if k > 0 and method in curve[str(k)]] ax.plot(xs, [curve[str(k)][method]["all"] for k in xs], style, marker="o", label=method) best = [ (int(k), v["best_heuristic"]) for k, v in sorted(gates.items(), key=lambda i: int(i[0])) ] ax.plot( [k for k, _ in best], [curve[str(k)][m]["all"] for k, m in best], "s", label="BEST_HEURISTIC (K=1,2,4)", ) ax.axhline(curve["0"]["NO_DEPTH"]["all"], color="grey", lw=1, label="NO_EXTRA_DEPTH") ax.set_xlabel("depth budget K (requested candidate frames)") ax.set_ylabel("query AbsRel (ALL), scene mean") ax.legend(fontsize=8) fig.tight_layout() fig.savefig(folder / "1_budget_curve.png", dpi=150) plt.close(fig) fig, axes = plt.subplots(1, len(gates), figsize=(4 * len(gates), 3.5), squeeze=False) for ax, (k, entry) in zip(axes[0], sorted(gates.items(), key=lambda i: int(i[0])), strict=True): rows = scenes[int(k)] gap = sorted( rows[entry["best_heuristic"]]["all"][s] - rows["ORACLE"]["all"][s] for s in rows["ORACLE"]["all"] ) ax.bar(range(len(gap)), gap, color="tab:blue") ax.set_title(f"K={k}: {entry['best_heuristic']} - ORACLE") ax.set_xlabel("scene (sorted)") axes[0][0].set_ylabel("AbsRel gap (ALL)") fig.tight_layout() fig.savefig(folder / "2_oracle_gap_per_scene.png", dpi=150) plt.close(fig) fig, ax = plt.subplots(figsize=(6, 3.5)) positions = util["oracle_best1_positions"] ax.bar([int(p) for p in positions], list(positions.values()), color="tab:blue") ax.set_xlabel("pool index of the oracle's best single candidate (0 = earliest)") ax.set_ylabel("episodes") fig.tight_layout() fig.savefig(folder / "3_oracle_frame_positions.png", dpi=150) plt.close(fig) pts = util["_points"] for _, file, feats in ( ("uncovered", "4_utility_vs_uncovered.png", ("uncovered",)), ("pose", "5_utility_vs_pose.png", ("translation_m", "rotation_deg")), ): fig, axes = plt.subplots(1, len(feats), figsize=(4.5 * len(feats), 3.5), squeeze=False) for ax, feat in zip(axes[0], feats, strict=True): ax.scatter([p[feat] for p in pts], [p["utility"] for p in pts], s=4, alpha=0.4) ax.set_xlabel(feat) ax.set_ylabel("single-step utility (ALL AbsRel reduction)") fig.tight_layout() fig.savefig(folder / file, dpi=150) plt.close(fig) fig, ax = plt.subplots(figsize=(6, 3.5)) marg = util["_marginal"] keys = sorted(marg, key=int) ax.boxplot([marg[k] for k in keys], showfliers=False) ax.set_xticks(range(1, len(keys) + 1), [f"|S|={k}" for k in keys]) ax.set_ylabel("marginal utility of one more depth frame") fig.tight_layout() fig.savefig(folder / "6_utility_by_history.png", dpi=150) plt.close(fig) fig, ax = plt.subplots(figsize=(5, 4)) xs = [ p["far_coverage_gain"] for p in pts if p["far_coverage_gain"] is not None and p["far_improvement"] is not None ] ys = [ p["far_improvement"] for p in pts if p["far_coverage_gain"] is not None and p["far_improvement"] is not None ] ax.scatter(xs, ys, s=4, alpha=0.4) ax.set_xlabel("FAR coverage gain of one candidate") ax.set_ylabel("FAR AbsRel improvement") fig.tight_layout() fig.savefig(folder / "7_far_coverage_vs_improvement.png", dpi=150) plt.close(fig) def analyze_cohort(cohort): episodes_ = load(cohort) scenes = scene_table(episodes_) deployable = tuple(h.DEPLOYABLE) pool_min = min(len(m["pool"]) for m in episodes_.values()) budgets = tuple(k for k in PRIMARY_BUDGETS if k <= pool_min) gates = headroom(scenes, budgets, deployable) util = utility_structure(episodes_) curve = budget_curve({k: v for k, v in scenes.items() if k <= pool_min}) replay = [d for m in episodes_.values() for d in m["replay_k0"].values()] integrity = { "episodes": len(episodes_), "scenes": len({m["scene_id"] for m in episodes_.values()}), "selector_audits_ok": all(m["selector_audits_ok"] for m in episodes_.values()), "teacher_purposes": sorted( {(t[0], tuple(t[1])) for m in episodes_.values() for t in m["teacher_reads"]}, key=str ), "replay_k0_checked": len(replay), "replay_k0_max_abs_diff": max((d["max_abs_diff_harmonic"] for d in replay), default=None), "replay_k0_near_equal": all(d["near_equal"] for d in replay) if replay else None, } return episodes_, scenes, gates, util, curve, integrity def analyze(): verify_lock() results = {} for cohort in COHORTS: if (cohort == "DEV") != REHEARSAL or not (OUT / "raw" / cohort / "_done.json").exists(): continue episodes_, scenes, gates, util, curve, integrity = analyze_cohort(cohort) folder = OUT / "figures" / cohort figures(curve, scenes, util, gates, folder) results[cohort] = { "use": COHORTS[cohort][3], "gates": gates, "budget_curve": curve, "utility": {k: v for k, v in util.items() if not k.startswith("_")}, "integrity": integrity, } primary = results.get(PRIMARY) if primary is None: raise FileNotFoundError("The primary cohort has not been computed") supported = [k for k, v in primary["gates"].items() if v["headroom_supported"]] status = "SUPPORTED" if supported else "NOT_ESTABLISHED" write( OUT / "oracle_results.json", { c: { k: { "ORACLE": r["budget_curve"][k].get("ORACLE"), "XORACLE": r["budget_curve"][k].get("XORACLE"), } for k in r["budget_curve"] } for c, r in results.items() }, ) write( OUT / "heuristic_results.json", { c: { k: {m: v for m, v in r["budget_curve"][k].items() if m not in DIAGNOSTIC} for k in r["budget_curve"] } for c, r in results.items() }, ) write(OUT / "budget_curve.json", {c: r["budget_curve"] for c, r in results.items()}) write( OUT / "bootstrap_results.json", {c: {"gates": r["gates"], "utility": r["utility"]} for c, r in results.items()}, ) summary = final_summary(results, status, supported) write(OUT / "final_summary.json", summary) # determinism: a second full pass must give identical numbers again = {c: analyze_cohort(c)[2] for c in results} deterministic = all( json.dumps(again[c], sort_keys=True) == json.dumps(results[c]["gates"], sort_keys=True) for c in results ) integrity_ok = ( deterministic and all(r["integrity"]["selector_audits_ok"] for r in results.values()) and all( r["integrity"]["replay_k0_max_abs_diff"] is None or ( r["integrity"]["replay_k0_max_abs_diff"] <= REPLAY_TOLERANCE_M and r["integrity"]["replay_k0_near_equal"] ) for r in results.values() ) ) write( OUT / "integrity.json", { "deterministic_reanalysis": deterministic, "replay_tolerance_m": REPLAY_TOLERANCE_M, "final_integrity": "PASS" if integrity_ok else "FAIL", **{c: r["integrity"] for c, r in results.items()}, "final_holdout_touched": False, }, ) print_summary(summary, integrity_ok) def final_summary(results, status, supported): p = results[PRIMARY] curve, gates, util = p["budget_curve"], p["gates"], p["utility"] def val(k, m): entry = curve.get(str(k), {}).get(m) return None if entry is None else round(entry["all"], 4) competitive = {k: v["best_heuristic"] for k, v in gates.items()} return { "N_SCENES": p["integrity"]["scenes"], "N_EPISODES": p["integrity"]["episodes"], "N_CANDIDATES_PER_SCENE": 9, "BUDGETS": list(PRIMARY_BUDGETS), "NO_DEPTH_ABSREL": val(0, "NO_DEPTH"), "RANDOM_K1": val(1, "RANDOM_EXPECTED"), "UNIFORM_K1": val(1, "UNIFORM"), **{ f"BEST_HEURISTIC_K{k}": val(k, gates[str(k)]["best_heuristic"]) for k in PRIMARY_BUDGETS }, **{f"ORACLE_K{k}": val(k, "ORACLE") for k in PRIMARY_BUDGETS}, **{f"XORACLE_K{k}": val(k, "XORACLE") for k in PRIMARY_BUDGETS}, **{ f"ORACLE_MINUS_BEST_HEURISTIC_K{k}": round( gates[str(k)]["oracle_minus_best"]["mean"], 4 ) for k in PRIMARY_BUDGETS }, "HEADROOM_CI": { k: [round(x, 4) for x in gates[k]["xoracle_minus_best"]["ci95"]] for k in gates }, "HEADROOM_FRACTION": { k: None if gates[k]["headroom_fraction"] is None else round(gates[k]["headroom_fraction"], 3) for k in gates }, "XORACLE_GATE": {k: gates[k]["xoracle_gate_status"] for k in gates}, "HEADROOM_STATUS": status, "HEADROOM_SUPPORTED_BUDGETS": supported, "MOST_COMPETITIVE_HEURISTIC": competitive, "UTILITY_SCENE_DEPENDENT": util["utility_scene_dependent"], "UTILITY_PREDICTABILITY_DIAGNOSTIC": f"{util['predictability_label']} (best feature " f"{util['best_feature']})", "POLICY_TRAINING_ALLOWED": status == "SUPPORTED", "SECONDARY": { c: { k: { "best": v["best_heuristic"], "fraction": v["headroom_fraction"], "xoracle_gate": v["xoracle_gate_status"], "supported": v["headroom_supported"], } for k, v in r["gates"].items() } for c, r in results.items() if c != PRIMARY }, "FINAL_HOLDOUT_TOUCHED": False, "NEW_DATA_DOWNLOADED": False, "GITHUB_PUSHED": False, "REPORT_DIR": str(OUT), } def print_summary(s, deterministic): tests = read(OUT / "tests.json")["status"] if (OUT / "tests.json").exists() else "NOT_RECORDED" lines = ["=" * 60, "MEASUREMENT PLANNING ORACLE V1 FINAL", "=" * 60, ""] for key in ( "N_SCENES", "N_CANDIDATES_PER_SCENE", "BUDGETS", "NO_DEPTH_ABSREL", "RANDOM_K1", "UNIFORM_K1", "BEST_HEURISTIC_K1", "ORACLE_K1", "BEST_HEURISTIC_K2", "ORACLE_K2", "BEST_HEURISTIC_K4", "ORACLE_K4", "ORACLE_MINUS_BEST_HEURISTIC_K1", "ORACLE_MINUS_BEST_HEURISTIC_K2", "ORACLE_MINUS_BEST_HEURISTIC_K4", "HEADROOM_CI", "HEADROOM_STATUS", "MOST_COMPETITIVE_HEURISTIC", "UTILITY_SCENE_DEPENDENT", "UTILITY_PREDICTABILITY_DIAGNOSTIC", "POLICY_TRAINING_ALLOWED", ): lines.append(f"{key}={s[key]}") lines += [ "", "FINAL_HOLDOUT_TOUCHED=false", "NEW_DATA_DOWNLOADED=false", "GITHUB_PUSHED=false", "", f"TESTS={tests}", f"FINAL_INTEGRITY={'PASS' if deterministic else 'FAIL'}", "", f"REPORT_DIR={s['REPORT_DIR']}", "=" * 60, ] text = "\n".join(lines) (OUT / "final_summary.txt").write_text(text + "\n") if REHEARSAL: print(f"REHEARSAL_ANALYSIS_DONE (numbers withheld) integrity={deterministic}", flush=True) return print(text, flush=True) def record_tests(): result = subprocess.run( [sys.executable, "-m", "pytest", "-q", "tests/test_measurement_planning_oracle_v1.py"], cwd=REPO, capture_output=True, text=True, env={**os.environ, "CUDA_VISIBLE_DEVICES": ""}, ) write( OUT / "tests.json", { "status": "PASS" if result.returncode == 0 else "FAIL", "summary": result.stdout.strip().splitlines()[-1:], "returncode": result.returncode, }, ) print(result.stdout[-2000:], flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("stage", choices=("lock", "tests", "compute", "analyze")) parser.add_argument("--cohort", choices=tuple(COHORTS)) parser.add_argument("--workers", type=int, default=16) args = parser.parse_args() if args.stage == "lock": lock() elif args.stage == "tests": record_tests() elif args.stage == "compute": compute(args.cohort, args.workers) else: analyze() if __name__ == "__main__": main()