diff --git a/README.md b/README.md index 8e93371..6f1e3b9 100644 --- a/README.md +++ b/README.md @@ -21,14 +21,15 @@ KV260 DPU support

-AnomaVision is a computer vision project for finding **defects and unusual patterns** in images. +AnomaVision is a production-oriented computer vision toolkit for detecting **defects and unusual patterns** from normal images. -It supports two anomaly detection methods: +It supports three anomaly detection methods: -- **PaDiM** — a simple and fast baseline. -- **PatchCore** — a lightweight memory-based method. +- **PaDiM** — a simple, fast feature-distribution baseline. +- **PatchCore** — a lightweight memory-based method designed for efficient inference. +- **EfficientAD** — a lightweight student-teacher method designed for fast industrial anomaly detection. -You only need **normal (`good`) images** to train the anomaly detector. +Training requires only **normal (`good`) images**. Labeled test images can then be used for evaluation, threshold calibration, and production model selection.

Open the AnomaVision live demo @@ -39,50 +40,46 @@ You only need **normal (`good`) images** to train the anomaly detector. ## What can AnomaVision do? - Train anomaly detection models using normal images. -- Detect image-level anomalies. -- Create anomaly heatmaps showing where the problem is. -- Export models to **ONNX, OpenVINO, and TensorRT**. +- Detect image-level anomalies and generate anomaly heatmaps. +- Evaluate anomaly detection and localization performance. +- Calibrate anomaly thresholds from validation data. +- Export models to **ONNX, OpenVINO, and TensorRT** where supported. +- Run production model selection with **Production Autopilot**. - Export and compile **PaDiM and PatchCore to XModel for the AMD/Xilinx Kria KV260**. ## Quick start ### 1. Install -#### Option A — From Source (development) +#### Option A — From Source ```bash git clone https://github.com/DeepKnowledge1/AnomaVision.git cd AnomaVision -# Create and activate a virtual environment uv venv --python 3.11 .venv source .venv/bin/activate # Windows: .venv\Scripts\Activate.ps1 -# Install with your hardware extra uv sync --extra cpu # CPU uv sync --extra cu121 # CUDA 12.1 ``` ---- - -#### Option B — From PyPI (production / quick start) +#### Option B — From PyPI ```bash -# CPU · Mac, CI runners, edge devices uv pip install "anomavision[cpu]" -# NVIDIA GPU · pick your CUDA version -uv pip install "anomavision[cu118]" # CUDA 11.8 -uv pip install "anomavision[cu121]" # CUDA 12.1 -uv pip install "anomavision[cu124]" # CUDA 12.4 +# NVIDIA GPU +uv pip install "anomavision[cu118]" +uv pip install "anomavision[cu121]" +uv pip install "anomavision[cu124]" ``` - For other environments, see [Installation](docs/installation.md). -### 2. Prepare your images +### 2. Prepare your dataset -Use a simple MVTec-style folder structure: +Use an MVTec-style structure: ```text dataset/ @@ -100,19 +97,22 @@ dataset/ └── good/ ``` -Training uses the **good** images. Test images can contain defects. +Training uses only `train/good`. Test images may contain defects. ### 3. Train -Create or edit `config.yml` and point `dataset_path` to your dataset. +Create or edit `config.yml` and set `dataset_path` to your dataset. + +Select the algorithm in the configuration: -Then run: +```yaml +algorithm: padim # padim | patchcore | efficientad +``` ```bash anomavision train --config config.yml ``` -PaDiM is the default model. For PatchCore, set `algorithm: patchcore` in the configuration. ### 4. Detect @@ -122,52 +122,72 @@ anomavision detect --config config.yml --img_path ./dataset/bottle/test ### 5. Export -For a portable model, ONNX is a good place to start: - ```bash anomavision export --config config.yml --format onnx ``` -For more export options, see [Export and deployment](docs/production_deployment.md). - -## KV260 support - -AnomaVision also supports a **Vitis AI workflow for PaDiM and PatchCore on the AMD/Xilinx Kria KV260**. - -The workflow is: - -```text -PyTorch → INT8 quantization → XModel → KV260 DPU compilation -``` - -Both PaDiM and PatchCore currently compile with **1 DPU subgraph** in the KV260 compiler. - -The complete setup and commands are in: - -**[KV260 XModel Guide](docs/kv260_xmodel.md)** - -> XModel compilation has been validated in the Vitis AI environment. Final on-device KV260 validation requires the physical hardware. - +See [Export and deployment](docs/production_deployment.md) for deployment-specific options. ## 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. +Production Autopilot compares trained anomaly models on the **same validation data**, measures their performance and latency on the target device, calibrates thresholds, and selects the best candidate for deployment. -Train both candidate models first, then run the complete labeled split on CPU: +PaDiM, PatchCore, and EfficientAD can be supplied as independent candidate models. The model paths are provided directly through the CLI: ```bash anomavision autopilot \ --config config.yml \ --padim_model ./distributions/padim/bottle/anomav_exp/model.pt \ --patchcore_model ./distributions/patchcore/bottle/anomav_exp/model.pt \ + --efficientad_model ./distributions/efficientad/bottle/anomav_exp/model.pt \ --device cpu \ --validation_split 1.0 \ --target_latency_ms 50 \ --output_dir ./production_package ``` -Open `production_package/production_autopilot_report.html` to see the selected model, AUROC, calibrated threshold, localization diagnostics, memory, median latency, P95 latency, and deployment recommendation. The package also contains `deployment_manifest.json`, `localization_report.md`, and the selected model artifact. See [`docs/production_deployment.md`](docs/production_deployment.md) for GPU, TensorRT, INT8, and packaging details. +### How selection works + +For every supplied model, Autopilot: + +1. Evaluates the model on the validation split. +2. Calibrates an image-level anomaly threshold. +3. Measures inference latency on the selected device. +4. Calculates image-level and pixel-level metrics when localization maps are available. +5. Checks localization quality and false-positive behavior. +6. Applies the target latency constraint when selecting the production candidate. +7. Packages the selected model and writes a deployment manifest. + +If multiple models satisfy the latency target, the model with the strongest image-level AUROC is preferred, with latency used as a tie-breaker. + +### Output + +Autopilot creates a production package containing: + +```text +production_package/ +├── model.pt +├── deployment_manifest.json +├── localization_report.md +└── production_autopilot_report.html +``` + +The HTML report is a self-contained dashboard showing the candidate comparison, selected model, AUROC, calibrated threshold, latency, localization diagnostics, and deployment recommendation. + + +## KV260 support +AnomaVision supports a **Vitis AI workflow for PaDiM and PatchCore on the AMD/Xilinx Kria KV260**. + +```text +PyTorch → INT8 quantization → XModel → KV260 DPU compilation +``` + +Both PaDiM and PatchCore currently compile with **1 DPU subgraph** in the KV260 compiler. + +See the complete [KV260 XModel Guide](docs/kv260_xmodel.md). + +> XModel compilation has been validated in the Vitis AI environment. Final on-device KV260 validation requires the physical hardware. ## Documentation @@ -186,8 +206,6 @@ Open `production_package/production_autopilot_report.html` to see the selected m ## Python example -You can also use AnomaVision directly from Python: - ```python import torch from torch.utils.data import DataLoader diff --git a/anomavision/__init__.py b/anomavision/__init__.py index 6eff355..117757e 100644 --- a/anomavision/__init__.py +++ b/anomavision/__init__.py @@ -9,6 +9,7 @@ """ from .algorithm.common.feature_extraction import ResnetEmbeddingsExtractor +from .algorithm.efficientad import EfficientAD from .algorithm.padim import Padim from .algorithm.patchcore import PatchCore from .datasets.dataset import AnodetDataset diff --git a/anomavision/algorithm/efficientad/__init__.py b/anomavision/algorithm/efficientad/__init__.py new file mode 100644 index 0000000..470fc91 --- /dev/null +++ b/anomavision/algorithm/efficientad/__init__.py @@ -0,0 +1,5 @@ +"""Lightweight EfficientAD anomaly detection.""" + +from .efficientad import EfficientAD, build_efficientad_from_stats + +__all__ = ["EfficientAD", "build_efficientad_from_stats"] diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py new file mode 100644 index 0000000..e7a4ee3 --- /dev/null +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -0,0 +1,210 @@ +"""Lightweight EfficientAD-style teacher-student anomaly detection.""" + +from __future__ import annotations + +from typing import Dict, List, Optional, Tuple + +import torch +import torch.nn as nn +import torch.nn.functional as F + +from ..common.feature_extraction import ResnetEmbeddingsExtractor + + +class _Student(nn.Module): + """Small CNN that learns normal ResNet layer-1 features.""" + + def __init__(self, channels: int = 64) -> None: + super().__init__() + self.net = nn.Sequential( + nn.Conv2d(3, 32, 5, stride=2, padding=2), + nn.ReLU(inplace=True), + nn.Conv2d(32, channels, 5, stride=2, padding=2), + nn.ReLU(inplace=True), + nn.Conv2d(channels, channels, 3, padding=1), + ) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + return self.net(x) + + +class EfficientAD(torch.nn.Module): + """Minimal teacher-student EfficientAD implementation.""" + + def __init__( + self, + backbone: str = "resnet18", + device: torch.device = torch.device("cpu"), + layer_indices: Optional[List[int]] = None, + epochs: int = 5, + learning_rate: float = 1e-3, + threshold_percentile: float = 99.5, + ) -> None: + super().__init__() + if backbone != "resnet18": + raise ValueError( + "EfficientAD lightweight supports backbone='resnet18' only." + ) + self.device = torch.device(device) + self.backbone = backbone + self.layer_indices = [0] + self.epochs = max(1, int(epochs)) + self.learning_rate = float(learning_rate) + self.threshold_percentile = float(threshold_percentile) + if not 0.0 < self.threshold_percentile <= 100.0: + raise ValueError("threshold_percentile must be in (0, 100].") + self.threshold = 0.0 + + self.teacher = ResnetEmbeddingsExtractor(backbone, self.device) + for parameter in self.teacher.parameters(): + parameter.requires_grad_(False) + self.teacher.eval() + + self.student = _Student(64).to(self.device) + self._fitted = False + + @torch.no_grad() + def _teacher_features(self, batch: torch.Tensor) -> torch.Tensor: + features, width, height = self.teacher(batch, layer_indices=[0]) + return features.reshape(batch.shape[0], width, height, 64).permute(0, 3, 1, 2) + + def _loss(self, teacher: torch.Tensor, batch: torch.Tensor) -> torch.Tensor: + teacher = F.normalize(teacher, dim=1) + student = F.normalize(self.student(batch), dim=1) + return F.mse_loss(student, teacher) + + @torch.no_grad() + def _raw_scores( + self, batch: torch.Tensor, teacher: Optional[torch.Tensor] = None + ) -> Tuple[torch.Tensor, torch.Tensor]: + batch = batch.to(self.device, non_blocking=True) + if teacher is None: + teacher = self._teacher_features(batch) + teacher = F.normalize(teacher, dim=1) + student = F.normalize(self.student(batch), dim=1) + score = (teacher - student).pow(2).mean(dim=1) + + # Mean of the highest-scoring 1% of locations is more robust than a + # single maximum while remaining sensitive to small defects. + flat = score.flatten(1) + k = max(1, int(flat.shape[1] * 0.01)) + image_scores = flat.topk(k, dim=1).values.mean(dim=1) + + score_map = F.interpolate( + score.unsqueeze(1), + size=batch.shape[-2:], + mode="bilinear", + align_corners=False, + ).squeeze(1) + return image_scores, score_map + + def fit( + self, dataloader: torch.utils.data.DataLoader, extractions: int = 1 + ) -> None: + """Train the student and calibrate an adaptive threshold on normal images.""" + optimizer = torch.optim.Adam(self.student.parameters(), lr=self.learning_rate) + + # Compute frozen teacher features once. This avoids running ResNet for + # every epoch and makes CPU training considerably faster. + cached = [] + self.teacher.eval() + with torch.no_grad(): + for item in dataloader: + batch = item[0] if isinstance(item, (tuple, list)) else item + batch = batch.to(self.device, non_blocking=True) + cached.append((batch.detach(), self._teacher_features(batch).detach())) + + self.student.train() + for _ in range(self.epochs): + for batch, teacher in cached: + optimizer.zero_grad(set_to_none=True) + loss = self._loss(teacher, batch) + loss.backward() + optimizer.step() + + self.student.eval() + + # Calibrate from normal training scores using the exact score that is + # used for inference. + normal_scores = [] + for batch, teacher in cached: + scores, _ = self._raw_scores(batch, teacher) + normal_scores.append(scores.cpu()) + self.threshold = float( + torch.quantile(torch.cat(normal_scores), self.threshold_percentile / 100.0) + ) + self.threshold = max(self.threshold, 1e-8) + self._fitted = True + self.eval() + + @torch.no_grad() + def forward( + self, x: torch.Tensor, return_map: bool = True, export: bool = False + ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: + if not self._fitted: + raise RuntimeError("EfficientAD is not fitted. Call fit() first.") + image_scores, score_map = self._raw_scores(x) + + # Normalize scores by the training-derived threshold. This embeds the + # adaptive threshold into the exported model: >= 1.0 means anomalous. + scale = image_scores.new_tensor(self.threshold) + image_scores = image_scores / scale + score_map = score_map / scale + return image_scores, score_map if return_map else None + + def predict( + self, batch: torch.Tensor, export: bool = False + ) -> Tuple[torch.Tensor, torch.Tensor]: + return self.forward(batch, return_map=True, export=export) + + def to_device(self, device: torch.device) -> None: + self.device = torch.device(device) + self.teacher.to_device(self.device) + self.student.to(self.device) + + def save_statistics(self, path: str, half: Optional[bool] = False) -> None: + if not self._fitted: + raise RuntimeError("EfficientAD is not fitted. Call fit() first.") + student_state = { + key: ( + value.detach().cpu().half() + if half and value.is_floating_point() + else value.detach().cpu() + ) + for key, value in self.student.state_dict().items() + } + torch.save( + { + "student": student_state, + "backbone": self.backbone, + "layer_indices": self.layer_indices, + "epochs": self.epochs, + "learning_rate": self.learning_rate, + "threshold_percentile": self.threshold_percentile, + "threshold": self.threshold, + "model_type": "efficientad", + "dtype": "fp16" if half else "fp32", + }, + path, + ) + + +def build_efficientad_from_stats( + stats: Dict, device: str = "cpu", force_precision: Optional[str] = None +) -> EfficientAD: + """Build EfficientAD from a compact statistics artifact.""" + model = EfficientAD( + backbone=str(stats.get("backbone", "resnet18")), + device=torch.device(device), + layer_indices=[0], + epochs=int(stats.get("epochs", 5)), + learning_rate=float(stats.get("learning_rate", 1e-3)), + threshold_percentile=float(stats.get("threshold_percentile", 99.5)), + ) + state = stats["student"] + model.student.load_state_dict({key: value.float() for key, value in state.items()}) + model.student.eval() + model.threshold = max(float(stats.get("threshold", 0.0)), 1e-8) + model._fitted = True + model.eval() + return model diff --git a/anomavision/autopilot.py b/anomavision/autopilot.py index 9ac4ad2..642db7f 100644 --- a/anomavision/autopilot.py +++ b/anomavision/autopilot.py @@ -10,7 +10,7 @@ import time from html import escape from pathlib import Path -from typing import Any, Dict, Iterable, Optional +from typing import Any, Dict, Optional import numpy as np import torch @@ -27,9 +27,10 @@ make_localization_mask, ) +ALGORITHMS = ("padim", "patchcore", "efficientad") + def create_parser(add_help: bool = True) -> argparse.ArgumentParser: - """Build the ``anomavision autopilot`` argument parser.""" parser = argparse.ArgumentParser( description="Select, calibrate, profile, and package a production anomaly model.", add_help=add_help, @@ -55,18 +56,19 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: default=None, help="PatchCore model artifact (.pt/.pth/.onnx).", ) + parser.add_argument( + "--efficientad_model", + type=str, + default=None, + help="EfficientAD model artifact (.pt/.pth/.onnx).", + ) parser.add_argument("--device", choices=["auto", "cpu", "cuda"], default="auto") parser.add_argument("--batch_size", type=int, default=1) parser.add_argument("--num_workers", type=int, default=0) parser.add_argument("--warmup", type=int, default=3) parser.add_argument("--timing_batches", type=int, default=20) parser.add_argument("--target_latency_ms", type=float, default=None) - parser.add_argument( - "--validation_split", - type=float, - default=1.0, - help="Fraction of the complete labeled test split used for calibration; 1.0 uses every sample.", - ) + parser.add_argument("--validation_split", type=float, default=1.0) parser.add_argument("--output_dir", type=str, default="./production_package") parser.add_argument("--copy_config", action="store_true", default=True) return parser @@ -94,9 +96,8 @@ def _profile_model( timing_batches: int, ) -> Dict[str, Any]: wrapper = ModelWrapper(model_path, device) - iterator = iter(dataloader) try: - first = next(iterator) + first = next(iter(dataloader)) except StopIteration: wrapper.close() raise ValueError("The evaluation dataset is empty.") @@ -105,10 +106,10 @@ def _profile_model( wrapper.predict(first_batch) if device.startswith("cuda") and torch.cuda.is_available(): torch.cuda.synchronize() + timings = [] all_scores, all_maps, all_labels, all_masks = [], [], [], [] - count = 0 - for item in dataloader: + for count, item in enumerate(dataloader): batch, _, labels, masks = item batch = batch.to(device) measure = count < max(1, timing_batches) @@ -124,8 +125,8 @@ def _profile_model( all_maps.extend(_to_numpy(maps)) all_labels.extend(_to_numpy(labels).reshape(-1).tolist()) all_masks.extend(_to_numpy(masks)) - count += 1 wrapper.close() + scores_np = np.asarray(all_scores, dtype=np.float32) labels_np = np.asarray(all_labels, dtype=np.int64) maps_np = ( @@ -142,10 +143,11 @@ def _profile_model( image_auroc = ( image_metrics.get("auc_score") if len(np.unique(labels_np)) > 1 else None ) - pixel_auroc = None + masks_np = np.asarray(all_masks, dtype=np.float32) if masks_np.ndim == 4 and masks_np.shape[1] == 1: masks_np = masks_np[:, 0] + pixel_auroc = None localization = { "available": bool(len(maps_np) == len(labels_np) and maps_np.ndim == 3), "non_empty_fraction": None, @@ -166,16 +168,16 @@ def _profile_model( roc_auc_score(masks_np.reshape(-1) > 0.5, maps_np.reshape(-1)) ) except ValueError: - pixel_auroc = None + pass image_metrics["image_auroc"] = image_auroc image_metrics["pixel_auroc"] = pixel_auroc - anomaly_labels = (scores_np >= threshold).astype(np.uint8) + if localization["available"]: + anomaly_labels = (scores_np >= threshold).astype(np.uint8) loc_masks = make_localization_mask(maps_np, anomaly_labels).astype(bool) non_empty = loc_masks.reshape(len(loc_masks), -1).any(axis=1) area = loc_masks.reshape(len(loc_masks), -1).mean(axis=1) - anomaly_idx = labels_np == 1 - normal_idx = labels_np == 0 + anomaly_idx, normal_idx = labels_np == 1, labels_np == 0 localization["non_empty_fraction"] = float(non_empty.mean()) localization["mean_mask_area_fraction"] = float(area.mean()) localization["anomaly_non_empty_fraction"] = ( @@ -190,31 +192,26 @@ def _profile_model( localization["normal_mean_mask_area_fraction"] = ( float(area[normal_idx].mean()) if normal_idx.any() else None ) - if pixel_auroc is not None: - localization["verdict"] = ( + localization["verdict"] = ( + ( "healthy" if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 else "review false positives" ) - else: - localization["verdict"] = "maps available; pixel AUROC unavailable" - median_ms = ( - float(np.median(timings) * 1000 / max(1, dataloader.batch_size)) - if timings - else 0.0 - ) - p95_ms = ( - float(np.percentile(timings, 95) * 1000 / max(1, dataloader.batch_size)) - if timings - else 0.0 - ) + if pixel_auroc is not None + else "maps available; pixel AUROC unavailable" + ) + + batch_size = max(1, dataloader.batch_size or 1) + median_ms = float(np.median(timings) * 1000 / batch_size) if timings else 0.0 + p95_ms = float(np.percentile(timings, 95) * 1000 / batch_size) if timings else 0.0 return { "model_path": str(Path(model_path).resolve()), "model_format": Path(model_path).suffix.lower(), "threshold": float(threshold), "threshold_f1": float(threshold_f1), "metrics": { - k: (float(v) if isinstance(v, (float, np.floating)) else v) + k: float(v) if isinstance(v, (float, np.floating)) else v for k, v in image_metrics.items() }, "latency_ms": {"median": median_ms, "p95": p95_ms}, @@ -248,25 +245,21 @@ def _select( def _write_report(manifest: Dict[str, Any], output_dir: Path) -> None: - """Write a self-contained HTML dashboard and a Markdown fallback report.""" selected = manifest["selected_model"] selected_result = manifest["candidates"][selected] target = manifest.get("target_latency_ms") - cards = [] - rows = [] + cards, rows = [], [] for name, result in manifest["candidates"].items(): - metrics = result["metrics"] - loc = result["localization"] - is_selected = name == selected - status = "Selected" if is_selected else "Candidate" - status_class = "selected" if is_selected else "candidate" + metrics, loc = result["metrics"], result["localization"] + active = name == selected + status = "Selected" if active else "Candidate" cards.append( - f'

