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