"""V4 plane-sweep carrier: geometric correctness of the cue and the matched single-factor design.""" import importlib.util import json import math from pathlib import Path import torch from test_mechanism_training import observation from mcss.dynamic.carrier import DynamicSceneCarrier from mcss.dynamic.types import OnlineObservation, hash_scene_state, hash_value from mcss.geometry import transform_cameras from mcss.mechanism_pilot.plane_sweep_carrier import ( VARIANT_SPECS, PlaneSweepCarrier, build_state, load_checkpoint, make_carrier, parameter_hash, plane_sweep_statistics, save_checkpoint, training_loss, ) from mcss.types import Cameras NEAR, FAR = 1.0, 4.0 def texture(x, y): return torch.stack( ( 0.5 + 0.4 * torch.sin(9 * x + 4 * y), 0.5 + 0.4 * torch.cos(11 * x - 3 * y), 0.5 + 0.4 * torch.sin(13 * y + 2 * x), ) ) def plane_view(center_x, frame, height=64, width=80, plane_z=2.0, focal=80.0): """Analytic render of a textured fronto-parallel plane z=plane_z seen along +z.""" cx, cy = (width - 1) / 2, (height - 1) / 2 intrinsics = torch.tensor([[focal, 0, cx], [0, focal, cy], [0, 0, 1.0]]) v, u = torch.meshgrid( torch.arange(height, dtype=torch.float32), torch.arange(width, dtype=torch.float32), indexing="ij", ) x = center_x + (u - cx) / focal * plane_z y = (v - cy) / focal * plane_z pose = torch.eye(4) pose[0, 3] = center_x return OnlineObservation( "plane", frame, texture(x, y), Cameras(intrinsics, pose, (height, width)) ) def test_plane_sweep_localizes_a_known_surface_and_its_free_space(): # 4x area downsample to 16x20; baselines give 2-4 px disparity steps between test depths. views = [plane_view(x, i) for i, x in enumerate((0.0, -0.4, 0.4))] depths = torch.tensor([1.2, 1.6, 2.0, 2.4, 3.2]) points = torch.stack((torch.zeros(5), torch.zeros(5), depths), -1) stats = plane_sweep_statistics( views, points, NEAR, FAR, torch.tensor(math.log(0.01)), planes=16, size=(16, 20) ) surface, free, valid = stats.unbind(-1) assert torch.isfinite(stats).all() assert torch.all(valid == 1.0) assert surface[2] > 2 * max(surface[0], surface[4]) assert free[0] > 0.8 and free[4] < 0.2 assert free[0] > free[2] > free[4] def test_single_view_contributes_no_cue_and_temperature_gets_gradient(): views = [plane_view(x, i) for i, x in enumerate((0.0, -0.4, 0.4))] points = torch.tensor([[0.0, 0.0, 2.0], [0.1, 0.0, 1.5]]) log_tau = torch.tensor(math.log(0.01), requires_grad=True) single = plane_sweep_statistics(views[:1], points, NEAR, FAR, log_tau, planes=16, size=(16, 20)) assert torch.equal(single, torch.zeros_like(single)) stats = plane_sweep_statistics(views, points, NEAR, FAR, log_tau, planes=16, size=(16, 20)) stats[:, 0].sum().backward() assert log_tau.grad is not None and torch.isfinite(log_tau.grad) and log_tau.grad != 0 def context(): torch.set_num_threads(1) observations = [observation(i) for i in (0, 1, 2)] 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 observations, bounds, cameras def test_sweep_carrier_starts_exactly_equal_to_the_dense_baseline(): observations, bounds, _ = context() base = make_carrier("C0", 11, "cpu", 0.2, 6.0) sweep = make_carrier("C1", 11, "cpu", 0.2, 6.0) assert type(base) is DynamicSceneCarrier and isinstance(sweep, PlaneSweepCarrier) assert parameter_hash(base) == parameter_hash(sweep) assert parameter_hash(make_carrier("C2", 11, "cpu", 0.2, 6.0)) == parameter_hash(base) assert sum(p.numel() for p in base.parameters()) == 5125 assert sum(p.numel() for p in sweep.parameters()) == 5174 first, _ = build_state(base, observations, bounds, "same") second, _ = build_state(sweep, observations, bounds, "same") assert hash_scene_state(first) == hash_scene_state(second) def test_surface_loss_gradient_reaches_the_sweep_bypass(): observations, bounds, cameras = context() carrier = make_carrier("C1", 5, "cpu", 0.2, 6.0) state, _ = build_state(carrier, observations, bounds, "grad") loss, _ = training_loss( state, cameras, torch.full((1, 1, 3, 8, 8), 0.6), torch.full((1, 1, 1, 8, 8), 2.0), torch.arange(64), "C1", ) loss.backward() grad = carrier.sweep_weight.grad assert grad is not None and torch.isfinite(grad).all() and grad.abs().sum() > 0 def test_variant_losses_follow_the_frozen_mapping(): assert {v: s["loss"] for v, s in VARIANT_SPECS.items()} == {"C0": "C1", "C1": "C1", "C2": "C0"} assert {v: s["sweep"] for v, s in VARIANT_SPECS.items()} == { "C0": False, "C1": True, "C2": True, } def test_checkpoint_roundtrip_keeps_class_and_weights(tmp_path): for variant in ("C0", "C1"): carrier = make_carrier(variant, 3, "cpu", 0.2, 6.0) with torch.no_grad(): for p in carrier.parameters(): p.add_(0.01) path = tmp_path / f"{variant}.pt" save_checkpoint(path, carrier, variant=variant, seed=3, step=1, lock_sha256="x") restored, payload = load_checkpoint(path, "cpu") assert type(restored) is type(carrier) assert hash_value(restored.state_dict()) == hash_value(carrier.state_dict()) assert payload["variant"] == variant def script(name): path = Path(__file__).parents[1] / "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 test_v4_runner_trains_only_v4_scripts(tmp_path, monkeypatch): module = script("run_plane_sweep_experiment.py") seen = [] monkeypatch.setattr(module, "execute", lambda root, label, args, **kwargs: seen.append(args[0])) config = { "seeds": [1], "variants": ["C0", "C1", "C2"], "primary_pair": ["C0", "C1"], "parallel_workers": 3, "infrastructure_retries": 2, } module.train_all(tmp_path, config, env=None, times={}) assert seen == ["scripts/train_plane_sweep_carrier.py"] * 3 def test_v4_report_uses_sweep_semantics(tmp_path): from test_geometry_carrier_statistics import fixture, region_fixture data = fixture() values = { "scene_split": data[3], "training_contract": {"seeds": data[4]["seeds"], "steps": 2000}, "raw/dev_query_results": data[0], "raw/dev_context_results": data[1], "raw/dev_matched_results": data[2], "raw/dev_region_results": region_fixture(data), "raw/dev_static_matched_results": data[4]["static_matched_rows"], "audit/dev_state_use": data[4]["state_use_audit"], "training_curves": {"training": [], "dev": []}, "cost_analysis": {"variants": {v: {"training_seconds": 1.0} for v in ("C0", "C1")}}, } for name, value in values.items(): path = tmp_path / (name + ".json") path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value)) script("analyze_plane_sweep_carrier.py").run(tmp_path) fields = script("report_plane_sweep_carrier.py").run(tmp_path) assert fields["SWEEP_GAIN"] == "0.010000" assert fields["SWEEP_STATUS"] == "SUPPORTED" assert fields["SCENE_SPECIFICITY_STATUS"] == "SUPPORTED" assert fields["INTERPRETATION_BRANCH"] == "A" assert fields["FRESH_QUALIFICATION_INCLUDED"] is False assert not fields["DYNAMIC_TTT_NEXT_STAGE_ALLOWED"] text = (tmp_path / "README.md").read_text() assert "SWEEP_GAIN" in text and all(f"**{i}. " in text for i in range(1, 12))