{escape(name.upper())}{status}
' + f'
{escape(name.upper())}{status}
' f'
{_format_metric(metrics.get("image_auroc"))} image AUROC
' f'
{_format_metric(metrics.get("pixel_auroc"))}pixel AUROC
{result["latency_ms"]["p95"]:.1f} msp95 latency
{_format_percent(loc.get("anomaly_non_empty_fraction"))}anomaly coverage
' ) rows.append( - f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("anomaly_non_empty_fraction"))}{_format_percent(loc.get("normal_false_positive_fraction"))}{result["threshold"]:.6f}' + f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("anomaly_non_empty_fraction"))}{_format_percent(loc.get("normal_false_positive_fraction"))}{result["threshold"]:.6f}' ) target_text = ( f"under {target:.1f} ms p95" @@ -274,41 +267,30 @@ def _write_report(manifest: Dict[str, Any], output_dir: Path) -> None: else "with the strongest measured accuracy/latency balance" ) environment_json = escape(json.dumps(manifest["environment"], indent=2)) - html = f""" - -AnomaVision Production Autopilot -
AnomaVision / Production Autopilot

Deployment confidence, before production.

Calibrated thresholds, hardware-aware profiling, and localization health checks in one reproducible report.

Selected: {escape(selected)}Class: {escape(str(manifest["dataset"]["class_name"]))}Samples: {manifest["dataset"]["samples"]}Device: {escape(str(manifest["environment"].get("device", "unknown")))}
Recommendation
Deploy {escape(selected)} with threshold {selected_result["threshold"]:.6f}. It was selected {escape(target_text)}. Recheck this threshold on a production validation set before release.

