"""EXP-3D-READOUT-METHOD-V1, Amendment 3 (post hoc; AMENDMENT_3.md): the premise test of Candidate A in an explicit geometric form, a free-space lower bound from the arrived measurements. Deployable (the 7 arrived RGB-D frames and the query camera only), non-learned, CPU, applied to the saved Phase 1 predictions; no training. Carving: a point X is free space when some arrived frame j sees it in front of its measured surface, |X - c_j| < D_j(pi_j(X)) (1 - REL) - ABS (REL = 0.05, ABS = 0.05 m; D_j = measured ray distance, valid pixels) A query ray o + t d is sampled at T = 128 log-spaced distances in [0.1, 20] m. Its free-space lower bound t_lb is the last sample of the carved prefix (every sample up to it is free space; 0 when the first sample is not carved): a static surface cannot lie there. The deployable readout X_FS clamps a prediction X on the protocol's uncovered pixels (TTT alpha < 0.5) to max(X, t_lb). Predictions (float16, as saved): B7 (V2's recipe), B5 (the strongest simple control), R7. Descriptive: how often t_lb constrains, how often a prediction or the ground truth lies inside carved space (the latter = carving errors), per region. Run from the repository root with the project environment (CUDA hidden, PYTHONPATH=src): python rm_amend3.py ROOT smoke (code check on the smoke fold FIT_SYSTEM2, a training cohort -> smoke/amend3) python rm_amend3.py ROOT lock (hashes AMENDMENT_3.md and this file; refuses when results exist) python rm_amend3.py ROOT run (refuses without the lock or when a locked file changed) """ import hashlib import json import sys import time from collections import defaultdict from pathlib import Path import numpy as np import torch ROOT, STAGE = Path(sys.argv[1]).resolve(), sys.argv[2] sys.path.insert(0, str(ROOT / "code")) import rm_common as C # noqa: E402 (sets the import paths) import rm_stats as RS # noqa: E402 (reads sys.argv[1:]; its main() does not run) from mcss.geometry import generate_rays, project_world # noqa: E402 REL, ABS = 0.05, 0.05 T = np.geomspace(0.1, 20.0, 128) METHODS = ("B7", "B5", "R7") LOCKED = ("AMENDMENT_3.md", "code/rm_amend3.py") LOCK_COHORTS = {"CAJU": ("CAJU_PILOT", "CAJU_SEL", "CAJU_EVAL")} def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def lock_episodes(cohort): locks = json.loads((C.OUT / C.LOCKS[cohort]).read_text())["cohorts"] out, records = {}, {} for name in LOCK_COHORTS.get(cohort, (cohort,)): out.update({f"{e['scene_id']}/{e['role']}": e for e in locks[name]["episodes"]}) records.update({r["scene_id"]: r for r in json.loads(Path(locks[name]["manifest"]).read_text())["scenes"]}) return out, records def arrived(teacher, episode): """(camera, measured ray distance [H, W], camera centre) of the 7 arrived frames.""" frames = [] for ids in (episode["context"], episode["stream"]): ctx = teacher.frames(ids) for obs, depth in zip(ctx, ctx.depths): cam = obs.camera.to(dtype=torch.float64) frames.append((cam, torch.as_tensor(depth, dtype=torch.float64).numpy(), cam.c2w[:3, 3].numpy())) return frames def carved_samples(frames, o, d): """[P, len(T)] free-space flags of the samples along the rays o + t d.""" pts = (o[:, None, :] + d[:, None, :] * T[None, :, None]).reshape(-1, 3) carved = np.zeros(len(pts), bool) tp = torch.as_tensor(pts) for cam, depth, centre in frames: pix, _, inside = project_world(tp, cam) pix, inside = np.nan_to_num(pix.numpy(), nan=-1.0, posinf=-1.0, neginf=-1.0), inside.numpy() # behind: not inside h, w = cam.image_size u = np.clip(np.round(pix[:, 0]).astype(np.int64), 0, w - 1) v = np.clip(np.round(pix[:, 1]).astype(np.int64), 0, h - 1) meas = depth[v, u] ok = inside & np.isfinite(meas) & (meas > 0) carved |= ok & (np.linalg.norm(pts - centre, axis=1) < meas * (1.0 - REL) - ABS) return carved.reshape(-1, len(T)) def in_carved(carved, t): k = np.clip(np.searchsorted(T, t), 1, len(T) - 1) k = np.where(np.abs(np.log(T[k - 1]) - np.log(t)) < np.abs(np.log(T[k]) - np.log(t)), k - 1, k) return carved[np.arange(len(t)), k] def lock(): if (ROOT / "results/amend3").exists() or (ROOT / "AMENDMENT_3.lock.json").exists(): raise SystemExit("cannot lock: amendment 3 results or a lock exist already") rec = {"amendment": 3, "post_hoc": True, "deployable": True, "locked_at": time.strftime("%Y-%m-%d %H:%M:%S %z"), "files": {n: sha(ROOT / n) for n in LOCKED}, "primary": "B7 - B7_FS, pooled 350 places (> 0: the free-space lower bound lowers V2's error)", "decision": {"A_PREMISE_FREE_SPACE=SUPPORTED": "mean >= 0.005 and CI lower > 0", "A_PREMISE_FREE_SPACE=NOT_SUPPORTED": "otherwise"}, "carving": {"rel": REL, "abs_m": ABS, "samples": len(T), "range_m": [0.1, 20.0]}} (ROOT / "AMENDMENT_3.lock.json").write_text(json.dumps(rec, indent=1) + "\n") (ROOT / "audit/amendment3_locked_at.txt").write_text(f"{rec['locked_at']} {sha(ROOT / 'AMENDMENT_3.lock.json')}\n") print("AMEND3_LOCKED", rec["locked_at"], sha(ROOT / "AMENDMENT_3.lock.json")) def run(smoke): if not smoke: rec = json.loads((ROOT / "AMENDMENT_3.lock.json").read_text()) for name, h in rec["files"].items(): if sha(ROOT / name) != h: raise PermissionError(f"{name} changed since the amendment 3 lock") torch.set_num_threads(8) out_dir = ROOT / ("smoke/amend3" if smoke else "results/amend3") out_dir.mkdir(parents=True, exist_ok=smoke) access, started, entries = [], time.time(), [] desc = defaultdict(lambda: defaultdict(float)) for fold in RS.FOLDS: arrays, rows = C.load_cache(fold) fj = json.loads((RS.RES / f"fold_{fold}.json").read_text())["queries"] if [e["key"] for e in fj] != [r["key"] for r in rows]: raise ValueError(f"{fold}: prediction order differs from the cache") saved = np.load(RS.RES / f"preds_{fold}.npz") preds = {m: saved[m] for m in METHODS} gt = C.load_gt(fold, rows, access) episodes, records = lock_episodes(fold) members = defaultdict(list) for i, r in enumerate(rows): members[r["episode"]].append(i) for ep, idx in members.items(): e = episodes[ep] teacher = C.S.TeacherAccess(records[e["scene_id"]], [rows[i]["query"] for i in idx], C.S.IMAGE_SIZE, C.S.REPO, access) frames = arrived(teacher, e) for i in idx: cam = teacher.query_camera(rows[i]["query"]).to(dtype=torch.float64) o, d = (a.numpy().reshape(-1, 3) for a in generate_rays(cam)) unc = (arrays["ttt_a"][i] < C.ALPHA_MIN).ravel() p = np.flatnonzero(unc) carved = carved_samples(frames, o[p], d[p]) first_open = np.where((~carved).any(1), (~carved).argmax(1), len(T)) t_lb = np.where(first_open > 0, T[np.maximum(first_open - 1, 0)], 0.0) g = gt[i].astype(np.float64) masks = C.strata_masks({"near": arrays["near7"][i], "cov": arrays["ttt_a"][i] >= C.ALPHA_MIN}, g, rows[i]["hit7"]) entry = {"key": rows[i]["key"], "place": rows[i]["place"], "episode": rows[i]["episode"], "fold": fold, "hit7": rows[i]["hit7"], "methods": {}} gv = g.ravel()[p] valid = gv > 0 for m in METHODS: x = np.maximum(preds[m][i].astype(np.float64), 1e-3) xf = x.ravel().copy() xf[p] = np.maximum(xf[p], t_lb) entry["methods"][m] = C.region_sums(x, g, masks) entry["methods"][m + "_FS"] = C.region_sums(xf.reshape(x.shape), g, masks) xv = x.ravel()[p][valid] desc[fold][f"{m}_below_bound"] += float((xv < t_lb[valid]).sum()) desc[fold][f"{m}_in_carved"] += float(in_carved(carved[valid], xv).sum()) desc[fold]["uncovered_valid"] += float(valid.sum()) desc[fold]["constrained"] += float((t_lb[valid] > 0).sum()) desc[fold]["gt_below_bound"] += float((gv[valid] < t_lb[valid]).sum()) desc[fold]["gt_in_carved"] += float(in_carved(carved[valid], gv[valid]).sum()) desc[fold]["bound_over_gt_sum"] += float((t_lb[valid] / gv[valid]).sum()) entries.append(entry) print(f"{fold}: {len(rows)} queries ({time.time() - started:.0f}s)", flush=True) purposes = {a["purpose"] for a in access} if purposes - {"SEALED_QUERY_EVALUATION", "QUERY_CAMERA_TEACHER_ONLY", "ORACLE_TEACHER_ONLY"}: raise PermissionError(f"unexpected read purposes {sorted(purposes)}") (out_dir / "queries.json").write_text(json.dumps(entries) + "\n") names = tuple(n for m in METHODS for n in (m, m + "_FS")) table = RS.place_table(entries, names) means = {m: float(np.mean(list(table[m].values()))) for m in names} contrasts = {f"{m}_minus_{m}_FS": {**RS.short(RS.contrast(table, m, m + "_FS")), **RS.summary(table, m, m + "_FS")} for m in METHODS} p1 = contrasts["B7_minus_B7_FS"] label = "SUPPORTED" if p1["mean"] >= 0.005 and p1["ci95"][0] > 0 else "NOT_SUPPORTED" pooled = defaultdict(float) for f in desc.values(): for k, v in f.items(): pooled[k] += v describe = {} for name, s in list(desc.items()) + [("POOLED", pooled)]: n = max(s["uncovered_valid"], 1.0) describe[name] = {"uncovered_valid_pixels": int(s["uncovered_valid"]), "constrained_share": s["constrained"] / n, "mean_bound_over_gt": s["bound_over_gt_sum"] / n, "gt_below_bound_share": s["gt_below_bound"] / n, "gt_in_carved_share": s["gt_in_carved"] / n, **{f"{m}_below_bound_share": s[f"{m}_below_bound"] / n for m in METHODS}, **{f"{m}_in_carved_share": s[f"{m}_in_carved"] / n for m in METHODS}} regional = {m: {st: RS.pixel_absrel(entries, m, st)[0] for st in ("ALL", "UNC_NEAR", "UNC_FILL", "FILL_D0_8", "FILL_D8_32", "FILL_D32", "ZERO_HIT")} for m in names} res = {"places": len(table["B7"]), "queries": len(entries), "means": means, "contrasts": contrasts, "primary": {"contrast": p1, "label": "A_PREMISE_FREE_SPACE=" + label}, "describe": describe, "pooled_pixel_absrel": regional, "read_purposes": sorted(purposes), "reads": len(access), "carving": {"rel": REL, "abs_m": ABS, "samples": len(T), "range_m": [0.1, 20.0]}} dest = out_dir if smoke else ROOT (dest / "AMENDMENT_3_RESULTS.json").write_text(json.dumps(res, indent=1, sort_keys=True) + "\n") print("AMEND3", res["primary"]["label"], json.dumps(p1), flush=True) print(json.dumps(describe["POOLED"]), flush=True) if __name__ == "__main__": {"lock": lock, "run": lambda: run(False), "smoke": lambda: run(True)}[STAGE]()