streampetr-p150 / code /tt_streampetr /tests /test_reference_host.py

tt-model push streampetr-p150 (container)

ac1609a verified 25 days ago

13.1 kB

# SPDX-License-Identifier: Apache-2.0
"""The CPU reference (``tt_streampetr.reference``) without ONNX Runtime: the weights loader and graph checks, the
reference against the research ORT goldens (teacher-forcing records of the public sequences), and the TT port's
exact rewrites (``reference.port_form``) against the deployed form. Needs torch + onnx (tt-env with the dev overlay,
or the research venv) and the weights; skipped otherwise.
    TT_VISIBLE_DEVICES=none OMP_NUM_THREADS=3 python -m pytest -q code/tt_streampetr/tests/test_reference_host.py
Gates (PLAN.md 0.3 item 1): PCC >= 0.9999 before the data-dependent top-256 selection; after it the comparison is
teacher-forced (the golden's selection, recovered from its ``post_memory_embedding``) or set-based.
"""
from __future__ import annotations

import os
import re

import numpy as np
import pytest

torch = pytest.importorskip("torch")
pytest.importorskip("onnx")

from tt_streampetr.host.agreement import detection_f1, recover_topk_indices, topk_overlap  # noqa: E402
from tt_streampetr.host import postprocess as hp  # noqa: E402
from tt_streampetr.reference import port_form as pf  # noqa: E402
from tt_streampetr.reference.config import check_hf_params, default_config, load_hf_params  # noqa: E402
from tt_streampetr.reference.pipeline import to_numpy  # noqa: E402
from tt_streampetr.tests import paths  # noqa: E402
from tt_streampetr.ttaw.metrics import pcc  # noqa: E402

REF_PCC = 0.9999                       # PLAN.md 0.3 item 1
CFG = default_config()
THREADS = int(os.environ.get("STREAMPETR_TEST_THREADS", "3"))   # workspace rule: <= 4 CPU threads
PRE_SELECTION = ("all_cls_scores", "all_bbox_preds", "outs_dec", "reference_points")
POST_MEMORY = ("post_memory_embedding", "post_memory_reference_point", "post_memory_egopose", "post_memory_velo")


@pytest.fixture(scope="module")
def net():
    w = paths.weights_dir()
    if w is None:
        pytest.skip("camera_streampetr v1.0 weights not found (set STREAMPETR_WEIGHTS_DIR)")
    torch.set_num_threads(THREADS)
    from tt_streampetr.reference.model import StreamPETRNet

    return StreamPETRNet(w, CFG)


def _tf_records():
    out = []
    for run, frames in (("ps019", (0, 1, 40)), ("ns0103", (111,))):
        g = paths.public_golden(run)
        for f in frames:
            if g is not None and (g / f"tf_{f:04d}.npz").is_file():
                out.append((run, f))
    return out


def _head(net, rec, pe, **kw):
    pre = {k: rec[k] for k in rec.files if k.startswith("pre_")}
    with torch.no_grad():
        return to_numpy(net.head(rec["img_feats"], pe["pos_embed"], pe["cone"], rec["data_ego_pose"],
                                 rec["data_ego_pose_inv"], pre, **kw))


# ---------------------------------------------------------------------------------------------- weights / config

def test_weights_sha_and_graph(net):
    from tt_streampetr.reference.config import ONNX_FILES

    assert net.weights.sha256() == {name: sha for name, sha in ONNX_FILES.values()}
    net.weights.check_graph(CFG)
    assert net.weights.attention_scale() == np.float32(1 / np.sqrt(32))
    q = net.weights.learned_queries()
    assert q["reference_points"].shape == (1, 644, 3) and q["tgt"].shape == (1, 644, 256)


