File size: 8,372 Bytes
51defdc 8eeb550 51defdc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | # SPDX-License-Identifier: Apache-2.0
"""The host half of ``api.METEOR`` (no device, no weights): the identity, the runtime params (the C++ renderer's knobs,
``host.postprocess.PostConfig``), ``_prepare`` on the shipped sample (the golden feed, per-stream state) and
``_postprocess`` on synthetic outputs (the same ``host.result`` code as the CPU reference). The shipped sample is
the synthetic eight-camera frame; the PandaSet 019 frame 40 sample is staged (``tests/paths.py``).
TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_meteor/tests/test_api_host.py
"""
from __future__ import annotations
import json
from pathlib import Path
import numpy as np
import pytest
from tt_meteor import io as tio
from tt_meteor.api import KNOBS, METEOR, Output, load_sample
from tt_meteor.host.outputs import MeteorOutput
from tt_meteor.host.postprocess import PostConfig
from tt_meteor.reference import config as C
from tt_meteor.tests import paths as P
PKG = Path(__file__).resolve().parents[1]
@pytest.fixture
def model():
m = METEOR.__new__(METEOR) # the host half only: no weights, no device
m.compile_params = {}
m._build_host()
return m
def test_identity_and_params():
assert METEOR.MODEL_NAME == "meteor-p150" and METEOR.DEFAULT_REVISION == C.HF_REVISION
assert list(METEOR.CAMERA_ORDER) == list(C.CAMERAS) and METEOR.LABELS == ["VEHICLE", "VRU"]
assert Output is MeteorOutput and METEOR.EXTRA_INPUTS == ("ego_speed",)
cfg = PostConfig()
for name, (_, _, _, default) in METEOR.RUNTIME_PARAMS.items():
if name == "det2d_hide":
assert tuple(int(v) for v in default.split(",")) == cfg.det2d_hide
elif name != "heads":
assert getattr(cfg, name) == default, name
p = METEOR.validate_params({"vehicle_threshold": "0.5", "det2d_hide": "7,9"})
assert p["vehicle_threshold"] == 0.5 and p["det2d_hide"] == "7,9"
with pytest.raises(tio.InputError):
METEOR.validate_params({"bev_nms_iou": 1.5})
assert KNOBS.defaults().as_dict() == {"INPUT_NORM": "onnx", "DEPTH_MEAN_BINS": "log", "MAX_STREAMS": 16,
"IMAGE_PRECISION": "terms3", "CONV_L1": "all", "CONV_L1_ABH": "none",
"MM_CFG": "grid", "HOST_PRE": "rows",
"HOST_POST": "sparse", "SEG_FUSE": "inplace", "TERMS_EPI": "fused",
"TERMS_ADD": "split", "MEAN_CFG": "bmm",
"CONV_RM": "io", "STEM_POOL": "kernel",
"IMG_IN": "rows", "TERMS_ST2": "exact", "PACK": "nchw",
"REROW": "kernel", "ARGMAX": "kernel", "READ_JOIN": "views",
"CONV_T4": "refine", "WPASS": "grid",
"TERMS_T4": "t4", "LIFT": "mk3"}
def test_serve_env_pins_the_knob_defaults():
"""``tt-model.yaml`` ``serve.env`` pins every load-time knob at its default (the measured optimized path)."""
yaml = pytest.importorskip("yaml")
path = Path(__file__).resolve().parents[3] / "tt-model.yaml"
if not path.is_file():
pytest.skip("no tt-model.yaml next to code/ (installed package or container image)")
env = yaml.safe_load(path.read_text())["serve"]["env"]
pins = KNOBS.serve_env()
assert pins["METEOR_LIFT"] == "mk3"
assert {k: str(env.get(k)) for k in pins} == pins
def test_prepare_the_sample_and_streams(model):
gold = P.pandaset_golden("inputs")
kw = load_sample(P.PANDASET_SAMPLE)
prep = model._prepare(**kw)
assert prep["frame"].sha256() == gold["input_sha256"] and prep["stream_id"] == "pandaset_019"
st = prep["state"]
st.frames = 3
assert model._prepare(**kw)["state"] is st and st.frames == 3 # same stream id
assert model._prepare(**dict(kw, stream={"id": "pandaset_019", "reset": True}))["state"].frames == 0
with pytest.raises(tio.InputError):
model._prepare(points=np.zeros((3, 4), np.float32), **kw)
with pytest.raises(tio.InputError):
model._prepare(**dict(kw, ego_speed=None))
assert model.cfg.input_norm == "onnx" and model.cfg.depth_mean_bins == "log"
def test_prepare_the_shipped_sample(model):
"""The shipped synthetic sample: eight present cameras, its preset, the feed sha256 stored at generation."""
gold = P.shipped_inputs()
kw = load_sample(P.SHIPPED_SAMPLE)
assert kw["calibration"]["cameras"].keys() == set(C.CAMERAS)
prep = model._prepare(**kw)
assert prep["frame"].sha256() == gold["input_sha256"] and prep["stream_id"] == P.SHIPPED_NAME
assert prep["frame"].present.all() and gold["present"] == [1] * 8
def test_postprocess_synthetic(model):
kw = load_sample(P.SHIPPED_SAMPLE)
prep = model._prepare(**kw)
rng = np.random.default_rng(0)
raw = {"hm": np.full((1, 2, 400, 250), -9.0, np.float32), "reg": np.zeros((1, 6, 400, 250), np.float32),
"lane": rng.integers(0, 3, (1, 800, 500)).astype(np.uint8), "ego": np.zeros((1, 42), np.float32),
"tl": np.array([[3.0, 0.0, 0.0, 0.0]], np.float32), "stationary": rng.normal(0, 1, (1, 1, 400, 250)),
"traj": np.zeros((1, 39, 400, 250), np.float32), "lane_logit": rng.normal(0, 1, (1, 9, 800, 500)),
"depth_mean": np.full((1, 8, 54, 96), 5.0, np.float16)}
raw["hm"][0, 0, 150, 125] = 3.0
for si, s in enumerate(C.DET2D_STRIDES):
h, w = C.FEAT_H * 4 // s, C.FEAT_W * 4 // s
raw[f"hm2d_s{si}"] = np.full((1, 8, 10, h, w), -20.0, np.float32)
raw[f"reg2d_s{si}"] = np.zeros((1, 8, 4, h, w), np.float32)
out = model._postprocess(raw, prep, METEOR.validate_params({"seg_fuse": False}))
d = out.to_dict()
assert d["num_detections"] == 1 and d["detections"][0]["center"][0] == pytest.approx(80 - 150 * 0.4)
assert d["traffic_light"]["state"] == "none" and d["meta"]["stream_id"] == P.SHIPPED_NAME
assert prep["state"].frames == 1 and prep["state"].mode.prev == 0
out2 = model._postprocess(raw, prep, METEOR.validate_params({})) # seg fusion with the pose
assert prep["state"].seg.acc is not None and out2.lane.shape == (800, 500)
def test_stored_reference_bodies_agree():
"""``samples/<stem>.reference.json`` (the smoke test's reference) is the CPU reference's /predict body of the
shipped request, consistent with the decoded golden of tests/goldens, and compares clean with itself."""
from tt_meteor.server import smoke_test
gold = P.pandaset_golden("reference")
body = json.loads(P.PANDASET_REFERENCE.read_text())
assert body["model"] == "meteor-p150" and body["num_detections"] == len(gold["boxes3d"]) == 8
assert [round(d["score"], 5) for d in body["detections"]] == [round(b["score"], 5) for b in gold["boxes3d"]]
assert body["plan"]["mode"] == gold["plan_mode"] and len(body["trajectory"]) == 6
assert gold["input_sha256"] == P.pandaset_golden("inputs")["input_sha256"]
metrics, fails = smoke_test.ttaw_smoke.compare_with_reference(body, body, gates=smoke_test.REFERENCE_GATES)
assert fails == [] and metrics["detections"]["recall"] == 1.0 and metrics["lane"]["agreement"] == 1.0
assert set(smoke_test.REQUIRED_KEYS) <= set(body)
assert smoke_test.output_gates(body, "VEHICLE:4,VRU:4") == []
def test_shipped_reference_body():
"""The shipped sample's stored reference (the container smoke's): the documented fields, clean against itself,
and the smoke's default ``--expect`` holds on it."""
from tt_meteor.server import smoke_test
body = json.loads(P.SHIPPED_REFERENCE.read_text())
assert body["model"] == "meteor-p150" and body["frame_id"] == "base_link" and body["timing_ms"] == {}
assert set(smoke_test.REQUIRED_KEYS) <= set(body) and len(body["trajectory"]) == 6
assert body["meta"]["present"] == [True] * 8 and body["meta"]["v0"] == pytest.approx(8.0)
metrics, fails = smoke_test.ttaw_smoke.compare_with_reference(body, body, gates=smoke_test.REFERENCE_GATES)
assert fails == [] and metrics["lane"]["agreement"] == 1.0 and "trajectory" in metrics
assert "{{" not in smoke_test.DEFAULT_EXPECT
assert smoke_test.output_gates(body, smoke_test.DEFAULT_EXPECT) == []
assert smoke_test.DEFAULT_INPUT == P.SHIPPED_SAMPLE
|