fix(camera): blur CLI/self-verify calls undefined edge_detect -> NameError on NPU

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,

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(<design>, ...) -> 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")

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