tests/test_rgbd_long_data_carrier.py
"""V18 long training: variants, V11 C1 draw, checkpoints, control, labels and branches."""
import importlib.util
import json
from pathlib import Path
import pytest
import torch
from mcss.mechanism_pilot import rgbd_data_scale_carrier as v14
from mcss.mechanism_pilot import rgbd_long_data_carrier as v18
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
def script(name):
spec = importlib.util.spec_from_file_location(name.removesuffix(".py"), SCRIPTS / name)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_single_variant_is_the_v14_c1_recipe_on_train127():
assert v18.VARIANTS == ("L",) and v18.EXPERIMENT == "EXP-3D-RGBD-LONG-TRAIN-DATA-V18"
spec = v18.VARIANT_SPECS["L"]
assert spec["train"] == "TRAIN_EXT" and (spec["hidden"], spec["expansion"]) == (32, 64)
carrier = v18.make_carrier("L", 20260928, "cpu")
assert sum(p.numel() for p in carrier.parameters()) == 64238
@pytest.mark.parametrize("seed", [20260928, 20260929, 20260930])
def test_parameter_draw_equals_v14_c1_which_is_v11_c1(seed):
mine = v18.parameter_hash(v18.make_carrier("L", seed, "cpu"))
theirs = v14.parameter_hash(v14.make_carrier("C1", seed, "cpu"))
assert mine == theirs
def test_checkpoint_roundtrip_and_schema_guard(tmp_path):
carrier = v18.make_carrier("L", 20260928, "cpu")
path = tmp_path / "step_000100.pt"
digest = v18.save_checkpoint(
path, carrier, variant="L", seed=20260928, step=100, lock_sha256="x"
)
assert len(digest) == 64
restored, payload = v18.load_checkpoint(path, "cpu")
assert payload["step"] == 100
assert all(
torch.equal(a, b)
for a, b in zip(restored.state_dict().values(), carrier.state_dict().values(), strict=True)
)
with pytest.raises(FileExistsError):
v18.save_checkpoint(path, carrier, variant="L", seed=20260928, step=100, lock_sha256="x")
other = tmp_path / "v14.pt"
v14.save_checkpoint(
other,
v14.make_carrier("C0", 20260928, "cpu"),
variant="C0",
seed=20260928,
step=1,
lock_sha256="x",
)
with pytest.raises(PermissionError):
v18.load_checkpoint(other, "cpu")
def test_control_compares_logs_without_seconds_and_parameters_exactly(tmp_path):
control = script("check_rgbd_long_data_prefix.py")
rows = [
{"step": i, "loss": 1.0 / i, "seconds": 0.5, "terms": {"A": {"loss": 0.1}}}
for i in range(1, 7)
]
a, b, c = tmp_path / "a.jsonl", tmp_path / "b.jsonl", tmp_path / "c.jsonl"
a.write_text("\n".join(json.dumps(r) for r in rows) + "\n")
b.write_text("\n".join(json.dumps({**r, "seconds": 9.0}) for r in rows) + "\n")
changed = [dict(r) for r in rows]
changed[3]["loss"] += 1e-12
c.write_text("\n".join(json.dumps(r) for r in changed) + "\n")
assert control.compare_logs(a, b, steps=6)["equal"]
assert control.compare_logs(a, c, steps=6) == {"equal": False, "first_mismatch_step": 4}
assert control.compare_logs(a, b, steps=7)["equal"] is False
carrier = v18.make_carrier("L", 20260928, "cpu")
p, q = tmp_path / "p.pt", tmp_path / "q.pt"
v18.save_checkpoint(p, carrier, variant="L", seed=20260928, step=6000, lock_sha256="one")
v18.save_checkpoint(q, carrier, variant="L", seed=20260928, step=6000, lock_sha256="two")
assert control.compare_parameters(p, q)["equal"]
assert control.compare_parameters(tmp_path / "absent.pt", p)["equal"] is False
assert control.compare_logs(tmp_path / "absent.jsonl", a, steps=6)["equal"] is False
with torch.no_grad():
next(carrier.parameters()).add_(1.0)
r = tmp_path / "r.pt"
v18.save_checkpoint(r, carrier, variant="L", seed=20260928, step=6000, lock_sha256="one")
assert control.compare_parameters(p, r)["equal"] is False
def test_labels_cover_the_curve_on_primary_cohorts_only():
evaluator = script("evaluate_rgbd_long_data_eval.py")
config = {
"eval": {
"curve_steps_primary": [1000, 6000, 24000],
"steps_descriptive_cohorts": [6000, 24000],
}
}
assert evaluator.labels_for("EVAL_V3", config) == ["T1000", "T6000", "T24000", "SEL"]
assert evaluator.labels_for("FRESH_V2", config) == ["T6000", "T24000", "SEL"]
def rows(length_gain, hybrid_gain, scenes=40):
"""FAR(T6000) - FAR(T24000) = length_gain; harmonic - AHFILL8 = hybrid_gain (per scene)."""
out = []
for i in range(scenes):
jitter = 0.002 * ((i % 5) - 2)
cohort = "EVAL_V3" if i % 2 else "EVAL_V4"
values = {
("REPROJ_NN", 0): (0.32, 0.22, 0.52),
("REPROJ_HARMONIC", 0): (0.30, 0.20, 0.50),
("AHFILL8_T24000", 1): (0.30 - hybrid_gain - jitter, 0.20, 0.45),
("T6000", 1): (0.31, 0.21, 0.45),
("T24000", 1): (0.31, 0.21, 0.45 - length_gain - jitter),
}
for (method, seed), (all_, near, far) in values.items():
for role in ("A", "B"):
for qid in (14, 15):
out.append(
{
"cohort": cohort,
"scene_id": f"ai_{i:03d}_001",
"role": role,
"query_id": qid,
"method": method,
"seed": seed,
"absrel_ALL": all_,
"absrel_NEAR": near,
"absrel_FAR": far,
"far_fraction": 0.2,
}
)
return out
def test_branches_follow_p1_gate_and_p2_label():
evaluator = script("evaluate_rgbd_long_data_eval.py")
branch = lambda a, b: evaluator.analyze_rows(rows(a, b))["INTERPRETATION_BRANCH"] # noqa: E731
assert branch(0.03, 0.03) == "A"
assert branch(0.03, 0.0) == "B"
assert branch(0.0, 0.03) == "B"
assert branch(0.0, 0.0) == "C"
def test_prepare_needs_the_exact_directory_and_a_protocol(tmp_path):
prepare = script("prepare_rgbd_long_data_experiment.py")
with pytest.raises(PermissionError):
prepare.prepare(tmp_path / "EXP-OTHER")
root = tmp_path / prepare.EXPERIMENT
root.mkdir()
with pytest.raises(PermissionError):
prepare.prepare(root)
assert not (root / "preregistration.json").exists()
assert (
prepare.CHECKPOINT_STEPS[:12] == prepare.V11_CHECKPOINTS
and prepare.CHECKPOINT_STEPS[-1] == 24000
)