Home

AnomaVision Wiki

Welcome to the AnomaVision documentation! This wiki provides comprehensive guides, API references, and examples for using AnomaVision - a production-ready visual anomaly detection library.

📚 Table of Contents

Home

AnomaVision is a high-performance, production-ready visual anomaly detection library built on the state-of-the-art PaDiM algorithm. It provides enterprise-grade performance with research-level accuracy, supporting multiple export formats for deployment anywhere from edge devices to cloud infrastructure.

🔥 Key Features

🎯 Unmatched Performance: Optimized PaDiM implementation with CPU-first design
🔄 Multi-Format Support: PyTorch, ONNX, TorchScript, OpenVINO, and more
📦 Production Ready: Enterprise-grade deployment capabilities
🎨 Rich Visualizations: Comprehensive anomaly visualization tools
📏 Flexible Image Dimensions: Support for any image size and aspect ratio
⚡ Edge-Ready: Optimized for edge device deployment

🏆 Performance Highlights

2-4x smaller model files with statistics-only .pth format
CPU-optimized pipeline that works without GPU
Multi-format export from a single trained model
Plug-and-play loading with unified inference interface

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

Python: 3.9+
CUDA: 11.7+ for GPU acceleration (optional)
PyTorch: 2.0+ (automatically installed)

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

For full functionality, install additional backends:

