AnomaVision Wiki

📚 Table of Contents

Getting Started

Core Concepts

User Guides

Advanced Topics

API Reference

Development

Home

🔥 Key Features

🏆 Performance Highlights

Quick Start Guide

30-Second Setup

# Clone and install
git clone https://github.com/DeepKnowledge1/AnomaVision.git
cd AnomaVision
poetry install
poetry shell
# Verify installation
python -c "import anodet; print('🎉 AnomaVision installed successfully!')"

2-Minute Training

import anodet
import torch
from torch.utils.data import DataLoader
# Load your "good" training images
dataset = anodet.AnodetDataset(
 "path/to/train/good",
 resize=[256, 192], # Flexible width/height
 crop_size=[224, 224], # Final crop size
 normalize=True # ImageNet normalization
)
dataloader = DataLoader(dataset, batch_size=4)
# Initialize PaDiM with optimal settings
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = anodet.Padim(
 backbone='resnet18', # Fast and accurate
 device=device,
 layer_indices=[0, 1], # Multi-scale features
 feat_dim=100 # Optimal feature dimension
)
# Train the model
print("🚀 Training model...")
model.fit(dataloader)
# Save for production deployment
torch.save(model, "anomaly_detector.pt")
model.save_statistics("compact_model.pth", half=True) # 4x smaller!
print("✅ Model trained and saved!")

Instant Detection

# Load test data and detect anomalies
test_dataset = anodet.AnodetDataset("path/to/test/images")
test_dataloader = DataLoader(test_dataset, batch_size=4)
for batch, images, _, _ in test_dataloader:
 # Get anomaly scores and detailed heatmaps
 image_scores, score_maps = model.predict(batch)
 
 # Classify anomalies (threshold=13 works great for most cases)
 predictions = anodet.classification(image_scores, threshold=13)
 
 print(f"🔥 Anomaly scores: {image_scores.tolist()}")
 print(f"📋 Predictions: {predictions.tolist()}")
 break

Installation

Prerequisites

Method 1: Poetry (Recommended)

git clone https://github.com/DeepKnowledge1/AnomaVision.git
cd AnomaVision
poetry install
poetry shell

Method 2: pip

git clone https://github.com/DeepKnowledge1/AnomaVision.git
cd AnomaVision
pip install -r requirements.txt

Method 3: Development Setup

git clone https://github.com/DeepKnowledge1/AnomaVision.git
cd AnomaVision
poetry install --dev
pre-commit install

Verify Installation

python -c "import anodet; print('🎉 AnomaVision installed successfully!')"

Optional Dependencies

# ONNX Runtime
pip install onnxruntime-gpu # or onnxruntime for CPU
# OpenVINO
pip install openvino
# Development tools
pip install pytest black flake8 pre-commit

Basic Usage

Training a Model

import anodet
import torch
from torch.utils.data import DataLoader
# 1. Create dataset
dataset = anodet.AnodetDataset(
 "path/to/normal/images",
 resize=[224, 224],
 crop_size=[224, 224],
 normalize=True
)
dataloader = DataLoader(dataset, batch_size=8)
# 2. Initialize model
model = anodet.Padim(
 backbone='resnet18',
 device=torch.device('cuda'),
 layer_indices=[0, 1],
 feat_dim=100
)
# 3. Train
model.fit(dataloader)
# 4. Save
torch.save(model, "model.pt")
model.save_statistics("model.pth", half=True)

Running Inference

from anodet.inference.model.wrapper import ModelWrapper
# Load any supported format
model = ModelWrapper("model.onnx", device='cuda')
# Predict
scores, maps = model.predict(batch)
# Classify
predictions = anodet.classification(scores, threshold=13)
# Cleanup
model.close()

Command Line Interface

# Train a model
python train.py --config config.yml
# Run inference
python detect.py --model model.onnx --img_path test_images/
# Evaluate performance
python eval.py --model model.pt --dataset_path data/mvtec --class_name bottle
# Export to multiple formats
python export.py --model model.pt --format all

Architecture Overview

AnomaVision/
├── 🧠 anodet/ # Core AI library
│ ├── 📄 padim.py # PaDiM implementation
│ ├── 📄 padim_lite.py # Lightweight runtime module
│ ├── 📄 feature_extraction.py # ResNet feature extraction
│ ├── 📄 mahalanobis.py # Distance computation
│ ├── 📁 datasets/ # Dataset loaders with flexible sizing
│ ├── 📁 visualization/ # Rich visualization tools
│ ├── 📁 inference/ # Multi-format inference engine
│ │ ├── 📄 wrapper.py # Universal model wrapper
│ │ ├── 📄 modelType.py # Format detection
│ │ └── 📁 backends/ # Format-specific backends
│ │ ├── 📄 torch_backend.py # PyTorch support
│ │ ├── 📄 onnx_backend.py # ONNX Runtime support
│ │ ├── 📄 torchscript_backend.py # TorchScript support
│ │ └── 📄 openvino_backend.py # OpenVINO support
│ └── 📁 config/ # Configuration management
├── 📄 train.py # Training script
├── 📄 detect.py # Inference script
├── 📄 eval.py # Evaluation script
├── 📄 export.py # Multi-format export utilities
└── 📄 config.yml # Default configuration

Core Components

1. PaDiM Algorithm (padim.py)

2. PaDiM Lite (padim_lite.py)

3. Inference Engine (inference/)

4. Feature Extraction (feature_extraction.py)

5. Distance Computation (mahalanobis.py)

PaDiM Algorithm

How PaDiM Works

Key Advantages

Implementation Details

class Padim(torch.nn.Module):
 def __init__(self, backbone='resnet18', layer_indices=[0, 1], feat_dim=100):
 # Initialize ResNet feature extractor
 self.embeddings_extractor = ResnetEmbeddingsExtractor(backbone, device)
 
 # Set up feature selection
 self.layer_indices = layer_indices
 self.channel_indices = get_dims_indices(layer_indices, feat_dim, ...)
 
 def fit(self, dataloader):
 # Extract features from training data
 features = self.embeddings_extractor.from_dataloader(dataloader)
 
 # Compute statistics
 mean = torch.mean(features, dim=0)
 cov = pytorch_cov(features) + 0.01 * torch.eye(features.shape[2])
 cov_inv = torch.inverse(cov)
 
 # Store for inference
 self.mahalanobisDistance = MahalanobisDistance(mean, cov_inv)
 
 def predict(self, batch):
 # Extract features
 features, w, h = self.embeddings_extractor(batch)
 
 # Compute distances
 distances = self.mahalanobisDistance(features, w, h)
 
 # Return scores and maps
 image_scores = distances.flatten(1).max(1).values
 score_maps = F.interpolate(distances.unsqueeze(1), size=batch.shape[-2:])
 
 return image_scores, score_maps.squeeze(1)

Supported Formats

Format Comparison

PyTorch (.pt)

# Full model with training capabilities
model = torch.load("model.pt")
scores, maps = model.predict(batch)

Statistics (.pth)

# Compact statistics-only (2-4x smaller)
model = ModelWrapper("model.pth", device='cpu')
scores, maps = model.predict(batch)

ONNX (.onnx)

# Cross-platform deployment
model = ModelWrapper("model.onnx", device='cuda')
scores, maps = model.predict(batch)

TorchScript (.torchscript)

# Optimized Python deployment
model = ModelWrapper("model.torchscript", device='cuda')
scores, maps = model.predict(batch)

OpenVINO (.xml/.bin)

# Intel hardware optimization
model = ModelWrapper("model_openvino/model.xml", device='cpu')
scores, maps = model.predict(batch)

Universal Interface

from anodet.inference.model.wrapper import ModelWrapper
# Load any format
model = ModelWrapper(model_path, device)
# Same prediction interface
scores, maps = model.predict(batch)
# Same cleanup
model.close()

Image Processing

Flexible Dimensions

# Square resize and crop
dataset = anodet.AnodetDataset(
 image_path,
 resize=[224, 224],
 crop_size=[224, 224]
)
# Flexible width/height
dataset = anodet.AnodetDataset(
 image_path,
 resize=[256, 192], # Width, Height
 crop_size=[224, 224] # Final crop
)
# No cropping
dataset = anodet.AnodetDataset(
 image_path,
 resize=[320, 240],
 crop_size=None # Keep resize dimensions
)

Normalization Options

# ImageNet normalization (recommended)
dataset = anodet.AnodetDataset(
 image_path,
 normalize=True,
 mean=[0.485, 0.456, 0.406],
 std=[0.229, 0.224, 0.225]
)
# Custom normalization
dataset = anodet.AnodetDataset(
 image_path,
 normalize=True,
 mean=[0.5, 0.5, 0.5],
 std=[0.5, 0.5, 0.5]
)
# No normalization
dataset = anodet.AnodetDataset(
 image_path,
 normalize=False
)

