| |
| """Host post-processing (``host.postprocess``): the node's NMS-free decode and DetectedObjects conversion against the |
| research reference's ``postprocess_autoware`` on the research goldens, the shipped sample's stored CPU detections and |
| the public-data workflow's BEVFormer test-time decode (threshold 0). |
| |
| TT_VISIBLE_DEVICES=none python -m pytest -q -p fake_ttnn_plugin code/tt_bevformer/tests/test_host_postprocess.py |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import math |
|
|
| import numpy as np |
| import pytest |
|
|
| from tt_bevformer.host import postprocess as hp |
| from tt_bevformer.reference import config as C |
| from tt_bevformer.tests import paths |
|
|
|
|
| def _same_boxes(ours: hp.Detections, theirs: list, *, rot_key: str = "r") -> None: |
| """The research lists are float64 before the node's float casts: compare at float32 precision, same order.""" |
| assert len(ours) == len(theirs) |
| for i, d in enumerate(theirs): |
| assert int(ours.label[i]) == int(d["label"]), i |
| for k in ("x", "y", "z", "w", "l", "h", "vx", "vy", "score"): |
| assert float(getattr(ours, k)[i]) == pytest.approx(float(d[k]), rel=1e-5, abs=1e-5), (i, k) |
| assert float(ours.rot[i]) == pytest.approx(float(d[rot_key]), rel=1e-5, abs=1e-5) |
|
|
|
|
| @pytest.fixture(scope="module") |
| def sample0_outputs(): |
| golden = paths.research_file("golden", "sample0_golden.npz") |
| if golden is None: |
| pytest.skip("research/bevformer golden not in this workspace") |
| with np.load(golden) as g: |
| return {f: (g[f"f{f}_outputs_classes"], g[f"f{f}_outputs_coords"]) for f in (1, 2, 3)} |
|
|
|
|
| @pytest.mark.parametrize("frame", [1, 2, 3]) |
| def test_matches_research_postprocess_autoware(sample0_outputs, frame): |
| R = paths.import_research() |
| if R is None: |
| pytest.skip("research/bevformer not in this workspace") |
| cls, crd = sample0_outputs[frame] |
| ours = hp.decode(cls, crd) |
| theirs = R.postprocess_autoware(cls, crd) |
| _same_boxes(ours, theirs) |
| assert len(ours) == {1: 49, 2: 37, 3: 39}[frame] |
| np.testing.assert_allclose(ours.yaw_autoware, [math.remainder(d["yaw_msg"], 2 * math.pi) for d in theirs], |
| atol=2e-6) |
|
|
|
|
| def test_matches_the_shipped_expected_detections(): |
| p = paths.shippable() |
| if p is None: |
| pytest.skip("research/bevformer/public_data/shippable not in this workspace") |
| z = np.load(p / "expected_cpu_outputs.npz") |
| exp = json.loads((p / "expected_cpu_detections.json").read_text()) |
| for i, fr in enumerate(exp["frames"]): |
| ours = hp.decode(z["outputs_classes"][i], z["outputs_coords"][i]) |
| _same_boxes(ours, fr["detections"]) |
|
|
|
|
| def test_threshold_zero_is_bevformer_test_time_decode(sample0_outputs): |
| P = paths.import_public_common() |
| if P is None: |
| pytest.skip("research/bevformer/public_data not in this workspace") |
| cls, crd = sample0_outputs[2] |
| ours = hp.decode(cls, crd, score_threshold=0.0) |
| theirs = P.decode_boxes(cls, crd) |
| assert len(ours) == len(theirs) and len(ours) <= 300 |
| assert ours.query.tolist() == [b["query"] for b in theirs] and ours.label.tolist() == [b["label"] for b in theirs] |
| np.testing.assert_allclose(ours.yaw_physical, [b["yaw"] for b in theirs], atol=2e-6) |
|
|
|
|
| def test_conventions_labels_and_threshold_semantics(): |
| cls = np.full((900, 10), -20.0, np.float32) |
| crd = np.zeros((900, 10), np.float32) |
| s = np.array([0.9, 0.5, 0.2000000015, 0.19, 0.3]) |
| logit = np.log(s / (1 - s)).astype(np.float32) |
| for q, (lab, lg) in enumerate(zip([3, 0, 8, 1, 9], logit)): |
| cls[q, lab] = lg |
| crd[q] = [q, -q, math.log(2.0), math.log(4.0), 0.5, math.log(1.5), math.sin(0.3), math.cos(0.3), 1.0, -1.0] |
| crd[4, 0] = 61.3 |
| d = hp.decode(cls, crd) |
| assert d.label.tolist() == [3, 0] and d.query.tolist() == [0, 1] |
| assert d.autoware_labels == ["TRUCK", "CAR"] and d.class_names == ["bus", "car"] |
| assert d.z[0] == pytest.approx(0.5) and (d.w[0], d.l[0], d.h[0]) == pytest.approx((2.0, 4.0, 1.5)) |
| assert d.yaw_autoware[0] == pytest.approx(-0.3 + math.pi) |
| assert d.yaw_physical[0] == pytest.approx(-0.3 - math.pi / 2) |
| ba, bp = d.boxes("autoware"), d.boxes("physical") |
| assert ba.shape == (2, 7) and tuple(ba[0, 3:6]) == (2.0, 4.0, 1.5) and tuple(bp[0, 3:6]) == (4.0, 2.0, 1.5) |
| assert hp.decode(cls, crd, score_threshold=0.195).label.tolist() == [3, 0, 8] |
| assert len(hp.decode(cls, crd, max_num=1)) == 1 |
| recs = d.to_records() |
| assert recs[0]["autoware_label"] == "TRUCK" and recs[0]["class_name"] == "bus" and recs[0]["query"] == 0 |
| with pytest.raises(ValueError): |
| d.boxes("polar") |
| assert C.AUTOWARE_LABELS == ("CAR", "TRUCK", "TRUCK", "TRUCK", "TRUCK", "UNKNOWN", "MOTORCYCLE", "BICYCLE", |
| "PEDESTRIAN", "UNKNOWN") |
| assert hp.wrap_angle(np.array([math.pi, -math.pi, 3 * math.pi])).tolist() == pytest.approx([math.pi] * 3) |
|
|
|
|
| def test_shipped_small_goldens_are_consistent(): |
| """The bundle's small goldens (code/scripts/make_goldens.py): the stored last-layer outputs of the shipped |
| sequence decode to the stored detections, and frame 30's to the smoke test's stored /predict body.""" |
| paths.need_pandaset() |
| gdir = paths.PANDASET_GOLDENS |
| if not (gdir / "pandaset_019_f30-40.json").is_file(): |
| pytest.skip("small goldens not generated yet (code/scripts/make_goldens.py)") |
| small = json.loads((gdir / "pandaset_019_f30-40.json").read_text()) |
| with np.load(gdir / "pandaset_019_f30-40_last_layer.npz") as z: |
| last = {k: z[k] for k in z.files} |
| for fr in small["frames"]: |
| f = fr["frame"] |
| d = hp.decode(last[f"f{f}_outputs_classes"], last[f"f{f}_outputs_coords"]) |
| assert len(d) == fr["n_detections"] == len(fr["detections"]) |
| for ours, theirs in zip(d.to_records(), fr["detections"]): |
| assert ours["label"] == theirs["label"] and ours["query"] == theirs["query"] |
| for k in ("x", "y", "z", "w", "l", "h", "rot", "score"): |
| assert ours[k] == pytest.approx(theirs[k], abs=1e-5) |
| if f == 30: |
| body = hp.to_output(d, model="bevformer-p150").to_dict() |
| stored = json.loads((paths.PANDASET_SAMPLES / "pandaset_019_f30.reference.json").read_text()) |
| assert body == stored |
|
|
