Image-Based Anomaly Detection System

System Development детекции аномалий по изображениям

Детекция аномалий на изображениях — задача нахождения отклонений от нормального состояния без явного определения класса дефекта. Ключевое отличие от supervised детекции: модель обучается только на нормальных примерах, не требуя размеченных дефектов. Applications: промышленный контроль качества, медицинская диагностика, безопасность.

PatchCore: лучший метод для промышленных задач

from anomalib.models import PatchCore
from anomalib.data import Folder
from anomalib import TaskType
from lightning.pytorch import Trainer

# Датасет: только нормальные изображения в папке train/good/
datamodule = Folder(
    name='my_product',
    root='./dataset',
    normal_dir='train/good',
    abnormal_dir='test/defective',  # только для валидации
    task=TaskType.SEGMENTATION,
    image_size=(256, 256),
    train_batch_size=32,
    eval_batch_size=32
)

model = PatchCore(
    backbone='wide_resnet50_2',
    pre_trained=True,
    layers=['layer2', 'layer3'],  # какие слои ResNet использовать
    coreset_sampling_ratio=0.1,    # 10% от memory bank
    num_neighbors=9
)

trainer = Trainer(max_epochs=1)  # PatchCore не обучается градиентно
trainer.fit(model, datamodule)

Сравнение методов Anomaly Detection

EfficientAD — лучший выбор для production из-за скорости инференса при сохранении высокого AUROC.

Настройка порога обнаружения

class AnomalyDetector:
    def __init__(self, model_path: str, threshold: float = None):
        self.model = load_model(model_path)
        self.threshold = threshold or self._calibrate_threshold()

    def _calibrate_threshold(self, fp_rate: float = 0.01) -> float:
        """Порог при заданном False Positive Rate"""
        # Прогоняем нормальные изображения из валидационного набора
        val_scores = self._predict_normal_samples()
        # Порог на уровне (1 - fp_rate) квантиля
        threshold = np.quantile(val_scores, 1 - fp_rate)
        return float(threshold)

    def predict(self, image: np.ndarray) -> dict:
        result = self.model(image)
        score = float(result.anomaly_score)

        return {
            'anomaly_score': score,
            'is_anomaly': score > self.threshold,
            'severity': self._classify_severity(score),
            'heatmap': result.anomaly_map,
            'threshold': self.threshold
        }

    def _classify_severity(self, score: float) -> str:
        if score < self.threshold:
            return 'normal'
        elif score < self.threshold * 1.5:
            return 'minor'
        elif score < self.threshold * 2.5:
            return 'moderate'
        return 'severe'

Локализация аномалий

Тепловая карта (anomaly map) позволяет локализовать дефект на изображении:

import matplotlib.pyplot as plt
import matplotlib.cm as cm

def visualize_anomaly(original: np.ndarray,
                       anomaly_map: np.ndarray,
                       threshold: float) -> np.ndarray:
    # Нормализация тепловой карты
    heatmap_norm = (anomaly_map - anomaly_map.min()) / \
                   (anomaly_map.max() - anomaly_map.min() + 1e-8)

    # Colormap overlay
    heatmap_colored = cv2.applyColorMap(
        (heatmap_norm * 255).astype(np.uint8),
        cv2.COLORMAP_JET
    )

    # Маска аномальных регионов
    _, mask = cv2.threshold(
        (heatmap_norm * 255).astype(np.uint8),
        int(threshold * 255), 255, cv2.THRESH_BINARY
    )
    contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL,
                                    cv2.CHAIN_APPROX_SIMPLE)

    # Overlay на оригинал
    overlay = cv2.addWeighted(original, 0.6, heatmap_colored, 0.4, 0)
    cv2.drawContours(overlay, contours, -1, (0, 255, 0), 2)

    return overlay

Работа с новыми категориями продуктов

При добавлении нового продукта: собираем 50–200 нормальных изображений → инициализация нового PatchCore за 5–10 минут без GPU. Нет необходимости в переобучении с нуля.