File size: 12,606 Bytes
51defdc 8eeb550 | 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 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 | # 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)
|