def test_config_equals_the_deployed_parameter_files():
    w = paths.weights_dir()
    if w is None:
        pytest.skip("weights not found")
    check_hf_params(load_hf_params(w), CFG)
    pkg = paths.autoware_package_dir()
    if pkg is None:
        pytest.skip("upstream autoware_universe not present")
    text = (pkg / "config" / "ml_package_camera_streampetr.param.yaml").read_text()

    def floats(key):
        return [float(v) for v in re.search(key + r":\s*\[([^\]]*)\]", text).group(1).split(",")]

    def scalar(key):
        return float(re.search(key + r":\s*([-0-9.]+)", text).group(1))

    assert tuple(floats("confidence_threshold")[:5]) == CFG.score_thresholds
    assert tuple(floats("yaw_norm_thresholds")[:5]) == CFG.yaw_norm_thresholds
    assert scalar("circle_nms_dist_threshold") == CFG.circle_nms_dist_threshold
    assert scalar("iou_nms_search_distance_2d") == CFG.iou_nms_search_distance_2d
    assert scalar("iou_nms_threshold") == CFG.iou_nms_threshold
    node = (pkg / "config" / "camera_streampetr.param.yaml").read_text()
    assert float(re.search(r"max_camera_time_diff:\s*([0-9.]+)", node).group(1)) == CFG.max_camera_time_diff
    assert int(re.search(r"anchor_camera_id:\s*([0-9]+)", node).group(1)) == CFG.anchor_camera_id


# ---------------------------------------------------------------------------------------------- vs ORT goldens

def test_position_embedding_vs_research_golden(net):
    g = paths.public_golden("ps019")
    if g is None:
        pytest.skip("research/streampetr/public_data not present")
    pe = np.load(g / "pe_inputs_outputs.npz")
    with torch.no_grad():
        pos, cone = net.pe(pe["img_metas_pad"], pe["intrinsics"], pe["img2lidar"])
    assert pcc(pos.numpy(), pe["pos_embed"]) >= REF_PCC and np.abs(pos.numpy() - pe["pos_embed"]).max() < 1e-3
    assert np.abs(cone.numpy() - pe["cone"]).max() < 1e-5


@pytest.mark.parametrize("run,frame", _tf_records(), ids=[f"{r}-{f}" for r, f in _tf_records()])
def test_head_vs_research_tf_record(net, run, frame):
    g = paths.public_golden(run)
    rec, pe = np.load(g / f"tf_{frame:04d}.npz"), np.load(g / "pe_inputs_outputs.npz")
    out = _head(net, rec, pe)
    for k in PRE_SELECTION:                                         # before the selection: PCC
        p = pcc(out[k], rec[k])
        print(f"  {run}/{frame} {k:28s} pcc={p:.10f} max_abs={np.abs(out[k] - rec[k]).max():.2e}")
        assert p >= REF_PCC, k
    gold_idx = recover_topk_indices(rec["post_memory_embedding"], rec["outs_dec"])
    ov = topk_overlap(out["topk_idx"].reshape(-1), gold_idx)
    print(f"  top-256 overlap with the golden selection: {ov:.4f}")
    assert ov >= 0.99                                               # near-ties only (S:6)
    tf = _head(net, rec, pe, topk_idx=gold_idx)                     # teacher-forced selection
    for k in POST_MEMORY:
        assert pcc(tf[k], rec[k]) >= REF_PCC, k
    assert np.array_equal(tf["post_memory_timestamp"], rec["post_memory_timestamp"])


def test_backbone_vs_research_golden(net):
    g = paths.public_golden("ps019")
    if g is None or "img" not in np.load(g / "tf_0000.npz").files:
        pytest.skip("research/streampetr/public_data not present")
    rec = np.load(g / "tf_0000.npz")
    from tt_streampetr.ttaw.golden import TapRegistry

    taps = TapRegistry(include=("backbone.stage*", "neck.*", "img_feats"), exclude=("backbone.stage*.osa*",))
    with torch.no_grad():
        feats = net.backbone(rec["img"], taps).numpy()
    p = pcc(feats, rec["img_feats"])
    print(f"  img_feats pcc={p:.12f} max_abs={np.abs(feats - rec['img_feats']).max():.2e}")
    assert p >= REF_PCC
    assert set(taps.names()) == {"backbone.stage2", "backbone.stage3", "backbone.stage4", "backbone.stage5",
                                 "neck.lateral0", "neck.lateral1", "neck.merged", "img_feats"}
    assert taps["backbone.stage4"].shape == (5, 768, 30, 40) and taps["backbone.stage5"].shape == (5, 1024, 15, 20)


# ---------------------------------------------------------------------------------------------- port form

def _gold_pe():
    g = paths.public_golden("ps019")
    if g is None or not all((g / f"tf_{f:04d}.npz").is_file() for f in (0, 1, 2)):
        pytest.skip("research/streampetr/public_data not present")
    return g, np.load(g / "pe_inputs_outputs.npz")


