bevdet-p150 / code /tt_bevdet /tests /test_api_host.py

tt-model push bevdet-p150 (container)

d160c10 verified 12 days ago

17.6 kB

# SPDX-License-Identifier: Apache-2.0
"""The host half of ``api.BEVDet`` (``_prepare`` / ``_forward`` bookkeeping / ``_postprocess``, no device, no weights)
against the CPU reference: the shipped sample prepares to its crop, the reference head maps post-process to the stored
reference outputs (the smoke test's ``samples/synthetic_6cam.reference.json``; the staged PandaSet sample's
``pandaset_019_f38.reference.json`` and goldens in the workspace), the output rows carry the node's Autoware label and
twist, ``nms_rescale_factor`` is a validated per-request parameter, the stream table keeps up to ``MAX_STREAMS``
histories with LRU eviction, and a frame that fails after the device ring changed restarts its stream.
    TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_bevdet/tests/test_api_host.py
"""
from __future__ import annotations

import json
import math

import numpy as np
import pytest

from tt_bevdet import io as tio
from tt_bevdet.api import DEFAULT_NMS_RESCALE, KNOBS, BEVDet
from tt_bevdet.host import inputs as hin
from tt_bevdet.reference.config import AUTOWARE_LABELS, CLASS_NAMES, HEAD_NAMES, HEAD_CHANNELS
from tt_bevdet.tests import paths
from tt_bevdet.ttaw.server import smoke as ttaw_smoke

SHIPPED_REFERENCE = paths.SAMPLES / "synthetic_6cam.reference.json"
SHIPPED_HEADS = paths.GOLDENS_SMALL / "synthetic_6cam_heads.npz"


def make_model(**knobs):
    m = object.__new__(BEVDet)          # no from_pretrained: no weights, no device
    m.compile_params = dict(knobs)
    m._build_host()
    return m


@pytest.fixture
def model():
    return make_model()


def shipped_heads():
    """The CPU reference head maps of the shipped sample, rebuilt from their sparse copy (every channel at the cells
    the vendor decode keeps; the other cells far below the decode threshold): decoding them is exact."""
    with np.load(SHIPPED_HEADS) as z:
        rr, cc = z["cells"][:, 0], z["cells"][:, 1]
        heads = {}
        for k in HEAD_NAMES:
            full = np.zeros((1, HEAD_CHANNELS[k], 128, 128), np.float32)
            if k == "heatmap_0":
                full[:] = -30.0
            full[0][:, rr, cc] = z[k].T
            heads[k] = full
    return heads


def test_identity_and_params():
    assert BEVDet.CAMERA_ORDER == ("CAM_FRONT_LEFT", "CAM_FRONT", "CAM_FRONT_RIGHT", "CAM_BACK_LEFT", "CAM_BACK",
                                   "CAM_BACK_RIGHT")
    assert BEVDet.LABELS == list(CLASS_NAMES) and BEVDet.AUTOWARE_LABELS == list(AUTOWARE_LABELS)   # PLAN 9.3 Q1
    assert BEVDet.requires_calibration()
    p = BEVDet.validate_params({})
    assert p == {"score_threshold": 0.2, "decode_score_threshold": 0.1, "nms_iou_threshold": 0.2, "max_pre_nms": 500,
                 "max_detections": 500, "nms_rescale_factor": DEFAULT_NMS_RESCALE}
    assert DEFAULT_NMS_RESCALE == [1.0, 0.7, 0.7, 0.4, 0.55, 1.1, 1.0, 1.0, 1.5, 3.5]
    assert BEVDet.validate_params(p) == p                                     # idempotent (server, then model)
    for bad in ({"history_stride": 2}, {"scale_intrinsics": False}):          # LOAD knob / calibration key: no params
        with pytest.raises(tio.InputError, match="unknown parameter"):
            BEVDet.validate_params(bad)
    from tt_bevdet.tt.params import DEFAULT_PRECISION

    assert KNOBS.read(env={}).as_dict() == {"HISTORY_STRIDE": 1, "BEVPOOL": "host", "HOST_BEVPOOL": "numba", "HOST_POST": "numba",
                                            "HOST_PRE": "numba", "HOST_RESIZE": "gather", "HOST_THREADS": 0, "IOU_MODE": "vendor", "MAX_STREAMS": 4,
                                            "PRECISION": DEFAULT_PRECISION, "STEM_IN": "kin",
                                                "CONV_L1": "all,dec", "BACKBONE_RES": "next12", "RING": "tile", "DEC_MERGE": 1, "MM1X1": 1, "CONV_TUNE": 1, "HEAD2": "hs", "L1RES": 1, "DN_L1": 1, "PUSH_X": 1, "UPSAMPLE": "kernel", "UPK_TERMS": 1, "MMRES_MCAST": 1, "RES_L1": 1, "TSHARD": "all", "DEC_CHAIN": 1, "L3T": 1, "ENC_PRECISION": "", "DEC_PRECISION": ""}


