tests/test_rgbd_stream_write_v2.py
"""Core B V2: the V1 stream mechanics on the frozen 32^3 V9 carrier, and the V2 primary gates."""
import importlib.util
import json
from pathlib import Path
import pytest
import torch
from test_mechanism_training import observation
from test_rgbd_stream_write import stream_fixture
from mcss.dynamic.types import hash_scene_state, hash_value
from mcss.geometry import transform_cameras
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_stream_write as v1
from mcss.mechanism_pilot.rgbd_stream_evaluation_v2 import evaluate_streams
from mcss.mechanism_pilot.rgbd_stream_write_v2 import (
CONTROLS,
POLICIES,
StreamResolutionCarrier,
load_frozen_carrier,
load_write_rule,
make_write_rule,
save_write_rule,
stream_frame_ids,
training_loss,
unroll,
)
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 carrier(seed=5):
model = StreamResolutionCarrier(v8.carrier_config("C1"))
model.load_state_dict(v8.make_carrier("C1", seed, "cpu").state_dict())
with torch.no_grad():
model.depth_weight.fill_(0.05)
for parameter in model.parameters():
parameter.requires_grad_(False)
return model.eval()
def contexts(n_warm=3, n_stream=4):
torch.set_num_threads(1)
observations = [observation(i) for i in range(n_warm + n_stream)]
depths = torch.full((n_warm + n_stream, 8, 8), 2.0)
warm = v5.RGBDContext(observations[:n_warm], depths[:n_warm])
stream = v5.RGBDContext(observations[n_warm:], depths[n_warm:])
bounds = torch.tensor([[-2.0, -2.0, 0.1], [2.0, 2.0, 4.0]])
query = observation(9).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 warm, stream, bounds, cameras
def test_the_stream_rule_and_policies_are_core_b_v1s():
assert stream_frame_ids is v1.stream_frame_ids
assert POLICIES == v1.POLICIES and CONTROLS == v1.CONTROLS
record = {
"roles": {"context_a": [0, 1, 2], "context_b": [0, 3, 4], "primary_query": [8, 9]},
"frames": [{"frame_id": i} for i in (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)],
}
assert stream_frame_ids(record, "A") == (5, 6, 7, 10)
def test_no_stream_reproduces_the_static_v9_state():
warm, stream, bounds, _ = contexts()
model = carrier()
reference = v8.make_carrier("C1", 5, "cpu")
reference.load_state_dict(model.state_dict())
expected, _ = v8.build_state(reference, warm, bounds, "static")
rule = make_write_rule(model, 1)
state, _, fast, _ = unroll(model, rule, warm, stream, bounds, "NO_STREAM", "static")
assert tuple(state.spatial_shape) == (32, 32, 32)
assert hash_scene_state(state) == hash_scene_state(expected)
assert fast.step == 0 and float(fast.delta_fuse.abs().sum()) == 0.0
model.count_clamp = 3 # three views never exceed the clamp
clamped, _ = v8.build_state(model, warm, bounds, "static")
model.count_clamp = None
assert hash_scene_state(clamped) == hash_scene_state(expected)
def test_policies_write_only_the_matrices_they_name():
warm, stream, bounds, _ = contexts()
model, rule = carrier(), make_write_rule(carrier(), 2)
results = {p: unroll(model, rule, warm, stream, bounds, p, "e") for p in POLICIES}
off_fast = results["OFF"][2]
assert off_fast.step == 0 and float(off_fast.delta_complete.abs().sum()) == 0.0
assert results["ALL"][2].step == len(stream)
assert float(results["FUSE"][2].delta_complete.detach().abs().sum()) == 0.0
assert float(results["FUSE"][2].delta_fuse.detach().abs().sum()) > 0.0
assert float(results["COMPLETE"][2].delta_fuse.detach().abs().sum()) == 0.0
hashes = {p: hash_scene_state(r[0]) for p, r in results.items()}
assert hashes["OFF"] != hashes["NO_STREAM"] and hashes["ALL"] != hashes["OFF"]
assert hashes["OFF_CLAMP3"] != hashes["OFF"]
with pytest.raises(TypeError):
unroll(v1.StreamRGBDCarrier(v5.carrier_config()), rule, warm, stream, bounds, "OFF", "x")
def test_write_rule_learns_through_the_frozen_carrier(tmp_path):
warm, stream, bounds, cameras = contexts()
model, rule = carrier(), make_write_rule(carrier(), 3)
before = hash_value(model.state_dict())
state, _, _, _ = unroll(model, rule, warm, stream, bounds, "ALL", "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),
)
loss.backward()
assert any(p.grad is not None and p.grad.abs().sum() > 0 for p in rule.parameters())
assert all(p.grad is None for p in model.parameters())
assert hash_value(model.state_dict()) == before
path = tmp_path / "rule.pt"
save_write_rule(path, rule, seed=3, step=1, lock_sha256="x", carrier_sha256="y")
restored, payload = load_write_rule(path, "cpu")
assert hash_value(restored.state_dict()) == hash_value(rule.state_dict())
assert payload["schema"] == "mcss.rgbd_stream_write_rule.v2"
with pytest.raises(PermissionError):
v1.load_write_rule(path, "cpu")
def test_only_32_grid_v8_checkpoints_load_as_frozen_carriers(tmp_path):
for variant, ok in (("C1", True), ("C0", False)):
path = tmp_path / f"{variant}.pt"
v8.save_checkpoint(
path,
v8.make_carrier(variant, 7, "cpu"),
variant=variant,
seed=7,
step=1,
lock_sha256="x",
)
if ok:
model, _ = load_frozen_carrier(path, "cpu")
assert isinstance(model, StreamResolutionCarrier) and not model.training
assert not any(p.requires_grad for p in model.parameters())
else:
with pytest.raises(PermissionError):
load_frozen_carrier(path, "cpu")
def test_dev_evaluation_seals_every_policy_before_query_depth(tmp_path):
manifest = stream_fixture(tmp_path)
model = carrier()
rules = {"trained": make_write_rule(model, 4), "untrained": make_write_rule(model, 4)}
out = tmp_path / "evaluation"
rows = evaluate_streams(model, rules, manifest, tmp_path, out, 4, "cpu")
assert len(rows) == 2 * 2 * 2 * len((*POLICIES, *CONTROLS))
access = json.loads((out / "GT_access.json").read_text())
marker = next(i for i, e in enumerate(access) if e.get("event") == "ALL_STREAM_STATES_SEALED")
early = [e for e in access[:marker] if "depth" in e.get("channels", [])]
assert {e["purpose"] for e in early} == {"CONTEXT_ONLY_RGBD", "STREAM_RGBD"}
assert all(e["frame_id"] not in (8, 9) for e in early)
def rows_for(values):
rows = []
for seed in (1, 2):
for i in range(8):
for role in ("A", "B"):
for qid in (8, 9):
for policy, value in values.items():
rows.append(
{
"seed": seed,
"scene_id": f"s{i}",
"role": role,
"query_id": qid,
"policy": policy,
"depth_absrel": value + 0.002 * ((i + qid + seed) % 3),
"depth_delta1": 0.5,
}
)
return rows
def test_v2_primary_gates_set_the_branch():
analyze = script("analyze_rgbd_stream_write_v2.py").analyze
base = {
"NO_STREAM": 0.30,
"OFF": 0.34,
"OFF_CLAMP3": 0.31,
"FUSE": 0.30,
"COMPLETE": 0.32,
"ALL": 0.25,
"ALL_UNTRAINED": 0.34,
"ALL_WRONG_SCENE": 0.40,
}
result = analyze(rows_for(base), [1, 2], True, "OFF")
assert result["contrasts"]["BEYOND_BASELINE"]["mean"] == pytest.approx(0.09)
assert result["statuses"]["BEYOND_BASELINE_STATUS"] == "SUPPORTED"
assert result["statuses"]["SPECIFICITY_STATUS"] == "SUPPORTED"
assert result["INTERPRETATION_BRANCH"] == "A" and result["baseline_policy"] == "OFF"
# beats the baseline, but another scene's fast weights do just as well
result = analyze(rows_for({**base, "ALL_WRONG_SCENE": 0.25}), [1, 2], True, "OFF")
assert result["statuses"]["SPECIFICITY_STATUS"] != "SUPPORTED"
assert result["INTERPRETATION_BRANCH"] == "B"
# the clamp baseline beats the writes (the Core B V1 pattern)
result = analyze(rows_for({**base, "ALL": 0.33}), [1, 2], True, "OFF_CLAMP3")
assert result["contrasts"]["BEYOND_BASELINE"]["mean"] < 0
assert result["INTERPRETATION_BRANCH"] == "C"
with pytest.raises(ValueError):
analyze(rows_for(base), [1, 2], True, "NO_STREAM")
def test_runner_trains_only_v2_scripts_on_cpu(tmp_path, monkeypatch):
module = script("run_rgbd_stream_write_v2.py")
seen = []
monkeypatch.setattr(module, "execute", lambda root, label, args, env, times: seen.append(args))
module.train_all(tmp_path, {"seeds": [1, 2], "parallel_workers": 2}, None, {})
assert sorted(a[a.index("--seed") + 1] for a in seen) == ["1", "2"]
assert all(a[0] == "scripts/train_rgbd_stream_write_v2.py" for a in seen)
assert all(a[a.index("--device") + 1] == "cpu" for a in seen)