def test_relative_memory_chain_equals_deployed(net):
    """Frames 0-2 of PandaSet 019 (consecutive teacher-forcing records' img_feats): the deployed global-frame memory
    loop vs the port's previous-ego-frame state with M_k = E_k^-1 E_{k-1}."""
    from tt_streampetr.host.temporal import AutowareMemory, relative_motion

    g, pe = _gold_pe()
    mem = AutowareMemory(CFG)
    rel = pf.RelativeMemory.zeros()
    E_prev = None
    for f in (0, 1, 2):
        rec = np.load(g / f"tf_{f:04d}.npz")
        ts = np.float32(rec["ts"])
        pre = mem.pre_inputs(ts)
        with torch.no_grad():
            dep = to_numpy(net.head(rec["img_feats"], pe["pos_embed"], pe["cone"], rec["data_ego_pose"],
                                    rec["data_ego_pose_inv"], pre))
            M = np.eye(4) if E_prev is None else relative_motion(E_prev, rec["data_ego_pose_inv"])
            out, rel = pf.head_relative(net.head, rec["img_feats"], pe["pos_embed"], pe["cone"], M, rel,
                                        pre["pre_memory_timestamp"])
        out = to_numpy(out)
        for k in ("all_cls_scores", "all_bbox_preds", "outs_dec"):
            p = pcc(out[k], dep[k])
            print(f"  frame {f} {k:16s} pcc={p:.10f} max_abs={np.abs(out[k] - dep[k]).max():.2e}")
            assert p >= 0.99999, (f, k)
        assert topk_overlap(out["topk_idx"].reshape(-1), dep["topk_idx"].reshape(-1)) >= 0.99
        d = detection_f1(hp.autoware_postprocess(out["all_cls_scores"], out["all_bbox_preds"], CFG)["objects"],
                         hp.autoware_postprocess(dep["all_cls_scores"], dep["all_bbox_preds"], CFG)["objects"])
        assert d["f1"] == 1.0 and d["centre_err_max"] < 0.01
        mem.post_update(dep, ts)
        E_prev = rec["data_ego_pose"]


def test_teacher_forced_relative_state(net):
    """A golden ``pre_memory_*`` set converted to the port state (``global_to_relative``, M = I) gives the deployed
    outputs (the device's teacher-forcing input)."""
    g, pe = _gold_pe()
    rec = np.load(g / "tf_0040.npz") if (g / "tf_0040.npz").is_file() else np.load(g / "tf_0002.npz")
    pre = {k: rec[k] for k in rec.files if k.startswith("pre_")}
    st, M = pf.global_to_relative(pre, rec["data_ego_pose_inv"])
    with torch.no_grad():
        out, _ = pf.head_relative(net.head, rec["img_feats"], pe["pos_embed"], pe["cone"], M, st,
                                  rec["pre_memory_timestamp"])
    out = to_numpy(out)
    for k in ("all_cls_scores", "all_bbox_preds", "outs_dec"):
        assert pcc(out[k], rec[k]) >= REF_PCC, k


def test_hoisted_kv_and_calibration_cache_are_exact_rewrites(net):
    g, pe = _gold_pe()
    rec = np.load(g / "tf_0001.npz")
    head = net.head
    with torch.no_grad():
        mem_ego, mem_ref = head.memory_pre_update({k: rec[k] for k in rec.files if k.startswith("pre_")},
                                                  rec["data_ego_pose_inv"])
        memory, pos = head.tokens(rec["img_feats"], pe["pos_embed"], pe["cone"])
        coeffs = head.spatial_alignment_coeffs(pe["cone"])
        memory2, pos2 = head.tokens(rec["img_feats"], pe["pos_embed"], pe["cone"], sa_coeffs=coeffs)
        assert torch.equal(memory, memory2) and torch.equal(pos, pos2)            # cached gamma / beta: bit-exact
        t = head.temporal(mem_ref, mem_ego, rec["pre_memory_timestamp"], rec["pre_memory_embedding"],
                          rec["pre_memory_velo"])
        q = head.queries(t)
        a = head.decoder(q["tgt"], q["query_pos"], q["temp_memory"], q["temp_pos"], memory, pos)
        b = pf.decoder_hoisted(head, q["tgt"], q["query_pos"], q["temp_memory"], q["temp_pos"], memory, pos)
        p = pcc(a.numpy(), b.numpy())
        print(f"  hoisted K/V decoder: pcc={p:.12f} max_abs={(a - b).abs().max().item():.2e}")
        assert p >= 0.9999999 and (a - b).abs().max().item() < 1e-3
        w, bias = pf.merged_branch_first_layer(head)
        o = a[-1]
        merged = o @ w + bias
        assert torch.allclose(merged[..., :256], head.lin(o, head.cls[0]), atol=1e-4)
        assert torch.allclose(merged[..., 256:], head.lin(o, head.reg[0]), atol=1e-4)