def test_nms_rescale_factor_param():
    """``nms_rescale_factor`` (RT-host, PLAN.md 2.10): 10 factors or a partial {class: factor} mapping; validated."""
    full = [2.0] * 10
    assert BEVDet.validate_params({"nms_rescale_factor": full})["nms_rescale_factor"] == full
    part = BEVDet.validate_params({"nms_rescale_factor": {"pedestrian": 2.0, "traffic_cone": "4"}})["nms_rescale_factor"]
    assert part == DEFAULT_NMS_RESCALE[:8] + [2.0, 4.0]
    for bad in ([1.0] * 9, {"tram": 1.0}, [0.0] * 10, [1.0] * 9 + [math.inf], "1,2", [True] * 10, 3.0,
                [1.0] * 9 + [101.0]):
        with pytest.raises(tio.InputError, match="nms_rescale_factor"):
            BEVDet.validate_params({"nms_rescale_factor": bad})


def test_serve_env_pins_the_knob_defaults():
    """``tt-model.yaml`` ``serve.env`` pins every load-time knob at its default (the measured best / the decision)."""
    import yaml

    env = yaml.safe_load((paths.BUNDLE_DIR / "tt-model.yaml").read_text())["serve"]["env"]
    pins = KNOBS.serve_env()
    assert {k: str(env.get(k)) for k in pins} == pins


def test_prepare_shipped_sample(model):
    kw = hin.load_sample(paths.SHIPPED_SAMPLE)
    prep = model._prepare(**kw, stream={"id": "a"})
    assert prep["crop_u8"].shape == (6, 3, 256, 704) and prep["crop_u8"].dtype == np.uint8
    assert prep["tables"].stats()["M"] == 355907 and prep["cam_params"].shape == (1, 6, 27)
    plan = prep["plan"]
    assert plan.flag == 1 and plan.history == [-1] * 8 and np.array_equal(plan.transforms[0], [1, 0, 0, 0, 1, 0])
    with pytest.raises(tio.InputError, match="not a point cloud"):
        model._prepare(points=np.zeros((3, 4)), **kw)
    with pytest.raises(tio.InputError, match="scale_intrinsics"):
        model._prepare(images=kw["images"], calibration=dict(kw["calibration"], scale_intrinsics="no"))


def test_stream_table_lru(model):
    """Each id keeps its own history; up to MAX_STREAMS (4) ids at once, a fifth evicts the least recently used one,
    which starts fresh again (PLAN.md 9.3 Q6); ``reset`` is a scene change (flag 0)."""
    kw = hin.load_sample(paths.SHIPPED_SAMPLE)

    def frame(sid, **stream):
        prep = model._prepare(**kw, stream={"id": sid, **stream})
        prep["state"].end_frame()
        return prep["plan"]

    assert frame("a").history == [-1] * 8
    assert frame("a").history == [0] * 8                                     # the first save filled every slot
    for sid in ("b", "c", "d"):
        frame(sid)
    assert list(model._streams) == ["a", "b", "c", "d"]
    frame("a")                                                                # "a" becomes the most recent
    frame("e")                                                                # evicts "b", the least recent
    assert list(model._streams) == ["c", "d", "a", "e"]
    assert frame("b").history == [-1] * 8                                     # "b" restarts fresh (evicts "c")
    assert frame("a", reset=True).flag == 0
    info = model.stream_info()
    assert info["max_streams"] == 4 and info["ids"] == ["d", "e", "b", "a"] and "least recently used" in info["policy"]
    one = make_model(max_streams=1)
    one._prepare(**kw, stream={"id": "x"})["state"].end_frame()
    one._prepare(**kw, stream={"id": "y"})["state"].abort_frame()
    assert list(one._streams) == ["y"]


class FakeRunner:
    """The TraceRunner calls ``_forward`` makes; ``fail`` = the variant whose replay raises."""

    def __init__(self, fail=None):
        self.fail, self.calls = fail, []

    def __call__(self, variant, inputs=None, params=None):
        self.calls.append(variant)
        return np.zeros((1, 1, 4224, 224), np.float32)

    def upload(self, inputs=None, params=None):
        self.calls.append("upload")

    def replay(self, variant):
        self.calls.append(variant)
        if variant == self.fail:
            raise RuntimeError(f"replay {variant} failed")

    def read(self, variant):
        return np.zeros((1, 1, 16384, 20), np.float32)

    def save_state(self, name, bank):
        self.calls.append(f"save{bank}")

    def load_state(self, name, bank):
        self.calls.append(f"load{bank}")


