"""V15 width x data: only the width differs on TRAIN72+EXT; V11/V12 draws; gates, 2x2, runner.""" import ast import importlib.util from pathlib import Path import pytest import torch from test_mechanism_training import observation from mcss.dynamic.config import CarrierConfig from mcss.dynamic.types import hash_scene_state, hash_value from mcss.geometry import transform_cameras from mcss.mechanism_pilot import rgbd_completion_carrier as v11 from mcss.mechanism_pilot import rgbd_data_scale_carrier as v14 from mcss.mechanism_pilot import rgbd_evidence_carrier as v5 from mcss.mechanism_pilot import rgbd_resolution_carrier as v8 from mcss.mechanism_pilot import rgbd_width_carrier as v12 from mcss.mechanism_pilot.rgbd_wide_scale_carrier import ( VARIANT_SPECS, build_state, load_checkpoint, make_carrier, save_checkpoint, train_records, training_loss, ) from mcss.types import Cameras ROOT = Path(__file__).parents[1] def script(name): path = ROOT / "scripts" / name spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def context(): torch.set_num_threads(1) observations = [observation(i) for i in (0, 1, 2)] rgbd = v5.RGBDContext(observations, torch.full((3, 8, 8), 2.0)) bounds = torch.tensor([[-2.0, -2.0, 0.1], [2.0, 2.0, 4.0]]) query = observation(5).camera cameras = Cameras(query.intrinsics[None, None], query.c2w[None, None], (8, 8)) cameras = transform_cameras(cameras, torch.linalg.inv(observations[0].camera.c2w)) return rgbd, bounds, cameras def targets(): return torch.full((1, 1, 3, 8, 8), 0.6), torch.full((1, 1, 1, 8, 8), 2.0), torch.arange(64) def test_only_the_width_differs_and_the_draws_are_v11_c1_and_v12_c1(): assert {v: (s["hidden"], s["expansion"], s["train"]) for v, s in VARIANT_SPECS.items()} == { "C0": (32, 64, "TRAIN_EXT"), "C1": (64, 128, "TRAIN_EXT"), } narrow, wide = make_carrier("C0", 3, "cpu"), make_carrier("C1", 3, "cpu") assert hash_value(narrow.state_dict()) == hash_value( v11.make_carrier("C1", 3, "cpu").state_dict() ) assert hash_value(narrow.state_dict()) == hash_value( v14.make_carrier("C1", 3, "cpu").state_dict() ) assert hash_value(wide.state_dict()) == hash_value( v12.make_carrier("C1", 3, "cpu").state_dict() ) assert sum(p.numel() for p in narrow.parameters()) == 64238 assert sum(p.numel() for p in wide.parameters()) == 250542 def test_both_variants_train_on_the_shared_train_ext_set(): manifest = {"scenes": [{"scene_id": f"t{i:02d}", "split": "TRAIN"} for i in range(30)]} config = { "train_sets": { "TRAIN72": [f"t{i:02d}" for i in range(24)], "TRAIN_EXT": [f"t{i:02d}" for i in range(30)], } } for variant in VARIANT_SPECS: assert [r["scene_id"] for r in train_records(manifest, config, variant)] == config[ "train_sets" ]["TRAIN_EXT"] def test_width32_reproduces_the_v14_c1_state_and_loss_exactly(): rgbd, bounds, cameras = context() old, new = v14.make_carrier("C1", 11, "cpu"), make_carrier("C0", 11, "cpu") with torch.no_grad(): for carrier in (old, new): carrier.depth_weight.fill_(0.2) first, _ = v14.build_state(old, rgbd, bounds, "same") second, _ = build_state(new, rgbd, bounds, "same") assert hash_scene_state(first) == hash_scene_state(second) a, _ = v14.training_loss(first, cameras, *targets(), "C1") b, _ = training_loss(second, cameras, *targets(), "C0") assert torch.equal(a, b) def test_the_wide_carrier_trains_and_roundtrips_under_the_v15_schema_only(tmp_path): rgbd, bounds, cameras = context() carrier = make_carrier("C1", 5, "cpu") state, _ = build_state(carrier, rgbd, bounds, "wide") assert tuple(state.spatial_shape) == (32, 32, 32) loss, _ = training_loss(state, cameras, *targets(), "C1") loss.backward() assert carrier.refinement[0].weight.grad.abs().sum() > 0 path = tmp_path / "wide.pt" save_checkpoint(path, carrier, variant="C1", seed=5, step=1, lock_sha256="x") restored, payload = load_checkpoint(path, "cpu") assert payload["config"]["hidden_dim"] == 64 assert hash_value(restored.state_dict()) == hash_value(carrier.state_dict()) for other in (v12.load_checkpoint, v14.load_checkpoint): with pytest.raises(PermissionError): other(path, "cpu") old = tmp_path / "v12.pt" v12.save_checkpoint( old, v12.make_carrier("C1", 5, "cpu"), variant="C1", seed=5, step=1, lock_sha256="v12" ) with pytest.raises(PermissionError): load_checkpoint(old, "cpu") odd = v8.ResolutionRGBDCarrier( CarrierConfig(grid_size=(32, 32, 32), token_count=32**3, hidden_dim=16, expansion_dim=32) ) with pytest.raises(ValueError): build_state(odd, rgbd, bounds, "odd") def test_dev_curve_evaluations_do_not_save_states(): source = (ROOT / "src/mcss/mechanism_pilot/rgbd_wide_scale_evaluation.py").read_text() guarded = [ node for node in ast.walk(ast.parse(source)) if isinstance(node, ast.If) and isinstance(node.test, ast.Name) and node.test.id == "diagnostics" and any("states.pt" in ast.unparse(child) for child in node.body) ] assert len(guarded) == 1 assert 'sha(out / "states.pt") if diagnostics else None' in source def make_row(base, noise, method, seed, far, total): return { **base, "method": method, "seed": seed, "absrel_ALL": total + noise, "absrel_NEAR": 0.2, "absrel_FAR": far + noise, "depth_absrel": total + noise, } def rows_for(c0_far, c1_far, harmonic_all, hfill_all, w32_t72_far=0.40, w64_t72_far=0.40): rows = [] cohorts = {"EVAL_V3": "v3", "EVAL_V4": "v4", "EVAL_FAR": "v3"} for cohort, prefix in cohorts.items(): for i in range(4): for role in ("A", "B"): for qid in (8, 9): noise = 0.002 * ((i + qid) % 3) base = { "cohort": cohort, "scene_id": f"{prefix}_s{i}", "role": role, "query_id": qid, "far_fraction": 0.2, } rows.append(make_row(base, noise, "CONSTANT", 0, 0.5, 0.5)) rows.append(make_row(base, noise, "REPROJ_NN", 0, 0.46, harmonic_all + 0.01)) rows.append(make_row(base, noise, "REPROJ_HARMONIC", 0, 0.45, harmonic_all)) for seed in (1, 2): cells = {"C0": c0_far, "C1": c1_far} if cohort != "EVAL_FAR": cells |= { "W32_T72": w32_t72_far, "W64_T72": w64_t72_far, "C0_LAST": c0_far, "C1_LAST": c1_far, } for method, far in cells.items(): total = hfill_all if method.startswith("C1") else hfill_all + 0.01 rows.append(make_row(base, noise, method, seed, far, 0.33)) rows.append(make_row(base, noise, f"HFILL8_{method}", seed, far, total)) if method in ("C0", "C1"): rows.append( make_row(base, noise, f"AHFILL8_{method}", seed, far, total) ) return rows def test_eval_gates_and_the_width_by_data_interaction(): analyze_rows = script("evaluate_rgbd_wide_scale_eval.py").analyze_rows result = analyze_rows(rows_for(0.44, 0.39, 0.28, 0.25, w32_t72_far=0.42, w64_t72_far=0.42)) contrasts = result["primary"]["contrasts"] assert result["primary"]["n_scenes"] == 8 assert contrasts["WIDTH_AT_SCALE_GAIN"]["gain"]["mean"] == pytest.approx(0.05) assert contrasts["INTERACTION"]["gain"]["mean"] == pytest.approx(-0.05) assert contrasts["DATA_AT_W32_GAIN"]["gain"]["mean"] == pytest.approx(-0.02) assert result["cohorts"]["EVAL_FAR"]["contrasts"]["INTERACTION"]["gain"] is None assert ( result["statuses"]["WIDTH_AT_SCALE_STATUS"], result["statuses"]["HARMONIC_HYBRID_LABEL"], ) == ("SUPPORTED", "ABOVE") assert result["INTERPRETATION_BRANCH"] == "A" result = analyze_rows(rows_for(0.44, 0.39, 0.24, 0.25)) assert result["statuses"]["HARMONIC_HYBRID_LABEL"] == "BELOW" assert result["INTERPRETATION_BRANCH"] == "B" assert analyze_rows(rows_for(0.39, 0.44, 0.24, 0.25))["INTERPRETATION_BRANCH"] == "C" def test_runner_trains_every_width_and_seed_concurrently_on_cpu(tmp_path, monkeypatch): module = script("run_rgbd_wide_scale_experiment.py") seen = [] monkeypatch.setattr(module, "execute", lambda root, label, args, **kwargs: seen.append(args)) config = { "seeds": [1, 2, 3], "variants": ["C0", "C1"], "primary_pair": ["C0", "C1"], "parallel_workers": 6, "infrastructure_retries": 2, } module.train_all(tmp_path, config, env=None, times={}) assert len(seen) == 6 and {a[0] for a in seen} == {"scripts/train_rgbd_wide_scale_carrier.py"} pairs = {(a[a.index("--variant") + 1], a[a.index("--seed") + 1]) for a in seen} assert pairs == {(v, s) for v in ("C0", "C1") for s in ("1", "2", "3")}