feat(hailo): add PatchCore Hailo export on latest main

.gitignore

@@ -369,3 +369,5 @@ tests/__pycache__/*
 compiled_patchcore_kv260/*
 quantize_result/*
 compiled_padim_kv260/*
+*.har
+*.hef

anomavision/inference/model/backends/hailo_backend.py

@@ -1,14 +1,19 @@
 """HailoRT runtime for complete AnomaVision anomaly HEFs.
 
-The HEF is expected to expose two outputs generated by ``hailo_export``:
-``image_scores`` and ``score_map``. Feature extraction and distance calculation
-must already be compiled into the HEF. This runtime intentionally contains no
-fallback CNN or CPU distance implementation, which prevents accidental partial
-quantization on Kria.
+The HEF is expected to expose ``image_scores`` and ``score_map``. Feature
+extraction and anomaly scoring are compiled into the HEF; this runtime only
+adapts the common AnomaVision inference input contract to HailoRT.
+
+The public backend contract matches the ONNX backend: input is a single
+ImageNet-normalized RGB tensor in NCHW float32. Hailo receives the same values
+in NHWC layout. No second resize or normalization is applied to tensors that
+have already passed through the AnomaVision dataset preprocessing pipeline.
 """
 
 from __future__ import annotations
 
+import ctypes
+import os
 from pathlib import Path
 from typing import Dict, Tuple
 
@@ -18,6 +23,79 @@
 from .base import InferenceBackend
 
 
+def _load_hailort():
+    """Load HailoRT and preload its native shared library when necessary."""
+    version = "5.3.0"
+    library_name = f"libhailort.so.{version}"
+    candidates = []
+
+    configured = os.environ.get("HAILORT_LIB_PATH")
+    if configured:
+        configured_path = Path(configured)
+        if configured_path.is_file():
+            candidates.append(configured_path)
+        elif configured_path.is_dir():
+            candidates.append(configured_path / library_name)
+
+    for directory in (
+        "/usr/lib",
+        "/usr/lib/aarch64-linux-gnu",
+        "/usr/lib/x86_64-linux-gnu",
+        "/usr/local/lib",
+    ):
+        candidates.append(Path(directory) / library_name)
+
+    for candidate in candidates:
+        if candidate.is_file():
+            try:
+                ctypes.CDLL(str(candidate), mode=ctypes.RTLD_GLOBAL)
+                break
+            except OSError as exc:
+                raise RuntimeError(
+                    f"Found {candidate}, but it could not be loaded: {exc}. "
+                    "Install the matching HailoRT native runtime and PCIe driver."
+                ) from exc
+
+    try:
+        from hailo_platform import (
+            HEF,
+            ConfigureParams,
+            FormatType,
+            HailoStreamInterface,
+            InferVStreams,
+            InputVStreamParams,
+            OutputVStreamParams,
+            VDevice,
+        )
+    except ModuleNotFoundError as exc:
+        raise RuntimeError(
+            "HailoRT Python bindings are not installed. Install the matching "
+            "HailoRT Python wheel (5.3.0) on the target system."
+        ) from exc
+    except ImportError as exc:
+        message = str(exc)
+        if "libhailort.so" in message:
+            raise RuntimeError(
+                "HailoRT Python bindings are installed, but the native "
+                f"{library_name} library is not available to the dynamic linker. "
+                "Install the matching HailoRT runtime package, or set "
+                "HAILORT_LIB_PATH to the directory/file containing "
+                f"{library_name}. The Python wheel alone is not sufficient."
+            ) from exc
+        raise RuntimeError(f"HailoRT could not be imported: {exc}") from exc
+
+    return {
+        "ConfigureParams": ConfigureParams,
+        "FormatType": FormatType,
+        "HEF": HEF,
+        "HailoStreamInterface": HailoStreamInterface,
+        "InputVStreamParams": InputVStreamParams,
+        "InferVStreams": InferVStreams,
+        "OutputVStreamParams": OutputVStreamParams,
+        "VDevice": VDevice,
+    }
+
+
 class HailoAnomalyRuntime:
     """Run a complete PaDiM or PatchCore HEF through HailoRT."""
 
@@ -26,59 +104,30 @@ def __init__(
         hef_path: str | Path,
         input_size: Tuple[int, int] = (224, 224),
         input_dtype: np.dtype = np.float32,
+        mean: Tuple[float, float, float] = (0.485, 0.456, 0.406),
+        std: Tuple[float, float, float] = (0.229, 0.224, 0.225),
     ) -> None:
-        """Load and configure a complete Hailo-8 HEF.
-
-        Args:
-            hef_path: Path to a HEF exposing ``image_scores`` and ``score_map``.
-            input_size: Fixed ``(height, width)`` expected by the HEF.
-            input_dtype: Host input dtype passed to HailoRT.
-
-        Raises:
-            RuntimeError: If HailoRT is unavailable or has no network group.
-            FileNotFoundError: If ``hef_path`` does not exist.
-            ValueError: If required anomaly outputs are missing.
-        """
-        try:
-            from hailo_platform import (
-                HEF,
-                ConfigureParams,
-                FormatType,
-                HailoStreamInterface,
-                InferVStreams,
-                InputVStreamParams,
-                OutputVStreamParams,
-                VDevice,
-            )
-        except ImportError as exc:  # pragma: no cover - depends on Kria image
-            raise RuntimeError(
-                "HailoRT is not installed. Install the HailoRT Python package on "
-                "the Kria K26 image before loading a HEF."
-            ) from exc
-
-        self._api = {
-            "ConfigureParams": ConfigureParams,
-            "FormatType": FormatType,
-            "HEF": HEF,
-            "HailoStreamInterface": HailoStreamInterface,
-            "InputVStreamParams": InputVStreamParams,
-            "InferVStreams": InferVStreams,
-            "OutputVStreamParams": OutputVStreamParams,
-            "VDevice": VDevice,
-        }
+        api = _load_hailort()
+        self._api = api
         self.hef_path = Path(hef_path)
         if not self.hef_path.exists():
             raise FileNotFoundError(self.hef_path)
         self.input_size = tuple(int(v) for v in input_size)
         self.input_dtype = input_dtype
-        self.device = VDevice()
-        self.hef = HEF(str(self.hef_path))
+        self.mean = np.asarray(mean, dtype=np.float32).reshape(1, 1, 3)
+        self.std = np.asarray(std, dtype=np.float32).reshape(1, 1, 3)
+        self.device = api["VDevice"]()
+        self.hef = api["HEF"](str(self.hef_path))
         self.network_groups = self.device.configure(self.hef)
         if not self.network_groups:
             raise RuntimeError(f"No network group found in {self.hef_path}")
         self.network_group = self.network_groups[0]
         self.network_group_params = self.network_group.create_params()
-        self.input_name = self.hef.get_input_vstream_infos()[0].name
+
+        input_infos = self.hef.get_input_vstream_infos()
+        if not input_infos:
+            raise ValueError(f"The HEF has no input stream: {self.hef_path}")
+        self.input_name = input_infos[0].name
         output_names = [info.name for info in self.hef.get_output_vstream_infos()]
         required = {"image_scores", "score_map"}
         missing = sorted(required.difference(output_names))
@@ -89,43 +138,64 @@ def __init__(
             )
         self.output_names = output_names
 
-    def _preprocess(self, image: Image.Image | np.ndarray | str | Path) -> np.ndarray:
-        """Convert an image path, PIL image, or RGB array to NCHW input."""
-        if isinstance(image, (str, Path)):
-            image = Image.open(image)
-        if isinstance(image, Image.Image):
-            image = np.asarray(image.convert("RGB"))
-        image = np.asarray(image)
-        if image.ndim != 3 or image.shape[2] != 3:
-            raise ValueError("image must be an HxWx3 RGB image")
-        image = np.asarray(
-            Image.fromarray(image.astype(np.uint8), mode="RGB").resize(
-                (self.input_size[1], self.input_size[0]), Image.Resampling.BILINEAR
-            ),
-            dtype=np.float32,
-        )
-        # Match AnomaVision's tensor contract: NCHW float RGB in [0, 1].
-        return np.transpose(image / 255.0, (2, 0, 1))[None].astype(self.input_dtype)
+    def _prepare_input(self, batch) -> np.ndarray:
+        """Convert AnomaVision NCHW input to the HEF's NHWC input."""
+        if isinstance(batch, Image.Image):
+            array = np.asarray(batch.convert("RGB"), dtype=np.uint8)
+            return self._preprocess_raw_hwc(array)[None]
 
-    def predict(
-        self, image: Image.Image | np.ndarray | str | Path
-    ) -> Dict[str, np.ndarray]:
-        """Run one image and return complete image and localization outputs.
+        array = np.asarray(batch)
+        if array.ndim == 4:
+            if array.shape[0] != 1:
+                raise ValueError("HailoBackend currently supports batch size 1")
+            array = array[0]
+
+        if array.ndim != 3:
+            raise ValueError("batch must be an HxWx3, 3xHxW, or single-image batch")
+
+        if array.shape[0] == 3:
+            if array.shape[1:] != self.input_size:
+                raise ValueError(
+                    f"Hailo input must be {self.input_size}, got {array.shape[1:]}"
+                )
+            return np.ascontiguousarray(
+                np.transpose(array, (1, 2, 0)), dtype=self.input_dtype
+            )
 
-        Args:
-            image: An image path, PIL RGB image, or HxWx3 RGB array.
+        if array.shape[2] == 3:
+            if np.issubdtype(array.dtype, np.integer):
+                return self._preprocess_raw_hwc(array)
+            if array.shape[:2] != self.input_size:
+                raise ValueError(
+                    f"Hailo input must be {self.input_size}, got {array.shape[:2]}"
+                )
+            return np.ascontiguousarray(array, dtype=self.input_dtype)
+
+        raise ValueError("batch must be RGB with three channels")
+
+    def _preprocess_raw_hwc(self, array: np.ndarray) -> np.ndarray:
+        """Preprocess a raw uint8 HWC RGB image exactly once."""
+        image = Image.fromarray(array.astype(np.uint8), mode="RGB").resize(
+            (self.input_size[1], self.input_size[0]), Image.Resampling.BILINEAR
+        )
+        normalized = np.asarray(image, dtype=np.float32) / 255.0
+        normalized = (normalized - self.mean) / self.std
+        return np.ascontiguousarray(normalized, dtype=self.input_dtype)
 
-        Returns:
-            A mapping containing ``image_scores`` and ``score_map`` arrays.
-        """
+    def predict(self, image) -> Dict[str, np.ndarray]:
+        """Run one image and return the HEF's complete anomaly outputs."""
         api = self._api
         input_params = api["InputVStreamParams"].make(
-            self.network_group, quantized=False, format_type=api["FormatType"].FLOAT32
+            self.network_group,
+            quantized=False,
+            format_type=api["FormatType"].FLOAT32,
         )
         output_params = api["OutputVStreamParams"].make(
-            self.network_group, quantized=False, format_type=api["FormatType"].FLOAT32
+            self.network_group,
+            quantized=False,
+            format_type=api["FormatType"].FLOAT32,
         )
-        tensor = self._preprocess(image)
+        tensor = self._prepare_input(image)
         with self.network_group.activate(self.network_group_params):
             with api["InferVStreams"](
                 self.network_group, input_params, output_params
@@ -162,28 +232,12 @@ def __init__(
         self.runtime = HailoAnomalyRuntime(model_path, input_size=input_size)
 
     def predict(self, batch) -> Tuple[np.ndarray, np.ndarray]:
-        """Run one image through the common backend contract.
-
-        Args:
-            batch: An HxWx3 RGB image or a single-image 1x3xHxW/1xHxWx3 batch.
-
-        Returns:
-            A tuple ``(image_scores, score_maps)`` as NumPy arrays.
-        """
-        array = np.asarray(batch)
-        if array.ndim == 4:
-            if array.shape[0] != 1:
-                raise ValueError("HailoBackend currently supports batch size 1")
-            array = (
-                np.transpose(array[0], (1, 2, 0)) if array.shape[1] == 3 else array[0]
-            )
-        elif array.ndim != 3:
-            raise ValueError("batch must be an HxWx3 or 1x3xHxW image")
-        result = self.runtime.predict(array)
+        """Run a single image through the HEF using the common backend contract."""
+        result = self.runtime.predict(batch)
         return result["image_scores"], result["score_map"]
 
     def warmup(self, batch=None, runs: int = 2) -> None:
-        """Warm up the device with a supplied image batch."""
+        """Warm up the device with a supplied preprocessed image batch."""
         if batch is None:
             raise ValueError("Hailo warmup requires a sample image batch")
         for _ in range(max(1, int(runs))):

anomavision/quantize/model/backends/hef/efficientad.py

@@ -0,0 +1,48 @@
+"""Fixed-shape EfficientAD graph for Hailo end-to-end deployment."""
+
+from __future__ import annotations
+
+from typing import Tuple
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+
+class EfficientADHailoGraph(nn.Module):
+    """Export-friendly EfficientAD inference graph for Hailo."""
+
+    def __init__(self, model: nn.Module, input_size: Tuple[int, int] = (224, 224)) -> None:
+        super().__init__()
+        if tuple(input_size) != (224, 224):
+            raise ValueError("Hailo EfficientAD export currently requires input_size=(224, 224)")
+
+        self.input_size = (224, 224)
+        self.teacher = model.teacher.eval()
+        self.student = model.student.eval()
+        self.register_buffer("map_mean", model.map_mean.detach().float().clone())
+        map_std = model.map_std.detach().float().clone().clamp_min(1e-6)
+        self.register_buffer("map_inv_std", map_std.reciprocal())
+
+        self.channel_mean = nn.Conv2d(112, 1, kernel_size=1, bias=False)
+        with torch.no_grad():
+            self.channel_mean.weight.fill_(1.0 / 112.0)
+        self.channel_mean.weight.requires_grad_(False)
+
+    def forward(self, image: torch.Tensor):
+        teacher_features = self.teacher(image)
+        student_features = self.student(image)
+        diff = student_features - teacher_features
+        squared = diff * diff
+
+        # Reduce the 112 feature channels with a fixed 1x1 convolution.
+        raw = self.channel_mean(squared)
+        raw = F.interpolate(
+            raw, size=(224, 224), mode="bilinear", align_corners=False
+        )
+        normalized = (raw - self.map_mean.unsqueeze(0)) * self.map_inv_std.unsqueeze(0)
+
+        image_scores = F.max_pool2d(
+            normalized, kernel_size=(224, 224), stride=(224, 224)
+        ).flatten(1)
+        return image_scores, normalized.squeeze(1)