# SPDX-License-Identifier: Apache-2.0 """Host post-processing (``host.postprocess``, ``host.result``, ``host.outputs``; no device): METEOR's C++ rules on synthetic heads (each rule in isolation), and the research decode (``research/meteor/public_data/scripts/ me_public_common.py``, an independent transliteration of ``decode.cpp`` / ``render.cpp``) on the full ORT goldens of four frames when the research tree is present. TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_meteor/tests/test_host_postprocess.py """ from __future__ import annotations import math import sys from pathlib import Path import numpy as np import pytest from tt_meteor.host import postprocess as P from tt_meteor.host.outputs import MeteorOutput from tt_meteor.host.preprocess import MeteorFrame from tt_meteor.host.result import build_output from tt_meteor.reference import config as C from tt_meteor.ttaw.server.smoke import decode_array RESEARCH = Path("/home/ubuntu/experiments/tt-models/research/meteor") F32 = np.float32 def logit(p): return float(np.log(p / (1 - p))) def heads(peaks=(), reg=None): hm = np.full((2, C.DET_H, C.DET_W), -10.0, F32) for c, r, q, p in peaks: hm[c, r, q] = logit(p) return hm, (np.zeros((6, C.DET_H, C.DET_W), F32) if reg is None else reg) def test_peaks_threshold_and_topk(): hm, reg = heads([(0, 100, 100, 0.9), (0, 100, 101, 0.8), (1, 50, 60, 0.149), (1, 50, 70, 0.151)]) boxes = P.decode_boxes_3d(hm, reg) assert [(b.cls, b.cell) for b in boxes] == [(0, (100, 100)), (1, (50, 70))] # neighbour is not a peak hm2, _ = heads([(0, 10, 10, 0.5), (0, 10, 11, 0.5)]) # equal neighbours: both peaks assert len(P.decode_boxes_3d(hm2, reg)) == 2 many = heads([(0, 2 * i, 3, 0.2 + 0.001 * i) for i in range(80)]) top = P.decode_boxes_3d(*many, topk=64) assert len(top) == 64 and top[0].score > top[-1].score def test_box_math_is_the_cpp_float32(): reg = np.zeros((6, C.DET_H, C.DET_W), F32) reg[:, 200, 125] = [0.3, -0.2, math.log(4.0), math.log(1.8), 1.0, 0.0] b = P.decode_boxes_3d(*heads([(0, 200, 125, 0.9)], reg))[0] assert b.x == float(F32(80.0) - (F32(200) + F32(0.3)) * F32(0.4)) assert b.y == float(F32(50.0) - (F32(125) + F32(-0.2)) * F32(0.4)) assert b.l == pytest.approx(4.0) and b.w == pytest.approx(1.8) and b.yaw == pytest.approx(math.pi / 2) def test_class_thresholds_and_range(): boxes = P.decode_boxes_3d(*heads([(0, 100, 50, 0.36), (0, 120, 50, 0.34), (1, 140, 50, 0.16), (0, 396, 50, 0.9)])) kept = P.filter_boxes_3d(boxes) assert sorted(b.cell for b in kept) == [(100, 50), (140, 50)] # x of row 396 = -78.4 < -78 def test_bev_nms_iou_and_containment(): big = P.Box3D(0, 0.9, 10.0, 0.0, 6.0, 3.0, 0.0, (0, 0)) inside = P.Box3D(0, 0.8, 10.0, 0.0, 2.0, 1.0, 0.3, (0, 0)) # IoU 0.11 < 0.3, containment 1.0 > 0.6 other_cls = P.Box3D(1, 0.7, 10.0, 0.0, 2.0, 1.0, 0.0, (0, 0)) far = P.Box3D(0, 0.6, 30.0, 0.0, 4.0, 2.0, 0.0, (0, 0)) shifted = P.Box3D(0, 0.5, 31.0, 0.0, 4.0, 2.0, 0.0, (0, 0)) # IoU 0.6 with far kept = P.bev_nms([shifted, far, other_cls, inside, big]) assert [b.score for b in kept] == [0.9, 0.7, 0.6] assert P.rotated_intersection(far, shifted) == pytest.approx(6.0, rel=1e-5) rot = P.Box3D(0, 0.5, 30.0, 0.0, 4.0, 2.0, math.pi / 2, (0, 0)) assert P.rotated_intersection(far, rot) == pytest.approx(4.0, rel=1e-5) def test_stationary_and_futures(): traj = np.zeros((C.TRAJ_C, C.DET_H, C.DET_W), F32) traj[36:39, 100, 50] = [0.0, 2.0, 1.0] # mode 1 traj[12:24, 100, 50] = np.arange(12, dtype=F32) * 0.01 # tiny motion: |wp6| < 0.5 m b = P.Box3D(0, 0.9, 80 - 100.5 * 0.4, 50 - 50.5 * 0.4, 4, 2, 0, (100, 50)) stat = np.zeros((C.DET_H, C.DET_W), F32) assert P.annotate_boxes([b], stat, traj) is False # constant map: unhealthy -> traj fallback assert b.stationary and b.future_mode == 1 np.testing.assert_allclose(b.future[0], [b.x + 0.0, b.y + 0.01], atol=1e-6) stat = np.random.default_rng(0).normal(0, 1, (C.DET_H, C.DET_W)).astype(F32) stat[100, 50] = -1.0 b2 = P.Box3D(0, 0.9, b.x, b.y, 4, 2, 0, (100, 50)) assert P.annotate_boxes([b2], stat, traj) is True and not b2.stationary assert P.stationary_healthy(np.full(10, 3.0)) is False def test_decode_2d_clip_peaks_hide(): out = {} for si, s in enumerate(C.DET2D_STRIDES): h, w = C.FEAT_H * 4 // s, C.FEAT_W * 4 // s hm = np.full((1, 8, 10, h, w), -20.0, F32) reg = np.zeros((1, 8, 4, h, w), F32) out[f"hm2d_s{si}"], out[f"reg2d_s{si}"] = hm, reg out["hm2d_s0"][0, 2, 1, 10, 20] = 80.0 # clipped to 50 before the peak test out["hm2d_s0"][0, 2, 1, 10, 21] = 60.0 # also 50 after clipping: a tie, both peaks out["hm2d_s1"][0, 2, 7, 5, 5] = 5.0 # road paint: hidden out["reg2d_s1"][0, 2, :, 5, 5] = [0.5, 0.25, math.log(2.0), math.log(3.0)] res = P.decode_boxes_2d(out) assert [(b.cls, b.stride) for b in res[2]] == [(1, 4), (1, 4)] and not any(res[i] for i in (0, 1, 3, 4, 5, 6, 7)) shown = P.decode_boxes_2d(out, keep_hidden=True)[2] paint = [b for b in shown if b.cls == 7][0] assert (paint.cx, paint.cy, paint.w, paint.h) == (5.25 * 8, 5.5 * 8, pytest.approx(16.0), pytest.approx(24.0)) def test_ground_point_and_unknown_obstacles(): # a level forward camera 2 m above the ground at the ego origin (optical z = ego x, optical y = -ego z) T_ego_cam = np.array([[0, 0, 1, 0], [-1, 0, 0, 0], [0, -1, 0, 2.0], [0, 0, 0, 1]], float) T = np.linalg.inv(T_ego_cam) K = np.array([[400.0, 0, 384], [0, 400, 216], [0, 0, 1]]) x, y = P.ground_point(384.0, 216.0 + 40.0, K, T) # 40 px below the horizon: z/x = 2/x = 40/400 assert (x, y) == (pytest.approx(20.0), pytest.approx(0.0)) assert P.ground_point(384.0, 200.0, K, T) is None # above the horizon Ks, Ts = np.stack([K] * 8), np.stack([T] * 8) boxes = [[P.Box2D(0, 0.9, 384.0, 236.0, 20.0, 40.0, 4), P.Box2D(0, 0.8, 386.0, 236.0, 20.0, 40.0, 4), P.Box2D(1, 0.9, 384.0, 236.0, 20.0, 40.0, 4)]] + [[] for _ in range(7)] unk = P.unknown_obstacles(boxes, None, Ks, Ts) assert len(unk) == 1 and unk[0]["x"] == pytest.approx(20.0) and unk[0]["camera"] == "CAM_FRONT_WIDE" def test_mode_selection(): ego = np.zeros(42, F32) ego[36:39] = [0.0, 0.9, 0.0] assert P.select_mode(ego)[0] == 0 # straight preference: margin 0.9 < 1.0 ego[36:39] = [0.0, 1.2, 0.0] assert P.select_mode(ego)[0] == 1 ego[36:39] = [0.0, 0.8, 0.9] assert P.select_mode(ego, prev_mode=1)[0] == 1 # 0.8 + 0.35 > 0.9, margin 1.15 >= 1 plan = P.plan_from_ego(ego, prev_mode=1) assert plan["mode_probs"].sum() == pytest.approx(1.0) and plan["path"].shape == (6, 2) def test_thin_road_edge(): lane = np.zeros((6, 6), np.uint8) lane[:, :2] = 1 # road lane[:, 2:4] = 6 # a 2 px edge band out = P.thin_road_edge(lane) assert (out[:, 2] == 6).all() and (out[:, 3] == 0).all() and (out[:, :2] == 1).all() def _full_outputs(): gold = [("ps019_f0020", RESEARCH / "public_data/golden/ps019/full_f0020.npz"), ("ps090_f0020", RESEARCH / "public_data/golden/ps090/full_f0020.npz"), ("ns0103_f0009", RESEARCH / "public_data/golden/ns0103/full_f0009.npz"), ("valday_f040", RESEARCH / "golden/valday_f040_ort.npz")] return [pytest.param(p, id=n) for n, p in gold] @pytest.mark.parametrize("path", _full_outputs()) def test_matches_the_research_decode(path): scripts = RESEARCH / "public_data" / "scripts" if not path.is_file() or not (scripts / "me_public_common.py").is_file(): pytest.skip("research goldens / scripts not present (they are not shipped)") pytest.importorskip("cv2") sys.path.insert(0, str(scripts)) import me_public_common as R g = np.load(path) out = {k: g[k] for k in g.files if not k.startswith("in_")} frame = MeteorFrame(g["in_imgs"], g["in_K"], g["in_T_cam_ego"], g["in_v0"], np.ones(8, bool)) res = build_output(out, frame, P.PostConfig(thin_road_edge=False), None) rd, healthy = R.decode_3d(out) assert healthy == res.stationary_healthy and len(rd) == len(res.boxes3d) for a, b in zip(sorted(res.boxes3d, key=lambda z: -z.score), rd): assert (a.cls, a.stationary, a.future_mode, tuple(a.cell)) == (b["cls"], b["stationary"], b["future_mode"], tuple(b["cell"])) assert abs(a.score - b["score"]) < 1e-6 and abs(a.x - b["x"]) < 1e-4 and abs(a.y - b["y"]) < 1e-4 assert abs(a.yaw - b["yaw"]) < 1e-5 and np.abs(np.asarray(a.future) - np.asarray(b["future"])).max() < 1e-3 r2 = R.decode_2d(out) for cam in range(C.N_CAMS): assert len(res.boxes2d[cam]) == len(r2[cam]), C.CAMERAS[cam] for a, b in zip(res.boxes2d[cam], r2[cam]): assert a.cls == b["cls"] and abs(a.score - b["score"]) < 1e-6, C.CAMERAS[cam] assert max(abs(a.cx - b["cx"]), abs(a.cy - b["cy"]), abs(a.w - b["w"]), abs(a.h - b["h"])) < 1e-3 assert res.plan["mode"] == R.select_mode(out["ego"]) np.testing.assert_array_equal(res.lane, out["lane"][0]) def test_output_json_and_npz(): hm, reg = heads([(0, 100, 100, 0.9)]) lane = np.random.default_rng(0).integers(0, 9, (1, 800, 500)).astype(np.uint8) ego = np.zeros((1, 42), F32) ego[0, :36] = np.arange(36) out = {"hm": hm[None], "reg": reg[None], "lane": lane, "ego": ego, "tl": np.array([[0.0, 3.0, 0.0, 0.0]], F32), "stationary": np.zeros((1, 1, 400, 250), F32), "traj": np.zeros((1, 39, 400, 250), F32), "seg2d": np.zeros((1, 8, 108, 192), np.uint8), "depth": np.ones((1, 8, 108, 192), np.uint8), "depth_mean": np.full((1, 8, 54, 96), 3.0, np.float16), "occ": np.zeros((1, 10, 16, 200, 200), F32), "risk": np.zeros((1, 1, 400, 250), F32), "lane_logit": np.zeros((1, 9, 800, 500), F32)} for si, s in enumerate(C.DET2D_STRIDES): h, w = C.FEAT_H * 4 // s, C.FEAT_W * 4 // s out[f"hm2d_s{si}"] = np.full((1, 8, 10, h, w), -20.0, F32) out[f"reg2d_s{si}"] = np.zeros((1, 8, 4, h, w), F32) frame = MeteorFrame(np.zeros((1, 8, 3, 432, 768), np.uint8), np.tile(np.eye(3, dtype=F32), (1, 8, 1, 1)), np.tile(np.eye(4, dtype=F32), (1, 8, 1, 1)), np.array([5.0], F32), np.ones(8, bool)) res = build_output(out, frame, P.PostConfig(thin_road_edge=False), None, heads=True) assert isinstance(res, MeteorOutput) d = res.to_dict() assert d["num_detections"] == 1 and d["detections"][0]["label"] == "VEHICLE" assert d["trajectory"] == np.arange(12, dtype=float).reshape(6, 2).tolist() and d["columns"] == ["x", "y"] assert d["traffic_light"]["state"] == "green" and "heads" not in d lane_back = decode_array(d["lane"]) assert lane_back.shape == (800, 500) and list(lane_back.values) == lane[0].ravel().tolist() n = res.to_dict("npz") assert set(n["heads"]) == {"seg2d", "depth", "depth_mean", "occupancy", "risk", "stationary"} assert decode_array(n["heads"]["occupancy"]).shape == (16, 200, 200) def test_candidates_sparse_matches_vec(): """OPT round 4 R4.4: ``HOST_POST=sparse`` gives exactly the ``vec`` candidates (scores, classes, rows, columns, order), with and without the clip, with plateaus / ties, border peaks and saturated logits.""" import numpy as np from tt_meteor.host import postprocess as PP rng = np.random.default_rng(3) for shape, clip in (((8, 10, 27, 48), 30.0), ((1, 2, 40, 25), None), ((3, 4, 16, 16), 2.0)): v = (rng.standard_normal(shape) * 3).astype(np.float32) v[..., 0, :] += 2.0 # border peaks v.flat[::97] = 40.0 # saturated (clipped) values v[..., 5:7, 5:7] = 1.25 # plateaus (ties) for thresh, topk in ((0.3, 48), (0.15, 64), (0.9, 5)): want = PP._candidates_nd(v, thresh, topk, clip=clip) got = PP._candidates_sparse(v, thresh, topk, clip=clip) assert got == want PP.set_host_post("sparse") assert PP.host_post_mode() == "sparse" PP.set_host_post(None)