# 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"]