meteor-p150 / code /tt_meteor /tests /test_host_postprocess.py

tt-model push meteor-p150 (container)

8eeb550 verified 6 days ago

12.6 kB

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

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