Transform Pipeline

Custom Transforms

from torchvision import transforms as T
# Create custom transforms
custom_transforms = T.Compose([
 T.Resize([256, 256]),
 T.CenterCrop([224, 224]),
 T.ColorJitter(brightness=0.1),
 T.ToTensor(),
 T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# Use with dataset
dataset = anodet.AnodetDataset(
 image_path,
 image_transforms=custom_transforms
)

Training Models

Command Line Training

# Basic training
python train.py \
 --dataset_path "data/bottle" \
 --class_name "bottle" \
 --model_data_path "./models/" \
 --backbone resnet18 \
 --batch_size 8 \
 --layer_indices 0 1 \
 --feat_dim 100
# Using config file (recommended)
python train.py --config config.yml

Configuration File Training

# Dataset configuration
dataset_path: "D:/01-DATA"
class_name: "bottle"
resize: [256, 224] # Width, Height
crop_size: [224, 224] # Final square crop
normalize: true
norm_mean: [0.485, 0.456, 0.406]
norm_std: [0.229, 0.224, 0.225]
# Model configuration
backbone: "resnet18"
feat_dim: 100
layer_indices: [0, 1]
batch_size: 8
# Output configuration
model_data_path: "./distributions/bottle_exp"
output_model: "padim_model.pt"
run_name: "bottle_experiment"

Programmatic Training

import anodet
import torch
from torch.utils.data import DataLoader
# Dataset setup
dataset = anodet.AnodetDataset(
 "path/to/train/good",
 resize=[224, 224],
 crop_size=[224, 224],
 normalize=True
)
dataloader = DataLoader(dataset, batch_size=8, shuffle=False)
# Model initialization
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = anodet.Padim(
 backbone='resnet18',
 device=device,
 layer_indices=[0, 1],
 feat_dim=100
)
# Training
print("Training model...")
model.fit(dataloader, extractions=1)
# Save multiple formats
torch.save(model, "full_model.pt")
model.save_statistics("stats_fp32.pth", half=False)
model.save_statistics("stats_fp16.pth", half=True)
print("Training completed!")

Training Parameters

MVTec Dataset Training

# MVTec dataset structure
dataset = anodet.MVTecDataset(
 "path/to/mvtec",
 class_name="bottle",
 is_train=True,
 resize=[224, 224],
 crop_size=[224, 224],
 normalize=True
)

mvtec/
├── bottle/
│ ├── train/
│ │ └── good/ # Normal training images
│ ├── test/
│ │ ├── good/ # Normal test images
│ │ └── broken_large/ # Anomalous test images
│ └── ground_truth/
│ └── broken_large/ # Pixel-level masks

Running Inference

Command Line Inference

# Basic inference
python detect.py \
 --model_data_path "./models/" \
 --model "padim_model.onnx" \
 --img_path "test_images/" \
 --batch_size 16 \
 --thresh 13 \
 --enable_visualization
# With config file
python detect.py --config config.yml

Programmatic Inference

from anodet.inference.model.wrapper import ModelWrapper
from torch.utils.data import DataLoader
import anodet
# Load model (any format)
model = ModelWrapper("model.onnx", device='cuda')
# Prepare test data
test_dataset = anodet.AnodetDataset("test_images/")
test_dataloader = DataLoader(test_dataset, batch_size=4)
# Run inference
for batch, images, _, _ in test_dataloader:
 # Get predictions
 image_scores, score_maps = model.predict(batch)
 
 # Classify anomalies
 predictions = anodet.classification(image_scores, threshold=13)
 
 print(f"Scores: {image_scores}")
 print(f"Predictions: {predictions}")
# Cleanup
model.close()

Batch Processing

def process_directory(model_path, image_dir, threshold=13):
 """Process all images in a directory."""
 model = ModelWrapper(model_path, device='cuda')
 
 dataset = anodet.AnodetDataset(image_dir)
 dataloader = DataLoader(dataset, batch_size=8)
 
 all_scores = []
 all_predictions = []
 
 for batch, _, _, _ in dataloader:
 scores, maps = model.predict(batch)
 predictions = anodet.classification(scores, threshold)
 
 all_scores.extend(scores.tolist())
 all_predictions.extend(predictions.tolist())
 
 model.close()
 return all_scores, all_predictions

Real-time Inference

import cv2
import numpy as np
from PIL import Image
def real_time_detection(model_path):
 """Real-time anomaly detection from webcam."""
 model = ModelWrapper(model_path, device='cuda')
 cap = cv2.VideoCapture(0)
 
 while True:
 ret, frame = cap.read()
 if not ret:
 break
 
 # Preprocess frame
 image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
 batch = anodet.to_batch([np.array(image)])
 
 # Detect anomalies
 scores, maps = model.predict(batch)
 prediction = anodet.classification(scores, threshold=13)[0]
 
 # Display result
 color = (0, 255, 0) if prediction == 1 else (0, 0, 255)
 cv2.putText(frame, f"Score: {scores[0]:.2f}", (10, 30), 
 cv2.FONT_HERSHEY_SIMPLEX, 1, color, 2)
 cv2.imshow('Anomaly Detection', frame)
 
 if cv2.waitKey(1) & 0xFF == ord('q'):
 break
 
 cap.release()
 cv2.destroyAllWindows()
 model.close()

Performance Optimization

# Warmup for consistent timing
model.warmup(sample_batch, runs=3)
# Batch processing for throughput
dataloader = DataLoader(dataset, batch_size=32, pin_memory=True)
# Use appropriate device
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = ModelWrapper(model_path, device=device)

Model Export

Export All Formats

# Export to all supported formats
python export.py \
 --model_data_path "./models/" \
 --model "padim_model.pt" \
 --format all \
 --opset 17 \
 --dynamic_batch
# Export specific format
python export.py \
 --model_data_path "./models/" \
 --model "padim_model.pt" \
 --format onnx \
 --opset 17

Programmatic Export

from export import ModelExporter
from pathlib import Path
# Initialize exporter
model_path = Path("models/padim_model.pt")
output_dir = Path("exported_models/")
exporter = ModelExporter(model_path, output_dir, logger)
# Export ONNX
onnx_path = exporter.export_onnx(
 input_shape=(1, 3, 224, 224),
 output_name="model.onnx",
 opset_version=17,
 dynamic_batch=True
)
# Export TorchScript
ts_path = exporter.export_torchscript(
 input_shape=(1, 3, 224, 224),
 output_name="model.torchscript",
 optimize=True
)
# Export OpenVINO
ov_path = exporter.export_openvino(
 input_shape=(1, 3, 224, 224),
 output_name="model_openvino",
 fp16=True,
 dynamic_batch=False
)

Export Configuration

# export section in config.yml
export:
 format: "all" # onnx, torchscript, openvino, all
 opset: 17 # ONNX opset version
 dynamic_batch: true # Allow dynamic batch size
 fp16: true # Use FP16 for OpenVINO
 optimize: true # TorchScript mobile optimization

Format-Specific Options

ONNX Export

onnx_path = exporter.export_onnx(
 input_shape=(1, 3, 224, 224),
 output_name="model.onnx",
 opset_version=17, # ONNX opset (11, 13, 15, 17)
 dynamic_batch=True # Allow variable batch size
)

TorchScript Export

ts_path = exporter.export_torchscript(
 input_shape=(1, 3, 224, 224),
 output_name="model.torchscript",
 optimize=True # Mobile optimization
)

OpenVINO Export

ov_path = exporter.export_openvino(
 input_shape=(1, 3, 224, 224),
 output_name="model_openvino",
 fp16=True, # Use FP16 precision
 dynamic_batch=False # Static batch for optimization
)

Statistics Export

# Save compact statistics (2-4x smaller)
model.save_statistics("model_fp32.pth", half=False) # Full precision
model.save_statistics("model_fp16.pth", half=True) # Half precision
# Load statistics
stats = model.load_statistics("model_fp16.pth", device='cpu', force_fp32=True)

Model Evaluation

Command Line Evaluation

# Evaluate on MVTec dataset
python eval.py \
 --model_data_path "./models/" \
 --model "padim_model.onnx" \
 --dataset_path "data/mvtec" \
 --class_name "bottle" \
 --batch_size 8
# With config file
python eval.py --config config.yml

Programmatic Evaluation

import anodet
from torch.utils.data import DataLoader
# Load test dataset
test_dataset = anodet.MVTecDataset(
 "data/mvtec",
 class_name="bottle",
 is_train=False, # Test set
 resize=[224, 224],
 crop_size=[224, 224],
 normalize=True
)
test_dataloader = DataLoader(test_dataset, batch_size=8)
# Load model
model = torch.load("padim_model.pt")
# Run evaluation
results = model.evaluate(test_dataloader)
images, targets, masks, scores, maps = results
# Visualize results
anodet.visualize_eval_data(targets, masks, scores, maps)

Performance Metrics

from sklearn.metrics import roc_auc_score, precision_recall_curve
# Image-level metrics
image_auroc = roc_auc_score(targets, scores)
precision, recall, thresholds = precision_recall_curve(targets, scores)
# Pixel-level metrics
pixel_auroc = roc_auc_score(masks.flatten(), maps.flatten())
print(f"Image AUROC: {image_auroc:.4f}")
print(f"Pixel AUROC: {pixel_auroc:.4f}")

Optimal Threshold

from anodet.test import optimal_threshold
# Find optimal threshold
precision, recall, threshold = optimal_threshold(targets, scores)
print(f"Optimal threshold: {threshold:.2f}")
print(f"Precision: {precision:.4f}")
print(f"Recall: {recall:.4f}")

Memory-Efficient Evaluation

# For large datasets
results = model.evaluate_memory_efficient(test_dataloader)
images, targets, masks, scores, maps = results

Cross-Format Evaluation

from anodet.inference.model.wrapper import ModelWrapper
# Compare different formats
formats = {
 'pytorch': 'model.pt',
 'onnx': 'model.onnx',
 'torchscript': 'model.torchscript',
 'openvino': 'model_openvino/model.xml'
}
results = {}
for name, path in formats.items():
 model = ModelWrapper(path, device='cpu')
 
 all_scores = []
 for batch, _, _, _ in test_dataloader:
 scores, _ = model.predict(batch)
 all_scores.extend(scores.tolist())
 
 auroc = roc_auc_score(targets, all_scores)
 results[name] = auroc
 model.close()
for name, auroc in results.items():
 print(f"{name}: {auroc:.4f}")

Configuration System

Configuration Structure

# =========================
# Dataset / preprocessing
# =========================
dataset_path: "D:/01-DATA"
class_name: "bottle"
resize: [224, 224] # [width, height] or single int
crop_size: [224, 224] # [width, height] or single int
normalize: true
norm_mean: [0.485, 0.456, 0.406] # ImageNet stats
norm_std: [0.229, 0.224, 0.225]
# =========================
# Model / training
# =========================
backbone: "resnet18" # resnet18, wide_resnet50
feat_dim: 100 # Feature dimension
layer_indices: [0, 1] # ResNet layers to use
model_data_path: "./distributions/exp"
output_model: "padim_model.pt"
batch_size: 8
device: "auto" # cpu, cuda, auto
# =========================
# Inference
# =========================
img_path: "test_images/"
thresh: 13.0 # Anomaly threshold
enable_visualization: true
save_visualizations: true
viz_output_dir: "./visualizations/"
viz_alpha: 0.6 # Heatmap transparency
viz_padding: 40 # Boundary padding
viz_color: "128,0,128" # RGB highlight color
# =========================
# Export
# =========================
format: "all" # onnx, torchscript, openvino, all
opset: 17 # ONNX opset version
dynamic_batch: true # Allow dynamic batch size
fp16: true # Use FP16 for OpenVINO
optimize: true # TorchScript optimization
# =========================
# Evaluation
# =========================
metrics: ["auroc", "pixel_auroc"]
val_batch_size: 8
memory_efficient: true
# =========================
# Logging
# =========================
log_level: "INFO" # DEBUG, INFO, WARNING, ERROR
detailed_timing: false

Using Configuration Files

# All scripts support config files
python train.py --config config.yml
python detect.py --config config.yml
python eval.py --config config.yml
python export.py --config config.yml

Command Line Override

# Config file + CLI overrides
python train.py --config config.yml --batch_size 16 --feat_dim 200

Programmatic Configuration

from anodet.config import load_config
from anodet.utils import merge_config
from easydict import EasyDict as edict
# Load configuration
config = load_config("config.yml")
# Merge with arguments
args = parse_args()
final_config = edict(merge_config(args, config))
# Use configuration
dataset = anodet.AnodetDataset(
 final_config.dataset_path,
 resize=final_config.resize,
 crop_size=final_config.crop_size,
 normalize=final_config.normalize
)

Configuration Validation

def validate_config(config):
 """Validate configuration parameters."""
 required_fields = ['dataset_path', 'backbone', 'batch_size']
 for field in required_fields:
 if not hasattr(config, field) or getattr(config, field) is None:
 raise ValueError(f"Required field '{field}' is missing")
 
 if config.backbone not in ['resnet18', 'wide_resnet50']:
 raise ValueError(f"Unsupported backbone: {config.backbone}")
 
 if config.batch_size <= 0:
 raise ValueError("Batch size must be positive")

Multi-Format Inference

Universal Interface

from anodet.inference.model.wrapper import ModelWrapper
# All formats use the same interface
models = {
 'pytorch': ModelWrapper("model.pt", device='cuda'),
 'statistics': ModelWrapper("model.pth", device='cuda'),
 'onnx': ModelWrapper("model.onnx", device='cuda'),
 'torchscript': ModelWrapper("model.torchscript", device='cuda'),
 'openvino': ModelWrapper("model_openvino/model.xml", device='cpu')
}
# Same prediction interface for all
for name, model in models.items():
 scores, maps = model.predict(batch)
 print(f"{name}: {scores.mean():.4f}")
 model.close()

Backend Details

PyTorch Backend

# Supports full models and statistics files
backend = TorchBackend("model.pt", device='cuda', use_amp=True)
# Automatic mixed precision for faster inference
scores, maps = backend.predict(batch)

ONNX Backend

# Cross-platform deployment
backend = OnnxBackend(
 "model.onnx", 
 device='cuda',
 intra_threads=4, # Parallel processing
 inter_threads=2 # Thread management
)
# Optimized execution providers
scores, maps = backend.predict(batch)

TorchScript Backend

# Optimized Python deployment
backend = TorchScriptBackend(
 "model.torchscript",
 device='cuda',
 num_threads=8 # CPU threading
)
# JIT compilation benefits
scores, maps = backend.predict(batch)

OpenVINO Backend

# Intel hardware optimization
backend = OpenVinoBackend(
 "model_openvino/model.xml",
 device='CPU', # CPU, GPU, AUTO
 num_threads=4 # CPU optimization
)
# Hardware-specific acceleration
scores, maps = backend.predict(batch)

Performance Comparison

import time
from anodet.general import Profiler
def benchmark_formats(batch, formats, runs=10):
 """Benchmark different model formats."""
 results = {}
 
 for name, model_path in formats.items():
 model = ModelWrapper(model_path, device='cuda')
 
 # Warmup
 model.warmup(batch, runs=3)
 
 # Benchmark
 profiler = Profiler()
 for _ in range(runs):
 with profiler:
 scores, maps = model.predict(batch)
 
 avg_time = profiler.get_avg_time_ms(runs)
 fps = profiler.get_fps(len(batch) * runs)
 
 results[name] = {
 'avg_time_ms': avg_time,
 'fps': fps,
 'scores_mean': scores.mean()
 }
 
 model.close()
 
 return results
# Run benchmark
formats = {
 'PyTorch': 'model.pt',
 'ONNX': 'model.onnx',
 'TorchScript': 'model.torchscript',
 'OpenVINO': 'model_openvino/model.xml'
}
results = benchmark_formats(test_batch, formats)
for name, metrics in results.items():
 print(f"{name}: {metrics['avg_time_ms']:.2f}ms, {metrics['fps']:.1f} FPS")

Device-Specific Optimization

def get_optimal_backend(model_path, device):
 """Select optimal backend based on device."""
 if device.startswith('cuda'):
 # NVIDIA GPU
 if model_path.endswith('.onnx'):
 return OnnxBackend(model_path, device='cuda')
 elif model_path.endswith('.engine'):
 return TensorRTBackend(model_path, device='cuda')
 else:
 return TorchBackend(model_path, device='cuda', use_amp=True)
 
 elif device == 'cpu':
 # CPU optimization
 if 'openvino' in model_path:
 return OpenVinoBackend(model_path, device='CPU', num_threads=8)
 elif model_path.endswith('.onnx'):
 return OnnxBackend(model_path, device='cpu', intra_threads=8)
 else:
 return TorchBackend(model_path, device='cpu')
 
 else:
 # Auto-select
 return ModelWrapper(model_path, device='auto')

Performance Optimization

Memory Optimization

# Use statistics files for minimal memory
model = ModelWrapper("model.pth", device='cpu') # 2-4x smaller
# Batch processing for throughput
dataloader = DataLoader(dataset, batch_size=32, pin_memory=True, num_workers=4)
# Memory-efficient evaluation
results = model.evaluate_memory_efficient(test_dataloader)

GPU Optimization

# Enable mixed precision
model = ModelWrapper("model.pt", device='cuda') # Automatic AMP
# Optimize GPU memory
torch.backends.cudnn.benchmark = True # Optimize for fixed input sizes
# Pin memory for faster transfers
dataloader = DataLoader(dataset, pin_memory=True)

CPU Optimization

# Use optimized backends
model = ModelWrapper("model_openvino/model.xml", device='CPU')
# Set thread count
torch.set_num_threads(8)
# Use ONNX with threading
backend = OnnxBackend("model.onnx", device='cpu', intra_threads=8, inter_threads=2)

Batch Size Optimization

def find_optimal_batch_size(model_path, sample_batch, device='cuda'):
 """Find optimal batch size for throughput."""
 model = ModelWrapper(model_path, device=device)
 
 batch_sizes = [1, 2, 4, 8, 16, 32, 64]
 best_fps = 0
 best_batch_size = 1
 
 for bs in batch_sizes:
 try:
 # Create test batch
 test_batch = sample_batch[:bs] if bs <= len(sample_batch) else \
 torch.cat([sample_batch] * (bs // len(sample_batch) + 1))[:bs]
 
 # Benchmark
 profiler = Profiler()
 for _ in range(10):
 with profiler:
 scores, maps = model.predict(test_batch)
 
 fps = profiler.get_fps(bs * 10)
 print(f"Batch size {bs}: {fps:.1f} FPS")
 
 if fps > best_fps:
 best_fps = fps
 best_batch_size = bs
 
 except RuntimeError: # OOM
 print(f"Batch size {bs}: OOM")
 break
 
 model.close()
 return best_batch_size, best_fps

Profiling and Monitoring

from anodet.general import Profiler
# Detailed profiling
profilers = {
 'inference': Profiler(),
 'postprocessing': Profiler(),
 'visualization': Profiler()
}
for batch, images, _, _ in dataloader:
 # Core inference
 with profilers['inference']:
 scores, maps = model.predict(batch)
 
 # Postprocessing
 with profilers['postprocessing']:
 predictions = anodet.classification(scores, threshold=13)
 maps = anodet.utils.adaptive_gaussian_blur(maps)
 
 # Visualization
 with profilers['visualization']:
 heatmaps = anodet.visualization.heatmap_images(images, maps)
# Print timing summary
for name, prof in profilers.items():
 print(f"{name}: {prof.accumulated_time*1000:.2f}ms")

Custom Datasets

Creating Custom Datasets

import os
from torch.utils.data import Dataset
from PIL import Image
import anodet
class CustomAnomalyDataset(Dataset):
 """Custom dataset for anomaly detection."""
 
 def __init__(self, root_dir, transform=None, is_train=True):
 self.root_dir = root_dir
 self.transform = transform or anodet.utils.create_image_transform()
 self.is_train = is_train
 
 # Load image paths
 self.image_paths = []
 self.labels = []
 
 if is_train:
 # Training: only normal images
 normal_dir = os.path.join(root_dir, 'normal')
 for img_name in os.listdir(normal_dir):
 if img_name.lower().endswith(('.png', '.jpg', '.jpeg')):
 self.image_paths.append(os.path.join(normal_dir, img_name))
 self.labels.append(0) # Normal = 0
 else:
 # Testing: both normal and anomalous
 for class_name in ['normal', 'anomaly']:
 class_dir = os.path.join(root_dir, class_name)
 if os.path.exists(class_dir):
 for img_name in os.listdir(class_dir):
 if img_name.lower().endswith(('.png', '.jpg', '.jpeg')):
 self.image_paths.append(os.path.join(class_dir, img_name))
 self.labels.append(0 if class_name == 'normal' else 1)
 
 def __len__(self):
 return len(self.image_paths)
 
 def __getitem__(self, idx):
 # Load image
 image_path = self.image_paths[idx]
 image = Image.open(image_path).convert('RGB')
 label = self.labels[idx]
 
 # Transform
 tensor = self.transform(image)
 
 # Return format compatible with AnomaVision
 return tensor, np.array(image), label, torch.zeros(1, tensor.shape[1], tensor.shape[2])
# Usage
train_dataset = CustomAnomalyDataset('data/custom', is_train=True)
test_dataset = CustomAnomalyDataset('data/custom', is_train=False)

Directory Structure Examples

Simple Structure

custom_dataset/
├── normal/ # Normal training images
│ ├── image1.jpg
│ ├── image2.jpg
│ └── ...
└── anomaly/ # Anomalous test images
 ├── defect1.jpg
 ├── defect2.jpg
 └── ...

MVTec-like Structure

dataset/
├── class1/
│ ├── train/
│ │ └── good/
│ ├── test/
│ │ ├── good/
│ │ └── defect_type/
│ └── ground_truth/
│ └── defect_type/
└── class2/
 └── ...

Industrial Dataset Example

class IndustrialDataset(Dataset):
 """Dataset for industrial anomaly detection."""
 
 def __init__(self, root_dir, product_type, transform=None):
 self.root_dir = root_dir
 self.product_type = product_type
 self.transform = transform
 
 # Load metadata
 metadata_path = os.path.join(root_dir, 'metadata.csv')
 self.metadata = pd.read_csv(metadata_path)
 
 # Filter by product type
 self.data = self.metadata[self.metadata['product'] == product_type]
 
 def __getitem__(self, idx):
 row = self.data.iloc[idx]
 
 # Load image
 image_path = os.path.join(self.root_dir, 'images', row['filename'])
 image = Image.open(image_path).convert('RGB')
 
 # Load mask if available
 if pd.notna(row['mask_filename']):
 mask_path = os.path.join(self.root_dir, 'masks', row['mask_filename'])
 mask = Image.open(mask_path).convert('L')
 else:
 mask = Image.new('L', image.size, 0)
 
 # Apply transforms
 if self.transform:
 image = self.transform(image)
 mask = self.transform(mask)
 
 return image, np.array(image), row['label'], mask

Data Augmentation

from torchvision import transforms as T
# Training augmentation
train_transform = T.Compose([
 T.Resize([256, 256]),
 T.RandomRotation(10),
 T.ColorJitter(brightness=0.1, contrast=0.1),
 T.RandomHorizontalFlip(0.5),
 T.CenterCrop([224, 224]),
 T.ToTensor(),
 T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# Test transform (no augmentation)
test_transform = T.Compose([
 T.Resize([224, 224]),
 T.ToTensor(),
 T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# Create datasets
train_dataset = CustomAnomalyDataset('data/', transform=train_transform, is_train=True)
test_dataset = CustomAnomalyDataset('data/', transform=test_transform, is_train=False)

Visualization Tools

Basic Visualizations

import anodet.visualization as viz
import matplotlib.pyplot as plt
# Load test data and run inference
model = anodet.Padim(backbone='resnet18', device='cpu')
# ... train model ...
test_dataset = anodet.AnodetDataset("test_images/")
dataloader = DataLoader(test_dataset, batch_size=4)
for batch, images, _, _ in dataloader:
 # Get predictions
 image_scores, score_maps = model.predict(batch)
 
 # Apply Gaussian blur to score maps
 score_maps = anodet.utils.adaptive_gaussian_blur(score_maps, kernel_size=33, sigma=4)
 
 # Classify
 score_map_classifications = anodet.classification(score_maps, threshold=13)
 image_classifications = anodet.classification(image_scores, threshold=13)
 
 # Generate visualizations
 boundary_images = viz.framed_boundary_images(
 images, score_map_classifications, image_classifications, padding=40
 )
 
 heatmap_images = viz.heatmap_images(
 images, score_maps, alpha=0.6
 )
 
 highlighted_images = viz.highlighted_images(
 images, score_map_classifications, color=(255, 0, 0), alpha=0.5
 )
 
 break

Comprehensive Visualization

def create_anomaly_visualization(model, test_image_path, threshold=13):
 """Create comprehensive anomaly visualization."""
 # Load and preprocess image
 image = Image.open(test_image_path).convert('RGB')
 transform = anodet.utils.create_image_transform(resize=[224, 224], normalize=True)
 batch = transform(image).unsqueeze(0)
 
 # Run inference
 image_scores, score_maps = model.predict(batch)
 score_maps = anodet.utils.adaptive_gaussian_blur(score_maps)
 
 # Classifications
 score_map_class = anodet.classification(score_maps, threshold)
 image_class = anodet.classification(image_scores, threshold)
 
 # Create visualization
 fig, axes = plt.subplots(2, 3, figsize=(15, 10))
 
 # Original image
 axes[0, 0].imshow(image)
 axes[0, 0].set_title('Original Image')
 axes[0, 0].axis('off')
 
 # Score map
 axes[0, 1].imshow(score_maps[0], cmap='hot')
 axes[0, 1].set_title(f'Anomaly Score Map\nMax Score: {score_maps[0].max():.2f}')
 axes[0, 1].axis('off')
 
 # Heatmap overlay
 heatmap = viz.heatmap_image(np.array(image), score_maps[0], alpha=0.6)
 axes[0, 2].imshow(heatmap)
 axes[0, 2].set_title('Heatmap Overlay')
 axes[0, 2].axis('off')
 
 # Boundary detection
 boundary = viz.boundary_image(np.array(image), score_map_class[0])
 axes[1, 0].imshow(boundary)
 axes[1, 0].set_title('Boundary Detection')
 axes[1, 0].axis('off')
 
 # Highlighted anomalies
 highlighted = viz.highlighted_image(np.array(image), score_map_class[0])
 axes[1, 1].imshow(highlighted)
 axes[1, 1].set_title('Highlighted Anomalies')
 axes[1, 1].axis('off')
 
 # Classification result
 result_color = 'red' if image_class[0] == 0 else 'green'
 result_text = 'ANOMALY' if image_class[0] == 0 else 'NORMAL'
 axes[1, 2].text(0.5, 0.5, result_text, transform=axes[1, 2].transAxes,
 fontsize=24, ha='center', va='center', color=result_color,
 bbox=dict(boxstyle='round', facecolor='white', alpha=0.8))
 axes[1, 2].set_title(f'Classification\nScore: {image_scores[0]:.2f}')
 axes[1, 2].axis('off')
 
 plt.tight_layout()
 return fig

Real-time Visualization

import cv2
import numpy as np
def real_time_visualization(model_path, camera_index=0):
 """Real-time anomaly detection with visualization."""
 model = ModelWrapper(model_path, device='cuda')
 cap = cv2.VideoCapture(camera_index)
 
 while True:
 ret, frame = cap.read()
 if not ret:
 break
 
 # Preprocess
 rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
 image = Image.fromarray(rgb_frame)
 batch = anodet.to_batch([np.array(image)])
 
 # Inference
 scores, maps = model.predict(batch)
 prediction = anodet.classification(scores, threshold=13)[0]
 
 # Create heatmap overlay
 heatmap = viz.heatmap_image(rgb_frame, maps[0], alpha=0.4)
 heatmap_bgr = cv2.cvtColor(heatmap, cv2.COLOR_RGB2BGR)
 
 # Add text overlay
 color = (0, 255, 0) if prediction == 1 else (0, 0, 255)
 text = f"Score: {scores[0]:.2f} - {'NORMAL' if prediction == 1 else 'ANOMALY'}"
 cv2.putText(heatmap_bgr, text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 
 1, color, 2, cv2.LINE_AA)
 
 # Display
 cv2.imshow('Anomaly Detection', heatmap_bgr)
 
 if cv2.waitKey(1) & 0xFF == ord('q'):
 break
 
 cap.release()
 cv2.destroyAllWindows()
 model.close()

Batch Visualization

def visualize_batch_results(model, dataloader, save_dir=None):
 """Visualize results for a batch of images."""
 for batch_idx, (batch, images, _, _) in enumerate(dataloader):
 # Run inference
 image_scores, score_maps = model.predict(batch)
 score_maps = anodet.utils.adaptive_gaussian_blur(score_maps)
 
 # Classifications
 score_map_classifications = anodet.classification(score_maps, threshold=13)
 image_classifications = anodet.classification(image_scores, threshold=13)
 
 # Create grid visualization
 batch_size = len(images)
 fig, axes = plt.subplots(batch_size, 4, figsize=(16, 4*batch_size))
 
 for i in range(batch_size):
 # Original
 axes[i, 0].imshow(images[i])
 axes[i, 0].set_title(f'Original {i+1}')
 axes[i, 0].axis('off')
 
 # Heatmap
 heatmap = viz.heatmap_image(images[i], score_maps[i], alpha=0.6)
 axes[i, 1].imshow(heatmap)
 axes[i, 1].set_title(f'Heatmap\nScore: {image_scores[i]:.2f}')
 axes[i, 1].axis('off')
 
 # Boundary
 boundary = viz.boundary_image(images[i], score_map_classifications[i])
 axes[i, 2].imshow(boundary)
 axes[i, 2].set_title('Boundaries')
 axes[i, 2].axis('off')
 
 # Highlighted
 highlighted = viz.highlighted_image(images[i], score_map_classifications[i])
 axes[i, 3].imshow(highlighted)
 axes[i, 3].set_title('Highlighted')
 axes[i, 3].axis('off')
 
 plt.tight_layout()
 
 if save_dir:
 os.makedirs(save_dir, exist_ok=True)
 plt.savefig(f"{save_dir}/batch_{batch_idx}.png", dpi=150, bbox_inches='tight')
 
 plt.show()
 break # Only visualize first batch

Custom Visualization Functions

def create_custom_heatmap(image, score_map, colormap='jet', alpha=0.6):
 """Create custom heatmap with different colormaps."""
 import cv2
 
 # Normalize score map
 normalized_scores = (score_map - score_map.min()) / (score_map.max() - score_map.min())
 normalized_scores = (normalized_scores * 255).astype(np.uint8)
 
 # Apply colormap
 colormap_dict = {
 'jet': cv2.COLORMAP_JET,
 'hot': cv2.COLORMAP_HOT,
 'cool': cv2.COLORMAP_COOL,
 'viridis': cv2.COLORMAP_VIRIDIS,
 'plasma': cv2.COLORMAP_PLASMA
 }
 
 colored_heatmap = cv2.applyColorMap(normalized_scores, colormap_dict[colormap])
 colored_heatmap = cv2.cvtColor(colored_heatmap, cv2.COLOR_BGR2RGB)
 
 # Blend with original image
 blended = cv2.addWeighted(image, 1-alpha, colored_heatmap, alpha, 0)
 
 return blended
def plot_score_distribution(scores, threshold=13, title="Score Distribution"):
 """Plot distribution of anomaly scores."""
 plt.figure(figsize=(10, 6))
 
 # Histogram
 plt.hist(scores, bins=50, alpha=0.7, edgecolor='black')
 
 # Threshold line
 plt.axvline(threshold, color='red', linestyle='--', linewidth=2, 
 label=f'Threshold: {threshold}')
 
 # Statistics
 plt.axvline(np.mean(scores), color='green', linestyle='-', linewidth=2,
 label=f'Mean: {np.mean(scores):.2f}')
 plt.axvline(np.median(scores), color='blue', linestyle='-', linewidth=2,
 label=f'Median: {np.median(scores):.2f}')
 
 plt.xlabel('Anomaly Score')
 plt.ylabel('Frequency')
 plt.title(title)
 plt.legend()
 plt.grid(True, alpha=0.3)
 plt.show()

Core Classes

Padim Class

class Padim(torch.nn.Module):
 """
 PaDiM anomaly detection model.

 Args:
 backbone (str): ResNet architecture ('resnet18', 'wide_resnet50')
 device (torch.device): Computation device
 layer_indices (List[int]): ResNet layers to extract features from [0-3]
 feat_dim (int): Target feature dimension after channel selection
 channel_indices (torch.Tensor): Specific channel indices (optional)
 layer_hook (Callable): Function to apply to extracted features (optional)
 """
 
 def __init__(self, backbone='resnet18', device=torch.device('cpu'), 
 layer_indices=[0, 1], feat_dim=50, **kwargs):
 pass
 
 def fit(self, dataloader, extractions=1):
 """
 Fit model to normal training data.

 Args:
 dataloader: PyTorch DataLoader with normal images
 extractions: Number of passes through data (for augmentation)
 """
 pass
 
 def predict(self, batch, export=False):
 """
 Predict anomaly scores and maps.

 Args:
 batch (torch.Tensor): Input images (B, C, H, W)
 export (bool): Use export-friendly computation paths

 Returns:
 Tuple[torch.Tensor, torch.Tensor]: (image_scores, score_maps)
 """
 pass
 
 def evaluate(self, dataloader):
 """
 Evaluate model on test data.

 Args:
 dataloader: Test data loader

 Returns:
 Tuple: (images, targets, masks, scores, maps)
 """
 pass
 
 def save_statistics(self, path, half=False):
 """
 Save model statistics for deployment.

 Args:
 path (str): Output file path
 half (bool): Use FP16 precision for smaller files
 """
 pass
 
 @staticmethod
 def load_statistics(path, device='cpu', force_fp32=True):
 """
 Load model statistics from file.

 Args:
 path (str): Statistics file path
 device (str): Target device
 force_fp32 (bool): Convert to FP32 for computation

 Returns:
 dict: Statistics dictionary
 """
 pass

ModelWrapper Class

class ModelWrapper:
 """
 Universal model wrapper for all supported formats.

 Args:
 model_path (str): Path to model file
 device (str): Target device ('cpu', 'cuda', 'auto')
 """
 
 def __init__(self, model_path, device='cuda'):
 pass
 
 def predict(self, batch):
 """
 Run inference on input batch.

 Args:
 batch: Input tensor or numpy array

 Returns:
 Tuple[np.ndarray, np.ndarray]: (scores, maps)
 """
 pass
 
 def close(self):
 """Release model resources."""
 pass
 
 def warmup(self, batch=None, runs=2):
 """
 Warm up model for consistent performance.

 Args:
 batch: Sample input for warmup
 runs (int): Number of warmup iterations
 """
 pass

Dataset Classes

class AnodetDataset(Dataset):
 """
 Flexible dataset for anomaly detection with configurable image processing.

 Args:
 image_directory_path (str): Path to directory containing images
 mask_directory_path (str, optional): Path to masks directory
 resize (Union[int, Tuple[int, int]]): Resize dimensions
 crop_size (Union[int, Tuple[int, int]], optional): Crop dimensions
 normalize (bool): Apply ImageNet normalization
 mean (List[float]): Normalization mean values
 std (List[float]): Normalization std values
 """
 
 def __init__(self, image_directory_path, mask_directory_path=None,
 resize=224, crop_size=224, normalize=True, **kwargs):
 pass
 
 def __len__(self):
 """Return dataset size."""
 pass
 
 def __getitem__(self, idx):
 """
 Get item by index.

 Returns:
 Tuple: (tensor, image_array, classification, mask)
 """
 pass
class MVTecDataset(Dataset):
 """
 MVTec anomaly detection dataset loader.

 Args:
 dataset_path (str): Path to MVTec dataset root
 class_name (str): Class name (must be in CLASS_NAMES)
 is_train (bool): Load training or test data
 resize (Union[int, Tuple[int, int]]): Resize dimensions
 crop_size (Union[int, Tuple[int, int]], optional): Crop dimensions
 normalize (bool): Apply ImageNet normalization
 """
 
 def __init__(self, dataset_path, class_name, is_train=True, **kwargs):
 pass

Utility Functions

Image Processing

def create_image_transform(resize=224, crop_size=None, normalize=True, 
 mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]):
 """
 Create configurable image transform pipeline.

 Args:
 resize: Size to resize to
 crop_size: Size to crop to (optional)
 normalize: Whether to apply normalization
 mean: Normalization mean values
 std: Normalization std values

 Returns:
 torchvision.transforms.Compose: Transform pipeline
 """
 pass
def to_batch(images, transforms=None, device='cpu', **kwargs):
 """
 Convert list of numpy images to PyTorch tensor batch.

 Args:
 images (List[np.ndarray]): List of images
 transforms: Optional transforms to apply
 device: Target device

 Returns:
 torch.Tensor: Batch tensor
 """
 pass
def adaptive_gaussian_blur(input_array, kernel_size=33, sigma=4):
 """
 Apply Gaussian blur with automatic backend selection.

 Args:
 input_array: Input tensor or array
 kernel_size (int): Blur kernel size
 sigma (float): Gaussian sigma

 Returns:
 Blurred array (same type as input)
 """
 pass

Classification and Scoring

def classification(image_scores, thresh):
 """
 Classify images based on anomaly scores.

 Args:
 image_scores: Anomaly scores (tensor or numpy array)
 thresh (float): Classification threshold

 Returns:
 Classifications (same type as input): 0=anomaly, 1=normal
 """
 pass
def image_score(patch_scores):
 """
 Calculate image-level scores from patch scores.

 Args:
 patch_scores (torch.Tensor): Patch-level scores

 Returns:
 torch.Tensor: Image-level scores
 """
 pass

Statistical Functions

def pytorch_cov(tensor, rowvar=True, bias=False):
 """
 Estimate covariance matrix (equivalent to np.cov).

 Args:
 tensor (torch.Tensor): Input tensor
 rowvar (bool): Whether rows are variables
 bias (bool): Whether to use bias correction

 Returns:
 torch.Tensor: Covariance matrix
 """
 pass
def mahalanobis(mean, cov_inv, batch):
 """
 Calculate Mahalanobis distance.

 Args:
 mean (torch.Tensor): Mean vectors
 cov_inv (torch.Tensor): Inverse covariance matrices
 batch (torch.Tensor): Input batch

 Returns:
 torch.Tensor: Mahalanobis distances
 """
 pass

Logging and Profiling

def setup_logging(log_level="INFO"):
 """
 Setup logging configuration.

 Args:
 log_level (str): Logging level

 Returns:
 logging.Logger: Configured logger
 """
 pass
def get_logger(name=None):
 """
 Get logger for specific module.

 Args:
 name (str): Logger name (use __name__ from calling module)

 Returns:
 logging.Logger: Module logger
 """
 pass
class Profiler:
 """
 Performance profiler for timing measurements.

 Usage:
 with Profiler() as prof:
 # code to profile
 print(f"Elapsed: {prof.elapsed_time:.3f}s")
 """
 
 def __init__(self, accumulated_time=0.0):
 pass
 
 def __enter__(self):
 """Start timing."""
 pass
 
 def __exit__(self, exc_type, exc_value, traceback):
 """Stop timing and accumulate."""
 pass
 
 def get_fps(self, num_samples):
 """Calculate FPS from accumulated time."""
 pass
 
 def get_avg_time_ms(self, num_operations):
 """Get average time per operation in milliseconds."""
 pass

Configuration Options

Complete Configuration Reference

# =========================
# Dataset Configuration
# =========================
dataset_path: "D:/01-DATA" # Root dataset directory
class_name: "bottle" # MVTec class name
img_path: "test_images/" # Test images path (for inference)
# =========================
# Image Processing
# =========================
resize: [224, 224] # Resize dimensions [width, height]
crop_size: [224, 224] # Crop dimensions [width, height]
normalize: true # Enable ImageNet normalization
norm_mean: [0.485, 0.456, 0.406] # RGB normalization means
norm_std: [0.229, 0.224, 0.225] # RGB normalization stds
# =========================
# Model Configuration
# =========================
backbone: "resnet18" # resnet18, wide_resnet50
feat_dim: 100 # Feature dimension after selection
layer_indices: [0, 1] # ResNet layers to extract [0-3]
device: "auto" # cpu, cuda, auto
batch_size: 8 # Batch size for training/inference
# =========================
# Training Configuration
# =========================
model_data_path: "./distributions/exp" # Model output directory
output_model: "padim_model.pt" # Model filename
run_name: "experiment_1" # Experiment name
epochs: 1 # Training epochs (usually 1 for PaDiM)
extractions: 1 # Dataset passes
# =========================
# Inference Configuration
# =========================
thresh: 13.0 # Anomaly classification threshold
num_workers: 1 # DataLoader workers
pin_memory: true # Enable pinned memory
# =========================
# Visualization Configuration
# =========================
enable_visualization: true # Enable result visualization
save_visualizations: false # Save visualization images
viz_output_dir: "./visualizations/" # Visualization output directory
viz_alpha: 0.6 # Heatmap overlay transparency
viz_padding: 40 # Boundary visualization padding
viz_color: "128,0,128" # RGB highlight color
# =========================
# Export Configuration
# =========================
format: "onnx" # onnx, torchscript, openvino, all
opset: 17 # ONNX opset version
dynamic_batch: true # Allow dynamic batch sizes
fp16: true # Use FP16 precision (OpenVINO)
optimize: false # TorchScript mobile optimization
# =========================
# Evaluation Configuration
# =========================
metrics: ["auroc", "pixel_auroc"] # Evaluation metrics
memory_efficient: true # Use memory-efficient evaluation
# =========================
# System Configuration
# =========================
log_level: "INFO" # DEBUG, INFO, WARNING, ERROR, CRITICAL
detailed_timing: false # Enable detailed performance timing
overwrite: false # Overwrite existing experiment directories

Parameter Validation

def validate_config(config):
 """Validate configuration parameters."""
 
 # Required parameters
 required = ['dataset_path', 'backbone', 'batch_size']
 for param in required:
 if not hasattr(config, param) or getattr(config, param) is None:
 raise ValueError(f"Required parameter '{param}' is missing")
 
 # Backbone validation
 valid_backbones = ['resnet18', 'wide_resnet50']
 if config.backbone not in valid_backbones:
 raise ValueError(f"backbone must be one of {valid_backbones}")
 
 # Layer indices validation
 if config.layer_indices:
 valid_layers = [0, 1, 2, 3]
 for layer in config.layer_indices:
 if layer not in valid_layers:
 raise ValueError(f"layer_indices must be subset of {valid_layers}")
 
 # Batch size validation
 if config.batch_size <= 0:
 raise ValueError("batch_size must be positive")
 
 # Feature dimension validation
 if config.feat_dim <= 0:
 raise ValueError("feat_dim must be positive")
 
 # Threshold validation
 if hasattr(config, 'thresh') and config.thresh < 0:
 raise ValueError("thresh must be non-negative")
 
 # Image processing validation
 if config.resize:
 if isinstance(config.resize, (list, tuple)):
 if len(config.resize) != 2 or any(x <= 0 for x in config.resize):
 raise ValueError("resize must be [width, height] with positive values")
 elif not isinstance(config.resize, int) or config.resize <= 0:
 raise ValueError("resize must be positive integer or [width, height]")
 
 return True

Environment-Specific Configs

# Development configuration
development:
 log_level: "DEBUG"
 batch_size: 2
 detailed_timing: true
 enable_visualization: true
 save_visualizations: true
# Production configuration 
production:
 log_level: "WARNING"
 batch_size: 32
 detailed_timing: false
 enable_visualization: false
 save_visualizations: false
 device: "cuda"
# Edge deployment configuration
edge:
 backbone: "resnet18"
 feat_dim: 50
 batch_size: 1
 device: "cpu"
 format: "onnx"
 fp16: true

Contributing

Development Setup

# Fork and clone
git clone https://github.com/yourusername/AnomaVision.git
cd AnomaVision
# Setup development environment
poetry install --dev
pre-commit install
# Create feature branch
git checkout -b feature/awesome-improvement

Development Guidelines

Code Style

Type Hints

# Add type hints to all new functions
def process_batch(
 batch: torch.Tensor, 
 model: Padim, 
 threshold: float = 13.0
) -> Tuple[np.ndarray, np.ndarray]:
 """Process batch with type hints."""
 pass

Documentation

def new_function(param1: str, param2: int = 10) -> bool:
 """
 Brief description of the function.

 Longer description explaining the purpose, behavior, and any important
 details about the function implementation.

 Args:
 param1 (str): Description of parameter 1
 param2 (int, optional): Description of parameter 2. Defaults to 10.

 Returns:
 bool: Description of return value

 Raises:
 ValueError: When parameter validation fails
 RuntimeError: When operation cannot be completed

 Example:
 >>> result = new_function("test", 20)
 >>> print(result)
 True
 """
 pass

Testing

# Add pytest tests for new functionality
def test_new_function():
 """Test new function with various inputs."""
 # Test normal case
 result = new_function("valid_input")
 assert result is True
 
 # Test edge cases
 with pytest.raises(ValueError):
 new_function("")
 
 # Test with different parameters
 result = new_function("test", param2=5)
 assert isinstance(result, bool)

Contribution Workflow

Code Review Process

PR Requirements

Review Checklist

Areas for Contribution

High Priority

Medium Priority

Low Priority

Getting Help

Testing

Running Tests

# Run all tests
poetry run pytest
# Run with coverage
poetry run pytest --cov=anodet --cov-report=html
# Run specific test files
poetry run pytest tests/test_padim.py
poetry run pytest tests/test_backends.py
# Run with verbose output
poetry run pytest -v
# Run tests matching pattern
poetry run pytest -k "test_inference"

Test Structure

tests/
├── conftest.py # Shared fixtures and configuration
├── test_padim.py # Core PaDiM functionality
├── test_backends.py # Multi-format backend tests
├── test_datasets.py # Dataset loading tests
├── test_feature_extraction.py # Feature extraction tests
├── test_mahalanobis.py # Distance computation tests
├── test_export_load_model.py # Export functionality tests
├── test_visualization.py # Visualization tests
└── test_inference_utils.py # Utility function tests

Key Test Fixtures

# From conftest.py
@pytest.fixture
def test_device():
 """Provide test device (CPU for CI compatibility)."""
 return torch.device("cpu")
@pytest.fixture
def sample_images_dir():
 """Create temporary directory with sample images."""
 pass
@pytest.fixture
def trained_padim_model(sample_dataloader, test_device):
 """Train a minimal PaDiM model for testing."""
 pass
@pytest.fixture
def sample_batch(sample_dataloader):
 """Get a single batch for testing."""
 pass

Writing Tests

Basic Test Example

def test_padim_initialization(test_device):
 """Test PaDiM model initialization."""
 model = anodet.Padim(
 backbone="resnet18", 
 device=test_device, 
 layer_indices=[0, 1], 
 feat_dim=50
 )
 
 assert model.device == test_device
 assert model.layer_indices == [0, 1]
 assert model.embeddings_extractor.backbone_name == "resnet18"

Parametrized Tests

@pytest.mark.parametrize(
 "backbone,expected_layers",
 [
 ("resnet18", [0, 1]),
 ("wide_resnet50", [0, 1, 2]),
 ]
)
def test_different_backbones(test_device, backbone, expected_layers):
 """Test PaDiM with different backbone architectures."""
 model = anodet.Padim(
 backbone=backbone,
 device=test_device,
 layer_indices=expected_layers
 )
 assert model.embeddings_extractor.backbone_name == backbone

Integration Tests

def test_full_pipeline(sample_dataloader, test_device):
 """Test complete training and inference pipeline."""
 # Initialize model
 model = anodet.Padim(backbone="resnet18", device=test_device)
 
 # Train
 model.fit(sample_dataloader)
 
 # Test inference
 batch, _, _, _ = next(iter(sample_dataloader))
 scores, maps = model.predict(batch)
 
 # Validate outputs
 assert scores.shape == (batch.size(0),)
 assert maps.shape == (batch.size(0), batch.size(2), batch.size(3))

Test Categories

Unit Tests

Integration Tests

Property-Based Tests

from hypothesis import given, strategies as st
@given(
 batch_size=st.integers(min_value=1, max_value=16),
 height=st.integers(min_value=32, max_value=512),
 width=st.integers(min_value=32, max_value=512)
)
def test_predict_with_random_inputs(trained_padim_model, batch_size, height, width):
 """Test prediction with random input dimensions."""
 batch = torch.randn(batch_size, 3, height, width)
 scores, maps = trained_padim_model.predict(batch)
 
 assert scores.shape == (batch_size,)
 assert maps.shape == (batch_size, height, width)
 assert torch.all(scores >= 0)
 assert torch.all(maps >= 0)

Continuous Integration

# .github/workflows/ci.yml
name: CI
on:
 push:
 branches: [develop, main]
 pull_request:
 branches: [develop, main]
jobs:
 tests:
 runs-on: ubuntu-latest
 strategy:
 matrix:
 python-version: ['3.9', '3.10', '3.11']
 
 steps:
 - uses: actions/checkout@v4
 - name: Set up Python
 uses: actions/setup-python@v5
 with:
 python-version: ${{ matrix.python-version }}
 
 - name: Install dependencies
 run: |
 python -m pip install --upgrade pip
 pip install poetry
 poetry install
 
 - name: Run tests
 run: poetry run pytest --cov=anodet
 
 - name: Upload coverage
 uses: codecov/codecov-action@v3

Performance Tests

def test_inference_performance(trained_padim_model, sample_batch):
 """Test inference performance meets requirements."""
 from anodet.general import Profiler
 
 batch, _, _, _ = sample_batch
 
 # Warmup
 for _ in range(3):
 _ = trained_padim_model.predict(batch)
 
 # Benchmark
 profiler = Profiler()
 runs = 10
 
 for _ in range(runs):
 with profiler:
 scores, maps = trained_padim_model.predict(batch)
 
 avg_time = profiler.get_avg_time_ms(runs)
 fps = profiler.get_fps(len(batch) * runs)
 
 # Performance assertions
 assert avg_time < 1000, f"Inference too slow: {avg_time:.2f}ms"
 assert fps > 1, f"FPS too low: {fps:.2f}"

Development Setup

Prerequisites

Initial Setup

# Clone repository
git clone https://github.com/DeepKnowledge1/AnomaVision.git
cd AnomaVision
# Install development dependencies
poetry install --dev
# Activate virtual environment
poetry shell
# Install pre-commit hooks
pre-commit install
# Verify installation
python -c "import anodet; print('✅ Development setup complete')"

Development Tools

Code Formatting

# Format code with Black
black .
# Sort imports with isort
isort .
# Lint with flake8
flake8 anodet/
# Type checking with mypy (optional)
mypy anodet/

Pre-commit Hooks

# .pre-commit-config.yaml
repos:
 - repo: https://github.com/psf/black
 rev: 23.9.1
 hooks:
 - id: black
 language_version: python3.9
 - repo: https://github.com/pycqa/isort
 rev: 5.12.0
 hooks:
 - id: isort
 - repo: https://github.com/pycqa/flake8
 rev: 6.1.0
 hooks:
 - id: flake8
 args: [--max-line-length=88, --extend-ignore=E203,W503]

IDE Configuration

VS Code Settings

{
 "python.defaultInterpreterPath": ".venv/bin/python",
 "python.formatting.provider": "black",
 "python.linting.enabled": true,
 "python.linting.flake8Enabled": true,
 "python.testing.pytestEnabled": true,
 "python.testing.pytestArgs": ["tests/"],
 "files.exclude": {
 "**/__pycache__": true,
 "**/*.pyc": true,
 ".pytest_cache": true,
 ".coverage": true,
 "htmlcov": true
 }
}

PyCharm Configuration

Debug Configuration

# Debug training script
python -m pdb train.py --config debug_config.yml
# Debug with VS Code
# Add to launch.json:
{
 "name": "Debug Training",
 "type": "python",
 "request": "launch",
 "program": "train.py",
 "args": ["--config", "debug_config.yml"],
 "console": "integratedTerminal"
}

Development Workflow

# 1. Create feature branch
git checkout -b feature/new-awesome-feature
# 2. Make changes and test frequently
poetry run pytest tests/test_new_feature.py
# 3. Run full test suite
poetry run pytest
# 4. Check code quality
black .
flake8 anodet/
isort .
# 5. Commit changes
git add .
git commit -m "feat: add awesome new feature"
# 6. Push and create PR
git push origin feature/new-awesome-feature

Building Documentation

# Install documentation dependencies
poetry install --extras docs
# Build docs locally
cd docs/
make html
# Serve docs locally
python -m http.server 8000 -d _build/html/

Package Building

# Build package
poetry build
# Check package
poetry run twine check dist/*
# Install locally for testing
pip install dist/anomavision-*.whl

Troubleshooting

Common Issues

# Verify Python path
python -c "import sys; print(sys.path)"
# Reinstall in development mode
poetry install --dev

# Check CUDA availability
python -c "import torch; print(torch.cuda.is_available())"
# Use CPU-only for development
export CUDA_VISIBLE_DEVICES=""

# Run tests with verbose output
poetry run pytest -v -s
# Run specific failing test
poetry run pytest tests/test_specific.py::test_function -v

# Run tests with limited parallelism
poetry run pytest -x --tb=short
# Use smaller test datasets
export ANOMAVISION_TEST_SIZE=small

FAQ

General Questions

Installation & Setup

# Reinstall with poetry
poetry install --dev
poetry shell
# Or use pip in a fresh environment
pip uninstall anomavision
pip install -r requirements.txt

pip install openvino # Intel optimization
pip install onnxruntime-gpu # ONNX with GPU support
pip install tensorrt # NVIDIA optimization (future)

Training & Models

# Any aspect ratio and size
dataset = anodet.AnodetDataset(
 path,
 resize=[640, 480], # Width x Height
 crop_size=[224, 224] # Final processing size
)

Inference & Deployment

# 1. Use appropriate backend
model = ModelWrapper("model_openvino.xml", device='CPU')
# 2. Batch processing
dataloader = DataLoader(dataset, batch_size=32)
# 3. Warmup model
model.warmup(sample_batch, runs=3)
# 4. Pin memory for GPU
dataloader = DataLoader(dataset, pin_memory=True)

Performance & Optimization

# For Intel hardware
model = ModelWrapper("model_openvino.xml", device='CPU')
# For NVIDIA GPUs
model = ModelWrapper("model.onnx", device='cuda')

# Find optimal batch size
optimal_bs = find_optimal_batch_size(model_path, sample_batch)

# 2-4x smaller, faster loading
model = ModelWrapper("model.pth", device='cpu')

Troubleshooting

# Use smaller batch size
dataloader = DataLoader(dataset, batch_size=1)
# Use CPU backend
model = ModelWrapper(model_path, device='cpu')
# Use memory-efficient evaluation
results = model.evaluate_memory_efficient(dataloader)

# Ensure consistent precision
model.save_statistics("model.pth", half=False) # Use FP32
# Load with force_fp32
stats = Padim.load_statistics("model.pth", force_fp32=True)
# Use consistent export settings
exporter.export_onnx(dynamic_batch=False, opset_version=17)

# Use CPU-only tests
@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available")
def test_gpu_function():
 pass

# Use smaller test datasets
@pytest.fixture
def small_dataset():
 return create_small_test_dataset(size=10)

# Install all test dependencies
poetry install --dev --extras test

# Check feature extraction
features, w, h = model.embeddings_extractor(batch)
print(f"Features shape: {features.shape}")
# Check distance computation
distances = model.mahalanobisDistance(features, w, h)
print(f"Distance range: {distances.min():.3f} - {distances.max():.3f}")

# Load reference model
ref_model = torch.load("reference_model.pt")
ref_scores, ref_maps = ref_model.predict(batch)
# Compare results
score_diff = torch.abs(scores - ref_scores).max()
print(f"Max score difference: {score_diff:.6f}")

Advanced Usage

# Custom layer hook
def custom_hook(features):
 # Apply custom processing
 return F.normalize(features, dim=1)
model = anodet.Padim(
 backbone='resnet18',
 layer_hook=custom_hook,
 layer_indices=[0, 1, 2]
)
# Custom distance computation
class CustomDistance(MahalanobisDistance):
 def forward(self, features, width, height, **kwargs):
 # Custom distance calculation
 return custom_distance_function(features)
# Replace distance module
model.mahalanobisDistance = CustomDistance(mean, cov_inv)

# MLflow integration
import mlflow
with mlflow.start_run():
 # Train model
 model = anodet.Padim()
 model.fit(dataloader)
 
 # Log metrics
 scores, maps = model.predict(test_batch)
 mlflow.log_metric("avg_score", scores.mean().item())
 
 # Log model
 mlflow.pytorch.log_model(model, "anomaly_model")
# Weights & Biases integration
import wandb
wandb.init(project="anomaly-detection")
wandb.log({"avg_score": scores.mean().item()})
wandb.save("model.pt")

# Process video frame by frame
import cv2
def process_video(video_path, model_path):
 model = ModelWrapper(model_path, device='cuda')
 cap = cv2.VideoCapture(video_path)
 
 frame_scores = []
 while True:
 ret, frame = cap.read()
 if not ret:
 break
 
 # Convert frame to batch
 rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
 batch = anodet.to_batch([rgb_frame])
 
 # Detect anomalies
 scores, maps = model.predict(batch)
 frame_scores.append(scores[0])
 
 return frame_scores

Community & Support

Changelog

Version 2.0.46 (Current)

🚀 New Features

⚡ Performance Improvements

🛠️ Infrastructure

📚 Documentation

🐛 Bug Fixes

License

MIT License
Copyright (c) 2025 DeepKnowledge Contributors
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

Third-Party Licenses

Citation

@software{anomavision2025,
 title={AnomaVision: Edge-Ready Visual Anomaly Detection},
 author={DeepKnowledge Contributors},
 year={2025},
 url={https://github.com/DeepKnowledge1/AnomaVision},
 version={2.0.46},
 note={High-performance anomaly detection library optimized for edge deployment}
}

Contact

🤝 Community

💼 Enterprise

🌟 Acknowledgments

Stop settling for slow, bloated solutions. Experience the future of edge-ready anomaly detection.

Get Started Documentation Star Us

🏆 Benchmark Results Don't Lie: AnomaVision Wins 10/10 Metrics

Deploy fast. Detect better. Scale everywhere.

Made with ❤️ for the edge AI community

Last updated: January 2025 | AnomaVision v2.0.46