# 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 is built with a modular architecture designed for production deployment:

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`)

Full-featured PaDiM implementation
Training and inference capabilities
Statistics saving for deployment

2. PaDiM Lite (`padim_lite.py`)

Lightweight runtime for deployment
Loads from statistics-only files
Minimal memory footprint

3. Inference Engine (`inference/`)

ModelWrapper: Unified interface for all formats
Backends: Format-specific implementations
ModelType: Automatic format detection

4. Feature Extraction (`feature_extraction.py`)

ResNet backbone support (ResNet18, Wide-ResNet50)
Multi-scale feature extraction
Optimized concatenation and processing

5. Distance Computation (`mahalanobis.py`)

Efficient Mahalanobis distance calculation
Memory-optimized chunked computation
ONNX export compatibility

PaDiM Algorithm

PaDiM (Patch Distribution Modeling) is a state-of-the-art anomaly detection algorithm that models the distribution of features at each spatial location.

How PaDiM Works

Feature Extraction: Extract multi-scale features from pre-trained ResNet
Statistical Modeling: Compute mean and covariance for each spatial location
Anomaly Scoring: Calculate Mahalanobis distance to detect deviations

Key Advantages

No additional training: Uses pre-trained features
Pixel-level detection: Provides detailed anomaly maps
Robust performance: Works across various domains
Interpretable results: Clear statistical foundation

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

AnomaVision supports multiple model formats for flexible deployment:

Format	Status	Use Case	Language Support
PyTorch	✅ Ready	Development & Research	Python
Statistics (.pth)	✅ Ready	Ultra-compact deployment (2-4x smaller)	Python
ONNX	✅ Ready	Cross-platform deployment	Python, C++
TorchScript	✅ Ready	Production Python deployment	Python
OpenVINO	✅ Ready	Intel hardware optimization	Python
TensorRT	🚧 Coming Soon	NVIDIA GPU acceleration	Python

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

All formats use the same interface through ModelWrapper:

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

AnomaVision provides flexible image processing with configurable dimensions:

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

The processing pipeline follows this order:

Load: PIL.Image.open() and convert to RGB
Resize: Resize to specified dimensions
Crop: Center crop to final size (optional)
ToTensor: Convert to PyTorch tensor
Normalize: Apply normalization (optional)

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

Create config.yml:

# 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

Parameter	Description	Default	Recommended
`backbone`	Feature extractor	`resnet18`	`resnet18` for speed, `wide_resnet50` for accuracy
`layer_indices`	ResNet layers	`[0]`	`[0, 1]` for best balance
`feat_dim`	Feature dimensions	`50`	`100-200` depending on complexity
`batch_size`	Training batch size	`2`	Largest that fits in memory
`extractions`	Dataset passes	`1`	`2-3` for data augmentation

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
)

Expected structure:

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

AnomaVision uses a flexible YAML-based configuration system that supports both command-line arguments and configuration files.

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

AnomaVision provides a unified interface for all supported model formats through the ModelWrapper class.

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

AnomaVision provides comprehensive visualization tools for anomaly detection results.

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

We welcome contributions to AnomaVision! Here's how to get involved:

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

Follow PEP 8 with 88-character line limit
Use Black for code formatting: black .
Use flake8 for linting: flake8 anodet/
Use isort for import sorting: isort .

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

Create Issue: Describe the bug or feature request
Fork Repository: Create your own fork

Create Branch: Use descriptive branch names

git checkout -b feature/add-tensorrt-backend
git checkout -b bugfix/fix-memory-leak
git checkout -b docs/update-api-reference

Make Changes: Implement your improvements
Add Tests: Ensure adequate test coverage
Update Docs: Update documentation as needed

Run Tests: Ensure all tests pass

poetry run pytest
poetry run black .
poetry run flake8 anodet/

Commit Changes: Use conventional commit messages

git commit -m "feat: add TensorRT backend support"
git commit -m "fix: resolve memory leak in batch processing"
git commit -m "docs: update API reference for new features"

Push Branch: Push to your fork
Create PR: Submit pull request with detailed description

Code Review Process

PR Requirements

Review Checklist

Code quality and readability
Test coverage and quality
Documentation completeness
Performance implications
Backward compatibility
Security considerations

Areas for Contribution

High Priority

TensorRT Backend: Complete TensorRT implementation
Performance Optimization: Memory and speed improvements
Additional Algorithms: Beyond PaDiM (PatchCore, etc.)
Mobile Deployment: iOS/Android optimization

Medium Priority

Visualization Enhancements: Interactive visualizations
Data Augmentation: Advanced augmentation techniques
Distributed Training: Multi-GPU support
Model Compression: Quantization and pruning

Low Priority

Additional Datasets: Support for more dataset formats
Cloud Integration: AWS/Azure/GCP deployment tools
Web Interface: Browser-based demo
Benchmarking Suite: Comprehensive performance benchmarks

Getting Help

Discord/Slack: Join our community chat (if available)
GitHub Discussions: Ask questions and discuss ideas
Email: Contact maintainers directly for complex issues
Documentation: Check existing docs before asking

Testing

AnomaVision uses pytest for comprehensive testing across all components.

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

Individual function testing
Component isolation
Edge case validation
Error handling verification

Integration Tests

End-to-end pipeline testing
Component interaction validation
Format compatibility testing
Performance regression detection

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

Python 3.9+
Git
Poetry (recommended) or pip
CUDA 11.7+ (optional, for GPU development)

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

Set interpreter to poetry virtual environment
Enable Black as code formatter
Configure pytest as test runner
Set up flake8 as external tool

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

Import errors after installation:

# Verify Python path
python -c "import sys; print(sys.path)"

# Reinstall in development mode
poetry install --dev

CUDA-related errors:

# Check CUDA availability
python -c "import torch; print(torch.cuda.is_available())"

# Use CPU-only for development
export CUDA_VISIBLE_DEVICES=""

Test failures:

# Run tests with verbose output
poetry run pytest -v -s

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

Memory issues during testing:

# Run tests with limited parallelism
poetry run pytest -x --tb=short

# Use smaller test datasets
export ANOMAVISION_TEST_SIZE=small

FAQ

General Questions

Q: What makes AnomaVision different from other anomaly detection libraries?

A: AnomaVision is specifically designed for production deployment with:

2-4x smaller model files through statistics-only storage
Multi-format export from a single trained model
CPU-first design that works without GPU requirements
Unified inference interface across all formats
Enterprise-grade performance with research-level accuracy

Q: Can I use AnomaVision without a GPU?

A: Yes! AnomaVision is designed with a CPU-first approach. All functionality works on CPU-only machines, and we provide optimized backends like OpenVINO for Intel hardware acceleration.

Q: How does AnomaVision compare to Anomalib?

A: AnomaVision wins on 10/10 performance metrics in our benchmarks. Key advantages:

Faster inference times
Smaller model files
Better memory efficiency
More deployment options
Simpler API

Installation & Setup

Q: I'm getting import errors after installation. What should I do?

A: Try these solutions:

# Reinstall with poetry
poetry install --dev
poetry shell

# Or use pip in a fresh environment
pip uninstall anomavision
pip install -r requirements.txt

Q: How do I install optional dependencies like OpenVINO?

A: Install additional backends as needed:

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

Training & Models

Q: How much training data do I need?

A: PaDiM requires only normal (non-anomalous) training data. Typical requirements:

Minimum: 50-100 normal images
Recommended: 200-500 normal images
Optimal: 1000+ normal images

Q: What image sizes does AnomaVision support?

A: AnomaVision supports flexible image dimensions:

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

Q: Can I use custom datasets?

A: Yes! AnomaVision supports:

Custom directory structures
Any image format (PNG, JPG, JPEG)
Flexible preprocessing pipelines
Custom normalization parameters

Inference & Deployment

Q: Which export format should I use for production?

A: Choose based on your deployment target:

ONNX: Universal deployment, cross-platform
OpenVINO: Intel hardware (CPU/GPU)
TorchScript: Python production environments
Statistics (.pth): Smallest files, Python-only

Q: How do I optimize inference speed?

A: Use these optimization strategies:

# 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)

Q: Can I run AnomaVision on edge devices?

A: Yes! AnomaVision is optimized for edge deployment:

Compact .pth files (2-4x smaller)
CPU-optimized inference
Low memory footprint
ONNX export for embedded systems

Performance & Optimization

Q: My inference is slow. How can I speed it up?

A: Try these optimizations:

Use optimal model format:

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

# For NVIDIA GPUs
model = ModelWrapper("model.onnx", device='cuda')

Optimize batch size:

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

Use statistics files:

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

Q: How much memory does AnomaVision use?

A: Memory usage depends on configuration:

Statistics files: ~10-50MB
Full models: ~100-500MB
Runtime memory: ~200MB-2GB (depends on batch size)

Troubleshooting

Q: I'm getting CUDA out of memory errors.

A: Reduce memory usage:

# 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)

Q: My exported ONNX model gives different results than PyTorch.

A: This is usually due to precision differences:

# 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)

Q: Tests are failing in CI/CD. What should I check?

A: Common CI issues and solutions:

CUDA not available:

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

Memory limitations:

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

Missing dependencies:

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

Q: How do I debug model predictions?

A: Use these debugging techniques:

Visualize intermediate results:

# 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}")

Compare with reference:

# 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

Q: Can I modify the PaDiM algorithm?

A: Yes! AnomaVision is designed for customization:

# 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)

Q: How do I integrate AnomaVision with MLOps pipelines?

A: AnomaVision integrates well with MLOps tools:

# 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")

Q: Can I use AnomaVision for video anomaly detection?

A: Currently, AnomaVision focuses on image anomaly detection. For video:

# 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

Q: How can I contribute to AnomaVision?

A: We welcome contributions! See our Contributing section for details:

Code contributions: Bug fixes, new features, optimizations
Documentation: Improve docs, add examples, write tutorials
Testing: Add test cases, improve coverage
Feedback: Report bugs, suggest features, share use cases

Q: Where can I get help?

A: Multiple support channels available:

GitHub Issues: Bug reports and feature requests
GitHub Discussions: Questions and community support
Email: Direct contact with maintainers
Documentation: Comprehensive guides and examples

Q: Is AnomaVision suitable for commercial use?

A: Yes! AnomaVision is released under the MIT License, allowing commercial use. We also offer:

Enterprise support: Custom development and consulting
Training workshops: Team training and best practices
Performance optimization: Custom optimizations for your use case

Changelog

Version 2.0.46 (Current)

🚀 New Features

Statistics-only models: 2-4x smaller .pth files with PadimLite runtime
CPU-first design: Full functionality without GPU requirements
Multi-format export: Single trained model → ONNX, TorchScript, OpenVINO
Unified inference: ModelWrapper provides consistent API across formats
Flexible image processing: Support for any aspect ratio and dimensions

⚡ Performance Improvements

Optimized feature extraction: Faster ResNet processing
Memory-efficient evaluation: Reduced memory usage for large datasets
Chunked distance computation: Better memory management
GPU memory optimization: Automatic mixed precision support

🛠️ Infrastructure

Enhanced testing: Comprehensive test suite with 90%+ coverage
CI/CD pipeline: Automated testing and quality checks
Configuration system: YAML-based configuration with CLI override
Profiling tools: Built-in performance measurement

📚 Documentation

Complete API reference: Detailed documentation for all classes
Usage examples: Real-world examples and tutorials
Performance guides: Optimization recommendations
Deployment guides: Production deployment best practices

🐛 Bug Fixes

Fixed memory leaks in batch processing
Resolved ONNX export compatibility issues
Fixed visualization rendering problems
Corrected device placement inconsistencies

License

AnomaVision is released under the MIT 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

AnomaVision builds upon excellent open-source projects:

PyTorch: BSD-style license
torchvision: BSD 3-Clause license
NumPy: BSD license
Pillow: HPND license
OpenCV: Apache 2.0 license
ONNX Runtime: MIT license
OpenVINO: Apache 2.0 license

Citation

If AnomaVision helps your research or project, please cite:

@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

💬 GitHub Discussions: Community Forum
🐛 Issues: Bug Reports & Features
📧 Email: deepp.knowledge@gmail.com
📖 Documentation: Wiki

💼 Enterprise

For enterprise deployments, custom integrations, or commercial support:

🏢 Enterprise Consulting: Custom development and optimization
🎓 Training Workshops: Team training and best practices
🔧 Custom Development: Tailored solutions for your use case
⚡ Performance Optimization: Hardware-specific optimizations

🌟 Acknowledgments

Special thanks to:

PaDiM Authors: For the original algorithm (Defard et al.)
PyTorch Team: For the excellent deep learning framework
ONNX Community: For cross-platform deployment standards
Intel: For OpenVINO optimization toolkit
Contributors: All community members who help improve AnomaVision

🚀 Ready to Transform Your Anomaly Detection?

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

🏆 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

Home

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

2-Minute Training

Instant Detection

Installation

Prerequisites

Method 1: Poetry (Recommended)

Method 2: pip

Method 3: Development Setup

Verify Installation

Optional Dependencies

Basic Usage

Training a Model

Running Inference

Command Line Interface

Architecture Overview

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

Supported Formats

Format Comparison

PyTorch (.pt)

Statistics (.pth)

ONNX (.onnx)

TorchScript (.torchscript)

OpenVINO (.xml/.bin)

Universal Interface

Image Processing

Flexible Dimensions

Normalization Options

Transform Pipeline

Custom Transforms

Training Models

Command Line Training

Configuration File Training

Programmatic Training

Training Parameters

MVTec Dataset Training

Running Inference

Command Line Inference

Programmatic Inference

Batch Processing

Real-time Inference

Performance Optimization

Model Export

Export All Formats

Programmatic Export

Export Configuration

Format-Specific Options

ONNX Export

TorchScript Export

OpenVINO Export

Statistics Export

Model Evaluation

Command Line Evaluation

Programmatic Evaluation

Performance Metrics

Optimal Threshold

Memory-Efficient Evaluation

Cross-Format Evaluation

Configuration System

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`)