File size: 8,457 Bytes
fe51a63 | 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 | # SPDX-License-Identifier: Apache-2.0
"""Host tests of the API wiring (no device): identity and runtime parameters; the host half of the hooks
(``_build_host``, ``_prepare``, ``_postprocess``) with the fp32 torch reference network in the device's place,
against the stored CPU references of the shipped synthetic samples and the staged PandaSet sample (compared with the container smoke's own comparator,
``ttaw.server.smoke``); the calibration contract (crop transform required while the box is on, presets, overrides,
refusals); a frame the node refuses (profile check) gives an empty output with the node's message.
TT_VISIBLE_DEVICES=none python -m pytest -q -p fake_ttnn_plugin code/tt_frnet/tests/test_api_host.py
"""
from __future__ import annotations
import json
import numpy as np
import pytest
from tt_frnet import io as tio
from tt_frnet.api import LABELS, FRNet
from tt_frnet.host.postprocess import RUNTIME_PARAMS, FRNetSegmentation
from tt_frnet.tests import _data
from tt_frnet.ttaw.server import smoke
def test_identity_and_runtime_params():
assert tuple(FRNet.VARIANTS) == ("ot128", "qt128") and FRNet.DEFAULT_VARIANT == "ot128"
assert tuple(FRNet.LABELS) == LABELS and len(LABELS) == 27 and FRNet.LABELS[0] == "drivable_surface"
assert FRNet.RUNTIME_PARAMS is RUNTIME_PARAMS and FRNet.EXTRA_INPUTS == ()
assert FRNet.DEFAULT_REVISION == "a53a1c11b0b28d31fd50a03abef5db0e830a5b3c" and FRNet.DEFAULT_TAG == "v2.0"
p = FRNet.validate_params({})
assert p == {"class_probability_threshold": 0.05, "filter_classes": "drivable_surface"}
assert FRNet.validate_params({"class_probability_threshold": "0.5"})["class_probability_threshold"] == 0.5
assert FRNet.validate_params({"filter_classes": ""})["filter_classes"] == ""
for bad in ({"class_probability_threshold": 1.5}, {"class_probability_threshold": True}, {"score_threshold": 0.1}):
with pytest.raises(tio.InputError):
FRNet.validate_params(bad)
def _host_model(variant: str = "ot128") -> FRNet:
m = FRNet.__new__(FRNet) # no device: the host half only
m.weights_path = _data.weights_dir()
m.variant = variant
m._build_host()
return m
@pytest.fixture(scope="module")
def FRNetTorch():
torch = pytest.importorskip("torch")
pytest.importorskip("onnx")
from tt_frnet.reference.model import FRNetTorch
torch.set_num_threads(4)
return FRNetTorch
@pytest.mark.parametrize("case", ["pandaset.ot128", "pandaset.qt128", "shipped.ot128", "shipped.qt128"])
def test_hooks_reproduce_the_stored_reference(FRNetTorch, case):
"""The API hooks with the torch fp32 reference in place of the device reproduce the stored ONNX Runtime reference
bodies: the staged PandaSet sample and the shipped synthetic samples (the container smoke's references)."""
kind, _, variant = case.partition(".")
m = _host_model(variant)
if kind == "pandaset":
sample, calib, ref_path = (_data.need_pandaset(), _data.SAMPLE_CALIB,
_data.SAMPLE_DIR / f"{_data.SAMPLE_STEM}.{variant}.reference.json")
else:
sample, calib = _data.shipped(variant), _data.shipped_calib(variant)
ref_path = _data.SAMPLES / f"synthetic_{variant}.{variant}.reference.json"
prep = m._prepare(points=sample, calibration=calib)
assert prep.skipped is None and prep.plan is not None and prep.plan.n == prep.inputs.num_points
inp = prep.inputs
probs = FRNetTorch(m.frnet_weights)(inp.points, inp.coors, inp.voxel_coors, inp.inverse_map)
out = m._postprocess(probs, prep, m.validate_params({}))
assert isinstance(out, FRNetSegmentation) and out.frame_id == "sensor" and out.model == "frnet-p150"
body = out.to_dict()
ref = json.loads(ref_path.read_text())
metrics, fails = smoke.compare_with_reference(body, ref, gates={"min_label_agreement": 0.9999,
"max_abs_err": 1e-4})
assert fails == [], (fails, metrics)
assert body["num_points"] == ref["num_points"] and body["num_points_raw"] == ref["num_points_raw"]
assert body["meta"]["variant"] == variant and body["meta"]["ego_crop"]["enabled"]
assert body["meta"]["num_unique_coors"] == ref["meta"]["num_unique_coors"]
np.testing.assert_array_equal(out.input_index, np.flatnonzero(inp.keep_mask))
def test_prepare_calibration_contract():
m = _host_model()
pts = np.c_[np.random.default_rng(0).uniform(-40, 40, (8000, 3)), np.full(8000, 7.0)].astype(np.float32)
with pytest.raises(tio.InputError, match="this model needs 'points'"):
m._prepare(points=None)
with pytest.raises(tio.InputError, match="ego crop box"):
m._prepare(points=pts) # the box is on, no transform: refused
with pytest.raises(tio.InputError, match="one sensor-frame sweep"):
m._prepare(points=pts, sweeps=[{"points": pts}], calibration={"ego_crop_box": None})
with pytest.raises(tio.InputError, match=r"\['images'\]"):
m._prepare(points=pts, images=np.zeros((2, 2, 3), np.uint8), calibration={"ego_crop_box": None})
with pytest.raises(tio.InputError, match="calibration must be an object"):
m._prepare(points=pts, calibration=[1, 2])
assert m._prepare(points=pts, sweeps=[], stream=None, calibration={"ego_crop_box": None}).inputs is not None
with pytest.raises(tio.InputError, match="unknown keys"):
m._prepare(points=pts, calibration={"T_lidar": np.eye(4).tolist()})
with pytest.raises(tio.InputError, match="proper rotation"):
m._prepare(points=pts, calibration={"T_base_link_from_sensor": (2 * np.eye(4))[:3].tolist()}) # 3x4, scaled
with pytest.raises(tio.InputError, match="preset"):
m._prepare(points=pts, calibration={"preset": "no_such_rig"})
off = m._prepare(points=pts, calibration={"ego_crop_box": None})
assert off.crop["enabled"] is False and off.inputs.num_points_raw == 8000
on = m._prepare(points=pts, calibration={"T_base_link_from_sensor": {"x": 0, "y": 0, "z": 0, "roll": 0,
"pitch": 0, "yaw": 0},
"ego_crop_box": [-10, -10, -10, 10, 10, 10]})
inside = (np.abs(pts[:, :3]) < 10).all(1)
assert on.crop["enabled"] and on.inputs.num_points_raw == int((~inside).sum())
preset = m._prepare(points=_data.shipped("ot128"), calibration=_data.shipped_calib("ot128"))
assert preset.crop["enabled"] and preset.inputs.num_points_raw == 52613 and preset.inputs.num_points == 52613
_data.need_pandaset()
preset = m._prepare(points=_data.SAMPLE, calibration=_data.SAMPLE_CALIB)
assert preset.inputs.num_points_raw == 106501 and preset.inputs.num_points == 145961
def test_a_frame_the_node_refuses_gives_an_empty_output():
m = _host_model()
pts = np.c_[np.random.default_rng(1).uniform(-40, 40, (2000, 3)), np.zeros(2000)].astype(np.float32)
prep = m._prepare(points=pts, calibration={"ego_crop_box": None})
assert prep.skipped is not None and "in input pointcloud does not match the profile" in prep.skipped
raw = m._forward(prep) # no device touched
out = m._postprocess(raw, prep, m.validate_params({}))
body = out.to_dict()
assert raw is None and body["num_points"] == 0 and "does not match the profile" in body["meta"]["skipped"]
with pytest.raises(tio.InputError, match="filter_classes"):
m._postprocess(raw, prep, m.validate_params({"filter_classes": "road"}))
def test_filter_params_change_only_the_filter_mask():
m = _host_model("qt128")
prep = m._prepare(points=_data.shipped("qt128"), calibration=_data.shipped_calib("qt128"))
rng = np.random.default_rng(2)
probs = rng.dirichlet(np.ones(27), size=prep.inputs.num_points).astype(np.float32)
a = m._postprocess(probs, prep, m.validate_params({}))
b = m._postprocess(probs, prep, m.validate_params({"class_probability_threshold": 0.9,
"filter_classes": "drivable_surface,sidewalk"}))
c = m._postprocess(probs, prep, m.validate_params({"filter_classes": ""}))
np.testing.assert_array_equal(a.label_ids, b.label_ids)
assert c.filter_keep.all() and b.filter_keep.sum() > a.filter_keep.sum()
assert b.meta["filter"]["classes"] == ["drivable_surface", "sidewalk"]
|