class FakeTT:
    curr_shape = (1, 128, 128, 96)

    def unpack_encode(self, enc):
        return np.zeros((6, 16, 44, 128), np.float32), np.zeros((6, 16, 44, 96), np.float32)

    def unpack_heads(self, heads):
        return {}

    def decode_variant(self, transforms, history, flag, taps=False):
        return "decode_taps" if taps else "decode"

    def after_variant(self, v, decode_variant):
        return v


def _forward(model, sid, **stream):
    kw = hin.load_sample(paths.SHIPPED_SAMPLE)
    prep = model._prepare(**kw, stream={"id": sid, **stream})
    prep.update(gates=(np.zeros((6, 512), np.float32), np.zeros((6, 512), np.float32)),
                enc_in={"img": np.zeros((1, 1, 6 * 256 * 704, 3), np.float32)},
                pool=lambda d, f, layout=None, out_layout=None, **kw: np.zeros((1, 128, 128, 80), np.float32))
    return model._forward(prep)


def test_failed_frame_keeps_device_ring_and_host_plan_consistent(model):
    """VERIFY_PORT.md L11: a failure before any ring update drops only the frame (history kept); once ``fill`` /
    ``push`` has started, the ring no longer matches the plan, so the stream restarts fresh (its next frame never
    reads the ring and its ``fill`` overwrites every slot)."""
    model.tt, model.runner = FakeTT(), FakeRunner()
    _forward(model, "s")                                                      # frame 0: decode, then fill
    assert model.runner.calls[-2:] == ["decode", "fill"] and model._streams["s"].frame == 1
    model.runner = FakeRunner(fail="decode")                                  # fails before any ring update
    with pytest.raises(RuntimeError, match="decode"):
        _forward(model, "s")
    assert model._streams["s"].frame == 1 and model._streams["s"]._pending is None   # history kept
    model.runner = FakeRunner(fail="decode")                                  # a reset frame: fill BEFORE decode
    with pytest.raises(RuntimeError, match="decode"):
        _forward(model, "s", reset=True)
    assert model.runner.calls[-2:] == ["fill", "decode"] and "s" not in model._streams   # restarted
    model.runner = FakeRunner(fail="push")                                    # fails during the push
    _forward(model, "t")
    with pytest.raises(RuntimeError, match="push"):
        _forward(model, "t")
    assert "t" not in model._streams
    model.runner = FakeRunner()
    _forward(model, "s")                                                      # the restarted stream: a fresh frame
    assert model.runner.calls[-2:] == ["decode", "fill"] and model._streams["s"].frame == 1


def test_stage_timings_do_not_collide_with_the_server_keys(model):
    """The served ``timing_ms`` is the model's plus the server's own ``decode`` (base64 + images), ``model_call`` and
    ``total`` (``ttaw.server``): a model stage under one of those names would be overwritten in the response, as the
    BEV decoder segment (``decode``) was until the release."""
    model.tt, model.runner = FakeTT(), FakeRunner()
    raw = _forward(model, "timing")
    assert set(raw["timing_ms"]) == {"encode", "bevpool", "bev_decode"}
    assert not set(raw["timing_ms"]) & {"decode", "model_call", "total"}


def test_bank_switch_failure_reloads_on_the_next_frame():
    """MAX_STREAMS > 1: while a bank loads, no stream owns the ring, so a failed load is redone on the next frame."""
    m = make_model()
    m.tt, m.runner = FakeTT(), FakeRunner()
    _forward(m, "a")
    _forward(m, "b")                                                          # save a's ring, b gets a new bank
    assert "save0" in m.runner.calls and m._active == "b"

    class LoadFails(FakeRunner):
        def load_state(self, name, bank):
            raise RuntimeError("load failed")

    m.runner = LoadFails()
    with pytest.raises(RuntimeError, match="load failed"):
        _forward(m, "a")
    assert m._active is None and m._streams["a"].frame == 1                   # history kept, ring owned by nobody
    m.runner = FakeRunner()
    _forward(m, "a")
    assert m.runner.calls[0] == "load0" and m._active == "a"                  # reloaded (b's ring was saved before)


def test_output_rows_carry_autoware_label_and_twist(model):
    """``label`` = network class, ``autoware_label`` = the node's label, ``twist`` = the node's formula."""
    from tt_bevdet.host.outputs import BEVDetDetections, node_twist

    boxes = np.array([[10, 0, 0.8, 4.5, 1.9, 1.6, 0.3], [5, 5, 0.4, 0.4, 0.4, 0.7, 0.0]], np.float32)
    det = BEVDetDetections(boxes, np.array([0.5, 0.9], np.float32), np.array([0, 9]),
                           velocities=np.array([[3.0, 4.0], [0.0, 0.0]], np.float32), labels=BEVDet.LABELS,
                           autoware_labels=BEVDet.AUTOWARE_LABELS, model="m")
    rows = det.to_dicts()
    assert [(r["label"], r["autoware_label"]) for r in rows] == [("traffic_cone", "UNKNOWN"), ("car", "CAR")]
    assert rows[1]["twist"] == {"linear_x": 5.0, "angular_z": round(2 * (math.atan2(4.0, 3.0) - float(np.float32(0.3))),
                                                                     4)}
    assert node_twist(0.0, 0.0, 0.5) == {"linear_x": 0.0, "angular_z": -1.0}
    body = det.to_dict("npz")
    assert set(body["arrays"]) == {"boxes", "scores", "label_ids", "velocities", "twist"}