Candidate overview

Measured on the same validation data
{"".join(cards)}
-

Detailed comparison

Higher AUROC and anomaly coverage are better; lower false positives and latency are better
{"".join(rows)}
ModelImage AUROCPixel AUROCMedian msP95 msAnomaly coverageNormal false positivesThreshold
-

Localization health

Selected model maps available{"PASS" if selected_result["localization"]["available"] else "CHECK"}
Anomaly images with localization{_format_percent(selected_result["localization"].get("anomaly_non_empty_fraction"))}
Normal images with false-positive maps{_format_percent(selected_result["localization"].get("normal_false_positive_fraction"))}
Anomaly mean mask area{_format_percent(selected_result["localization"].get("anomaly_mean_mask_area_fraction"))}
Localization verdict{escape(str(selected_result["localization"].get("verdict", "N/A")))}

Anomaly coverage measures detected defect images. Normal false positives should remain low.

Deployment artifact

Artifact{escape(str(manifest["selected_artifact"]))}
Format{escape(str(selected_result["model_format"]))}
Preprocessing{escape(str(manifest["preprocessing"].get("resize")))} px
Target latency{escape(str(target)) if target is not None else "not set"}
+

Detailed comparison

Higher AUROC is better; lower false positives and latency are better
{"".join(rows)}
ModelImage AUROCPixel AUROCMedian msP95 msAnomaly coverageNormal false positivesThreshold
+

