tests/test_rgbd_stream_ttt_v11.py
"""Core B V11: online writes per arriving frame, one-shot rows, the sealed run and the branch."""
import copy
import importlib.util
import json
from pathlib import Path
import pytest
import torch
from test_rgbd_stream_write import stream_fixture
from test_rgbd_stream_write_v3 import carrier, contexts
from mcss.dynamic.types import hash_scene_state, hash_value
from mcss.mechanism_pilot import rgbd_stream_ttt_v4 as v4
from mcss.mechanism_pilot import rgbd_stream_ttt_v7 as v7
from mcss.mechanism_pilot import rgbd_stream_ttt_v11 as v11
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 test_one_shot_rows_are_core_b_v7s():
for name in ("NO_STREAM", "OFF", "TTT_100"):
assert v11.POLICY_SPECS[name] == v7.POLICY_SPECS[name]
assert v11.POLICY_SPECS["ONLINE_100"] == (True, "online", 100)
assert v11.POLICY_SPECS["ONLINE_NEW_100"] == (True, "online_new", 100)
assert v11.POLICY_SPECS["ALL_TTT_100"] == (True, "all", 100)
assert v11.CONTROLS == {"ONLINE_100_WRONG_SCENE": "ONLINE_100"}
def test_one_shot_policies_unroll_exactly_as_core_b_v7():
warm, stream, bounds, _ = contexts()
model = carrier()
specs = {"NO_STREAM": (False, None, 0), "OFF": (True, None, 0), "TTT_100": (True, "stream", 3)}
for policy in specs:
mine = v11.unroll_v11(model, warm, stream, bounds, policy, "e", 7, specs)
theirs = v7.unroll_v7(model, warm, stream, bounds, policy, "e", 7, specs)
assert hash_scene_state(mine[0]) == hash_scene_state(theirs[0])
assert mine[4] == pytest.approx(theirs[4])
def test_online_write_on_a_single_frame_is_the_one_shot_write():
warm, stream, bounds, _ = contexts(n_stream=1)
model = carrier()
batch = v4.cached_episode(model, warm, stream, bounds, "e", True)
fast_b, losses_b = v4.ttt_fast(model, batch, stream, 3, 1e-3, v4.episode_generator(7, "g"))
online = v4.cached_episode(model, warm, stream, bounds, "e", False)
fast_o, losses_o = v11.ttt_online(
model, online, stream, 3, 1e-3, v4.episode_generator(7, "g"), True
)
assert losses_o == losses_b and fast_o.step == fast_b.step == 3
assert torch.equal(fast_o.delta_fuse, fast_b.delta_fuse)
assert torch.equal(fast_o.delta_complete, fast_b.delta_complete)
with torch.no_grad():
assert hash_scene_state(online.run(fast_o, "materialize")) == hash_scene_state(
batch.run(fast_b, "materialize")
)
def test_online_policies_cache_every_frame_and_differ_only_in_replay():
warm, stream, bounds, _ = contexts()
model = carrier()
before = hash_value(model.state_dict())
specs = {name: (s, sup, 8) for name, (s, sup, _) in v11.POLICY_SPECS.items()}
replay = v11.unroll_v11(model, warm, stream, bounds, "ONLINE_100", "e", 7, specs)
newest = v11.unroll_v11(model, warm, stream, bounds, "ONLINE_NEW_100", "e", 7, specs)
both = v11.unroll_v11(model, warm, stream, bounds, "ALL_TTT_100", "e", 7, specs)
ids = [o.frame_id for o in warm] + [o.frame_id for o in stream]
for result in (replay, newest, both):
assert result[3].source_frames == ids and result[2].step == 8 and len(result[4]) == 8
assert hash_scene_state(replay[0]) != hash_scene_state(newest[0])
assert hash_value(model.state_dict()) == before
uneven = {"ONLINE_100": (True, "online", 6)}
with pytest.raises(ValueError):
v11.unroll_v11(model, warm, stream, bounds, "ONLINE_100", "e", 7, uneven)
again = v11.unroll_v11(model, warm, stream, bounds, "ONLINE_100", "e", 7, specs)
assert again[4] == pytest.approx(replay[4])
def test_two_cohorts_are_sealed_before_query_depth(tmp_path):
torch.set_num_threads(1)
manifest = stream_fixture(tmp_path)
dev = [r for r in manifest["scenes"] if r["split"] == "DEV"]
cohorts = {}
for name, record in zip(("EVAL_V3", "EVAL_V4"), dev, strict=True):
cohort = copy.deepcopy(manifest)
cohort["scenes"] = [{**copy.deepcopy(record), "split": name}]
cohorts[name] = (cohort, tmp_path)
expected = {name: name for name in cohorts}
specs = {name: (s, sup, min(steps, 4)) for name, (s, sup, steps) in v11.POLICY_SPECS.items()}
out = tmp_path / "evaluation"
rows = v11.evaluate_v11(carrier(), cohorts, expected, out, 7, "cpu", specs)
assert len(rows) == 2 * 2 * 2 * (len(specs) + len(v11.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)
meta = json.loads((out / "state_hashes.json").read_text())
donors = {m["scene_id"]: m["donor_scene_id"] for m in meta.values() if "donor_scene_id" in m}
assert donors == {"capacity0": "capacity1", "capacity1": "capacity0"}
for m in meta.values():
if m["policy"] == "ALL_TTT_100":
assert m["supervised_frame_ids"] == m["warmup_frame_ids"] + m["stream_frame_ids"]
if m["policy"] in ("ONLINE_100", "ONLINE_NEW_100"):
assert m["supervised_frame_ids"] == m["stream_frame_ids"] and m["fast_step"] == 4
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,
"cohort": "EVAL_V3",
"scene_id": f"s{i}",
"role": role,
"query_id": qid,
"policy": policy,
"depth_absrel": value + 0.002 * ((i + qid + seed) % 3),
}
)
return rows
def test_branches_follow_the_online_gain_and_the_midpoint():
analyze = script("run_rgbd_stream_ttt_online_v11.py").analyze
base = {
"NO_STREAM": 0.30,
"OFF": 0.24,
"TTT_100": 0.22,
"ONLINE_100": 0.222,
"ONLINE_NEW_100": 0.23,
"ALL_TTT_100": 0.215,
"ONLINE_100_WRONG_SCENE": 0.25,
}
result = analyze(rows_for(base), [1, 2])
assert result["INTERPRETATION_BRANCH"] == "A"
assert result["statuses"]["P3_REPLAY_GAIN_STATUS"] == "SUPPORTED"
assert result["contrasts"]["ONLINE_RETAINS_HALF"]["mean"] == pytest.approx(0.008)
assert result["contrasts"]["ONLINE_VS_ONE_SHOT"]["mean"] == pytest.approx(0.002)
weak = {**base, "ONLINE_100": 0.235}
assert analyze(rows_for(weak), [1, 2])["INTERPRETATION_BRANCH"] == "B"
none = {**base, "ONLINE_100": 0.24, "ONLINE_NEW_100": 0.24}
result = analyze(rows_for(none), [1, 2])
assert result["INTERPRETATION_BRANCH"] == "C"
assert result["statuses"]["P3_REPLAY_GAIN_STATUS"] != "SUPPORTED"