import hashlib import importlib.util import json from pathlib import Path import numpy as np import pytest SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" def load(name, filename): spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module replication = load("replication", "evaluate_rgbd_replication_eval_v4.py") inpaint = load("inpaint_for_replication_test", "run_rgbd_inpaint_baselines.py") def rows_for(cohort, scenes, rng, sign=1): rows = [] base = { "REPROJ_NN": 0.30, "REPROJ_HARMONIC": 0.28, "REPROJ_TELEA": 0.29, "C0": 0.33, "C1": 0.32, } for s in range(scenes): for role in ("A", "B"): for q in (1, 2): common = {"cohort": cohort, "scene_id": f"s{s}", "role": role, "query_id": q} common["far_fraction"] = 0.2 for method, value in base.items(): for seed in (0,) if method.startswith("REPROJ") else (1, 2): noise = rng.normal(0, 0.003) rows.append( { **common, "method": method, "seed": seed, "depth_absrel": value + noise, "absrel_ALL": value + noise, "absrel_NEAR": value - noise, "absrel_FAR": value + 2 * noise, } ) for method in ("FILL8_C0", "FILL8_C1", "HFILL8_C0", "HFILL8_C1", "TFILL8_C1"): for seed in (1, 2): value = 0.28 - sign * (0.02 + abs(rng.normal(0, 0.003))) rows.append( { **common, "method": method, "seed": seed, "depth_absrel": value, "absrel_ALL": value, "absrel_NEAR": value, "absrel_FAR": value, } ) return rows def test_contrasts_and_block_equal_the_frozen_inpaint_v1_analysis(): rng = np.random.default_rng(3) rows = rows_for("EVAL_V3", 12, rng) + rows_for("DEV", 4, rng) frozen = inpaint.analyze(rows)["cohorts"]["EVAL_V3"] mine = replication.inpaint_block([r for r in rows if r["cohort"] == "EVAL_V3"]) assert set(mine["contrasts"]) == set(frozen["contrasts"]) for name, entry in frozen["contrasts"].items(): assert entry["definition"] == mine["contrasts"][name]["definition"] assert entry["gain"] == mine["contrasts"][name]["gain"] assert entry.get("label") == mine["contrasts"][name].get("label") assert frozen["absolute_absrel"] == mine["absolute_absrel"] assert frozen["n_scenes"] == mine["n_scenes"] == 12 def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def test_lock_requires_the_frozen_plan_and_a_prepared_cohort(tmp_path, monkeypatch): root = tmp_path / replication.EXPERIMENT root.mkdir() (root / "PROTOCOL.md").write_text("protocol") (root / replication.PLAN).write_text("a different plan") v11, inp, eval_v4 = tmp_path / "v11", tmp_path / "inpaint", tmp_path / "eval_v4" for directory in (v11, inp, eval_v4): directory.mkdir() (v11 / "training_contract.json").write_text("{}") (inp / "lock.json").write_text("{}") for name in ("manifest_eval_v4.json", "candidate_lock.json"): (eval_v4 / name).write_text("{}") (eval_v4 / "preparation_integrity.json").write_text(json.dumps({"status": "PASS"})) with pytest.raises(PermissionError): replication.lock(root, v11, inp, eval_v4) monkeypatch.setattr(replication, "PLAN_SHA256", sha(root / replication.PLAN)) (eval_v4 / "preparation_integrity.json").write_text(json.dumps({"status": "BLOCKED"})) with pytest.raises(PermissionError): replication.lock(root, v11, inp, eval_v4) (eval_v4 / "preparation_integrity.json").write_text(json.dumps({"status": "PASS"})) replication.lock(root, v11, inp, eval_v4) lock = json.loads((root / "lock.json").read_text()) assert lock["v11_results_existed_at_lock"] is False and lock["final_holdout_opened"] is False assert set(lock["input_sha256"]) >= {"plan", "script", "eval_v4_manifest", "inpaint_lock"} with pytest.raises(FileExistsError): replication.lock(root, v11, inp, eval_v4) def test_score_and_analyze_on_a_synthetic_seal(tmp_path): rng = np.random.default_rng(5) root = tmp_path / replication.EXPERIMENT (root / "raw" / "predictions").mkdir(parents=True) seeds = [1, 2] contract = tmp_path / "contract.json" contract.write_text(json.dumps({"seeds": seeds})) scenes, entries = [], {} for s in range(6): sid = f"ai_{s:03d}_001" frames = [] for fid in (10, 11): gt = rng.uniform(1, 5, (1, 32, 40)).astype(np.float32) np.save(tmp_path / f"{sid}_{fid}.npy", gt) frames.append({"frame_id": fid, "depth": str(tmp_path / f"{sid}_{fid}.npy")}) for role in ("A", "B"): hit = rng.random((32, 40)) < 0.5 hit[5:25, 10:30] = False arrays = { "hit": hit, "near": inpaint.near_mask(hit), "REPROJ_NN": rng.uniform(1, 5, (32, 40)), "REPROJ_HARMONIC": rng.uniform(1, 5, (32, 40)), "REPROJ_TELEA": rng.uniform(1, 5, (32, 40)), } for variant in ("C0", "C1"): for seed in seeds: arrays[f"{variant}_{seed}"] = rng.uniform(1, 5, (32, 40)) path = root / "raw" / "predictions" / f"{sid}_{role}_{fid}.npz" np.savez_compressed(path, **arrays) entries[f"{sid}/{role}/{fid}"] = {"path": str(path), "sha256": sha(path)} scenes.append( { "scene_id": sid, "split": "EVAL_V4", "frames": frames, "roles": {"primary_query": [10, 11]}, } ) manifest = tmp_path / "manifest_eval_v4.json" manifest.write_text(json.dumps({"scenes": scenes, "image_size": [32, 40]})) lock = { "input_sha256": { "eval_v4_manifest": {"path": str(manifest), "sha256": sha(manifest)}, "v11_contract": {"path": str(contract), "sha256": sha(contract)}, } } (root / "lock.json").write_text(json.dumps(lock)) replay = {"scenes": 2, "arrays_compared": 10, "max_abs_deviation": 0.0} (root / "raw" / "prediction_seal.json").write_text( json.dumps({"predictions": entries, "replay_on_eval_v3": replay}) ) rows = replication.score(root) per_query = 3 + 2 * len(seeds) * 4 assert len(rows) == 6 * 2 * 2 * per_query result = replication.analyze(root) assert result["eval_v4"]["n_scenes"] == 6 assert result["REPLICATION_BRANCH"] in {"A", "B", "C"} assert result["V11_BRANCH_ON_EVAL_V4"] in {"A", "B", "C"} assert (root / "README.md").exists() and (root / "terminal_summary.txt").exists() entry = next(iter(entries.values())) Path(entry["path"]).write_bytes(b"tampered") with pytest.raises(PermissionError): replication.score(root)