"""The sar3d processor on the real CUDA backend: notebook reference and contracts. Every case runs build/native_tests/backprojection_fixture (BACKPROJECTION_FIXTURE overrides it). Without a usable GPU the processor publishes sar3d_status 3: then only that warning and the box rules, which are host code, are checked, and the remaining tests are skipped. The notebook is needed only to regenerate the references with generate_backprojection_reference.py. """ from __future__ import annotations import copy import json import os from pathlib import Path import re import subprocess import unittest import numpy as np ROOT = Path(__file__).resolve().parents[2] FIXTURE = Path(os.environ.get("BACKPROJECTION_FIXTURE", ROOT / "build/native_tests/backprojection_fixture")) REFERENCE = Path(__file__).with_name("sar3d_reference.npz") VIEWS = ("sar3d_front", "sar3d_side", "sar3d_top", "sar3d_oblique") def run(request: dict | list[dict]) -> dict[str, dict]: output = subprocess.run([str(FIXTURE)], input=json.dumps(request), text=True, capture_output=True, timeout=30, check=True) result = json.loads(output.stdout) if isinstance(request, list): result = result[-1] return {item["name"]: item for item in result["payloads"]} def status(payloads: dict[str, dict]) -> tuple[int, str]: table = payloads["sar3d_status"]["table"] return int(table[0]), bytes(map(int, table[1:])).decode("utf-8") def reference_cases(archive) -> list[dict]: """Requests of the reference archive, one per notebook parameter case.""" cases = [] for case in json.loads(archive["cases"].item()): request = json.loads(archive["request"].item()) request.update(case["settings"]) for member, windows in zip([*request["previous"], request["latest"]], case["windows"], strict=True): for trace, (start, end) in zip(member["traces"], windows, strict=True): trace.update(start=start, end=end) cases.append(request) return cases class Sar3dTest(unittest.TestCase): @classmethod def setUpClass(cls) -> None: with np.load(REFERENCE, allow_pickle=False) as archive: cls.reference = {key: archive[key] for key in archive.files} cls.cases = reference_cases(cls.reference) cls.scene_payloads = run(cls.cases[0]) code, message = status(cls.scene_payloads) cls.unavailable = message if code == 3 else None def scene(self, **settings) -> dict: if self.unavailable: self.skipTest(self.unavailable) request = copy.deepcopy(self.cases[0]) request.update(settings) return request def test_unavailable_gpu_is_a_visible_warning(self) -> None: code, message = status(self.scene_payloads) if self.unavailable: self.assertIn("CUDA", message) self.assertFalse(any(item["image"] for item in self.scene_payloads.values())) else: self.assertEqual(code, 0, message) def test_cuda_matches_notebook(self) -> None: if self.unavailable: self.skipTest(self.unavailable) for index, request in enumerate(self.cases): with self.subTest(case=index): payloads = run(request) self.assertEqual(status(payloads)[0], 0, status(payloads)) for view, horizontal, vertical in (("front", "x", "y"), ("side", "z", "y"), ("top", "x", "z")): payload = payloads[f"sar3d_{view}"] expected = self.reference[f"case_{index}_{view}"] np.testing.assert_allclose(payload["x"], self.reference[f"case_{index}_{horizontal}"], rtol=2e-6, atol=1e-6) np.testing.assert_allclose(payload["y"], self.reference[f"case_{index}_{vertical}"], rtol=2e-6, atol=1e-6) np.testing.assert_allclose(np.asarray(payload["image"]).reshape(expected.shape), expected, rtol=2e-3, atol=3e-4) widths = self.reference.get(f"case_{index}_target_width") if widths is None: self.assertNotIn("sar3d_target_width", payloads) else: np.testing.assert_allclose(payloads["sar3d_target_width"]["table"], widths, rtol=2e-3, atol=2e-4) peak = payloads["sar3d_peak"]["table"] expected_peak = self.reference[f"case_{index}_peak"] np.testing.assert_allclose(peak[1:], expected_peak[1:], rtol=0, atol=2e-6) np.testing.assert_allclose(peak[0], expected_peak[0], rtol=2e-3, atol=0) self.check_display(index, payloads) def check_display(self, index: int, payloads: dict[str, dict]) -> None: """Background, CLEAN objects and boxes against sar_prepare_display. Steps at the noise level may pick a neighbouring A-scan sample where two samples are equal to float32 precision; the reference scenes were checked to keep their decisions with the kernels compiled without fused multiply-adds. """ background = self.reference.get(f"case_{index}_background") if background is None: self.assertNotIn("sar3d_background", payloads) else: np.testing.assert_allclose(payloads["sar3d_background"]["table"], background, rtol=2e-3, atol=0) objects = self.reference.get(f"case_{index}_objects") if objects is None: self.assertNotIn("sar3d_objects", payloads) self.assertNotIn("sar3d_marks", payloads) else: table = np.reshape(payloads["sar3d_objects"]["table"], (-1, 4)) np.testing.assert_allclose(table[:, :3], objects[:, :3], rtol=0, atol=2e-6) np.testing.assert_allclose(table[:, 3], objects[:, 3], rtol=2e-3, atol=0) np.testing.assert_allclose(np.reshape(payloads["sar3d_marks"]["table"], (-1, 7)), self.reference[f"case_{index}_marks"], rtol=0, atol=2e-6) clean = self.reference.get(f"case_{index}_clean") if clean is not None: message = status(payloads)[1] match = re.search(r"\((\d+) subtractions?, residual maximum ([\d.]+) backgrounds\)", message) assert match is not None, message self.assertEqual(int(match[1]), clean[0]) self.assertAlmostEqual(float(match[2]), clean[1], delta=0.051) def test_clean_needs_a_one_frame_aperture(self) -> None: # The actual aperture decides, as in sar_clean_frame: three frames skip CLEAN, while the # first frame of a run is a one-frame aperture even with SAR_FRAME_RADIUS = 2. skipped = run(self.scene(clean=True)) code, message = status(skipped) self.assertEqual(code, 0, message) self.assertIn("CLEAN skipped: it needs a one-frame aperture, this one has 3 frames", message) self.assertNotIn("sar3d_objects", skipped) code, message = status(run(self.scene(clean=True, previous=[]))) self.assertEqual(code, 0, message) self.assertIn("; CLEAN: ", message) def test_boxes_follow_the_notebook_rules(self) -> None: # Host code only, so it runs without a GPU as well. request = json.loads(self.reference["marks_request"].item()) output = subprocess.run([str(FIXTURE)], input=json.dumps(request), text=True, capture_output=True, timeout=30, check=True) np.testing.assert_allclose(json.loads(output.stdout)["marks"], self.reference["marks"], rtol=0, atol=1e-9) def test_invalid_input_is_an_error_with_a_reason(self) -> None: def no_traces(request): request["latest"]["traces"] = [] def empty_window(request): request["latest"]["traces"][0]["end"] = request["latest"]["traces"][0]["start"] def falling_frequency(request): request["latest"]["traces"][0]["frequency"].reverse() def overflow(request): for trace in request["latest"]["traces"]: samples = np.asarray(trace["s21"]) trace["s21"] = (samples / np.abs(samples).max() * 1e38).tolist() for name, settings, damage in ( ("no traces", {}, no_traces), ("empty capture window", {}, empty_window), ("falling frequency", {}, falling_frequency), ("volume beyond float32", {"previous": [], "radius": 0, "normalize": False, "equalize": False}, overflow), ("grid beyond 32-bit indexing", {"voxel": [1e-8] * 3}, None), ("vertical look angle", {"pitch": 90}, None), ("result larger than the ring slot", {"slot": 1024}, None), ): with self.subTest(name): request = self.scene(**settings) if damage: damage(request) code, message = status(run(request)) self.assertEqual(code, 2, message) self.assertTrue(message) def test_views_are_normalized_at_the_peak_of_the_reference_planes(self) -> None: # Fine grids spread every Y plane over all threads of the per-plane maximum. for voxel in (.05, .06, .07, .08, .09, .11): with self.subTest(voxel=voxel): payloads = run(self.scene(voxel=[voxel] * 3)) self.assertEqual(status(payloads)[0], 0, status(payloads)) front = payloads["sar3d_front"] x, y = np.asarray(front["x"]), np.asarray(front["y"]) image = np.where(((y >= y[0] + .3) & (y <= y[-1] - .3))[:, None], np.reshape(front["image"], (len(y), len(x))), -1.) row, column = np.unravel_index(np.argmax(image), image.shape) self.assertAlmostEqual(float(image[row, column]), 1., places=5) np.testing.assert_allclose(payloads["sar3d_peak"]["table"][1:3], [x[column], y[row]], rtol=0, atol=1e-6) def test_cached_aperture_updates_when_previous_mean_context_changes(self) -> None: request = self.scene(subtract_mean=True, mean_count=1) shorter = copy.deepcopy(request) shorter["previous"] = shorter["previous"][-1:] cached, fresh = run([request, shorter]), run(shorter) self.assertEqual(status(cached)[0], 0, status(cached)) for name in VIEWS: np.testing.assert_allclose(cached[name]["image"], fresh[name]["image"], rtol=2e-3, atol=3e-4) def test_empty_signal_is_reported_without_a_peak(self) -> None: request = self.scene(previous=[], radius=0) for trace in request["latest"]["traces"]: trace["s21"] = [[0, 0]] * len(trace["frequency"]) payloads = run(request) self.assertEqual(status(payloads)[0], 1) self.assertNotIn("sar3d_peak", payloads) self.assertNotIn("sar3d_background", payloads) for name in VIEWS: self.assertFalse(np.any(payloads[name]["image"])) def test_weak_signal_stays_finite(self) -> None: request = self.scene(previous=[], radius=0, normalize=False) for trace in request["latest"]["traces"]: trace["s21"] = (np.asarray(trace["s21"]) * 1e-38).tolist() payloads = run(request) self.assertEqual(status(payloads)[0], 0, status(payloads)) for name in VIEWS: values = np.asarray(payloads[name]["image"]) self.assertTrue(np.isfinite(values).all()) self.assertGreater(values.max(), 0) self.assertLess(values.max(), 1e-10) class GprTest(unittest.TestCase): def test_changing_tx_rx_counts_rebuilds_the_cached_grid(self) -> None: with np.load(REFERENCE.with_name("gpr_reference.npz"), allow_pickle=False) as reference: base = json.loads(reference["case_0_request"].item()) requests = [] for tx, rx in ((2, 1), (1, 2)): request = copy.deepcopy(base) request.update(tx=[[i, 0, 0, 0] for i in range(tx)], rx=[[i, 0, 0, 0] for i in range(rx)]) request["latest"]["traces"] = [trace for trace in request["latest"]["traces"] if trace["output"] < tx and trace["input"] < rx] requests.append(request) np.testing.assert_allclose(run(requests)["gpr_accumulator"]["image"], run(requests[-1])["gpr_accumulator"]["image"]) def test_coherent_ground_bp_matches_horns_and_notebook_equalizer(self) -> None: with np.load(REFERENCE.with_name("gpr_reference.npz"), allow_pickle=False) as reference: for i in range(3): with self.subTest(case=i): request = json.loads(reference[f"case_{i}_request"].item()) payload = run(request)["gpr_accumulator"] expected = reference[f"case_{i}_image"] np.testing.assert_allclose(np.asarray(payload["image"]).reshape(expected.shape), expected, rtol=2e-4, atol=3e-5) np.testing.assert_allclose(payload["y"], np.linspace(.1, request["max_depth"], expected.shape[0]), rtol=1e-6, atol=1e-7) if __name__ == "__main__": unittest.main()