Детекція аномалій у відео через AI для мобільних додатків
Детекція аномалій — це задача без чітких міток для «поганих» подій. Не можна заздалегідь перелічити всі можливі аномалії, особливо в системах безпеки або моніторингу виробництва. Це призводить до архітектурного вибору: неконтрольовані або напів-контрольовані підходи, на відміну від класичної детекції об'єктів.
Визначення аномалії: постановка задачі
Перед кодуванням встановіть чіткі вимоги зі стейкхолдерами:
- Просторова аномалія: об'єкт у зоні, де його не повинно бути (людина в серверній кімнаті)
- Поведінкова аномалія: нормальний об'єкт діє незвично (людина бігає, де всі ходять; машина їде в обернену сторону)
- Часова аномалія: подія відбувається у невдалий час (рух у робочі часи)
- Технічна аномалія: обладнання працює неправильно (вібрація, дим, іскри)
Кожен тип потребує різної архітектури.
Архітектура: двохрівневий підхід
// iOS: комбінований детектор аномалій
class VideoAnomalyDetector {
// Рівень 1: детерміновані правила (швидко, дешево)
private let rulesEngine: AnomalyRulesEngine
// Рівень 2: AI для неочевидних аномалій (лише якщо правила пройдуть)
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 для поведінкових аномалій
Для поведінкових аномалій використовуйте підхід autoencoder: навчіть на нормальній поведінці; аномалія = висока помилка реконструкції.
# Навчання 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])
Поріг визначається на валідаційній множині нормальних сцен (99-й перцентиль помилки реконструкції → межа норми).
На мобільному, цей 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% перекриття)
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 тижнів плюс час на збір нормальних даних і навчання моделі.