| |
| """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) |
| 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) |
| 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()}) |
| 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) |
| 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"] |
|
|