def test_postprocess_reproduces_the_shipped_reference(model):
    """The CPU reference heads of the shipped sample, post-processed by the API with the smoke protocol's params,
    give exactly the stored reference body the container smoke compares with; at the default params, the published
    rows. Also the smoke's agreement rule on identical bodies."""
    if not SHIPPED_REFERENCE.is_file() or not SHIPPED_HEADS.is_file():
        pytest.fail("the shipped sample's stored reference outputs are missing (code/scripts/make_sample.py)")
    from tt_bevdet.server import smoke_test

    body = json.loads(SHIPPED_REFERENCE.read_text())
    heads = shipped_heads()
    kw = hin.load_sample(paths.SHIPPED_SAMPLE)
    prep = model._prepare(**kw, stream={"id": "smoke"})
    out = model._postprocess(heads, prep, BEVDet.validate_params(smoke_test.SMOKE_PARAMS))
    assert out.to_dict()["detections"] == body["detections"] and body["model"] == BEVDet.MODEL_NAME
    published = model._postprocess(heads, prep, BEVDet.validate_params({})).to_dicts()
    assert published == [d for d in body["detections"] if d["score"] >= 0.2]
    expect = dict(part.split(":") for part in smoke_test.DEFAULT_EXPECT.split(","))
    for label, n in expect.items():
        assert sum(1 for d in published if d["label"] == label) >= int(n), label
    assert min(d["score"] for d in published) > 0.3                           # far from the 0.1 / 0.2 thresholds
    assert ttaw_smoke.find_reference(paths.SHIPPED_SAMPLE) == SHIPPED_REFERENCE
    metrics, fails = smoke_test.compare_with_reference(body, body)
    assert fails == [] and metrics["recall"] == metrics["precision"] == metrics["strict_recall"] == 1.0


def test_postprocess_reproduces_the_pandaset_reference(model):
    """Workspace only: the research goldens' head maps of PandaSet f38 post-process to the staged sample's stored
    reference (smoke protocol) and goldens (default params)."""
    golden = paths.golden_dir() / "pandaset_019_autoware" / "f38.npz"
    if not golden.is_file():
        pytest.skip("reference goldens not generated (code/scripts/make_goldens.py)")
    pdir = paths.require_pandaset()
    small = paths.pandaset_golden()
    ref_body = pdir / "pandaset_019_f38.reference.json"
    if small is None or not ref_body.is_file():
        pytest.fail("the PandaSet stored outputs are missing: make_goldens.py --sets pandaset --no-big")
    with np.load(golden) as g:
        heads = {k: g[f"head.{k}"] for k in HEAD_NAMES}
    kw = hin.load_sample(pdir / "pandaset_019_f38.json")
    prep = model._prepare(**kw, stream={"id": "smoke"})
    out = model._postprocess(heads, prep, BEVDet.validate_params({"score_threshold": 0.1}))
    assert out.to_dict()["detections"] == json.loads(ref_body.read_text())["detections"]
    rec = json.loads(small.read_text())["frames"]["f38"]
    assert model._postprocess(heads, prep, BEVDet.validate_params({})).to_dicts() == rec["detections"]
    assert out.meta["class_names"][2] == "construction_vehicle" and out.meta["stream"]["flag"] == 1
    assert prep["tables"].stats()["M"] == 316530


