Реализация AI-детекции аномалий на видео в мобильном приложении
Детекция аномалий — задача без чёткой разметки «плохих» событий. Нельзя заранее перечислить все возможные аномалии, особенно в системах безопасности или мониторинга производства. Отсюда — архитектурный выбор: unsupervised или semi-supervised подход, в отличие от классической детекции объектов.
Что считать аномалией: постановка задачи
До написания кода — точная постановка с заказчиком:
- Пространственная аномалия: объект в зоне, где его не должно быть (человек в серверной комнате)
- Поведенческая аномалия: нормальный объект ведёт себя необычно (человек бежит там, где все ходят; машина едет в обратную сторону)
- Временная аномалия: событие происходит не в то время (движение в нерабочие часы)
- Техническая аномалия: оборудование работает нештатно (вибрация, дым, искры)
Каждый тип требует разной архитектуры.
Архитектура: двухуровневый подход
// iOS: комбинированный детектор аномалий
class VideoAnomalyDetector {
// Уровень 1: детерминированные правила (быстро, дёшево)
private let rulesEngine: AnomalyRulesEngine
// Уровень 2: AI для неочевидных аномалий (только при прохождении rule-фильтра)
private let aiDetector: MLAnomalyDetector
func analyze(frame: CVPixelBuffer, timestamp: Date) async -> AnomalyResult {
// Правила первыми — они быстрее и точнее для известных сценариев
let ruleViolations = rulesEngine.check(frame: frame, timestamp: timestamp)
if !ruleViolations.isEmpty {
return AnomalyResult(detected: true,
type: .ruleViolation,
violations: ruleViolations,
confidence: 1.0) // детерминировано
}
// AI — для неизвестных паттернов
return await aiDetector.detect(frame: frame, timestamp: timestamp)
}
}
Детерминированные правила
class AnomalyRulesEngine {
struct RestrictedZone {
let polygon: [CGPoint] // нормализованные координаты
let schedule: WorkSchedule? // nil = всегда ограничена
let name: String
}
private let restrictedZones: [RestrictedZone]
private let personDetector: VNCoreMLModel // лёгкий YOLOv8n
func check(frame: CVPixelBuffer, timestamp: Date) -> [RuleViolation] {
let persons = detectPersons(frame)
var violations: [RuleViolation] = []
for person in persons {
let personCenter = person.boundingBox.center
for zone in restrictedZones {
// Нельзя входить в зону вне расписания
if zone.polygon.contains(personCenter) {
if let schedule = zone.schedule, !schedule.isActive(at: timestamp) {
violations.append(RuleViolation(
type: .unauthorizedZoneAccess,
zone: zone.name,
timestamp: timestamp
))
} else if zone.schedule == nil {
violations.append(RuleViolation(type: .restrictedZone, zone: zone.name))
}
}
}
}
return violations
}
}
AI-детекция: autoencoder для поведенческих аномалий
Для поведенческих аномалий — подход на основе автоэнкодера: обучаем на нормальном поведении, аномалия = высокая reconstruction error.
# Обучение autoencoder на нормальных видео-фрагментах
import torch
import torch.nn as nn
class VideoAnomalyAutoencoder(nn.Module):
"""
Входной тензор: [batch, frames, height, width, channels]
Обучается только на НОРМАЛЬНЫХ сценах
Аномалия: reconstruction_error > threshold
"""
def __init__(self, input_shape=(16, 64, 64, 3)):
super().__init__()
self.encoder = nn.Sequential(
nn.Conv3d(3, 32, kernel_size=(3,3,3), padding=1),
nn.ReLU(),
nn.MaxPool3d((1,2,2)),
nn.Conv3d(32, 64, kernel_size=(3,3,3), padding=1),
nn.ReLU(),
nn.MaxPool3d((2,2,2)),
)
self.decoder = nn.Sequential(
nn.ConvTranspose3d(64, 32, kernel_size=(3,3,3),
stride=(2,2,2), padding=1, output_padding=1),
nn.ReLU(),
nn.ConvTranspose3d(32, 3, kernel_size=(3,3,3),
stride=(1,2,2), padding=1, output_padding=(0,1,1)),
nn.Sigmoid()
)
def forward(self, x):
z = self.encoder(x)
return self.decoder(z)
def anomaly_score(self, x):
reconstructed = self(x)
# MSE по пространству и времени
return ((x - reconstructed) ** 2).mean(dim=[1,2,3,4])
Порог аномального score определяется на валидационной выборке нормальных сцен (99-й перцентиль reconstruction error → граница нормы).
На мобильном устройстве этот autoencoder конвертируется в CoreML / TFLite. Размер существенно меньше YOLOv8: 5–15 МБ.
Мобильный инференс: обработка скользящего окна
// iOS: анализ видеопотока скользящим окном из 16 кадров
class SlidingWindowAnalyzer {
private var frameBuffer: CircularBuffer<CVPixelBuffer> = CircularBuffer(capacity: 16)
private var frameCounter = 0
private let stepSize = 8 // новое окно каждые 8 кадров (50% overlap)
func addFrame(_ frame: CVPixelBuffer) async -> AnomalyScore? {
frameBuffer.append(frame)
frameCounter += 1
// Анализ каждые stepSize кадров
guard frameCounter % stepSize == 0,
frameBuffer.count == 16 else { return nil }
return try? await computeAnomalyScore(frames: Array(frameBuffer))
}
private func computeAnomalyScore(frames: [CVPixelBuffer]) async throws -> AnomalyScore {
let tensor = prepareTensor(frames) // [1, 16, 64, 64, 3]
let output = try autoencoderModel.prediction(input: tensor)
let score = output.anomalyScore.floatValue
return AnomalyScore(
value: score,
isAnomaly: score > anomalyThreshold,
frameWindow: frames
)
}
}
Алерты и реакция
// Android: многоуровневая система алертов
sealed class AnomalyAlert {
data class Warning(val message: String, val score: Float) : AnomalyAlert()
data class Critical(val message: String, val violations: List<RuleViolation>) : AnomalyAlert()
}
class AlertManager(private val notificationManager: NotificationManager) {
private val cooldownMap = mutableMapOf<String, Long>()
private val alertCooldownMs = 30_000L // не спамить: не чаще раза в 30 сек
fun emit(alert: AnomalyAlert, alertKey: String) {
val lastAlertTime = cooldownMap[alertKey] ?: 0L
if (System.currentTimeMillis() - lastAlertTime < alertCooldownMs) return
cooldownMap[alertKey] = System.currentTimeMillis()
when (alert) {
is AnomalyAlert.Warning -> showLocalNotification(alert.message, priority = LOW)
is AnomalyAlert.Critical -> {
showLocalNotification(alert.message, priority = HIGH)
sendWebhook(alert) // интеграция с внешней системой безопасности
}
}
}
}
Ориентиры по срокам
Детекция зональных нарушений с детерминированными правилами (без AI autoencoder) — 1–2 недели. Полная система с autoencoder для поведенческих аномалий, скользящим окном, многоуровневыми алертами, интеграцией с системой безопасности/мониторинга и поддержкой iOS + Android — 2–4 недели плюс время на сбор нормальных данных и обучение модели.