tests/test_rgbd_width_carrier.py
"""V12 width 64: only the width differs; width 32 is V11 C1; EVAL-V4 gates and runner."""
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_evidence_carrier as v5
from mcss.mechanism_pilot import rgbd_resolution_carrier as v8
from mcss.mechanism_pilot.rgbd_width_carrier import (
VARIANT_SPECS,
build_state,
load_checkpoint,
load_v11_checkpoint,
make_carrier,
save_checkpoint,
training_loss,
)
from mcss.types import Cameras
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 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_width32_is_the_v11_c1_draw():
assert {v: (s["hidden"], s["expansion"]) for v, s in VARIANT_SPECS.items()} == {
"C0": (32, 64),
"C1": (64, 128),
}
assert {s["train"] for s in VARIANT_SPECS.values()} == {"TRAIN72"}
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 sum(p.numel() for p in narrow.parameters()) == 64238
assert sum(p.numel() for p in wide.parameters()) == 250542
assert wide._candidate_points.shape[0] == 32**3
def test_width32_reproduces_the_v11_c1_state_and_loss_exactly():
rgbd, bounds, cameras = context()
old, new = v11.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, _ = v11.build_state(old, rgbd, bounds, "same")
second, _ = build_state(new, rgbd, bounds, "same")
assert hash_scene_state(first) == hash_scene_state(second)
a, _ = v11.training_loss(first, cameras, *targets(), "C1")
b, _ = training_loss(second, cameras, *targets(), "C0")
assert torch.equal(a, b)
def test_the_widest_carrier_trains_and_roundtrips(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())
with pytest.raises(PermissionError):
v11.load_checkpoint(path, "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_v11_c1_checkpoints_load_with_the_v11_loader_and_build_under_the_v12_guard(tmp_path):
rgbd, bounds, _ = context()
carrier = v11.make_carrier("C1", 7, "cpu")
path = tmp_path / "v11.pt"
v11.save_checkpoint(path, carrier, variant="C1", seed=7, step=1, lock_sha256="v11")
restored, payload = load_v11_checkpoint(path, "cpu")
assert payload["schema"] == v11.SCHEMA
with pytest.raises(PermissionError):
load_checkpoint(path, "cpu")
first, _ = v11.build_state(restored, rgbd, bounds, "control")
second, _ = build_state(restored, rgbd, bounds, "control")
assert hash_scene_state(first) == hash_scene_state(second)
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(control_far, c1_far, harmonic_all, hfill_all, scenes=8):
rows = []
for i in range(scenes):
for role in ("A", "B"):
for qid in (8, 9):
noise = 0.002 * ((i + qid) % 3)
base = {"scene_id": f"s{i}", "role": role, "query_id": qid, "far_fraction": 0.2}
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):
rows.append(make_row(base, noise, "V11C1", seed, control_far, 0.33))
rows.append(make_row(base, noise, "C0", seed, control_far, 0.33))
rows.append(make_row(base, noise, "C1", seed, c1_far, 0.33))
rows.append(
make_row(base, noise, "HFILL8_V11C1", seed, control_far, hfill_all + 0.01)
)
rows.append(make_row(base, noise, "HFILL8_C1", seed, c1_far, hfill_all))
rows.append(make_row(base, noise, "FILL8_C1", seed, c1_far, hfill_all + 0.005))
return rows
def test_eval_v4_gates_set_the_branch():
analyze_rows = script("evaluate_rgbd_width_eval_v4.py").analyze_rows
result = analyze_rows(rows_for(0.36, 0.31, 0.28, 0.25))
assert result["WIDTH64_COMPLETION_GAIN"]["mean"] == pytest.approx(0.05)
assert result["statuses"]["WIDTH64_COMPLETION_STATUS"] == "SUPPORTED"
assert result["statuses"]["HARMONIC_HYBRID_LABEL"] == "ABOVE"
assert result["INTERPRETATION_BRANCH"] == "A"
result = analyze_rows(rows_for(0.36, 0.31, 0.24, 0.25))
assert result["statuses"]["HARMONIC_HYBRID_LABEL"] == "BELOW"
assert result["INTERPRETATION_BRANCH"] == "B"
assert analyze_rows(rows_for(0.31, 0.36, 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_width_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_width_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")}
assert all(a[a.index("--device") + 1] == "cpu" for a in seen)