# SPDX-License-Identifier: Apache-2.0 """Host pre-processing (``host.preprocess`` / ``host.inputs``) against the research reference and the SPEC. TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_bevdet/tests/test_host_preprocess.py The research reference (``research/bevdet/scripts/bevdet_common.py``: ``node_image_transport`` + ``preprocess``, verified against the golden network inputs, SPEC section 6) is the oracle; those tests skip where the workspace research tree is absent (e.g. inside the container image). """ from __future__ import annotations import json import numpy as np import pytest from tt_bevdet.host import inputs as hin from tt_bevdet.host import preprocess as hp from tt_bevdet.reference.config import BEVDetConfig, default_config from tt_bevdet.tests import paths CFG = default_config() @pytest.fixture(scope="module") def bc(): mod = paths.import_research() if mod is None: pytest.skip("research reference research/bevdet/scripts not present") return mod @pytest.fixture(scope="module") def research_cfg(bc): yaml = pytest.importorskip("yaml") cfg_dir = paths.autoware_config_dir() if cfg_dir is None: pytest.skip("autoware_universe checkout not present") return bc.Cfg(yaml.safe_load((cfg_dir / "bevdet_r50_4dlongterm_depth.yaml").read_text())) def test_gather_index_matches_spec(): rows, cols = hp.gather_index(CFG) assert rows[:3].tolist() == [318, 320, 323] and rows[-1] == 898 and len(set(rows.tolist())) == 256 assert cols[:4].tolist() == [0, 2, 5, 7] and cols[-1] == 1598 and len(set(cols.tolist())) == 704 assert float(CFG.resize) == float(np.float32(0.44)) def test_crop_and_normalize_equal_research(bc, research_cfg): rng = np.random.default_rng(1) planes = rng.integers(0, 256, size=(6, 3, 900, 1600), dtype=np.uint8) crop = hp.crop_u8(planes, CFG) assert crop.shape == (6, 3, 256, 704) and crop.dtype == np.uint8 np.testing.assert_array_equal(hp.normalize(crop, CFG), bc.preprocess(planes, research_cfg)) @pytest.mark.parametrize("hw", [(900, 1600), (1080, 1920), (720, 1280)]) def test_node_transport_and_crop_equal_research(bc, research_cfg, hw): pytest.importorskip("cv2") rng = np.random.default_rng(hw[0]) rgb = [rng.integers(0, 256, size=hw + (3,), dtype=np.uint8) for _ in range(2)] bgr = [np.ascontiguousarray(x[:, :, ::-1]) for x in rgb] ours = np.stack([hp.node_transport(x, CFG) for x in rgb]) np.testing.assert_array_equal(ours, bc.node_image_transport(bgr, research_cfg)) fast = np.stack([hp.crop_from_rgb(x, CFG) for x in rgb]) np.testing.assert_array_equal(fast, hp.crop_u8(ours, CFG)) prepared = hp.prepare_images(rgb, CFG) np.testing.assert_array_equal(prepared.crop_u8, fast) assert prepared.scale[0] == (1600 / hw[1], 900 / hw[0]) def test_scale_intrinsics_pixel_centre_convention(): K = np.array([[1000.0, 0.0, 959.5], [0.0, 1000.0, 539.5], [0.0, 0.0, 1.0]]) k2 = hp.scale_intrinsics(K, 1600 / 1920, 900 / 1080) assert k2[0, 0] == pytest.approx(1000 * 1600 / 1920) and k2[1, 1] == pytest.approx(1000 * 900 / 1080) assert k2[0, 2] == pytest.approx(799.5) and k2[1, 2] == pytest.approx(449.5) # centre stays the centre def test_prepare_frame_rejects_bad_camera_sets(): img = np.zeros((900, 1600, 3), np.uint8) cal = {"cameras": {c: {"intrinsics": np.eye(3).tolist(), "T_ref_from_camera": np.eye(4).tolist()} for c in CFG.camera_order}} from tt_bevdet.ttaw.io import InputError with pytest.raises(InputError, match="missing"): hin.prepare_frame({c: img for c in CFG.camera_order[:5]}, cal, CFG) with pytest.raises(InputError, match="needs 'images'"): hin.prepare_frame(None, cal, CFG) with pytest.raises(InputError, match="needs 'intrinsics'"): hin.prepare_frame({c: img for c in CFG.camera_order}, None, CFG) ok = hin.prepare_frame({c: img for c in reversed(CFG.camera_order)}, cal, CFG) assert [c.name for c in ok.cameras] == list(CFG.camera_order) and ok.crop_u8.shape == (6, 3, 256, 704) def test_shipped_sample_decodes_to_its_crop(): """The shipped synthetic sample (PNG, lossless) decodes to exactly the uint8 crop it was generated for (sha256 in the manifest), with the calibration of its preset (``calib/synthetic_6cam.json``, 1600x900: K unscaled).""" import hashlib man = json.loads(paths.SHIPPED_SAMPLE.read_text()) kw = hin.load_sample(paths.SHIPPED_SAMPLE) frame = hin.prepare_frame(kw["images"], kw["calibration"], CFG) assert frame.crop_u8.shape == (6, 3, 256, 704) assert hashlib.sha256(np.ascontiguousarray(frame.crop_u8).tobytes()).hexdigest() == man["crop_u8_sha256"] preset = json.loads((paths.PKG_DIR / "calib" / "synthetic_6cam.json").read_text()) for i, name in enumerate(CFG.camera_order): np.testing.assert_array_equal(frame.calib.K[i], np.asarray(preset["cameras"][name]["intrinsics"], np.float32)) T = np.asarray(preset["cameras"][name]["T_ref_from_camera"], np.float64) np.testing.assert_array_equal(frame.calib.R[i], T[:3, :3].astype(np.float32)) np.testing.assert_array_equal(frame.calib.t[i], T[:3, 3].astype(np.float32)) def test_pandaset_sample_matches_research_inputs(): """The PandaSet sample (staged, not shipped) decodes (PIL) to the exact crops and calibration of the research manifest ``autoware__ps019_f38-40`` (OpenCV decode + cv::resize + K scaled to 1600x900).""" pytest.importorskip("cv2") pub = paths.public_data() sample = paths.require_pandaset() / "pandaset_019_f38.json" kw = hin.load_sample(sample) frame = hin.prepare_frame(kw["images"], kw["calibration"], CFG) assert frame.crop_u8.shape == (6, 3, 256, 704) ref = pub / "pandaset" / "inputs" / "autoware__ps019_f38-40" / "f38.npz" if pub else None if ref is None or not ref.is_file(): pytest.skip("research PandaSet inputs not present") with np.load(ref) as g: np.testing.assert_array_equal(frame.crop_u8, g["crop_u8"]) np.testing.assert_array_equal(frame.calib.K, g["K"].astype(np.float32)) np.testing.assert_array_equal(frame.calib.R, g["cam2ego_R"].astype(np.float32)) np.testing.assert_array_equal(frame.calib.t, g["cam2ego_t"].astype(np.float32)) quirk = hin.prepare_frame(kw["images"], kw["calibration"], CFG, scale_intrinsics=False) raw = json.loads(sample.read_text())["calibration"] # inline: the preset's cameras np.testing.assert_array_equal(quirk.calib.K[1], np.asarray(raw["cameras"]["CAM_FRONT"]["intrinsics"], np.float32)) def test_config_matches_autoware_files(): yaml = pytest.importorskip("yaml") cfg_dir = paths.autoware_config_dir() if cfg_dir is None: pytest.skip("autoware_universe checkout not present") cfg = BEVDetConfig.from_yaml(cfg_dir / "bevdet_r50_4dlongterm_depth.yaml", cfg_dir / "bevdet.param.yaml") assert cfg.summary() == CFG.summary() model = yaml.safe_load((cfg_dir / "bevdet_r50_4dlongterm_depth.yaml").read_text()) assert model == CFG.model_yaml # the embedded copy is the file assert (cfg.n_depth, cfg.bev_xyz, cfg.feat_hw, cfg.bev_in_channels) == (118, (128, 128, 1), (16, 44), 720) assert cfg.autoware_labels == ("CAR", "TRUCK", "UNKNOWN", "BUS", "TRAILER", "UNKNOWN", "MOTORCYCLE", "BICYCLE", "PEDESTRIAN", "UNKNOWN")