Fix/kv260 padim xmodel

README.md

@@ -34,6 +34,7 @@ AnomaVision supports **PaDiM** and lightweight **PatchCore**, image-level scores
 - Use **PaDiM** for the default fast baseline or **PatchCore** for a compact nearest-neighbor memory bank.
 - Run inference through PyTorch, ONNX Runtime, OpenVINO, or native TensorRT.
 - Export FP16 or calibrated INT8 TensorRT engines for NVIDIA production deployments.
+- **Export and compile PaDiM and PatchCore to Vitis AI XModel for the AMD/Xilinx Kria KV260.**
 
 Benchmark results and reproduction details are documented in [`docs/benchmark.md`](docs/benchmark.md). Treat benchmark numbers as workload-specific, and reproduce them on your own hardware before making production claims.
 
@@ -112,6 +113,71 @@ anomavision train --help
 anomavision export --help
 ```
 
+## KV260 XModel deployment
+
+AnomaVision also provides a Vitis AI workflow for deploying **PaDiM** and **PatchCore** on the **AMD/Xilinx Kria KV260** DPU.
+
+The KV260 workflow is:
+
+**PyTorch model → Vitis AI INT8 quantization → XModel → KV260 DPU compilation**
+
+The detailed, copy-ready guide is available in [`docs/kv260_xmodel.md`](docs/kv260_xmodel.md).
+
+The guide covers:
+
+- Vitis AI 3.5 environment setup
+- PaDiM INT8 calibration and test mode
+- PatchCore INT8 calibration and test mode
+- XModel validation
+- KV260 DPU compilation with `vai_c_xir`
+- Expected output files and troubleshooting notes
+
+### PaDiM quick example
+
+```bash
+python quantize_padim_kv260.py \
+  --model distributions/padim/bottle/anomav_exp/model.pt \
+  --calibration-dir /workspace/dataset/bottle/train/good \
+  --output-dir compiled_padim_kv260 \
+  --quant_mode calib
+
+python quantize_padim_kv260.py \
+  --model distributions/padim/bottle/anomav_exp/model.pt \
+  --calibration-dir /workspace/dataset/bottle/train/good \
+  --output-dir compiled_padim_kv260 \
+  --quant_mode test
+
+vai_c_xir \
+  -x compiled_padim_kv260/PadimKV260_int.xmodel \
+  -a /opt/vitis_ai/compiler/arch/DPUCZDX8G/KV260/arch.json \
+  -o compiled_padim_kv260/compiled \
+  -n PadimKV260
+```
+
+### PatchCore quick example
+
+```bash
+python quantize_patchcore_kv260.py \
+  --model distributions/patchcore/bottle/anomav_exp/model.pt \
+  --calibration-dir /workspace/dataset/bottle/train/good \
+  --output-dir compiled_patchcore_kv260 \
+  --quant_mode calib
+
+python quantize_patchcore_kv260.py \
+  --model distributions/patchcore/bottle/anomav_exp/model.pt \
+  --calibration-dir /workspace/dataset/bottle/train/good \
+  --output-dir compiled_patchcore_kv260 \
+  --quant_mode test
+
+vai_c_xir \
+  -x compiled_patchcore_kv260/PatchCoreKV260_int.xmodel \
+  -a /opt/vitis_ai/compiler/arch/DPUCZDX8G/KV260/arch.json \
+  -o compiled_patchcore_kv260/compiled \
+  -n PatchCoreKV260
+```
+
+> **Note:** KV260 XModel generation and compilation are intended for a Linux/Vitis AI environment. The final on-device KV260 validation is a separate step from XModel generation and compiler validation.
+
 ## Production Autopilot
 
 **Production Autopilot is the easiest way to move from two trained models to one deployable choice.** It compares PaDiM and ultra-light PatchCore on the same labeled test split, calibrates a separate threshold for each, profiles median and P95 latency on your hardware, checks localization health, and packages the selected artifact with a self-contained HTML dashboard.
@@ -163,6 +229,7 @@ PatchCore compares image patches with a compact normal-feature memory bank. Its
 | CLI and configuration | [`docs/cli.md`](docs/cli.md), [`docs/config.md`](docs/config.md) |
 | Python API | [`docs/api.md`](docs/api.md) |
 | PatchCore and TensorRT deployment | [`docs/production_deployment.md`](docs/production_deployment.md) |
+| **KV260 PaDiM/PatchCore XModel** | [`docs/kv260_xmodel.md`](docs/kv260_xmodel.md) |
 | Runnable CPU, PatchCore, and TensorRT examples | [`examples/README.md`](examples/README.md) |
 | Benchmark methodology | [`docs/benchmark.md`](docs/benchmark.md) |
 | Troubleshooting | [`docs/troubleshooting.md`](docs/troubleshooting.md) |

anomavision/inference/model/backends/k260_backend.py

@@ -12,51 +12,83 @@
 
 
 class KV260Backend(InferenceBackend):
-    """Run an AMD Vitis AI XModel through VART on a KV260/K26 target."""
+    """Run an AMD Vitis AI XModel through VART on a KV260/K26 target.
+
+    Compiled AnomaVision XModels can contain both DPU and CPU subgraphs.  When
+    the Vitis AI GraphRunner is available, use it so the complete graph is
+    executed instead of accidentally running only the first child subgraph.
+    """
 
     def __init__(
         self,
         model_path: str | Path,
         device: str = "k260",
         input_size: tuple[int, int] = (224, 224),
     ) -> None:
-        """Load an AMD Vitis AI XModel through the VART runtime.
-
-        Args:
-            model_path: Path to an XModel exposing image score and map outputs.
-            device: Board identifier retained for backend-factory consistency.
-            input_size: Fixed ``(height, width)`` used for RGB preprocessing.
-
-        Raises:
-            FileNotFoundError: If ``model_path`` does not exist.
-            RuntimeError: If VART/XIR is unavailable or no runnable subgraph exists.
-            ValueError: If the XModel does not expose at least two outputs.
-        """
+        """Load an AMD Vitis AI XModel through the VART runtime."""
         del device
         self.model_path = Path(model_path)
         if not self.model_path.is_file():
             raise FileNotFoundError(self.model_path)
+
         try:
             import vart
             import xir
         except ImportError as exc:  # pragma: no cover - target-only dependency
             raise RuntimeError(
                 "Vitis AI VART/XIR is required for KV260 XModel inference."
             ) from exc
+
         self._vart = vart
         self.input_size = tuple(int(value) for value in input_size)
-        graph = xir.Graph.deserialize(str(self.model_path))
-        runners = graph.get_root_subgraph().children_topological_sort()
-        if not runners:
-            raise RuntimeError(f"No runnable subgraph found in {self.model_path}")
-        self.runner = vart.Runner.create_runner(runners[0], "run")
-        self.input_tensor = self.runner.get_input_tensors()[0]
-        self.output_tensors = self.runner.get_output_tensors()
+        self.graph = xir.Graph.deserialize(str(self.model_path))
+        self.runner = None
+        self._graph_runner = False
+
+        # A compiled XModel may be split into DPU + CPU subgraphs. GraphRunner
+        # is the Vitis AI API intended for executing such complete graphs.
+        try:
+            from vitis_ai_library import GraphRunner
+
+            self.runner = GraphRunner.create_graph_runner(self.graph)
+            self._graph_runner = True
+            self.input_tensors = [
+                buffer.get_tensor() for buffer in self.runner.get_inputs()
+            ]
+            self.output_tensors = [
+                buffer.get_tensor() for buffer in self.runner.get_outputs()
+            ]
+        except (ImportError, AttributeError, RuntimeError) as exc:
+            # Keep a VART-only fallback for single-DPU XModels. This is useful
+            # for minimal graphs, but cannot execute CPU post-processing nodes.
+            children = self.graph.get_root_subgraph().toposort_child_subgraph()
+            dpu_children = [
+                child
+                for child in children
+                if child.has_attr("device") and child.get_attr("device") == "DPU"
+            ]
+            if len(dpu_children) != 1:
+                raise RuntimeError(
+                    "This XModel contains multiple subgraphs and requires "
+                    "Vitis AI GraphRunner (vitis_ai_library)."
+                ) from exc
+
+            self.runner = vart.Runner.create_runner(dpu_children[0], "run")
+            self.input_tensors = self.runner.get_input_tensors()
+            self.output_tensors = self.runner.get_output_tensors()
+
+        if not self.input_tensors:
+            raise RuntimeError(f"No input tensor found in {self.model_path}")
         if len(self.output_tensors) < 2:
             raise ValueError(
-                "KV260 XModel must expose image_scores and score_map outputs"
+                "KV260 XModel must expose image_scores and score_map outputs. "
+                "A backbone-only XModel is not a complete PaDiM/PatchCore model."
             )
-        self.output_names = [tensor.name for tensor in self.output_tensors]
+
+        self.input_tensor = self.input_tensors[0]
+        self.output_names = [
+            getattr(tensor, "name", "") for tensor in self.output_tensors
+        ]
         self.input_shape = tuple(int(value) for value in self.input_tensor.dims)
 
     @staticmethod
@@ -81,33 +113,62 @@ def _preprocess(self, image: Image.Image | np.ndarray | str | Path) -> np.ndarra
             )
         )
         batch = resized[None]
-        if len(self.input_shape) == 4 and self.input_shape[1] == 3:
+        if len(self.input_shape) == 4 and self.input_shape[-1] == 3:
+            # Vitis AI DPU tensors are normally NHWC.
+            pass
+        elif len(self.input_shape) == 4 and self.input_shape[1] == 3:
             batch = np.transpose(batch, (0, 3, 1, 2))
-        if np.issubdtype(self.input_tensor.dtype, np.floating):
+        if np.issubdtype(np.dtype(self.input_tensor.dtype), np.floating):
             return (batch.astype(np.float32) / 255.0).astype(self.input_tensor.dtype)
         return batch.astype(self.input_tensor.dtype)
 
+    @staticmethod
+    def _copy_to_tensor_buffer(buffer: Any, data: np.ndarray) -> None:
+        """Copy a NumPy array into a VART TensorBuffer."""
+        target = np.asarray(buffer)
+        if target.shape != data.shape:
+            data = data.reshape(target.shape)
+        np.copyto(target, data, casting="unsafe")
+
     def predict(self, batch: Any) -> ScoresMaps:
-        """Run one image through VART and return anomaly scores and maps.
+        """Run one image through the complete XModel graph."""
+        input_data = self._preprocess(batch)
 
-        Args:
-            batch: An image path, PIL image, or RGB NumPy array.
+        if self._graph_runner:
+            input_buffers = self.runner.get_inputs()
+            output_buffers = self.runner.get_outputs()
+            self._copy_to_tensor_buffer(input_buffers[0], input_data)
+            for buffer in input_buffers:
+                buffer.sync_for_write(
+                    0,
+                    buffer.get_tensor().get_data_size() // buffer.get_tensor().dims[0],
+                )
+            job_id, status = self.runner.execute_async(input_buffers, output_buffers)
+            if status != 0:
+                raise RuntimeError(f"KV260 GraphRunner execution failed: {status}")
+            status = self.runner.wait(job_id)
+            if status != 0:
+                raise RuntimeError(f"KV260 GraphRunner wait failed: {status}")
+            for buffer in output_buffers:
+                buffer.sync_for_read(
+                    0,
+                    buffer.get_tensor().get_data_size() // buffer.get_tensor().dims[0],
+                )
+            outputs = [np.asarray(buffer).copy() for buffer in output_buffers]
+        else:
+            outputs = [
+                np.empty(tuple(int(value) for value in tensor.dims), dtype=tensor.dtype)
+                for tensor in self.output_tensors
+            ]
+            result = self.runner.execute_async([input_data], outputs)
+            job_id = result[0] if isinstance(result, tuple) else result
+            self.runner.wait(job_id)
 
-        Returns:
-            A tuple ``(image_scores, score_maps)`` as NumPy arrays.
-        """
-        input_data = self._preprocess(batch)
-        outputs = [
-            np.empty(tuple(int(value) for value in tensor.dims), dtype=tensor.dtype)
-            for tensor in self.output_tensors
-        ]
-        job_id = self.runner.execute_async([input_data], outputs)
-        self.runner.wait(job_id)
         image_index = next(
             (
                 index
                 for index, name in enumerate(self.output_names)
-                if "image_score" in name
+                if "image_score" in name or "score" in name
             ),
             0,
         )
@@ -125,7 +186,7 @@ def predict(self, batch: Any) -> ScoresMaps:
         )
 
     def close(self) -> None:
-        """Release the VART runner and allow its resources to be reclaimed."""
+        """Release the VART/GraphRunner resources."""
         self.runner = None