diff --git a/camera/blur_pipeline.py b/camera/blur_pipeline.py index 540b411..7525b32 100644 --- a/camera/blur_pipeline.py +++ b/camera/blur_pipeline.py @@ -304,6 +304,41 @@ def _edge_detect_ref(rgba_uint8, height, width): return _add_weighted_cv_ref(rgba_uint8, mask_rgba, 1, 1, 0) +def _gaussian_blur3x3_ref(gray_uint8, height, width): + """Numpy equivalent of the on-device Gaussian blur stage in ``blur()``. + + ``filter_kernel_buff`` holds ``[[256,512,256],[512,1024,512],[256,512,256]]`` + (int16). Per the module docstring that's the unity-gain kernel + ``[[1,2,1],[2,4,2],[1,2,1]]`` (sum 16) left-shifted by 8 for fixed-point + precision, right-shifted back down by 12 total after the convolution -- + the extra precision bits cancel out, so the net math is a plain weighted + mean with BORDER_REPLICATE, same border handling ``_filter2d_cv_ref`` + uses for the (unrelated) Laplacian edge kernel above. + """ + img = gray_uint8.reshape(height, width).astype(np.float32) + padded = np.pad(img, 1, mode="edge") + kernel = np.array([[1, 2, 1], [2, 4, 2], [1, 2, 1]], dtype=np.float32) / 16.0 + out = np.zeros((height, width), dtype=np.float32) + for dy in range(3): + for dx in range(3): + out += kernel[dy, dx] * padded[dy : dy + height, dx : dx + width] + return np.clip(np.round(out), 0, 255).astype(np.uint8) + + +def _blur_ref(rgba_uint8, height, width): + """End-to-end reference mirroring the ``blur()`` design's actual pipeline: + rgba2gray -> 3x3 Gaussian blur -> gray2rgba, combined via add_weighted + with alpha=1 (register value 16384 == Q14 unity gain, see the + "GAIN FIX" module docstring) and beta=0 (register value 0) -- unlike + edge_detect's highlighted-edges-over-original composite, the original + RGBA input is NOT blended back in; the output is the blur alone. + """ + gray = _rgba2gray_ref(rgba_uint8, height, width) + blurred = _gaussian_blur3x3_ref(gray, height, width) + mask_rgba = _gray2rgba_ref(blurred) + return _add_weighted_cv_ref(mask_rgba, rgba_uint8, 1, 0, 0) + + # Matches the C++ test.cpp's ``epsilon = 2.0`` tolerance on # ``error_per_pixel = sum(abs(actual - golden)) / num_pixels``. _EPSILON = 2.0 @@ -317,10 +352,10 @@ def _run_and_verify(opts): b_t = iron.zeros(16 * 16, dtype=np.int32, device="npu") out_t = iron.zeros(tensor_size, dtype=np.int8, device="npu") - edge_detect(in_t, b_t, out_t, **_compile_kwargs(opts)) + blur(in_t, b_t, out_t, **_compile_kwargs(opts)) in_uint8 = in_np.view(np.uint8) - expected_uint8 = _edge_detect_ref(in_uint8, opts.height, opts.width) + expected_uint8 = _blur_ref(in_uint8, opts.height, opts.width) actual = out_t.numpy().view(np.uint8) n_diff = int(np.sum(actual != expected_uint8)) @@ -339,7 +374,7 @@ def _run_and_verify(opts): def main(): opts = _make_argparser().parse_args() run_design_cli( - edge_detect, + blur, opts, compile_kwargs=_compile_kwargs, run_and_verify=_run_and_verify, diff --git a/camera/test_blur_pipeline_cli_symbol.py b/camera/test_blur_pipeline_cli_symbol.py new file mode 100644 index 0000000..1b9d653 --- /dev/null +++ b/camera/test_blur_pipeline_cli_symbol.py @@ -0,0 +1,68 @@ +# Regression guard for the blur_pipeline CLI / self-verify design symbol. +# +# blur_pipeline.py was adapted from AMD's `edge_detect` IRON example. The design +# it actually defines is `blur` (see `def blur(...)` and +# `from blur_pipeline import blur` in npu_camera_daemon.py / test_blur.py), but +# `_run_and_verify()` and `main()` still called the old name `edge_detect`, +# which is never defined or imported in this module. Because the module imports +# `aie.iron` at the top, a host without the NPU toolchain fails at that import +# first (so CI's import-boundary check stays green) -- but on a real NPU box +# `python blur_pipeline.py` reaches main() and dies with +# `NameError: name 'edge_detect' is not defined` instead of running the design. +# +# This test is pure-AST (no aie/NPU toolchain, no numpy) so it runs anywhere, +# and it asserts the design symbol the CLI hands to the NPU is one the module +# actually defines. +import ast +import builtins +import os + +HERE = os.path.dirname(os.path.abspath(__file__)) +SRC = os.path.join(HERE, "blur_pipeline.py") + + +def _module_bound_names(tree): + names = set() + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + names.add(node.name) + elif isinstance(node, (ast.Import, ast.ImportFrom)): + for alias in node.names: + names.add(alias.asname or alias.name.split(".")[0]) + elif isinstance(node, ast.Assign): + for tgt in node.targets: + if isinstance(tgt, ast.Name): + names.add(tgt.id) + return names + + +def _cli_referenced_names(tree): + """Bare names called (or handed to run_design_cli) inside the CLI paths.""" + refs = set() + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name in ("_run_and_verify", "main"): + for sub in ast.walk(node): + if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Name): + refs.add(sub.func.id) + # run_design_cli(, ...) -> first positional arg is the design fn + if sub.func.id == "run_design_cli" and sub.args and isinstance(sub.args[0], ast.Name): + refs.add(sub.args[0].id) + return refs + + +def test_cli_design_symbol_is_defined(): + tree = ast.parse(open(SRC, encoding="utf-8").read()) + bound = _module_bound_names(tree) + assert "blur" in bound, "expected the design function `blur` to be defined" + refs = _cli_referenced_names(tree) + assert "edge_detect" not in refs, ( + "blur_pipeline.py references undefined `edge_detect` (leftover from the " + "AMD edge_detect example) in its CLI/self-verify path; it should call `blur`" + ) + undefined = {r for r in refs if r not in bound and not hasattr(builtins, r)} + assert not undefined, f"CLI/self-verify references undefined names: {sorted(undefined)}" + + +if __name__ == "__main__": + test_cli_design_symbol_is_defined() + print("OK: blur_pipeline CLI design symbol is defined") diff --git a/camera/test_blur_pipeline_golden_ref.py b/camera/test_blur_pipeline_golden_ref.py new file mode 100644 index 0000000..d0ad9d4 --- /dev/null +++ b/camera/test_blur_pipeline_golden_ref.py @@ -0,0 +1,206 @@ +# Regression tests for blur_pipeline's self-verify golden reference. +# +# _run_and_verify() used to compare the blur() design's output against +# _edge_detect_ref() -- a Laplacian edge-detect + threshold + highlight +# pipeline that shares no math with the Gaussian blur the design actually +# computes. #2 fixed the CLI NameError crash (edge_detect -> blur) but left +# that mismatch in place, so on real NPU hardware --verify would go from +# "crashes" to "runs and reports the blur output as wrong" (it doesn't +# match an edge map, because it isn't one). +# +# blur_pipeline.py imports aie.iron (the real NPU/MLIR-AIE toolchain) at +# module level for the @iron.jit design, so it can't be imported on a host +# without that toolchain (this test runner included) unless those names +# exist. The stand-ins below only need to satisfy import-time references +# (the jit decorator, and names used directly as type annotations); they +# are never called, because these tests only exercise the plain-numpy +# _..._ref() helpers, never the @iron.jit design itself or _run_and_verify. +import os +import sys +import types + +import numpy as np + +HERE = os.path.dirname(os.path.abspath(__file__)) +SRC = os.path.join(HERE, "blur_pipeline.py") + + +def _install_aie_stubs(): + if "aie.iron" in sys.modules: + return + + class _Subscriptable: + """Stands in for CompileTime so `CompileTime[int]` evaluates fine + as a bare annotation expression at function-definition time.""" + + def __getitem__(self, item): + return self + + def _identity_jit(*_args, **_kwargs): + def _decorator(fn): + return fn + + return _decorator + + aie = types.ModuleType("aie") + aie_iron = types.ModuleType("aie.iron") + aie_iron.jit = _identity_jit + aie_iron.Buffer = object + aie_iron.CompileTime = _Subscriptable() + aie_iron.In = object() + aie_iron.Out = object() + aie_iron.ObjectFifo = object + aie_iron.Program = object + aie_iron.Runtime = object + aie_iron.Worker = object + aie_iron.kernels = types.SimpleNamespace( + rgba2gray=lambda **kw: None, + filter2d=lambda **kw: None, + threshold=lambda **kw: None, + gray2rgba=lambda **kw: None, + add_weighted=lambda **kw: None, + ) + aie_iron.get_current_device = lambda: None + + aie_iron_controlflow = types.ModuleType("aie.iron.controlflow") + aie_iron_controlflow.range_ = range + + aie_utils = types.ModuleType("aie.utils") + aie_utils_hostruntime = types.ModuleType("aie.utils.hostruntime") + aie_utils_hostruntime_argparse = types.ModuleType("aie.utils.hostruntime.argparse") + aie_utils_hostruntime_argparse.device_from_args = lambda *a, **kw: None + aie_utils_hostruntime_argparse.add_compile_args = lambda *a, **kw: None + aie_utils_hostruntime_cli = types.ModuleType("aie.utils.hostruntime.cli") + aie_utils_hostruntime_cli.run_design_cli = lambda *a, **kw: None + aie_utils_verify = types.ModuleType("aie.utils.verify") + aie_utils_verify.assert_pass = lambda *a, **kw: None + + aie.iron = aie_iron + aie.utils = aie_utils + aie_utils.hostruntime = aie_utils_hostruntime + aie_utils_hostruntime.argparse = aie_utils_hostruntime_argparse + aie_utils_hostruntime.cli = aie_utils_hostruntime_cli + aie_utils.verify = aie_utils_verify + + for name, mod in ( + ("aie", aie), + ("aie.iron", aie_iron), + ("aie.iron.controlflow", aie_iron_controlflow), + ("aie.utils", aie_utils), + ("aie.utils.hostruntime", aie_utils_hostruntime), + ("aie.utils.hostruntime.argparse", aie_utils_hostruntime_argparse), + ("aie.utils.hostruntime.cli", aie_utils_hostruntime_cli), + ("aie.utils.verify", aie_utils_verify), + ): + sys.modules[name] = mod + + +_install_aie_stubs() +sys.path.insert(0, HERE) +import blur_pipeline as bp # noqa: E402 + + +def test_run_and_verify_uses_blur_ref_not_edge_detect_ref(): + """AST guard mirroring test_blur_pipeline_cli_symbol.py's design-symbol + check, but for the golden-reference symbol: _run_and_verify() must + compare against _blur_ref(), never _edge_detect_ref().""" + import ast + + tree = ast.parse(open(SRC, encoding="utf-8").read()) + verify_fn = next( + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "_run_and_verify" + ) + called = { + sub.func.id + for sub in ast.walk(verify_fn) + if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Name) + } + assert "_blur_ref" in called, "_run_and_verify must call _blur_ref()" + assert "_edge_detect_ref" not in called, ( + "_run_and_verify still compares the blur design's output against " + "the unrelated Laplacian edge-detect reference" + ) + + +def test_gaussian_blur3x3_ref_impulse_response_matches_kernel_weights(): + """A single bright pixel in an otherwise-black 5x5 image, run through + the 3x3 unity-gain kernel [[1,2,1],[2,4,2],[1,2,1]]/16, must produce + exactly those weights (scaled by 255, rounded) centered on the impulse + -- pins down both the kernel values and the normalization shift.""" + gray = np.zeros((5, 5), dtype=np.uint8) + gray[2, 2] = 255 + out = bp._gaussian_blur3x3_ref(gray, 5, 5) + + expected = np.array( + [ + [1, 2, 1], + [2, 4, 2], + [1, 2, 1], + ], + dtype=np.float64, + ) * (255.0 / 16.0) + expected = np.round(expected).astype(np.uint8) + + np.testing.assert_array_equal(out[1:4, 1:4], expected) + # Everything outside the 3x3 footprint of the impulse stays exactly 0. + mask = np.ones((5, 5), dtype=bool) + mask[1:4, 1:4] = False + assert np.all(out[mask] == 0) + + +def test_gaussian_blur3x3_ref_border_replicate_matches_interior_symmetry(): + """BORDER_REPLICATE means a corner pixel's 3x3 neighborhood duplicates + the edge/corner, which for a uniform image must return that same value + unchanged (the kernel is unity-gain, so blurring a flat field is a + no-op everywhere, corners included).""" + gray = np.full((4, 4), 200, dtype=np.uint8) + out = bp._gaussian_blur3x3_ref(gray, 4, 4) + np.testing.assert_array_equal(out, gray) + + +def test_blur_ref_smooths_a_checkerboard_like_the_hardware_smoke_test(): + """Mirrors test_blur.py's real-hardware smoke assertion (blurred row + variance < 0.8x the sharp row variance) entirely in numpy, and confirms + the alpha=1/beta=0 add_weighted stage really drops the original color + input rather than blending it back in.""" + h, w = 16, 16 + rgba = np.zeros((h, w, 4), dtype=np.uint8) + rgba[:, :, 3] = 255 + mask = (np.add.outer(np.arange(h), np.arange(w))) % 2 == 0 + rgba[mask] = [255, 255, 255, 255] + flat = rgba.reshape(-1) + + out = bp._blur_ref(flat, h, w).reshape(h, w, 4) + + sharp_var = float(rgba[h // 2, :, 0].astype(np.float64).var()) + blur_var = float(out[h // 2, :, 0].astype(np.float64).var()) + assert blur_var < sharp_var * 0.8, ( + f"checkerboard not smoothed: sharp_var={sharp_var} blur_var={blur_var}" + ) + # R == G == B and alpha stays 255 (gray2rgba output), confirming the + # original per-channel color (all-white/all-black here, so this alone + # wouldn't catch a beta!=0 bug, but the smoothing check above does). + assert np.array_equal(out[..., 0], out[..., 1]) + assert np.array_equal(out[..., 0], out[..., 2]) + assert np.all(out[..., 3] == 255) + + +def test_blur_ref_differs_from_edge_detect_ref(): + """Sanity check that the two references are genuinely different + pipelines on the same input, not an accidental alias.""" + rng = np.random.default_rng(0) + rgba = rng.integers(0, 255, size=(8 * 8 * 4,), dtype=np.uint8) + blur_out = bp._blur_ref(rgba, 8, 8) + edge_out = bp._edge_detect_ref(rgba, 8, 8) + assert not np.array_equal(blur_out, edge_out) + + +if __name__ == "__main__": + test_run_and_verify_uses_blur_ref_not_edge_detect_ref() + test_gaussian_blur3x3_ref_impulse_response_matches_kernel_weights() + test_gaussian_blur3x3_ref_border_replicate_matches_interior_symmetry() + test_blur_ref_smooths_a_checkerboard_like_the_hardware_smoke_test() + test_blur_ref_differs_from_edge_detect_ref() + print("OK: blur_pipeline golden reference matches the actual blur design")