tests/test_rgbd_data_scale_carrier.py
"""V14 data scale: only the TRAIN set differs; both variants are the V11 C1 draw; gates, 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_evidence_carrier as v5
from mcss.mechanism_pilot import rgbd_resolution_carrier as v8
from mcss.mechanism_pilot.rgbd_data_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_both_variants_are_the_v11_c1_draw_and_differ_only_by_their_train_set():
assert {v: (s["hidden"], s["expansion"]) for v, s in VARIANT_SPECS.items()} == {
"C0": (32, 64),
"C1": (32, 64),
}
assert {v: s["train"] for v, s in VARIANT_SPECS.items()} == {"C0": "TRAIN72", "C1": "TRAIN_EXT"}
reference = hash_value(v11.make_carrier("C1", 3, "cpu").state_dict())
for variant in VARIANT_SPECS:
carrier = make_carrier(variant, 3, "cpu")
assert hash_value(carrier.state_dict()) == reference
assert sum(p.numel() for p in carrier.parameters()) == 64238
def test_train_records_follow_the_v14_table_and_not_the_v8_table():
manifest = {"scenes": [{"scene_id": f"t{i:02d}", "split": "TRAIN"} for i in range(30)]}
manifest["scenes"].append({"scene_id": "d00", "split": "DEV"})
config = {
"train_sets": {
"TRAIN72": [f"t{i:02d}" for i in range(24)],
"TRAIN_EXT": [f"t{i:02d}" for i in range(30)],
}
}
c0, c1 = train_records(manifest, config, "C0"), train_records(manifest, config, "C1")
assert [r["scene_id"] for r in c0] == config["train_sets"]["TRAIN72"]
assert [r["scene_id"] for r in c1] == config["train_sets"]["TRAIN_EXT"]
# The V8 helper resolves every variant to TRAIN72: reusing it would silently drop TRAIN-EXT.
assert len(v8.train_records(manifest, config, "C1")) == 24
config["train_sets"]["TRAIN_EXT"].append("missing")
with pytest.raises(PermissionError):
train_records(manifest, config, "C1")
def test_both_variants_reproduce_the_v11_c1_state_and_loss_exactly():
rgbd, bounds, cameras = context()
old = v11.make_carrier("C1", 11, "cpu")
with torch.no_grad():
old.depth_weight.fill_(0.2)
first, _ = v11.build_state(old, rgbd, bounds, "same")
a, _ = v11.training_loss(first, cameras, *targets(), "C1")
for variant in VARIANT_SPECS:
new = make_carrier(variant, 11, "cpu")
with torch.no_grad():
new.depth_weight.fill_(0.2)
second, _ = build_state(new, rgbd, bounds, "same")
assert hash_scene_state(first) == hash_scene_state(second)
b, _ = training_loss(second, cameras, *targets(), variant)
assert torch.equal(a, b)
def test_checkpoints_roundtrip_under_the_v14_schema_only(tmp_path):
rgbd, bounds, cameras = context()
carrier = make_carrier("C1", 5, "cpu")
state, _ = build_state(carrier, rgbd, bounds, "ext")
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 / "ext.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"] == 32 and payload["variant"] == "C1"
assert hash_value(restored.state_dict()) == hash_value(carrier.state_dict())
with pytest.raises(PermissionError):
v11.load_checkpoint(path, "cpu")
old = tmp_path / "v11.pt"
v11.save_checkpoint(
old, v11.make_carrier("C1", 5, "cpu"), variant="C1", seed=5, step=1, lock_sha256="v11"
)
with pytest.raises(PermissionError):
load_checkpoint(old, "cpu")
odd = v8.ResolutionRGBDCarrier(
CarrierConfig(grid_size=(32, 32, 32), token_count=32**3, hidden_dim=64, expansion_dim=128)
)
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_data_scale_evaluation.py").read_text()
saves = [
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(saves) == 1
assert source.count('out / "states.pt")') == 2 # the guarded save and the guarded hash
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):
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):
rows.append(make_row(base, noise, "C0", seed, c0_far, 0.33))
rows.append(make_row(base, noise, "C1", seed, c1_far, 0.33))
for prefix_method in ("HFILL8", "AHFILL8"):
rows.append(
make_row(
base,
noise,
f"{prefix_method}_C0",
seed,
c0_far,
hfill_all + 0.01,
)
)
rows.append(
make_row(
base, noise, f"{prefix_method}_C1", seed, c1_far, hfill_all
)
)
return rows
def test_eval_gates_on_the_pooled_primary_cohorts_set_the_branch():
analyze_rows = script("evaluate_rgbd_data_scale_eval.py").analyze_rows
result = analyze_rows(rows_for(0.36, 0.31, 0.28, 0.25))
assert result["primary"]["n_scenes"] == 8
assert set(result["cohorts"]) == {"EVAL_V3", "EVAL_V4", "EVAL_FAR"}
assert result["primary"]["contrasts"]["DATA_GAIN"]["gain"]["mean"] == pytest.approx(0.05)
assert result["statuses"]["DATA_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_train_set_and_seed_concurrently_on_cpu(tmp_path, monkeypatch):
module = script("run_rgbd_data_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_data_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")}
assert all(a[a.index("--device") + 1] == "cpu" for a in seen)