# 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)