def test_stem_normalisation_fold_needs_mean_padding(net):
    """``conv(W, (v - m) / s)`` == ``conv(W / s, v padded with m)`` + folded bias (float rounding only); a zero pad
    instead is wrong on the border (why the functional port keeps the normalisation on the host)."""
    import torch.nn.functional as F

    from tt_streampetr.ttaw.image_triangle import STREAMPETR_PRESETS

    w, b, s, p = net.backbone.params["/img_backbone/stem/stem_1/conv/Conv"]
    pr = STREAMPETR_PRESETS["autoware_main"]
    rng = np.random.default_rng(0)
    v = torch.from_numpy(rng.uniform(0, 255, size=(1, 3, 32, 48)).astype(np.float32))
    m = torch.from_numpy(pr.mean_f32()).view(1, 3, 1, 1)
    sd = torch.from_numpy(pr.std_f32()).view(1, 3, 1, 1)
    ref = F.conv2d((v - m) / sd, w, b, stride=s, padding=p)
    w2, b2 = pf.fold_normalisation_into_stem(w.numpy(), b.numpy(), pr.mean, pr.std)
    vp = torch.cat([torch.full((1, 1, 34, 50), float(c)) for c in pr.mean], 1)
    vp[:, :, 1:-1, 1:-1] = v
    fold = F.conv2d(vp, torch.from_numpy(w2), torch.from_numpy(b2), stride=s, padding=0)
    assert (fold - ref).abs().max().item() < 1e-3 * ref.abs().max().item()
    zero_pad = F.conv2d(v, torch.from_numpy(w2), torch.from_numpy(b2), stride=s, padding=p)
    assert (zero_pad - ref)[..., 0, :].abs().max().item() > 1.0              # the border breaks without it

	# SPDX-License-Identifier: Apache-2.0
	"""The CPU reference (``tt_streampetr.reference``) without ONNX Runtime: the weights loader and graph checks, the
	reference against the research ORT goldens (teacher-forcing records of the public sequences), and the TT port's
	exact rewrites (``reference.port_form``) against the deployed form. Needs torch + onnx (tt-env with the dev overlay,
	or the research venv) and the weights; skipped otherwise.

	TT_VISIBLE_DEVICES=none OMP_NUM_THREADS=3 python -m pytest -q code/tt_streampetr/tests/test_reference_host.py

	Gates (PLAN.md 0.3 item 1): PCC >= 0.9999 before the data-dependent top-256 selection; after it the comparison is
	teacher-forced (the golden's selection, recovered from its ``post_memory_embedding``) or set-based.
	"""
	from __future__ import annotations

	import os
	import re

	import numpy as np
	import pytest

	torch = pytest.importorskip("torch")
	pytest.importorskip("onnx")

	from tt_streampetr.host.agreement import detection_f1, recover_topk_indices, topk_overlap # noqa: E402
	from tt_streampetr.host import postprocess as hp # noqa: E402
	from tt_streampetr.reference import port_form as pf # noqa: E402
	from tt_streampetr.reference.config import check_hf_params, default_config, load_hf_params # noqa: E402
	from tt_streampetr.reference.pipeline import to_numpy # noqa: E402
	from tt_streampetr.tests import paths # noqa: E402
	from tt_streampetr.ttaw.metrics import pcc # noqa: E402

	REF_PCC = 0.9999 # PLAN.md 0.3 item 1
	CFG = default_config()
	THREADS = int(os.environ.get("STREAMPETR_TEST_THREADS", "3")) # workspace rule: <= 4 CPU threads
	PRE_SELECTION = ("all_cls_scores", "all_bbox_preds", "outs_dec", "reference_points")
	POST_MEMORY = ("post_memory_embedding", "post_memory_reference_point", "post_memory_egopose", "post_memory_velo")


	@pytest.fixture(scope="module")
	def net():
	w = paths.weights_dir()
	if w is None:
	pytest.skip("camera_streampetr v1.0 weights not found (set STREAMPETR_WEIGHTS_DIR)")
	torch.set_num_threads(THREADS)
	from tt_streampetr.reference.model import StreamPETRNet

	return StreamPETRNet(w, CFG)


	def _tf_records():
	out = []
	for run, frames in (("ps019", (0, 1, 40)), ("ns0103", (111,))):
	g = paths.public_golden(run)
	for f in frames:
	if g is not None and (g / f"tf_{f:04d}.npz").is_file():
	out.append((run, f))
	return out


	def _head(net, rec, pe, **kw):
	pre = {k: rec[k] for k in rec.files if k.startswith("pre_")}
	with torch.no_grad():
	return to_numpy(net.head(rec["img_feats"], pe["pos_embed"], pe["cone"], rec["data_ego_pose"],
	rec["data_ego_pose_inv"], pre, **kw))


	# ---------------------------------------------------------------------------------------------- weights / config

	def test_weights_sha_and_graph(net):
	from tt_streampetr.reference.config import ONNX_FILES

	assert net.weights.sha256() == {name: sha for name, sha in ONNX_FILES.values()}
	net.weights.check_graph(CFG)
	assert net.weights.attention_scale() == np.float32(1 / np.sqrt(32))
	q = net.weights.learned_queries()
	assert q["reference_points"].shape == (1, 644, 3) and q["tgt"].shape == (1, 644, 256)


	def test_config_equals_the_deployed_parameter_files():
	w = paths.weights_dir()
	if w is None:
	pytest.skip("weights not found")
	check_hf_params(load_hf_params(w), CFG)
	pkg = paths.autoware_package_dir()
	if pkg is None:
	pytest.skip("upstream autoware_universe not present")
	text = (pkg / "config" / "ml_package_camera_streampetr.param.yaml").read_text()

	def floats(key):
	return [float(v) for v in re.search(key + r":\s\[([^\]])\]", text).group(1).split(",")]

	def scalar(key):
	return float(re.search(key + r":\s*([-0-9.]+)", text).group(1))

	assert tuple(floats("confidence_threshold")[:5]) == CFG.score_thresholds
	assert tuple(floats("yaw_norm_thresholds")[:5]) == CFG.yaw_norm_thresholds
	assert scalar("circle_nms_dist_threshold") == CFG.circle_nms_dist_threshold
	assert scalar("iou_nms_search_distance_2d") == CFG.iou_nms_search_distance_2d
	assert scalar("iou_nms_threshold") == CFG.iou_nms_threshold
	node = (pkg / "config" / "camera_streampetr.param.yaml").read_text()
	assert float(re.search(r"max_camera_time_diff:\s*([0-9.]+)", node).group(1)) == CFG.max_camera_time_diff
	assert int(re.search(r"anchor_camera_id:\s*([0-9]+)", node).group(1)) == CFG.anchor_camera_id


	# ---------------------------------------------------------------------------------------------- vs ORT goldens

	def test_position_embedding_vs_research_golden(net):
	g = paths.public_golden("ps019")
	if g is None:
	pytest.skip("research/streampetr/public_data not present")
	pe = np.load(g / "pe_inputs_outputs.npz")
	with torch.no_grad():
	pos, cone = net.pe(pe["img_metas_pad"], pe["intrinsics"], pe["img2lidar"])
	assert pcc(pos.numpy(), pe["pos_embed"]) >= REF_PCC and np.abs(pos.numpy() - pe["pos_embed"]).max() < 1e-3
	assert np.abs(cone.numpy() - pe["cone"]).max() < 1e-5


	@pytest.mark.parametrize("run,frame", _tf_records(), ids=[f"{r}-{f}" for r, f in _tf_records()])
	def test_head_vs_research_tf_record(net, run, frame):
	g = paths.public_golden(run)
	rec, pe = np.load(g / f"tf_{frame:04d}.npz"), np.load(g / "pe_inputs_outputs.npz")
	out = _head(net, rec, pe)
	for k in PRE_SELECTION: # before the selection: PCC
	p = pcc(out[k], rec[k])
	print(f" {run}/{frame} {k:28s} pcc={p:.10f} max_abs={np.abs(out[k] - rec[k]).max():.2e}")
	assert p >= REF_PCC, k
	gold_idx = recover_topk_indices(rec["post_memory_embedding"], rec["outs_dec"])
	ov = topk_overlap(out["topk_idx"].reshape(-1), gold_idx)
	print(f" top-256 overlap with the golden selection: {ov:.4f}")
	assert ov >= 0.99 # near-ties only (S:6)
	tf = _head(net, rec, pe, topk_idx=gold_idx) # teacher-forced selection
	for k in POST_MEMORY:
	assert pcc(tf[k], rec[k]) >= REF_PCC, k
	assert np.array_equal(tf["post_memory_timestamp"], rec["post_memory_timestamp"])


	def test_backbone_vs_research_golden(net):
	g = paths.public_golden("ps019")
	if g is None or "img" not in np.load(g / "tf_0000.npz").files:
	pytest.skip("research/streampetr/public_data not present")
	rec = np.load(g / "tf_0000.npz")
	from tt_streampetr.ttaw.golden import TapRegistry

	taps = TapRegistry(include=("backbone.stage", "neck.", "img_feats"), exclude=("backbone.stage.osa",))
	with torch.no_grad():
	feats = net.backbone(rec["img"], taps).numpy()
	p = pcc(feats, rec["img_feats"])
	print(f" img_feats pcc={p:.12f} max_abs={np.abs(feats - rec['img_feats']).max():.2e}")
	assert p >= REF_PCC
	assert set(taps.names()) == {"backbone.stage2", "backbone.stage3", "backbone.stage4", "backbone.stage5",
	"neck.lateral0", "neck.lateral1", "neck.merged", "img_feats"}
	assert taps["backbone.stage4"].shape == (5, 768, 30, 40) and taps["backbone.stage5"].shape == (5, 1024, 15, 20)


	# ---------------------------------------------------------------------------------------------- port form

	def _gold_pe():
	g = paths.public_golden("ps019")
	if g is None or not all((g / f"tf_{f:04d}.npz").is_file() for f in (0, 1, 2)):
	pytest.skip("research/streampetr/public_data not present")
	return g, np.load(g / "pe_inputs_outputs.npz")


	def test_relative_memory_chain_equals_deployed(net):
	"""Frames 0-2 of PandaSet 019 (consecutive teacher-forcing records' img_feats): the deployed global-frame memory
	loop vs the port's previous-ego-frame state with M_k = E_k^-1 E_{k-1}."""
	from tt_streampetr.host.temporal import AutowareMemory, relative_motion

	g, pe = _gold_pe()
	mem = AutowareMemory(CFG)
	rel = pf.RelativeMemory.zeros()
	E_prev = None
	for f in (0, 1, 2):
	rec = np.load(g / f"tf_{f:04d}.npz")
	ts = np.float32(rec["ts"])
	pre = mem.pre_inputs(ts)
	with torch.no_grad():
	dep = to_numpy(net.head(rec["img_feats"], pe["pos_embed"], pe["cone"], rec["data_ego_pose"],
	rec["data_ego_pose_inv"], pre))
	M = np.eye(4) if E_prev is None else relative_motion(E_prev, rec["data_ego_pose_inv"])
	out, rel = pf.head_relative(net.head, rec["img_feats"], pe["pos_embed"], pe["cone"], M, rel,
	pre["pre_memory_timestamp"])
	out = to_numpy(out)
	for k in ("all_cls_scores", "all_bbox_preds", "outs_dec"):
	p = pcc(out[k], dep[k])
	print(f" frame {f} {k:16s} pcc={p:.10f} max_abs={np.abs(out[k] - dep[k]).max():.2e}")
	assert p >= 0.99999, (f, k)
	assert topk_overlap(out["topk_idx"].reshape(-1), dep["topk_idx"].reshape(-1)) >= 0.99
	d = detection_f1(hp.autoware_postprocess(out["all_cls_scores"], out["all_bbox_preds"], CFG)["objects"],
	hp.autoware_postprocess(dep["all_cls_scores"], dep["all_bbox_preds"], CFG)["objects"])
	assert d["f1"] == 1.0 and d["centre_err_max"] < 0.01
	mem.post_update(dep, ts)
	E_prev = rec["data_ego_pose"]


	def test_teacher_forced_relative_state(net):
	"""A golden ``pre_memory_*`` set converted to the port state (``global_to_relative``, M = I) gives the deployed
	outputs (the device's teacher-forcing input)."""
	g, pe = _gold_pe()
	rec = np.load(g / "tf_0040.npz") if (g / "tf_0040.npz").is_file() else np.load(g / "tf_0002.npz")
	pre = {k: rec[k] for k in rec.files if k.startswith("pre_")}
	st, M = pf.global_to_relative(pre, rec["data_ego_pose_inv"])
	with torch.no_grad():
	out, _ = pf.head_relative(net.head, rec["img_feats"], pe["pos_embed"], pe["cone"], M, st,
	rec["pre_memory_timestamp"])
	out = to_numpy(out)
	for k in ("all_cls_scores", "all_bbox_preds", "outs_dec"):
	assert pcc(out[k], rec[k]) >= REF_PCC, k


	def test_hoisted_kv_and_calibration_cache_are_exact_rewrites(net):
	g, pe = _gold_pe()
	rec = np.load(g / "tf_0001.npz")
	head = net.head
	with torch.no_grad():
	mem_ego, mem_ref = head.memory_pre_update({k: rec[k] for k in rec.files if k.startswith("pre_")},
	rec["data_ego_pose_inv"])
	memory, pos = head.tokens(rec["img_feats"], pe["pos_embed"], pe["cone"])
	coeffs = head.spatial_alignment_coeffs(pe["cone"])
	memory2, pos2 = head.tokens(rec["img_feats"], pe["pos_embed"], pe["cone"], sa_coeffs=coeffs)
	assert torch.equal(memory, memory2) and torch.equal(pos, pos2) # cached gamma / beta: bit-exact
	t = head.temporal(mem_ref, mem_ego, rec["pre_memory_timestamp"], rec["pre_memory_embedding"],
	rec["pre_memory_velo"])
	q = head.queries(t)
	a = head.decoder(q["tgt"], q["query_pos"], q["temp_memory"], q["temp_pos"], memory, pos)
	b = pf.decoder_hoisted(head, q["tgt"], q["query_pos"], q["temp_memory"], q["temp_pos"], memory, pos)
	p = pcc(a.numpy(), b.numpy())
	print(f" hoisted K/V decoder: pcc={p:.12f} max_abs={(a - b).abs().max().item():.2e}")
	assert p >= 0.9999999 and (a - b).abs().max().item() < 1e-3
	w, bias = pf.merged_branch_first_layer(head)
	o = a[-1]
	merged = o @ w + bias
	assert torch.allclose(merged[..., :256], head.lin(o, head.cls[0]), atol=1e-4)
	assert torch.allclose(merged[..., 256:], head.lin(o, head.reg[0]), atol=1e-4)


	def test_stem_normalisation_fold_needs_mean_padding(net):
	"""``conv(W, (v - m) / s)`` == ``conv(W / s, v padded with m)`` + folded bias (float rounding only); a zero pad
	instead is wrong on the border (why the functional port keeps the normalisation on the host)."""
	import torch.nn.functional as F

	from tt_streampetr.ttaw.image_triangle import STREAMPETR_PRESETS

	w, b, s, p = net.backbone.params["/img_backbone/stem/stem_1/conv/Conv"]
	pr = STREAMPETR_PRESETS["autoware_main"]
	rng = np.random.default_rng(0)
	v = torch.from_numpy(rng.uniform(0, 255, size=(1, 3, 32, 48)).astype(np.float32))
	m = torch.from_numpy(pr.mean_f32()).view(1, 3, 1, 1)
	sd = torch.from_numpy(pr.std_f32()).view(1, 3, 1, 1)
	ref = F.conv2d((v - m) / sd, w, b, stride=s, padding=p)
	w2, b2 = pf.fold_normalisation_into_stem(w.numpy(), b.numpy(), pr.mean, pr.std)
	vp = torch.cat([torch.full((1, 1, 34, 50), float(c)) for c in pr.mean], 1)
	vp[:, :, 1:-1, 1:-1] = v
	fold = F.conv2d(vp, torch.from_numpy(w2), torch.from_numpy(b2), stride=s, padding=0)
	assert (fold - ref).abs().max().item() < 1e-3 * ref.abs().max().item()
	zero_pad = F.conv2d(v, torch.from_numpy(w2), torch.from_numpy(b2), stride=s, padding=p)
	assert (zero_pad - ref)[..., 0, :].abs().max().item() > 1.0 # the border breaks without it