Localization health

Selected model maps available{"PASS" if selected_result["localization"]["available"] else "CHECK"}
Anomaly images with localization{_format_percent(selected_result["localization"].get("anomaly_non_empty_fraction"))}
Normal images with false-positive maps{_format_percent(selected_result["localization"].get("normal_false_positive_fraction"))}
Localization verdict{escape(str(selected_result["localization"].get("verdict", "N/A")))}

Deployment artifact

Artifact{escape(str(manifest["selected_artifact"]))}
Format{escape(str(selected_result["model_format"]))}
Preprocessing{escape(str(manifest["preprocessing"].get("resize")))} px
Target latency{escape(str(target)) if target is not None else "not set"}

Reproducibility environment

{environment_json}
""" (output_dir / "production_autopilot_report.html").write_text(html, encoding="utf-8") - - markdown = [ - "# AnomaVision Production Autopilot Report", - "", - f"**Selected model:** `{selected}`", - "", - "See `production_autopilot_report.html` for the full dashboard.", - "", - ] (output_dir / "localization_report.md").write_text( - "\\n".join(markdown), encoding="utf-8" + f"# AnomaVision Production Autopilot Report\n\n**Selected model:** `{selected}`\n\nSee `production_autopilot_report.html` for the full dashboard.\n", + encoding="utf-8", ) def run(args: argparse.Namespace) -> Dict[str, Any]: - """Run Autopilot and create a deployment package.""" cfg = load_config(args.config) dataset_path = args.dataset_path or cfg.get("dataset_path") or cfg.get("img_path") class_name = args.class_name or cfg.get("class_name") @@ -334,27 +316,33 @@ def run(args: argparse.Namespace) -> Dict[str, Any]: num_workers=args.num_workers, pin_memory=False, ) + candidates = {} for name, model_path in ( ("padim", args.padim_model), ("patchcore", args.patchcore_model), + ("efficientad", args.efficientad_model), ): if model_path: candidates[name] = _profile_model( model_path, dataloader, device, args.warmup, args.timing_batches ) if not candidates: - raise ValueError("Provide at least one of --padim_model or --patchcore_model.") + raise ValueError( + "Provide at least one model: --padim_model, --patchcore_model, or --efficientad_model." + ) + selected = _select(candidates, args.target_latency_ms) output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) selected_source = Path(candidates[selected]["model_path"]) packaged_model = output_dir / f"model{selected_source.suffix}" shutil.copy2(selected_source, packaged_model) + manifest = { - "schema_version": 2, + "schema_version": 3, "selected_model": selected, - "selected_artifact": str(packaged_model.name), + "selected_artifact": packaged_model.name, "dataset": { "path": str(Path(dataset_path).resolve()), "class_name": class_name, @@ -369,6 +357,7 @@ def run(args: argparse.Namespace) -> Dict[str, Any]: }, "candidates": candidates, "target_latency_ms": args.target_latency_ms, + "validation_split": args.validation_split, "environment": { "python": sys.version.split()[0], "platform": platform.platform(), @@ -379,6 +368,8 @@ def run(args: argparse.Namespace) -> Dict[str, Any]: (output_dir / "deployment_manifest.json").write_text( json.dumps(manifest, indent=2), encoding="utf-8" ) + if args.copy_config: + shutil.copy2(args.config, output_dir / Path(args.config).name) _write_report(manifest, output_dir) return manifest diff --git a/anomavision/inference/model/backends/torch_backend.py b/anomavision/inference/model/backends/torch_backend.py index 09b36be..578c87e 100644 --- a/anomavision/inference/model/backends/torch_backend.py +++ b/anomavision/inference/model/backends/torch_backend.py @@ -1,7 +1,5 @@ # inference/model/backends/torch_backend.py -""" -PyTorch backend implementation. -""" +"""PyTorch inference backend.""" from __future__ import annotations @@ -9,9 +7,8 @@ import torch -from anomavision.algorithm.padim.padim_lite import ( # stats-only .pth → runtime module - build_padim_from_stats, -) +from anomavision.algorithm.efficientad import build_efficientad_from_stats +from anomavision.algorithm.padim.padim_lite import build_padim_from_stats from anomavision.algorithm.patchcore import build_patchcore_from_stats from anomavision.utils import get_logger @@ -23,187 +20,85 @@ class TorchBackend(InferenceBackend): """Inference backend based on PyTorch.""" - def __init__( - self, - model_path: str, - device: str = "cpu", - *, - use_amp: bool = True, - ): - """Initialize PyTorch backend with automatic model type detection. - - Supports multiple PyTorch model formats including TorchScript, standard - PyTorch models, and PaDiM statistics files. Automatically handles model - preparation and optimization for inference. - - Args: - model_path (str): Path to PyTorch model file. Supported formats: - - TorchScript (.pts, .pt files) - - Standard PyTorch models (.pth with nn.Module) - - PaDiM statistics (.pth with mean/cov_inv dict) - device (str, optional): Target device. Automatically falls back to CPU - if CUDA is requested but unavailable. Defaults to "cpu". - use_amp (bool, optional): Enable automatic mixed precision (FP16) for - faster inference on supported GPUs. Defaults to True. - - Example: - >>> backend = TorchBackend("model.pts", "cuda", use_amp=True) - >>> backend = TorchBackend("stats.pth", "cpu") # PaDiM statistics - """ - - # --- Device selection: CPU-first; use CUDA only if requested & available + def __init__(self, model_path: str, device: str = "cpu", *, use_amp: bool = True): req = str(device or "cpu").lower() if req.startswith("cuda") and torch.cuda.is_available(): self.device = torch.device("cuda") else: - if req.startswith("cuda") and not torch.cuda.is_available(): + if req.startswith("cuda"): logger.warning("CUDA requested but not available; falling back to CPU.") self.device = torch.device("cpu") loaded_obj = None - - # 1) Try TorchScript first try: - logger.info("Trying torch.jit.load: %s", model_path) loaded_obj = torch.jit.load(model_path, map_location=self.device) logger.info("Loaded TorchScript model from %s", model_path) - except Exception as e_jit: - logger.info( - "torch.jit.load failed (%s). Falling back to torch.load.", str(e_jit) - ) + except Exception as exc: + logger.info("torch.jit.load failed (%s); falling back to torch.load.", exc) - # 2) Fallback: raw torch.load (may be nn.Module or a stats dict) if loaded_obj is None: - logger.info("Trying torch.load: %s", model_path) loaded_obj = torch.load( model_path, map_location=self.device, weights_only=False ) - logger.info("Loaded object type: %s", type(loaded_obj).__name__) - # 3) If it's a stats-only dict, build a PadimLite runtime module on CPU if isinstance(loaded_obj, dict) and { "mean", "cov_inv", "channel_indices", "layer_indices", "backbone", - }.issubset(loaded_obj.keys()): - logger.info("Detected PaDiM statistics artifact; building PadimLite.") + }.issubset(loaded_obj): model = build_padim_from_stats(loaded_obj, device=device) elif isinstance(loaded_obj, dict) and { "memory_bank", "layer_indices", "backbone", - }.issubset(loaded_obj.keys()): - logger.info("Detected PatchCore memory-bank artifact; building PatchCore.") + }.issubset(loaded_obj): model = build_patchcore_from_stats(loaded_obj, device=device) + elif isinstance(loaded_obj, dict) and { + "student", + "backbone", + "threshold", + }.issubset(loaded_obj): + logger.info("Detected EfficientAD statistics artifact.") + model = build_efficientad_from_stats(loaded_obj, device=device) else: model = loaded_obj - # 4) Unwrap DataParallel if present if hasattr(model, "module"): - logger.info("Unwrapping DataParallel container.") model = model.module - - # 5) Finalize if hasattr(model, "eval"): model.eval() - # Disable grads if parameters exist if hasattr(model, "parameters"): - for p in model.parameters(): - p.requires_grad_(False) + for parameter in model.parameters(): + parameter.requires_grad_(False) self.model = model - # AMP only makes sense on CUDA self.use_amp = bool(use_amp and self.device.type == "cuda") - def predict(self, batch: Batch) -> ScoresMaps: - """Run PyTorch inference with automatic mixed precision support. - - Executes inference using PyTorch with optional AMP acceleration. Handles - device placement and tensor conversion automatically. - - Args: - batch (Batch): Input batch of images. Converted to torch.Tensor if needed. - - Returns: - ScoresMaps: Tuple containing: - - scores (np.ndarray): Per-image anomaly scores - - maps (np.ndarray): Pixel-level anomaly maps - - Note: - Uses model.predict() method for consistent interface across all - model types and export formats. - - Example: - >>> batch = torch.randn(2, 3, 224, 224) - >>> scores, maps = backend.predict(batch) - """ - - logger.debug("Running inference via TorchBackend") - - if not isinstance(batch, torch.Tensor): - batch = torch.as_tensor(batch, dtype=torch.float32) - - batch = batch.to(self.device, non_blocking=True) - logger.debug("Torch input shape: %s", tuple(batch.shape)) - - autocast_ctx = ( + def _autocast(self): + return ( torch.autocast(device_type=self.device.type, dtype=torch.float16) - if self.use_amp and self.device.type == "cuda" + if self.use_amp else nullcontext() ) - with torch.inference_mode(), autocast_ctx: - # Always use .predict to match your runtime/export path + def predict(self, batch: Batch) -> ScoresMaps: + if not isinstance(batch, torch.Tensor): + batch = torch.as_tensor(batch, dtype=torch.float32) + batch = batch.to(self.device, non_blocking=True) + with torch.inference_mode(), self._autocast(): scores, maps = self.model.predict(batch) - - scores_np = scores.detach().cpu().numpy() - maps_np = maps.detach().cpu().numpy() - logger.debug("Torch output shapes: %s, %s", scores_np.shape, maps_np.shape) - return scores_np, maps_np + return scores.detach().cpu().numpy(), maps.detach().cpu().numpy() def close(self) -> None: - """Release PyTorch model and clear GPU memory. - - Removes model reference and triggers garbage collection. Essential for - preventing memory leaks in applications that load multiple models. - """ - self.model = None def warmup(self, batch, runs: int = 2) -> None: - """Warm up PyTorch backend with AMP support. - - Performs warmup inference runs using the same settings as production - inference, including automatic mixed precision if enabled. - - Args: - batch: Input batch for warmup. Converted to appropriate tensor format. - runs (int, optional): Number of warmup iterations. Defaults to 2. - - Example: - >>> warmup_batch = torch.randn(1, 3, 224, 224) - >>> backend.warmup(warmup_batch, runs=3) - """ - if not isinstance(batch, torch.Tensor): batch = torch.as_tensor(batch, dtype=torch.float32, device=self.device) else: batch = batch.to(self.device, non_blocking=True) - - autocast_ctx = ( - torch.autocast(device_type=self.device.type, dtype=torch.float16) - if self.use_amp and self.device.type == "cuda" - else nullcontext() - ) - - with torch.inference_mode(), autocast_ctx: + with torch.inference_mode(), self._autocast(): for _ in range(max(1, runs)): - _ = self.model.predict(batch) - - logger.info( - "TorchBackend warm-up completed (runs=%d, shape=%s).", - runs, - tuple(batch.shape), - ) + self.model.predict(batch) diff --git a/anomavision/train.py b/anomavision/train.py index 7761310..dc5c183 100644 --- a/anomavision/train.py +++ b/anomavision/train.py @@ -19,26 +19,22 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Train PaDiM (args OR config).", add_help=add_help ) - # meta parser.add_argument( "--config", type=str, default="config.yml", help="Path to config.yml/.json" ) - # dataset parser.add_argument( "--dataset_path", type=str, default=None, help='Path to the dataset folder containing "train/good" images.', ) - - # preprocessing parser.add_argument( "--resize", type=int, nargs="*", default=None, metavar=("W", "H"), - help="Resize before processing. Provide one value for a square resize (e.g., 256) or two values for width and height (e.g., 256 192). Omit to keep original size.", + help="Resize before processing.", ) parser.add_argument( "--crop_size", @@ -46,19 +42,19 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: nargs="*", default=None, metavar=("W", "H"), - help="Apply a center (or configured) crop. One value for a square crop (e.g., 224) or two for width and height (e.g., 224 224). Omit to disable cropping.", + help="Apply a center crop.", ) parser.add_argument( "--normalize", action="store_true", default=None, - help="Enable input normalization. If set, inputs are normalized using --norm_mean/--norm_std if provided (commonly ImageNet stats).", + help="Enable input normalization.", ) parser.add_argument( "--no_normalize", action="store_true", default=None, - help="Disable input normalization explicitly. If both --normalize and --no_normalize are set, this flag should take precedence in your code.", + help="Disable input normalization.", ) parser.add_argument( "--norm_mean", @@ -66,7 +62,7 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: nargs=3, default=None, metavar=("R", "G", "B"), - help="Per-channel RGB mean used when normalization is enabled. Example: 0.485 0.456 0.406.", + help="Per-channel RGB mean.", ) parser.add_argument( "--norm_std", @@ -74,128 +70,102 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: nargs=3, default=None, metavar=("R", "G", "B"), - help="Per-channel RGB standard deviation used when normalization is enabled. Example: 0.229 0.224 0.225.", + help="Per-channel RGB standard deviation.", ) - - # train parser.add_argument( "--backbone", type=str, choices=["resnet18", "wide_resnet50"], default=None, - help="Backbone network to use for feature extraction.", - ) - parser.add_argument( - "--batch_size", - type=int, - default=None, - help="Batch size used during training and inference.", + help="Backbone network.", ) + parser.add_argument("--batch_size", type=int, default=None, help="Batch size.") parser.add_argument( "--feat_dim", type=int, default=None, - help="Number of random feature dimensions to keep.", + help="Number of PaDiM feature dimensions to keep.", ) parser.add_argument( "--layer_indices", type=int, nargs="+", default=None, - help="List of layer indices to extract features from, e.g., 0 1 2.", + help="Backbone feature layers.", ) parser.add_argument( "--coreset_ratio", type=float, default=None, - help="PatchCore memory-bank fraction to retain (0, 1].", + help="PatchCore memory-bank fraction.", ) parser.add_argument( "--max_memory_patches", type=int, default=None, - help="Maximum PatchCore memory-bank size; omit for no cap.", + help="Maximum PatchCore memory-bank size.", ) parser.add_argument( - "--patch_grid", - type=int, - default=None, - help="PatchCore pooled grid size; use a smaller value for lower latency.", + "--patch_grid", type=int, default=None, help="PatchCore pooled grid size." ) parser.add_argument( "--search_chunk_size", type=int, default=None, - help="PatchCore query chunk size used to bound nearest-neighbor memory.", + help="PatchCore query chunk size.", ) parser.add_argument( "--coreset_method", type=str, choices=["kcenter", "random"], default=None, - help="PatchCore coreset selection strategy; kcenter is the diverse default.", + help="PatchCore coreset method.", ) parser.add_argument( - "--coreset_seed", - type=int, - default=None, - help="Seed used for deterministic PatchCore coreset selection.", + "--coreset_seed", type=int, default=None, help="PatchCore coreset seed." ) parser.add_argument( - "--output_model", - type=str, - default=None, - help="Filename to save the PT model.", + "--output_model", type=str, default=None, help="Filename to save the PT model." + ) + parser.add_argument("--run_name", type=str, default=None, help="Experiment name.") + parser.add_argument( + "--model_data_path", type=str, default=None, help="Model output directory." ) + parser.add_argument("--algorithm", type=str, default=None, help="Algorithm name.") parser.add_argument( - "--run_name", + "--log_level", type=str, + choices=["DEBUG", "INFO", "WARNING", "ERROR"], default=None, - help="Experiment name for this training run.", + help="Logging level.", ) parser.add_argument( - "--model_data_path", - type=str, + "--efficientad_epochs", + type=int, default=None, - help="Directory to save model distributions and PT file.", + help="EfficientAD student training epochs.", ) parser.add_argument( - "--algorithm", - type=str, + "--efficientad_learning_rate", + type=float, default=None, - help="Algorithm name (e.g., padim, patchcore).", + help="EfficientAD student learning rate.", ) parser.add_argument( - "--log_level", - type=str, - choices=["DEBUG", "INFO", "WARNING", "ERROR"], + "--efficientad_threshold_percentile", + type=float, default=None, - help="Logging level (default: INFO).", + help="Percentile of normal EfficientAD scores used for the adaptive threshold.", ) - return parser def run_training(args): - """ - Executes the training pipeline. - - Args: - args: Namespace object containing command line arguments. - - Returns: - padim (AnomaVision.Padim): The trained model object. - config (edict): The final merged configuration. - run_dir (Path): The directory where artifacts were saved. - dataloaders (dict): Dictionary containing the 'train' DataLoader. - """ cfg = load_config(args.config) - - # Merge config with CLI args config = edict(merge_config(args, cfg)) setup_logging(enabled=True, log_level=config.log_level, log_to_file=True) - logger = get_logger("anomavision.train") # Force it into anomavision hierarchy + logger = get_logger("anomavision.train") if not config.dataset_path: error_msg = "dataset.path is required (via --dataset_path or config.common.dataset_path)" @@ -203,7 +173,6 @@ def run_training(args): raise ValueError(error_msg) t0 = time.perf_counter() - logger.info( "Image processing: resize=%s, crop_size=%s, normalize=%s", config.resize, @@ -213,7 +182,6 @@ def run_training(args): if config.normalize: logger.info("Normalization: mean=%s, std=%s", config.norm_mean, config.norm_std) - # Resolve output run dir once run_dir = increment_path( Path(config.model_data_path) / config.algorithm @@ -223,20 +191,10 @@ def run_training(args): mkdir=True, ) - # === Dataset === - # Handle the 'class_name' logic safely. - # If dataset_path ends with the class name, use parent? - # Original logic assumes dataset_path is the container of class folders OR the class folder itself? - # Original code: os.path.join(realpath(dataset_path), config.class_name, "train", "good") - # This implies dataset_path is the root (e.g. MVTec root) and config.class_name is "bottle" - root = os.path.join( os.path.realpath(config.dataset_path), config.class_name, "train", "good" ) - if not os.path.isdir(root): - # Fallback check: maybe dataset_path ALREADY points to the class folder? - # This makes it more robust for different input styles potential_root = os.path.join( os.path.realpath(config.dataset_path), "train", "good" ) @@ -254,22 +212,16 @@ def run_training(args): mean=config.norm_mean, std=config.norm_std, ) - if len(ds) == 0: - error_msg = f"No training images found in {root}" - logger.error(error_msg) - raise ValueError(error_msg) + raise ValueError(f"No training images found in {root}") dl = DataLoader(ds, batch_size=int(config.batch_size), shuffle=False) logger.info("dataset: %d images | batch_size=%d", len(ds), config.batch_size) - # === Device === device = torch.device("cuda" if torch.cuda.is_available() else "cpu") logger.info( "device: %s (cuda_available=%s)", device.type, torch.cuda.is_available() ) - - # === Model & Train === logger.info( "cfg: algorithm=%s | backbone=%s | layers=%s", config.algorithm, @@ -289,6 +241,17 @@ def run_training(args): coreset_method=config.get("coreset_method", "kcenter"), coreset_seed=int(config.get("coreset_seed", 42)), ) + elif str(config.algorithm).lower() == "efficientad": + model = anomavision.EfficientAD( + backbone=config.backbone, + device=device, + layer_indices=config.get("layer_indices", [0]), + epochs=int(config.get("efficientad_epochs", 5)), + learning_rate=float(config.get("efficientad_learning_rate", 1e-3)), + threshold_percentile=float( + config.get("efficientad_threshold_percentile", 99.5) + ), + ) else: model = anomavision.Padim( backbone=config.backbone, @@ -301,11 +264,9 @@ def run_training(args): model.fit(dl) logger.info("fit: completed in %.2fs", time.perf_counter() - t_fit) - # === Save === model_path = Path(run_dir) / config.output_model torch.save(model, str(model_path)) - # Save a compact statistics/memory-bank artifact for deployment. stats_path = model_path.with_suffix(".pth") try: model.save_statistics(str(stats_path), half=True) @@ -313,13 +274,9 @@ def run_training(args): except Exception as e: logger.warning("saving slim statistics failed: %s", e) - # snapshot the effective configuration save_args_to_yaml(config, str(Path(run_dir) / "config.yml")) - logger.info("saved: model=%s, config=%s", model_path, Path(run_dir) / "config.yml") logger.info("=== Training done in %.2fs ===", time.perf_counter() - t0) - - # Return objects for external usage (e.g. MLOps pipeline) return model, config, run_dir, {"train": dl} @@ -327,17 +284,13 @@ def main(args=None): try: if args is None: args = create_parser().parse_args() - - # Optional git check — silently skipped if not in a repo or no network try: checker = GitStatusChecker() if checker.is_repo(): checker.check_status() except Exception: - pass # Never block training over a git check - + pass run_training(args) - except Exception: get_logger(__name__).exception("Fatal error during training.") sys.exit(1) diff --git a/config.yml b/config.yml index 0af6c53..937d4f9 100644 --- a/config.yml +++ b/config.yml @@ -1,124 +1,139 @@ # ========================= # Dataset / preprocessing (shared by train, detect, eval, stream) # ========================= -dataset_path: "D:/01-DATA" # Root dataset folder (MVTec-style: contains train/test subfolders) -class_name: "bottle" # Class name for MVTec dataset -resize: [224, 224] # Resize dimensions before processing [width, height] -crop_size: # Final crop size [width, height] -normalize: true # Whether to normalize images -no_normalize: false # Inverse flag (used by CLI) – keep in sync with "normalize" -norm_mean: [0.485, 0.456, 0.406] # Mean values for normalization (ImageNet default) -norm_std: [0.229, 0.224, 0.225] # Standard deviation for normalization (ImageNet default) +dataset_path: "D:/01-DATA" # Root dataset directory (contains train/test folders) +class_name: "bottle" # Dataset class to train/evaluate +resize: [224, 224] # Input image size [width, height] +crop_size: # Optional center crop size [width, height] +normalize: true # Apply ImageNet normalization +no_normalize: false # Explicitly disable normalization when true +norm_mean: [0.485, 0.456, 0.406] # Normalization mean for RGB channels +norm_std: [0.229, 0.224, 0.225] # Normalization standard deviation for RGB channels # ========================= # Model / training # ========================= -backbone: "resnet18" # Backbone CNN architecture (resnet18 | wide_resnet50) -algorithm: "padim" # Algorithm to use: padim or patchcore -coreset_ratio: 0.02 # Ultra-light PatchCore memory fraction -max_memory_patches: 2048 # Hard memory-bank cap for low latency -patch_grid: 14 # Pool feature maps to at most 14x14 patches -search_chunk_size: 1024 # Bound nearest-neighbor working memory -coreset_method: "kcenter" # PatchCore selection: kcenter or random -coreset_seed: 42 # Reproducible k-center initialization -feat_dim: 50 # Feature dimension size for embedding -layer_indices: [0] # Which backbone layers to extract features from (0,1,2,3) -model_data_path: "./distributions" # Path to store/load model-related data -model: "model.pt" # File name for saved model (used by detect/eval/export) -output_model: "model.pt" # File name for saving trained model (train.py expects this) -batch_size: 2 # Training/evaluation/inference batch size -device: "auto" # Device to run on: "cpu", "cuda", or "auto" +backbone: "resnet18" # Backbone network: resnet18 | wide_resnet50 +algorithm: "efficientad" # Anomaly algorithm: padim | patchcore | efficientad +coreset_ratio: 0.02 # PatchCore fraction of patches kept in memory +max_memory_patches: 2048 # Maximum number of PatchCore memory-bank patches +patch_grid: 14 # PatchCore feature-map grid size +search_chunk_size: 1024 # PatchCore nearest-neighbor search chunk size +coreset_method: "kcenter" # PatchCore coreset method: kcenter | random +coreset_seed: 42 # Random seed for PatchCore coreset selection +feat_dim: 50 # PaDiM number of feature dimensions to keep +layer_indices: [0] # Backbone feature layers used by PaDiM/PatchCore/EfficientAD +model_data_path: "./distributions" # Directory for trained models and statistics +model: "model.pt" # Model filename used by detect/eval/export +output_model: "model.pt" # Filename written by the train command +batch_size: 2 # Training/inference batch size +device: "auto" # Device to run on: auto | cpu | cuda + +# ========================= +# Lightweight EfficientAD +# ========================= +efficientad_epochs: 5 # Number of student training epochs +efficientad_learning_rate: 0.001 # Student optimizer learning rate +efficientad_threshold_percentile: 99.5 # Percentile of normal scores used to learn the adaptive threshold + +# ========================= +# Inference performance tests +# ========================= +# Opt-in benchmark used to catch inference regressions for every algorithm. +# Run with: pytest tests/test_inference_performance.py -s +inference_benchmark: + enabled: true # Set true to run model benchmarks + warmup_runs: 10 # Warm-up predictions excluded from timing + test_runs: 50 # Timed predictions per algorithm + batch_size: 2 # Batch size used by the benchmark + max_inference_ms: # Maximum allowed pure inference time per batch + padim: 100.0 + patchcore: 100.0 + efficientad: 150.0 + max_regression_percent: 25.0 # Allowed regression if a baseline is configured + baseline_inference_ms: # Optional per-algorithm baseline; null disables regression check + padim: null + patchcore: null + efficientad: 47.15 # ========================= # Logging / run metadata # ========================= -log_level: "INFO" # Logging level: DEBUG, INFO, WARNING, ERROR -run_name: "anomav_exp" # Name of experiment run (used for organizing results) -detailed_timing: false # Enable detailed timing measurements +log_level: "INFO" # Logging level: DEBUG | INFO | WARNING | ERROR +run_name: "anomav_exp" # Experiment name used in output directories +detailed_timing: false # Enable detailed timing information # ========================= -# Visualization (shared by detect & eval) +# Visualization (detect/eval) # ========================= -enable_visualization: true # Enable visualization during inference/evaluation -save_visualizations: true # Save visualization results to disk -viz_output_dir: "./visualizations/" # Directory to save visualization results -viz_alpha: 0.5 # Transparency factor for overlay heatmaps -viz_padding: 40 # Padding added around visualization -viz_color: "128,0,128" # RGB color for visualization overlays +enable_visualization: true # Enable anomaly visualization +save_visualizations: true # Save generated visualizations to disk +viz_output_dir: "./visualizations/" # Directory for visualization output +viz_alpha: 0.5 # Heatmap overlay transparency +viz_padding: 40 # Extra visualization border/padding +viz_color: "128,0,128" # RGB color used for anomaly overlays # ========================= -# Inference (detect.py) +# Inference (detect) # ========================= -img_path: "D:/01-DATA/test" # Path to test images for inference -thresh: null # Legacy fallback; prefer algorithm-specific thresholds -thresh_padim: 13.0 # PaDiM score threshold; null lets eval auto-select -thresh_patchcore: 0.25 # PatchCore score threshold; null lets eval auto-select -num_workers: 1 # Number of workers for dataloader -pin_memory: false # Use pinned memory for faster GPU transfers -overwrite: false # Overwrite existing run directory without auto-incrementing +img_path: "D:/01-DATA/test" # Input image or directory for detection +thresh: null # Generic legacy threshold fallback +thresh_padim: 13.0 # Fixed PaDiM anomaly threshold +thresh_patchcore: 0.25 # Fixed PatchCore anomaly threshold +thresh_efficientad: 1.0 # EfficientAD threshold; 1.0 means the model's adaptive training threshold +num_workers: 1 # Number of data-loader worker processes +pin_memory: false # Pin CPU memory for faster GPU transfers +overwrite: false # Overwrite an existing output directory # ========================= -# Evaluation (eval.py) +# Evaluation (eval) # ========================= -memory_efficient: true # Use memory efficient evaluation mode +memory_efficient: true # Reduce memory usage during evaluation # ========================= -# Export (export.py) -# ========================= -format: "all" # onnx, tensorrt, torchscript, openvino, all -opset: 18 # ONNX opset version -precision: "auto" # ONNX/TorchScript precision: auto, fp16, fp32 -tensorrt_precision: "fp16" # TensorRT precision: fp32, fp16, int8 -dynamic_batch: true # Allow dynamic batch size in exported model -static_batch: false # Disable dynamic batch size (if true) -min_batch: 1 # TensorRT dynamic profile minimum batch -opt_batch: 1 # TensorRT dynamic profile optimal batch -max_batch: 4 # TensorRT dynamic profile maximum batch -workspace_gb: 2.0 # TensorRT workspace limit in GiB -calib_dir: null # INT8 calibration image directory; auto-derived when null -calib_samples: 100 # Maximum real images used for INT8 calibration -quantize_dynamic: false # Also write dynamically quantized INT8 ONNX -quantize_static: false # Also write statically quantized INT8 ONNX -optimize: false # Enable mobile optimization for TorchScript -output_path: null # Optional explicit output filename -half: false # Legacy compatibility field -int8: false # Legacy compatibility field; use tensorrt_precision: int8 +# Export (export) +# ========================= +format: "all" # Export format: onnx | tensorrt | torchscript | openvino | all +opset: 18 # ONNX operator-set version +precision: "auto" # Export precision: auto | fp16 | fp32 +tensorrt_precision: "fp16" # TensorRT precision: fp32 | fp16 | int8 +dynamic_batch: true # Allow variable batch size in exported models +static_batch: false # Force a fixed batch size when true +min_batch: 1 # Minimum TensorRT dynamic batch size +opt_batch: 1 # Optimal TensorRT dynamic batch size +max_batch: 4 # Maximum TensorRT dynamic batch size +workspace_gb: 2.0 # TensorRT workspace memory limit in GiB +calib_dir: null # Directory containing INT8 calibration images +calib_samples: 100 # Maximum number of calibration images +quantize_dynamic: false # Also generate dynamically quantized INT8 ONNX +quantize_static: false # Also generate statically quantized INT8 ONNX +optimize: false # Enable TorchScript/mobile optimization +output_path: null # Optional explicit export output path +half: false # Legacy FP16 compatibility option +int8: false # Legacy INT8 compatibility option # ========================= -# Streaming Configuration +# Streaming # ========================= -stream_mode: false # true = real-time streaming, false = static dataset - +stream_mode: false # Enable real-time streaming mode stream_source: - type: "webcam" # webcam | video | mqtt | tcp - - # Webcam settings (type: webcam) - camera_id: 0 # Camera device index - - # Video file settings (type: video) - video_path: "path/to/video.mp4" # Path to video file - loop: false # Loop video when it ends - - # MQTT settings (type: mqtt) - broker: "localhost" # MQTT broker hostname/IP - port: 1883 # MQTT broker port - topic: "camera/frames" # Topic to subscribe to - client_id: null # Optional client ID - keepalive: 60 # Keepalive interval (seconds) - qos: 0 # QoS level (0, 1, or 2) - max_queue_size: 10 # Max buffered frames - read_timeout: 1.0 # Timeout for reading frames (seconds) - - # TCP settings (type: tcp) - host: "192.168.1.100" # TCP server hostname/IP - port: 8080 # TCP server port (matches code) - recv_timeout: 1.0 # Socket timeout for recv (seconds) - header_size: 4 # Length header size in bytes - max_message_size: 10485760 # Max payload size (10MB) - - -# Streaming processing settings -stream_max_frames: null # Max frames to process (null = infinite) -stream_display_fps: true # Show FPS during streaming -stream_save_detections: true # Save detected anomalies to disk -stream_detection_dir: "./stream_detections/" # Directory for saved detections + type: "webcam" # Source type: webcam | video | mqtt | tcp + camera_id: 0 # Webcam device index + video_path: "path/to/video.mp4" # Video file path + loop: false # Restart video when it reaches the end + broker: "localhost" # MQTT broker hostname/IP + port: 1883 # MQTT broker port + topic: "camera/frames" # MQTT topic containing frames + client_id: null # Optional MQTT client ID + keepalive: 60 # MQTT keepalive interval in seconds + qos: 0 # MQTT quality of service: 0 | 1 | 2 + max_queue_size: 10 # Maximum buffered streaming frames + read_timeout: 1.0 # Streaming read timeout in seconds + host: "192.168.1.100" # TCP server hostname/IP + recv_timeout: 1.0 # TCP receive timeout in seconds + header_size: 4 # TCP message length-header size in bytes + max_message_size: 10485760 # Maximum TCP message size in bytes +stream_max_frames: null # Maximum frames to process; null means unlimited +stream_display_fps: true # Display FPS during streaming +stream_save_detections: true # Save streaming anomaly detections +stream_detection_dir: "./stream_detections/" # Output directory for streaming detections diff --git a/tests/test_inference_performance.py b/tests/test_inference_performance.py new file mode 100644 index 0000000..2faed4d --- /dev/null +++ b/tests/test_inference_performance.py @@ -0,0 +1,138 @@ +"""Config-driven pure inference performance regression tests. + +These tests intentionally measure only ``model.predict(batch)``. Image loading, +preprocessing, postprocessing, visualization, and result accumulation are kept +outside the timed section so the numbers match the ``Pure inference FPS`` and +``Average inference time`` reported by ``anomavision.detect``. + +Enable them in ``config.yml`` with ``inference_benchmark.enabled: true``. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest +import torch +from torch.utils.data import DataLoader + +import anomavision +from anomavision.config import load_config +from anomavision.general import determine_device +from anomavision.inference.model.wrapper import ModelWrapper + +CONFIG_PATH = Path(__file__).resolve().parents[1] / "config.yml" +ALGORITHMS = ("padim", "patchcore", "efficientad") + + +def _cuda_sync(device: str) -> None: + """Synchronize CUDA before/after a timed inference call.""" + if device == "cuda" and torch.cuda.is_available(): + torch.cuda.synchronize() + + +def _load_benchmark_config() -> dict: + config = load_config(str(CONFIG_PATH)) + benchmark = config.get("inference_benchmark", {}) + if not benchmark.get("enabled", False): + pytest.skip( + "Inference benchmark disabled. Set inference_benchmark.enabled: true in config.yml." + ) + return config + + +@pytest.mark.parametrize("algorithm", ALGORITHMS) +def test_inference_performance(algorithm: str) -> None: + """Benchmark PaDiM, PatchCore, and EfficientAD independently.""" + config = _load_benchmark_config() + benchmark = config["inference_benchmark"] + + device = determine_device(str(config.get("device", "auto"))) + model_path = ( + Path(config["model_data_path"]) + / algorithm + / config["class_name"] + / config["run_name"] + / config["model"] + ) + dataset_path = Path(config["img_path"]) + + if not model_path.exists(): + pytest.skip(f"{algorithm}: model not found: {model_path}") + if not dataset_path.exists(): + pytest.skip(f"{algorithm}: dataset not found: {dataset_path}") + + batch_size = int(benchmark.get("batch_size", config.get("batch_size", 1))) + warmup_runs = int(benchmark.get("warmup_runs", 10)) + test_runs = int(benchmark.get("test_runs", 50)) + + if batch_size < 1 or warmup_runs < 0 or test_runs < 1: + raise ValueError( + "inference_benchmark.batch_size >= 1, warmup_runs >= 0, test_runs >= 1" + ) + + dataset = anomavision.AnodetDataset( + str(dataset_path), + resize=config.get("resize"), + crop_size=config.get("crop_size"), + normalize=config.get("normalize", True), + mean=config.get("norm_mean"), + std=config.get("norm_std"), + ) + dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=False) + + try: + first = next(iter(dataloader)) + except StopIteration: + pytest.fail(f"{algorithm}: benchmark dataset is empty: {dataset_path}") + + batch = first[0] + if device == "cuda": + batch = batch.half() + batch = batch.to(device) + + model = ModelWrapper(str(model_path), device) + try: + for _ in range(warmup_runs): + model.predict(batch) + _cuda_sync(device) + + timings_ms = [] + for _ in range(test_runs): + _cuda_sync(device) + start = time.perf_counter() + model.predict(batch) + _cuda_sync(device) + timings_ms.append((time.perf_counter() - start) * 1000.0) + + average_ms = sum(timings_ms) / len(timings_ms) + pure_fps = batch_size * 1000.0 / average_ms + + max_ms = benchmark.get("max_inference_ms", {}).get(algorithm) + baseline_ms = benchmark.get("baseline_inference_ms", {}).get(algorithm) + max_regression = float(benchmark.get("max_regression_percent", 25.0)) + + allowed_ms = None if max_ms is None else float(max_ms) + if baseline_ms is not None: + regression_limit = float(baseline_ms) * (1.0 + max_regression / 100.0) + allowed_ms = ( + regression_limit + if allowed_ms is None + else min(allowed_ms, regression_limit) + ) + + print( + f"\n{algorithm.upper()} | " + f"Pure inference FPS: {pure_fps:.2f} images/sec | " + f"Average inference time: {average_ms:.2f} ms/batch | " + f"Throughput: {pure_fps:.2f} images/sec (batch size: {batch_size})" + ) + + if allowed_ms is not None: + assert average_ms <= allowed_ms, ( + f"{algorithm} inference regression: {average_ms:.2f} ms/batch " + f"> allowed {allowed_ms:.2f} ms/batch." + ) + finally: + model.close()