tests/test_rgbd_stream_write.py
"""Core B V1: stream rule, frozen-carrier identity, write mechanics and the sealed DEV boundary."""
import copy
import importlib.util
import json
import shutil
from pathlib import Path
import numpy as np
import pytest
import torch
from test_geometry_carrier_evaluation import fixture as evaluation_fixture
from test_mechanism_training import observation
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.rgbd_stream_evaluation import evaluate_streams
from mcss.mechanism_pilot.rgbd_stream_write import (
CONTROLS,
POLICIES,
StreamRGBDCarrier,
StreamRGBDLoader,
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 = StreamRGBDCarrier(v5.carrier_config())
model.load_state_dict(v5.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 stream_fixture(root):
manifest, _ = evaluation_fixture(root)
for record in manifest["scenes"]:
base = next(f for f in record["frames"] if f["frame_id"] == 4)
for fid in (5, 6, 7, 10):
frame = copy.deepcopy(base)
frame["frame_id"] = fid
for key, suffix in (("rgb", ".png"), ("depth", ".npy")):
source = Path(base[key])
if source.exists():
target = source.with_name(f"{record['scene_id']}_{fid}{suffix}")
shutil.copyfile(source, target)
frame[key] = str(target)
pose = np.array(frame["c2w"])
pose[0, 3] += 0.01 * fid
frame["c2w"] = pose.tolist()
record["frames"].append(frame)
record["frames"].sort(key=lambda f: f["frame_id"])
return manifest
def test_stream_rule_uses_frame_ids_only_and_never_shortens():
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)
assert stream_frame_ids(record, "B") == (5, 6, 7, 10)
record["frames"] = record["frames"][:-1]
assert stream_frame_ids(record, "A") is None
# free frames may precede warmup frames; the stream spreads evenly over all of them
record = {
"roles": {"context_a": [0, 6, 7], "context_b": [0, 12, 13], "primary_query": [14, 15]},
"frames": [{"frame_id": i} for i in range(16)],
}
assert stream_frame_ids(record, "A") == (1, 4, 8, 11)
def test_frozen_carrier_and_no_stream_reproduce_the_static_v5_state():
warm, stream, bounds, _ = contexts()
model = carrier()
reference = v5.make_carrier("C1", 5, "cpu")
reference.load_state_dict(model.state_dict())
expected, _ = v5.build_state(reference, warm, bounds, "static")
rule = make_write_rule(model, 1)
state, _, fast, _ = unroll(model, rule, warm, stream, bounds, "NO_STREAM", "static")
assert hash_scene_state(state) == hash_scene_state(expected)
assert fast.step == 0 and float(fast.delta_fuse.abs().sum()) == 0.0
# three views never exceed the trained count, so the clamp is inert without a stream
model.count_clamp = 3
clamped, _ = v5.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"] # seven views exceed the trained count
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()
grads = [p.grad for p in rule.parameters()]
assert any(g is not None and g.abs().sum() > 0 for g in grads)
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["slow_carrier_updated"] is False
with pytest.raises(FileExistsError):
save_write_rule(path, rule, seed=3, step=1, lock_sha256="x", carrier_sha256="y")
def test_stream_loader_reads_stream_frames_only(tmp_path):
manifest = stream_fixture(tmp_path)
record = next(r for r in manifest["scenes"] if r["split"] == "DEV")
access = []
loader = StreamRGBDLoader(
record,
manifest["image_size"],
tmp_path,
"cpu",
access,
allowed_scene_ids=[record["scene_id"]],
holdout_scene_ids=["train_unused"],
)
stream = loader.stream("A")
assert tuple(o.frame_id for o in stream) == (5, 6, 7, 10)
assert {e["frame_id"] for e in access if "depth" in e.get("channels", [])} == {5, 6, 7, 10}
assert {e["purpose"] for e in access} == {"STREAM_RGBD"}
with pytest.raises(PermissionError):
StreamRGBDLoader(
record,
manifest["image_size"],
tmp_path,
"cpu",
[],
allowed_scene_ids=["other"],
holdout_scene_ids=[],
)
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")
names = (*POLICIES, *CONTROLS)
assert len(rows) == 2 * 2 * 2 * len(names)
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)
assert all(e.get("scene_id") != "train_unused" for e in access)
metadata = json.loads((out / "state_hashes.json").read_text())
donors = {m["donor_scene_id"] for k, m in metadata.items() if k.endswith("ALL_WRONG_SCENE")}
assert donors == {"capacity0", "capacity1"}
def test_analysis_gates_and_branches():
analyze = script("analyze_rgbd_stream_write.py").analyze
scenes = [f"s{i}" for i in range(8)]
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,
}
rows = []
for seed in (1, 2):
for i, sid in enumerate(scenes):
for role in ("A", "B"):
for qid in (8, 9):
for policy, value in base.items():
noise = 0.002 * ((i + qid + seed) % 3)
rows.append(
{
"seed": seed,
"scene_id": sid,
"role": role,
"query_id": qid,
"policy": policy,
"depth_absrel": value + noise,
"depth_delta1": 0.5,
}
)
result = analyze(rows, [1, 2], True)
assert result["INTERPRETATION_BRANCH"] == "A"
assert result["contrasts"]["WRITE_GAIN"]["mean"] == pytest.approx(0.09)
assert result["statuses"]["WRITE_SPECIFICITY_STATUS"] == "SUPPORTED"
for row in rows:
if row["policy"] == "ALL":
row["depth_absrel"] += 0.08 # still better than OFF, worse than NO_STREAM
result = analyze(rows, [1, 2], True)
assert result["statuses"]["WRITE_STATUS"] == "SUPPORTED"
assert result["statuses"]["STREAM_STATUS"] == "HARMFUL"
assert result["INTERPRETATION_BRANCH"] == "B"
assert analyze(rows, [1, 2], False)["INTERPRETATION_BRANCH"] == "C"
def test_runner_trains_only_stream_write_scripts_on_cpu(tmp_path, monkeypatch):
module = script("run_rgbd_stream_write.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.py" for a in seen)
assert all(a[a.index("--device") + 1] == "cpu" for a in seen)