diff --git a/README.md b/README.md
index 8e93371..b7d3e28 100644
--- a/README.md
+++ b/README.md
@@ -23,10 +23,11 @@
AnomaVision is a computer vision project for finding **defects and unusual patterns** in 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.
+- **EfficientAD** — a student/teacher model with a compact reconstruction branch for fast anomaly detection.
You only need **normal (`good`) images** to train the anomaly detector.
@@ -43,6 +44,7 @@ You only need **normal (`good`) images** to train the anomaly detector.
- Create anomaly heatmaps showing where the problem is.
- Export models to **ONNX, OpenVINO, and TensorRT**.
- Export and compile **PaDiM and PatchCore to XModel for the AMD/Xilinx Kria KV260**.
+- Switch between PaDiM, PatchCore, and EfficientAD without changing the CLI workflow.
## Quick start
@@ -53,31 +55,19 @@ You only need **normal (`good`) images** to train the anomaly detector.
```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
+source .venv/bin/activate # Windows: .venv\\Scripts\\Activate.ps1
uv sync --extra cpu # CPU
uv sync --extra cu121 # CUDA 12.1
```
----
-
#### Option B — From PyPI (production / quick start)
```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
+uv pip install "anomavision[cu121]"
```
-
For other environments, see [Installation](docs/installation.md).
### 2. Prepare your images
@@ -88,9 +78,6 @@ Use a simple MVTec-style folder structure:
dataset/
└── bottle/
├── ground_truth/
- │ ├── broken_large/
- │ ├── broken_small/
- │ └── contamination/
├── test/
│ ├── broken_large/
│ ├── broken_small/
@@ -102,33 +89,56 @@ dataset/
Training uses the **good** images. Test images can contain defects.
-### 3. Train
+### 3. Choose an algorithm
+
+The existing configuration format works unchanged:
+
+```yaml
+algorithm: padim
+```
+
+Switch to EfficientAD by changing one value:
+
+```yaml
+algorithm: efficientad
+```
+
+You can also use the native model selector:
+
+```yaml
+model:
+ name: efficientad
+```
-Create or edit `config.yml` and point `dataset_path` to your dataset.
+The CLI commands remain the same.
-Then run:
+### 4. Train
```bash
anomavision train --config config.yml
```
-PaDiM is the default model. For PatchCore, set `algorithm: patchcore` in the configuration.
+For a quick EfficientAD configuration, see [`examples/efficientad_cpu.yml`](examples/efficientad_cpu.yml).
-### 4. Detect
+### 5. Detect
```bash
-anomavision detect --config config.yml --img_path ./dataset/bottle/test
+anomavision detect --config config.yml --model model.pt --img_path ./test_images
```
-### 5. Export
+### 6. Export
-For a portable model, ONNX is a good place to start:
+```bash
+anomavision export --config config.yml --model model.pt --format onnx
+```
+
+### 7. Evaluate
```bash
-anomavision export --config config.yml --format onnx
+anomavision eval --config config.yml --model model.pt --class_name bottle
```
-For more export options, see [Export and deployment](docs/production_deployment.md).
+For EfficientAD-specific options and limitations, see [EfficientAD](docs/efficientad.md).
## KV260 support
@@ -142,13 +152,10 @@ 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)**
+The complete setup and commands are in the [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.
-
## 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.
@@ -156,18 +163,18 @@ The complete setup and commands are in:
Train both candidate models first, then run the complete labeled split on CPU:
```bash
-anomavision autopilot \
- --config config.yml \
- --padim_model ./distributions/padim/bottle/anomav_exp/model.pt \
- --patchcore_model ./distributions/patchcore/bottle/anomav_exp/model.pt \
- --device cpu \
- --validation_split 1.0 \
- --target_latency_ms 50 \
+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.onnx `
+ --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.
-
+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. See [`docs/production_deployment.md`](docs/production_deployment.md) for details.
## Documentation
@@ -176,6 +183,7 @@ Open `production_package/production_autopilot_report.html` to see the selected m
| Quick start | [`docs/quickstart.md`](docs/quickstart.md) |
| Installation | [`docs/installation.md`](docs/installation.md) |
| CLI and configuration | [`docs/cli.md`](docs/cli.md), [`docs/config.md`](docs/config.md) |
+| EfficientAD | [`docs/efficientad.md`](docs/efficientad.md) |
| Python API | [`docs/api.md`](docs/api.md) |
| KV260 / XModel | [`docs/kv260_xmodel.md`](docs/kv260_xmodel.md) |
| Production deployment | [`docs/production_deployment.md`](docs/production_deployment.md) |
@@ -186,23 +194,18 @@ 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
import anomavision
+from torch.utils.data import DataLoader
train_set = anomavision.AnodetDataset("./dataset/bottle/train/good")
-train_loader = DataLoader(train_set, batch_size=16, shuffle=False)
-
-model = anomavision.Padim(backbone="resnet18", device=torch.device("cpu"))
-model.fit(train_loader)
+train_loader = DataLoader(train_set, batch_size=1, shuffle=False)
-batch = next(iter(train_loader))
-if isinstance(batch, (tuple, list)):
- batch = batch[0]
+model = anomavision.EfficientAD(device=torch.device("cpu"))
+model.fit(train_loader, epochs=1)
+batch = next(iter(train_loader))[0]
scores, maps = model.predict(batch)
```
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..eb7f485
--- /dev/null
+++ b/anomavision/algorithm/efficientad/__init__.py
@@ -0,0 +1,5 @@
+"""EfficientAD anomaly detection algorithm."""
+
+from .efficientad import EfficientAD
+
+__all__ = ["EfficientAD"]
diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py
new file mode 100644
index 0000000..2cfae73
--- /dev/null
+++ b/anomavision/algorithm/efficientad/efficientad.py
@@ -0,0 +1,227 @@
+"""EfficientAD anomaly detection algorithm."""
+
+from __future__ import annotations
+
+from typing import Optional, Tuple
+
+import torch
+import torch.nn.functional as F
+from torch import nn
+from torchvision.models import EfficientNet_B0_Weights, efficientnet_b0
+
+
+class _FeatureTeacher(nn.Module):
+ def __init__(self, pretrained: bool = True) -> None:
+ super().__init__()
+ weights = EfficientNet_B0_Weights.DEFAULT if pretrained else None
+ net = efficientnet_b0(weights=weights)
+ self.features = nn.Sequential(*list(net.features[:6]))
+ self.out_channels = 112
+ for p in self.parameters():
+ p.requires_grad_(False)
+ self.eval()
+
+ @torch.no_grad()
+ def forward(self, x):
+ return self.features(x)
+
+
+class _Student(nn.Module):
+ def __init__(self, out_channels=112):
+ super().__init__()
+ self.net = nn.Sequential(
+ nn.Conv2d(3, 64, 3, 2, 1),
+ nn.BatchNorm2d(64),
+ nn.ReLU(inplace=True),
+ nn.Conv2d(64, 96, 3, 2, 1),
+ nn.BatchNorm2d(96),
+ nn.ReLU(inplace=True),
+ nn.Conv2d(96, 112, 3, 2, 1),
+ nn.BatchNorm2d(112),
+ nn.ReLU(inplace=True),
+ nn.Conv2d(112, out_channels, 3, 2, 1),
+ )
+
+ def forward(self, x):
+ return self.net(x)
+
+
+class EfficientAD(nn.Module):
+ """EfficientAD-compatible teacher/student anomaly detector for AnomaVision."""
+
+ def __init__(
+ self,
+ device=torch.device("cpu"),
+ model_size="s",
+ lr=1e-4,
+ weight_decay=1e-5,
+ pretrained_teacher=True,
+ teacher_weights=None,
+ feature_weight=1.0,
+ reconstruction_weight=0.0,
+ threshold_quantile=0.995,
+ ):
+ super().__init__()
+ model_size = str(model_size).lower()
+ if model_size not in {"s", "m", "small", "medium"}:
+ raise ValueError("EfficientAD model_size must be one of: s, m")
+ if not 0.0 < float(threshold_quantile) < 1.0:
+ raise ValueError("threshold_quantile must be between 0 and 1")
+ self.device = torch.device(device)
+ self.model_size = "m" if model_size in {"m", "medium"} else "s"
+ self.lr = float(lr)
+ self.weight_decay = float(weight_decay)
+ self.feature_weight = float(feature_weight)
+ self.reconstruction_weight = float(reconstruction_weight)
+ self.threshold_quantile = float(threshold_quantile)
+ self.teacher = _FeatureTeacher(pretrained_teacher)
+ if teacher_weights:
+ self.teacher.load_state_dict(
+ torch.load(teacher_weights, map_location="cpu", weights_only=False),
+ strict=False,
+ )
+ self.student = _Student(self.teacher.out_channels)
+ self.register_buffer("map_mean", torch.zeros(1, 224, 224))
+ self.register_buffer("map_std", torch.ones(1, 224, 224))
+ self.register_buffer("score_mean", torch.tensor(0.0))
+ self.register_buffer("score_std", torch.tensor(1.0))
+ self.register_buffer("threshold", torch.tensor(0.0))
+ self.register_buffer("trained", torch.tensor(False, dtype=torch.bool))
+ self.to(self.device)
+
+ def _normalise(self, x):
+ return x
+
+ @torch.no_grad()
+ def _raw_map(self, x, teacher=None):
+ teacher_features = (
+ self.teacher(self._normalise(x)) if teacher is None else teacher
+ )
+ student_features = self.student(self._normalise(x))
+ raw = (student_features - teacher_features).pow(2).mean(1, keepdim=True)
+ return F.interpolate(
+ raw, size=x.shape[-2:], mode="bilinear", align_corners=False
+ ).squeeze(1)
+
+ def forward(self, x, return_map=True, export=False):
+ del export
+ raw = self._raw_map(x)
+ normalized_map = (raw - self.map_mean.to(raw.device)) / self.map_std.to(
+ raw.device
+ ).clamp_min(1e-6)
+ scores = normalized_map.flatten(1).amax(1)
+ return scores, normalized_map if return_map else None
+
+ def fit(self, dataloader, epochs=1):
+ cached = []
+ self.teacher.eval()
+ with torch.no_grad():
+ for item in dataloader:
+ images = item[0] if isinstance(item, (tuple, list)) else item
+ images = images.to(self.device, non_blocking=True).float()
+ teacher = self.teacher(self._normalise(images)).detach().cpu()
+ cached.append((images.cpu(), teacher))
+ if not cached:
+ raise RuntimeError("EfficientAD training requires normal training images")
+
+ optimizer = torch.optim.AdamW(
+ self.student.parameters(), lr=self.lr, weight_decay=self.weight_decay
+ )
+ use_amp = self.device.type == "cuda"
+ scaler = torch.amp.GradScaler("cuda", enabled=use_amp)
+ self.student.train()
+ for _ in range(int(epochs)):
+ for images_cpu, teacher_cpu in cached:
+ images = images_cpu.to(self.device, non_blocking=True)
+ teacher = teacher_cpu.to(self.device, non_blocking=True)
+ optimizer.zero_grad(set_to_none=True)
+ with torch.autocast(
+ device_type="cuda", dtype=torch.float16, enabled=use_amp
+ ):
+ student = self.student(self._normalise(images))
+ loss = F.mse_loss(student.float(), teacher.float())
+ scaler.scale(loss).backward()
+ scaler.step(optimizer)
+ scaler.update()
+
+ self.student.eval()
+ maps = []
+ with torch.no_grad():
+ for images_cpu, teacher_cpu in cached:
+ images = images_cpu.to(self.device, non_blocking=True)
+ teacher = teacher_cpu.to(self.device, non_blocking=True)
+ maps.append(self._raw_map(images, teacher).float())
+ normal_maps = torch.cat(maps, 0)
+ mean = normal_maps.mean(0, keepdim=True)
+ std = normal_maps.std(0, unbiased=False, keepdim=True).clamp_min(1e-6)
+ normalized = (normal_maps - mean) / std
+ scores = normalized.flatten(1).amax(1)
+ self.map_mean.copy_(mean.cpu())
+ self.map_std.copy_(std.cpu())
+ self.score_mean.copy_(scores.mean().cpu())
+ self.score_std.copy_(scores.std(unbiased=False).clamp_min(1e-6).cpu())
+ self.threshold.copy_(torch.quantile(scores, self.threshold_quantile).cpu())
+ self.trained.fill_(True)
+ self.to(self.device)
+
+ def predict(self, batch, export=False):
+ if not export and not bool(self.trained.item()):
+ raise RuntimeError("EfficientAD model is not trained. Call fit() first.")
+ self.eval()
+ with torch.inference_mode():
+ return self.forward(
+ batch.to(self.device, non_blocking=True).float(), export=export
+ )
+
+ def to_device(self, device):
+ self.device = torch.device(device)
+ self.to(self.device)
+
+ def save_statistics(self, path: str, half: Optional[bool] = None):
+ if not bool(self.trained.item()):
+ raise RuntimeError("Model is not trained. Call fit() first.")
+ # Keep the calibrated threshold explicitly available to deployment tools.
+ state = {k: v.detach().cpu() for k, v in self.state_dict().items()}
+ torch.save(
+ {
+ "algorithm": "efficientad",
+ "model_state": state,
+ "model_size": self.model_size,
+ "threshold": float(self.threshold.detach().cpu().item()),
+ "threshold_quantile": self.threshold_quantile,
+ },
+ path,
+ )
+
+ @staticmethod
+ def load_statistics(path: str, device: str = "cpu"):
+ obj = torch.load(path, map_location="cpu", weights_only=False)
+ if isinstance(obj, EfficientAD):
+ obj.to_device(torch.device(device))
+ return obj
+ if isinstance(obj, dict) and obj.get("algorithm") == "efficientad":
+ model = EfficientAD(
+ device=torch.device(device),
+ model_size=obj.get("model_size", "s"),
+ pretrained_teacher=False,
+ threshold_quantile=obj.get("threshold_quantile", 0.995),
+ )
+ model.load_state_dict(obj["model_state"])
+ return model
+ raise ValueError("Not an EfficientAD artifact")
+
+
+def build_efficientad_from_stats(stats, device="cpu"):
+ if isinstance(stats, EfficientAD):
+ stats.to_device(torch.device(device))
+ return stats
+ if isinstance(stats, dict):
+ model = EfficientAD(
+ device=torch.device(device),
+ model_size=stats.get("model_size", "s"),
+ pretrained_teacher=False,
+ threshold_quantile=stats.get("threshold_quantile", 0.995),
+ )
+ model.load_state_dict(stats["model_state"])
+ return model
+ raise ValueError("Unsupported EfficientAD statistics artifact")
diff --git a/anomavision/autopilot.py b/anomavision/autopilot.py
index 9ac4ad2..940e68d 100644
--- a/anomavision/autopilot.py
+++ b/anomavision/autopilot.py
@@ -3,14 +3,14 @@
from __future__ import annotations
import argparse
+import html
import json
import platform
import shutil
import sys
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
@@ -29,31 +29,19 @@
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,
)
+ parser.add_argument("--config", required=True)
+ parser.add_argument("--dataset_path", default=None)
+ parser.add_argument("--class_name", default=None)
+ parser.add_argument("--padim_model", default=None)
+ parser.add_argument("--patchcore_model", default=None)
parser.add_argument(
- "--config", type=str, required=True, help="Base AnomaVision config file."
- )
- parser.add_argument(
- "--dataset_path", type=str, default=None, help="MVTec-style dataset root."
- )
- parser.add_argument(
- "--class_name", type=str, default=None, help="Dataset class to evaluate."
- )
- parser.add_argument(
- "--padim_model",
- type=str,
+ "--efficientad_model",
default=None,
- help="PaDiM model artifact (.pt/.pth/.onnx).",
- )
- parser.add_argument(
- "--patchcore_model",
- type=str,
- default=None,
- help="PatchCore model artifact (.pt/.pth/.onnx).",
+ 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)
@@ -61,13 +49,8 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser:
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("--output_dir", type=str, default="./production_package")
+ parser.add_argument("--validation_split", type=float, default=1.0)
+ parser.add_argument("--output_dir", default="./production_package")
parser.add_argument("--copy_config", action="store_true", default=True)
return parser
@@ -78,14 +61,32 @@ def _to_numpy(value: Any) -> np.ndarray:
return np.asarray(value)
-def _format_metric(value: Any) -> str:
- return "N/A" if value is None else f"{float(value):.4f}"
+def _fmt(value: Any, digits: int = 4) -> str:
+ return "N/A" if value is None else f"{float(value):.{digits}f}"
+
+
+def _format_metric(value: Any, digits: int = 4) -> str:
+ """Backward-compatible public formatter used by the autopilot tests/API."""
+ return _fmt(value, digits)
-def _format_percent(value: Any) -> str:
+def _pct(value: Any) -> str:
return "N/A" if value is None else f"{float(value):.1%}"
+def _json_default(value: Any) -> Any:
+ """Convert NumPy/PyTorch scalar values to JSON-safe Python values."""
+ if isinstance(value, np.generic):
+ return value.item()
+ if isinstance(value, torch.Tensor):
+ if value.ndim == 0:
+ return value.item()
+ return value.detach().cpu().tolist()
+ if isinstance(value, Path):
+ return str(value)
+ raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")
+
+
def _profile_model(
model_path: str,
dataloader: DataLoader,
@@ -94,60 +95,53 @@ def _profile_model(
timing_batches: int,
) -> Dict[str, Any]:
wrapper = ModelWrapper(model_path, device)
- iterator = iter(dataloader)
- try:
- first = next(iterator)
- except StopIteration:
- wrapper.close()
- raise ValueError("The evaluation dataset is empty.")
- first_batch = first[0].to(device)
- for _ in range(max(0, warmup)):
- 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:
- batch, _, labels, masks = item
- batch = batch.to(device)
- measure = count < max(1, timing_batches)
- if measure:
- start = time.perf_counter()
- scores, maps = wrapper.predict(batch)
+ scores_all, maps_all, labels_all, masks_all = [], [], [], []
+ try:
+ first = next(iter(dataloader))
+ first_batch = first[0].to(device)
+ for _ in range(max(0, warmup)):
+ wrapper.predict(first_batch)
if device.startswith("cuda") and torch.cuda.is_available():
torch.cuda.synchronize()
- if measure:
- timings.append(time.perf_counter() - start)
- all_scores.extend(_to_numpy(scores).reshape(-1).tolist())
- if maps is not None:
- 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 = (
- np.asarray(all_maps, dtype=np.float32)
- if all_maps
+ for index, (batch, _, labels, masks) in enumerate(dataloader):
+ batch = batch.to(device)
+ measure = index < max(1, timing_batches)
+ start = time.perf_counter() if measure else 0.0
+ scores, maps = wrapper.predict(batch)
+ if device.startswith("cuda") and torch.cuda.is_available():
+ torch.cuda.synchronize()
+ if measure:
+ timings.append(time.perf_counter() - start)
+ scores_all.extend(_to_numpy(scores).reshape(-1).tolist())
+ if maps is not None:
+ maps_all.extend(_to_numpy(maps))
+ labels_all.extend(_to_numpy(labels).reshape(-1).astype(int).tolist())
+ masks_all.extend(_to_numpy(masks))
+ finally:
+ wrapper.close()
+
+ scores = np.asarray(scores_all, dtype=np.float32)
+ labels = np.asarray(labels_all, dtype=np.int64)
+ maps = (
+ np.asarray(maps_all, dtype=np.float32)
+ if maps_all
else np.empty((0, 0, 0), dtype=np.float32)
)
- threshold, threshold_f1 = (
- find_optimal_threshold(labels_np, scores_np)
- if len(np.unique(labels_np)) > 1
- else (float(np.median(scores_np)), 0.0)
- )
- image_metrics = compute_metrics(labels_np, scores_np, thresh=threshold)
- image_auroc = (
- image_metrics.get("auc_score") if len(np.unique(labels_np)) > 1 else None
- )
+ masks = np.asarray(masks_all, dtype=np.float32)
+ if masks.ndim == 4 and masks.shape[1] == 1:
+ masks = masks[:, 0]
+
+ if len(np.unique(labels)) > 1:
+ threshold, threshold_f1 = find_optimal_threshold(labels, scores)
+ else:
+ threshold, threshold_f1 = (
+ float(np.median(scores)) if len(scores) else 0.0
+ ), 0.0
+ metrics = compute_metrics(labels, scores, thresh=threshold) if len(labels) else {}
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]
localization = {
- "available": bool(len(maps_np) == len(labels_np) and maps_np.ndim == 3),
+ "available": bool(len(maps) == len(labels) and maps.ndim == 3),
"non_empty_fraction": None,
"mean_mask_area_fraction": None,
"anomaly_non_empty_fraction": None,
@@ -158,71 +152,57 @@ def _profile_model(
}
if (
localization["available"]
- and masks_np.shape == maps_np.shape
- and np.unique(masks_np).size > 1
+ and masks.shape == maps.shape
+ and np.unique(masks).size > 1
):
try:
pixel_auroc = float(
- roc_auc_score(masks_np.reshape(-1) > 0.5, maps_np.reshape(-1))
+ roc_auc_score(masks.reshape(-1) > 0.5, maps.reshape(-1))
)
except ValueError:
- pixel_auroc = None
- image_metrics["image_auroc"] = image_auroc
- image_metrics["pixel_auroc"] = pixel_auroc
- anomaly_labels = (scores_np >= threshold).astype(np.uint8)
- if localization["available"]:
- 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
+ pass
+ loc_masks = make_localization_mask(
+ maps, (scores >= threshold).astype(np.uint8)
+ ).astype(bool)
+ flat = loc_masks.reshape(len(loc_masks), -1)
+ non_empty, area = flat.any(axis=1), flat.mean(axis=1)
+ anomaly, normal = labels == 1, labels == 0
localization["non_empty_fraction"] = float(non_empty.mean())
localization["mean_mask_area_fraction"] = float(area.mean())
- localization["anomaly_non_empty_fraction"] = (
- float(non_empty[anomaly_idx].mean()) if anomaly_idx.any() else None
- )
- localization["normal_false_positive_fraction"] = (
- float(non_empty[normal_idx].mean()) if normal_idx.any() else None
- )
- localization["anomaly_mean_mask_area_fraction"] = (
- float(area[anomaly_idx].mean()) if anomaly_idx.any() else None
- )
- 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"] = (
- "healthy"
- if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10
- else "review false positives"
+ if anomaly.any():
+ localization["anomaly_non_empty_fraction"] = float(
+ non_empty[anomaly].mean()
)
- 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
+ localization["anomaly_mean_mask_area_fraction"] = float(
+ area[anomaly].mean()
+ )
+ if normal.any():
+ localization["normal_false_positive_fraction"] = float(
+ non_empty[normal].mean()
+ )
+ localization["normal_mean_mask_area_fraction"] = float(area[normal].mean())
+ localization["verdict"] = (
+ "healthy"
+ if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10
+ else "review false positives"
+ )
+ metrics["image_auroc"] = (
+ float(metrics["auc_score"]) if metrics.get("auc_score") is not None else None
)
+ metrics["pixel_auroc"] = pixel_auroc
+ bs = max(1, int(dataloader.batch_size or 1))
+ median_ms = float(np.median(timings) * 1000 / bs) if timings else 0.0
+ p95_ms = float(np.percentile(timings, 95) * 1000 / bs) 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)
- for k, v in image_metrics.items()
- },
+ "metrics": metrics,
"latency_ms": {"median": median_ms, "p95": p95_ms},
- "throughput_images_per_second": (
- float(1000.0 / median_ms) if median_ms > 0 else 0.0
- ),
+ "throughput_images_per_second": float(1000 / median_ms) if median_ms else 0.0,
"localization": localization,
- "samples": int(len(labels_np)),
+ "samples": int(len(labels)),
}
@@ -232,83 +212,56 @@ def _select(
eligible = results
if target_latency_ms is not None:
eligible = {
- name: result
- for name, result in results.items()
- if result["latency_ms"]["p95"] <= target_latency_ms
+ n: r
+ for n, r in results.items()
+ if r["latency_ms"]["p95"] <= target_latency_ms
}
if not eligible:
eligible = results
return max(
eligible,
- key=lambda name: (
- eligible[name]["metrics"].get("image_auroc") or 0.0,
- -eligible[name]["latency_ms"]["p95"],
+ key=lambda n: (
+ eligible[n]["metrics"].get("image_auroc") or 0.0,
+ -eligible[n]["latency_ms"]["p95"],
),
)
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
cards.append(
- 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} ms p95 latency
{_format_percent(loc.get("anomaly_non_empty_fraction"))} anomaly coverage
{html.escape(name.upper())} {"SELECTED" if active else "CANDIDATE"}
{_fmt(metrics.get("image_auroc"))}Image AUROC
{_fmt(metrics.get("pixel_auroc"))} Pixel AUROC
{result["latency_ms"]["p95"]:.1f} ms P95 latency
{_pct(loc.get("anomaly_non_empty_fraction"))} Anomaly coverage
{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}{html.escape(name)} {_fmt(metrics.get("image_auroc"))} {_fmt(metrics.get("pixel_auroc"))} {result["latency_ms"]["median"]:.2f} {result["latency_ms"]["p95"]:.2f} {_pct(loc.get("anomaly_non_empty_fraction"))} {_pct(loc.get("normal_false_positive_fraction"))} {result["threshold"]:.6f}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 {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 Model Image AUROC Pixel AUROC Median ms P95 ms Anomaly coverage Normal false positives Threshold {"".join(rows)}
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"}
Reproducibility environment {environment_json} Generated by AnomaVision Production Autopilot · manifest schema {manifest["schema_version"]}
- """
- (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.",
- "",
- ]
+ env = html.escape(
+ json.dumps(manifest["environment"], indent=2, default=_json_default)
+ )
+ document = 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: {html.escape(selected)} Class: {html.escape(str(manifest["dataset"]["class_name"]))} Samples: {manifest["dataset"]["samples"]} Device: {html.escape(str(manifest["environment"].get("device","unknown")))}
Recommendation {html.escape(selected)} with threshold {selected_result["threshold"]:.6f}. It was selected {html.escape(target_text)}.
Candidate overview {"".join(cards)}
Detailed comparison Model Image AUROC Pixel AUROC Median ms P95 ms Anomaly coverage Normal images with false-positive maps Threshold {"".join(rows)}
Localization health Maps available {"PASS" if selected_result["localization"]["available"] else "CHECK"}
Anomaly localization {_pct(selected_result["localization"].get("anomaly_non_empty_fraction"))}
Normal false-positive maps {_pct(selected_result["localization"].get("normal_false_positive_fraction"))}
Mean anomaly mask area {_pct(selected_result["localization"].get("anomaly_mean_mask_area_fraction"))}
Verdict {html.escape(str(selected_result["localization"].get("verdict","N/A")))}
Deployment artifact Artifact {html.escape(str(manifest["selected_artifact"]))}
Format {html.escape(str(selected_result["model_format"]))}
Preprocessing {manifest["preprocessing"].get("resize",224)} px
Target latency {target if target is not None else "not set"}
Reproducibility environment {env} Generated by AnomaVision Production Autopilot · manifest schema {manifest["schema_version"]} = 0.0
+
+
+def test_efficientad_rejects_unknown_model_size():
+ try:
+ EfficientAD(pretrained_teacher=False, model_size="large")
+ except ValueError as exc:
+ assert "model_size" in str(exc)
+ else:
+ raise AssertionError("Expected invalid EfficientAD model_size to fail")
+
+
+def test_efficientad_rejects_invalid_threshold_quantile():
+ try:
+ EfficientAD(pretrained_teacher=False, threshold_quantile=1.0)
+ except ValueError as exc:
+ assert "threshold_quantile" in str(exc)
+ else:
+ raise AssertionError("Expected invalid threshold_quantile to fail")
diff --git a/tests/test_efficientad_config.py b/tests/test_efficientad_config.py
new file mode 100644
index 0000000..095be78
--- /dev/null
+++ b/tests/test_efficientad_config.py
@@ -0,0 +1,19 @@
+from pathlib import Path
+
+from anomavision.config import load_config
+
+
+def test_model_name_selects_algorithm(tmp_path: Path):
+ config_path = tmp_path / "config.yml"
+ config_path.write_text("model:\n name: efficientad\n")
+ config = load_config(str(config_path))
+ assert config["algorithm"] == "efficientad"
+ assert config["model"] is None
+
+
+def test_legacy_algorithm_config_is_unchanged(tmp_path: Path):
+ config_path = tmp_path / "config.yml"
+ config_path.write_text("algorithm: padim\nmodel: model.pt\n")
+ config = load_config(str(config_path))
+ assert config["algorithm"] == "padim"
+ assert config["model"] == "model.pt"
diff --git a/tests/test_efficientad_threshold.py b/tests/test_efficientad_threshold.py
new file mode 100644
index 0000000..f95c5eb
--- /dev/null
+++ b/tests/test_efficientad_threshold.py
@@ -0,0 +1,19 @@
+from pathlib import Path
+
+import torch
+
+from anomavision.efficientad_threshold import load_calibrated_threshold
+
+
+def test_load_calibrated_efficientad_threshold(tmp_path: Path):
+ model_path = tmp_path / "model.onnx"
+ stats_path = model_path.with_suffix(".pth")
+ torch.save(
+ {
+ "algorithm": "efficientad",
+ "model_state": {"threshold": torch.tensor(2.75)},
+ },
+ stats_path,
+ )
+
+ assert load_calibrated_threshold(model_path) == 2.75