def test_rescale_factor_changes_only_the_nms(model):
    """A larger factor for one class makes the Scale-NMS suppress more of its overlapping boxes; the published
    sizes stay the network's (divided back by the same factor)."""
    from tt_bevdet.host.postprocess import postprocess
    from tt_bevdet.reference.config import default_config

    cfg = default_config()
    heads = {k: np.zeros((1, HEAD_CHANNELS[k], 128, 128), np.float32) for k in HEAD_NAMES}
    heads["heatmap_0"][:] = -9.0
    for col, logit in ((64, 2.0), (65, 1.5)):                                 # two cars 0.8 m apart
        heads["heatmap_0"][0, 0, 64, col] = logit
        heads["dim_0"][0, :, 64, col] = np.log([0.6, 0.6, 1.5])               # small boxes: IoU 0 at factor 1
        heads["rot_0"][0, 1, 64, col] = 1.0
    base = postprocess(heads, cfg)
    wide = postprocess(heads, cfg, nms_rescale_factor=[3.0] + [1.0] * 9)
    assert len(base) == 2 and len(wide) == 1
    np.testing.assert_allclose(wide.boxes[0, 3:6], base.boxes[0, 3:6], rtol=1e-6)

	# SPDX-License-Identifier: Apache-2.0
	"""The host half of ``api.BEVDet`` (``_prepare`` / ``_forward`` bookkeeping / ``_postprocess``, no device, no weights)
	against the CPU reference: the shipped sample prepares to its crop, the reference head maps post-process to the stored
	reference outputs (the smoke test's ``samples/synthetic_6cam.reference.json``; the staged PandaSet sample's
	``pandaset_019_f38.reference.json`` and goldens in the workspace), the output rows carry the node's Autoware label and
	twist, ``nms_rescale_factor`` is a validated per-request parameter, the stream table keeps up to ``MAX_STREAMS``
	histories with LRU eviction, and a frame that fails after the device ring changed restarts its stream.

	TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_bevdet/tests/test_api_host.py
	"""
	from __future__ import annotations

	import json
	import math

	import numpy as np
	import pytest

	from tt_bevdet import io as tio
	from tt_bevdet.api import DEFAULT_NMS_RESCALE, KNOBS, BEVDet
	from tt_bevdet.host import inputs as hin
	from tt_bevdet.reference.config import AUTOWARE_LABELS, CLASS_NAMES, HEAD_NAMES, HEAD_CHANNELS
	from tt_bevdet.tests import paths
	from tt_bevdet.ttaw.server import smoke as ttaw_smoke

	SHIPPED_REFERENCE = paths.SAMPLES / "synthetic_6cam.reference.json"
	SHIPPED_HEADS = paths.GOLDENS_SMALL / "synthetic_6cam_heads.npz"


	def make_model(**knobs):
	m = object.__new__(BEVDet) # no from_pretrained: no weights, no device
	m.compile_params = dict(knobs)
	m._build_host()
	return m


	@pytest.fixture
	def model():
	return make_model()


	def shipped_heads():
	"""The CPU reference head maps of the shipped sample, rebuilt from their sparse copy (every channel at the cells
	the vendor decode keeps; the other cells far below the decode threshold): decoding them is exact."""
	with np.load(SHIPPED_HEADS) as z:
	rr, cc = z["cells"][:, 0], z["cells"][:, 1]
	heads = {}
	for k in HEAD_NAMES:
	full = np.zeros((1, HEAD_CHANNELS[k], 128, 128), np.float32)
	if k == "heatmap_0":
	full[:] = -30.0
	full[0][:, rr, cc] = z[k].T
	heads[k] = full
	return heads


	def test_identity_and_params():
	assert BEVDet.CAMERA_ORDER == ("CAM_FRONT_LEFT", "CAM_FRONT", "CAM_FRONT_RIGHT", "CAM_BACK_LEFT", "CAM_BACK",
	"CAM_BACK_RIGHT")
	assert BEVDet.LABELS == list(CLASS_NAMES) and BEVDet.AUTOWARE_LABELS == list(AUTOWARE_LABELS) # PLAN 9.3 Q1
	assert BEVDet.requires_calibration()
	p = BEVDet.validate_params({})
	assert p == {"score_threshold": 0.2, "decode_score_threshold": 0.1, "nms_iou_threshold": 0.2, "max_pre_nms": 500,
	"max_detections": 500, "nms_rescale_factor": DEFAULT_NMS_RESCALE}
	assert DEFAULT_NMS_RESCALE == [1.0, 0.7, 0.7, 0.4, 0.55, 1.1, 1.0, 1.0, 1.5, 3.5]
	assert BEVDet.validate_params(p) == p # idempotent (server, then model)
	for bad in ({"history_stride": 2}, {"scale_intrinsics": False}): # LOAD knob / calibration key: no params
	with pytest.raises(tio.InputError, match="unknown parameter"):
	BEVDet.validate_params(bad)
	from tt_bevdet.tt.params import DEFAULT_PRECISION

	assert KNOBS.read(env={}).as_dict() == {"HISTORY_STRIDE": 1, "BEVPOOL": "host", "HOST_BEVPOOL": "numba", "HOST_POST": "numba",
	"HOST_PRE": "numba", "HOST_RESIZE": "gather", "HOST_THREADS": 0, "IOU_MODE": "vendor", "MAX_STREAMS": 4,
	"PRECISION": DEFAULT_PRECISION, "STEM_IN": "kin",
	"CONV_L1": "all,dec", "BACKBONE_RES": "next12", "RING": "tile", "DEC_MERGE": 1, "MM1X1": 1, "CONV_TUNE": 1, "HEAD2": "hs", "L1RES": 1, "DN_L1": 1, "PUSH_X": 1, "UPSAMPLE": "kernel", "UPK_TERMS": 1, "MMRES_MCAST": 1, "RES_L1": 1, "TSHARD": "all", "DEC_CHAIN": 1, "L3T": 1, "ENC_PRECISION": "", "DEC_PRECISION": ""}


	def test_nms_rescale_factor_param():
	"""``nms_rescale_factor`` (RT-host, PLAN.md 2.10): 10 factors or a partial {class: factor} mapping; validated."""
	full = [2.0] * 10
	assert BEVDet.validate_params({"nms_rescale_factor": full})["nms_rescale_factor"] == full
	part = BEVDet.validate_params({"nms_rescale_factor": {"pedestrian": 2.0, "traffic_cone": "4"}})["nms_rescale_factor"]
	assert part == DEFAULT_NMS_RESCALE[:8] + [2.0, 4.0]
	for bad in ([1.0] * 9, {"tram": 1.0}, [0.0] * 10, [1.0] * 9 + [math.inf], "1,2", [True] * 10, 3.0,
	[1.0] * 9 + [101.0]):
	with pytest.raises(tio.InputError, match="nms_rescale_factor"):
	BEVDet.validate_params({"nms_rescale_factor": bad})


	def test_serve_env_pins_the_knob_defaults():
	"""``tt-model.yaml`` ``serve.env`` pins every load-time knob at its default (the measured best / the decision)."""
	import yaml

	env = yaml.safe_load((paths.BUNDLE_DIR / "tt-model.yaml").read_text())["serve"]["env"]
	pins = KNOBS.serve_env()
	assert {k: str(env.get(k)) for k in pins} == pins


	def test_prepare_shipped_sample(model):
	kw = hin.load_sample(paths.SHIPPED_SAMPLE)
	prep = model._prepare(**kw, stream={"id": "a"})
	assert prep["crop_u8"].shape == (6, 3, 256, 704) and prep["crop_u8"].dtype == np.uint8
	assert prep["tables"].stats()["M"] == 355907 and prep["cam_params"].shape == (1, 6, 27)
	plan = prep["plan"]
	assert plan.flag == 1 and plan.history == [-1] * 8 and np.array_equal(plan.transforms[0], [1, 0, 0, 0, 1, 0])
	with pytest.raises(tio.InputError, match="not a point cloud"):
	model._prepare(points=np.zeros((3, 4)), **kw)
	with pytest.raises(tio.InputError, match="scale_intrinsics"):
	model._prepare(images=kw["images"], calibration=dict(kw["calibration"], scale_intrinsics="no"))


	def test_stream_table_lru(model):
	"""Each id keeps its own history; up to MAX_STREAMS (4) ids at once, a fifth evicts the least recently used one,
	which starts fresh again (PLAN.md 9.3 Q6); ``reset`` is a scene change (flag 0)."""
	kw = hin.load_sample(paths.SHIPPED_SAMPLE)

	def frame(sid, **stream):
	prep = model._prepare(kw, stream={"id": sid, stream})
	prep["state"].end_frame()
	return prep["plan"]

	assert frame("a").history == [-1] * 8
	assert frame("a").history == [0] * 8 # the first save filled every slot
	for sid in ("b", "c", "d"):
	frame(sid)
	assert list(model._streams) == ["a", "b", "c", "d"]
	frame("a") # "a" becomes the most recent
	frame("e") # evicts "b", the least recent
	assert list(model._streams) == ["c", "d", "a", "e"]
	assert frame("b").history == [-1] * 8 # "b" restarts fresh (evicts "c")
	assert frame("a", reset=True).flag == 0
	info = model.stream_info()
	assert info["max_streams"] == 4 and info["ids"] == ["d", "e", "b", "a"] and "least recently used" in info["policy"]
	one = make_model(max_streams=1)
	one._prepare(**kw, stream={"id": "x"})["state"].end_frame()
	one._prepare(**kw, stream={"id": "y"})["state"].abort_frame()
	assert list(one._streams) == ["y"]


	class FakeRunner:
	"""The TraceRunner calls ``_forward`` makes; ``fail`` = the variant whose replay raises."""

	def __init__(self, fail=None):
	self.fail, self.calls = fail, []

	def __call__(self, variant, inputs=None, params=None):
	self.calls.append(variant)
	return np.zeros((1, 1, 4224, 224), np.float32)

	def upload(self, inputs=None, params=None):
	self.calls.append("upload")

	def replay(self, variant):
	self.calls.append(variant)
	if variant == self.fail:
	raise RuntimeError(f"replay {variant} failed")

	def read(self, variant):
	return np.zeros((1, 1, 16384, 20), np.float32)

	def save_state(self, name, bank):
	self.calls.append(f"save{bank}")

	def load_state(self, name, bank):
	self.calls.append(f"load{bank}")


	class FakeTT:
	curr_shape = (1, 128, 128, 96)

	def unpack_encode(self, enc):
	return np.zeros((6, 16, 44, 128), np.float32), np.zeros((6, 16, 44, 96), np.float32)

	def unpack_heads(self, heads):
	return {}

	def decode_variant(self, transforms, history, flag, taps=False):
	return "decode_taps" if taps else "decode"

	def after_variant(self, v, decode_variant):
	return v


	def _forward(model, sid, **stream):
	kw = hin.load_sample(paths.SHIPPED_SAMPLE)
	prep = model._prepare(kw, stream={"id": sid, stream})
	prep.update(gates=(np.zeros((6, 512), np.float32), np.zeros((6, 512), np.float32)),
	enc_in={"img": np.zeros((1, 1, 6 * 256 * 704, 3), np.float32)},
	pool=lambda d, f, layout=None, out_layout=None, **kw: np.zeros((1, 128, 128, 80), np.float32))
	return model._forward(prep)


	def test_failed_frame_keeps_device_ring_and_host_plan_consistent(model):
	"""VERIFY_PORT.md L11: a failure before any ring update drops only the frame (history kept); once ``fill`` /
	``push`` has started, the ring no longer matches the plan, so the stream restarts fresh (its next frame never
	reads the ring and its ``fill`` overwrites every slot)."""
	model.tt, model.runner = FakeTT(), FakeRunner()
	_forward(model, "s") # frame 0: decode, then fill
	assert model.runner.calls[-2:] == ["decode", "fill"] and model._streams["s"].frame == 1
	model.runner = FakeRunner(fail="decode") # fails before any ring update
	with pytest.raises(RuntimeError, match="decode"):
	_forward(model, "s")
	assert model._streams["s"].frame == 1 and model._streams["s"]._pending is None # history kept
	model.runner = FakeRunner(fail="decode") # a reset frame: fill BEFORE decode
	with pytest.raises(RuntimeError, match="decode"):
	_forward(model, "s", reset=True)
	assert model.runner.calls[-2:] == ["fill", "decode"] and "s" not in model._streams # restarted
	model.runner = FakeRunner(fail="push") # fails during the push
	_forward(model, "t")
	with pytest.raises(RuntimeError, match="push"):
	_forward(model, "t")
	assert "t" not in model._streams
	model.runner = FakeRunner()
	_forward(model, "s") # the restarted stream: a fresh frame
	assert model.runner.calls[-2:] == ["decode", "fill"] and model._streams["s"].frame == 1


	def test_stage_timings_do_not_collide_with_the_server_keys(model):
	"""The served ``timing_ms`` is the model's plus the server's own ``decode`` (base64 + images), ``model_call`` and
	``total`` (``ttaw.server``): a model stage under one of those names would be overwritten in the response, as the
	BEV decoder segment (``decode``) was until the release."""
	model.tt, model.runner = FakeTT(), FakeRunner()
	raw = _forward(model, "timing")
	assert set(raw["timing_ms"]) == {"encode", "bevpool", "bev_decode"}
	assert not set(raw["timing_ms"]) & {"decode", "model_call", "total"}


	def test_bank_switch_failure_reloads_on_the_next_frame():
	"""MAX_STREAMS > 1: while a bank loads, no stream owns the ring, so a failed load is redone on the next frame."""
	m = make_model()
	m.tt, m.runner = FakeTT(), FakeRunner()
	_forward(m, "a")
	_forward(m, "b") # save a's ring, b gets a new bank
	assert "save0" in m.runner.calls and m._active == "b"

	class LoadFails(FakeRunner):
	def load_state(self, name, bank):
	raise RuntimeError("load failed")

	m.runner = LoadFails()
	with pytest.raises(RuntimeError, match="load failed"):
	_forward(m, "a")
	assert m._active is None and m._streams["a"].frame == 1 # history kept, ring owned by nobody
	m.runner = FakeRunner()
	_forward(m, "a")
	assert m.runner.calls[0] == "load0" and m._active == "a" # reloaded (b's ring was saved before)


	def test_output_rows_carry_autoware_label_and_twist(model):
	"""``label`` = network class, ``autoware_label`` = the node's label, ``twist`` = the node's formula."""
	from tt_bevdet.host.outputs import BEVDetDetections, node_twist

	boxes = np.array([[10, 0, 0.8, 4.5, 1.9, 1.6, 0.3], [5, 5, 0.4, 0.4, 0.4, 0.7, 0.0]], np.float32)
	det = BEVDetDetections(boxes, np.array([0.5, 0.9], np.float32), np.array([0, 9]),
	velocities=np.array([[3.0, 4.0], [0.0, 0.0]], np.float32), labels=BEVDet.LABELS,
	autoware_labels=BEVDet.AUTOWARE_LABELS, model="m")
	rows = det.to_dicts()
	assert [(r["label"], r["autoware_label"]) for r in rows] == [("traffic_cone", "UNKNOWN"), ("car", "CAR")]
	assert rows[1]["twist"] == {"linear_x": 5.0, "angular_z": round(2 * (math.atan2(4.0, 3.0) - float(np.float32(0.3))),
	4)}
	assert node_twist(0.0, 0.0, 0.5) == {"linear_x": 0.0, "angular_z": -1.0}
	body = det.to_dict("npz")
	assert set(body["arrays"]) == {"boxes", "scores", "label_ids", "velocities", "twist"}


	def test_postprocess_reproduces_the_shipped_reference(model):
	"""The CPU reference heads of the shipped sample, post-processed by the API with the smoke protocol's params,
	give exactly the stored reference body the container smoke compares with; at the default params, the published
	rows. Also the smoke's agreement rule on identical bodies."""
	if not SHIPPED_REFERENCE.is_file() or not SHIPPED_HEADS.is_file():
	pytest.fail("the shipped sample's stored reference outputs are missing (code/scripts/make_sample.py)")
	from tt_bevdet.server import smoke_test

	body = json.loads(SHIPPED_REFERENCE.read_text())
	heads = shipped_heads()
	kw = hin.load_sample(paths.SHIPPED_SAMPLE)
	prep = model._prepare(**kw, stream={"id": "smoke"})
	out = model._postprocess(heads, prep, BEVDet.validate_params(smoke_test.SMOKE_PARAMS))
	assert out.to_dict()["detections"] == body["detections"] and body["model"] == BEVDet.MODEL_NAME
	published = model._postprocess(heads, prep, BEVDet.validate_params({})).to_dicts()
	assert published == [d for d in body["detections"] if d["score"] >= 0.2]
	expect = dict(part.split(":") for part in smoke_test.DEFAULT_EXPECT.split(","))
	for label, n in expect.items():
	assert sum(1 for d in published if d["label"] == label) >= int(n), label
	assert min(d["score"] for d in published) > 0.3 # far from the 0.1 / 0.2 thresholds
	assert ttaw_smoke.find_reference(paths.SHIPPED_SAMPLE) == SHIPPED_REFERENCE
	metrics, fails = smoke_test.compare_with_reference(body, body)
	assert fails == [] and metrics["recall"] == metrics["precision"] == metrics["strict_recall"] == 1.0


	def test_postprocess_reproduces_the_pandaset_reference(model):
	"""Workspace only: the research goldens' head maps of PandaSet f38 post-process to the staged sample's stored
	reference (smoke protocol) and goldens (default params)."""
	golden = paths.golden_dir() / "pandaset_019_autoware" / "f38.npz"
	if not golden.is_file():
	pytest.skip("reference goldens not generated (code/scripts/make_goldens.py)")
	pdir = paths.require_pandaset()
	small = paths.pandaset_golden()
	ref_body = pdir / "pandaset_019_f38.reference.json"
	if small is None or not ref_body.is_file():
	pytest.fail("the PandaSet stored outputs are missing: make_goldens.py --sets pandaset --no-big")
	with np.load(golden) as g:
	heads = {k: g[f"head.{k}"] for k in HEAD_NAMES}
	kw = hin.load_sample(pdir / "pandaset_019_f38.json")
	prep = model._prepare(**kw, stream={"id": "smoke"})
	out = model._postprocess(heads, prep, BEVDet.validate_params({"score_threshold": 0.1}))
	assert out.to_dict()["detections"] == json.loads(ref_body.read_text())["detections"]
	rec = json.loads(small.read_text())["frames"]["f38"]
	assert model._postprocess(heads, prep, BEVDet.validate_params({})).to_dicts() == rec["detections"]
	assert out.meta["class_names"][2] == "construction_vehicle" and out.meta["stream"]["flag"] == 1
	assert prep["tables"].stats()["M"] == 316530


	def test_rescale_factor_changes_only_the_nms(model):
	"""A larger factor for one class makes the Scale-NMS suppress more of its overlapping boxes; the published
	sizes stay the network's (divided back by the same factor)."""
	from tt_bevdet.host.postprocess import postprocess
	from tt_bevdet.reference.config import default_config

	cfg = default_config()
	heads = {k: np.zeros((1, HEAD_CHANNELS[k], 128, 128), np.float32) for k in HEAD_NAMES}
	heads["heatmap_0"][:] = -9.0
	for col, logit in ((64, 2.0), (65, 1.5)): # two cars 0.8 m apart
	heads["heatmap_0"][0, 0, 64, col] = logit
	heads["dim_0"][0, :, 64, col] = np.log([0.6, 0.6, 1.5]) # small boxes: IoU 0 at factor 1
	heads["rot_0"][0, 1, 64, col] = 1.0
	base = postprocess(heads, cfg)
	wide = postprocess(heads, cfg, nms_rescale_factor=[3.0] + [1.0] * 9)
	assert len(base) == 2 and len(wide) == 1
	np.testing.assert_allclose(wide.boxes[0, 3:6], base.boxes[0, 3:6], rtol=1e-6)