Video Analytics System for Security

System Development видеоаналитики для безопасности

Видеоаналитика для безопасности — это не запись видео для последующего ручного просмотра, а автоматический анализ потока в реальном времени с генерацией оповещений при значимых событиях. Разрыв между системой «камеры пишут на NVR» и «камеры анализируют и сигнализируют» — это переход к Active Security.

Architecture системы видеоаналитики

IP Cameras (RTSP)
    ↓
Stream Ingestion Layer (GStreamer/FFmpeg)
    ↓
AI Analytics Engine (GPU Cluster)
    ├── Object Detection (YOLOv8)
    ├── Object Tracking (ByteTrack)
    ├── Event Detection (Rules Engine)
    └── Behavior Analysis (ML Models)
    ↓
Event Processing (Kafka/Redis Streams)
    ↓
Alert & Response System
    ├── Security Dashboard
    ├── Mobile Notifications
    ├── Guard Dispatch
    └── Access Control Integration

from ultralytics import YOLO
import numpy as np
from dataclasses import dataclass, field

@dataclass
class SecurityEvent:
    event_type: str
    camera_id: str
    timestamp: str
    confidence: float
    bbox: list
    track_id: int | None = None
    metadata: dict = field(default_factory=dict)

class SecurityAnalyticsEngine:
    def __init__(self, camera_id: str, config: dict):
        self.camera_id = camera_id
        self.detector = YOLO(config['model_path'])
        self.tracker = config['tracker']
        self.event_rules = config['event_rules']
        self.zones = config['security_zones']

    def process_frame(self, frame: np.ndarray,
                       timestamp: str) -> list[SecurityEvent]:
        # Детекция и трекинг
        detections = self.detector(frame, conf=0.5, classes=[0])  # person only
        tracks = self.tracker.update(detections[0])

        events = []
        for track in tracks:
            # Проверка зон
            for zone_name, zone_polygon in self.zones.items():
                if self._in_zone(track.center, zone_polygon):
                    rule_result = self.event_rules.check(
                        zone_name, track, timestamp
                    )
                    if rule_result.triggered:
                        events.append(SecurityEvent(
                            event_type=rule_result.event_type,
                            camera_id=self.camera_id,
                            timestamp=timestamp,
                            confidence=rule_result.confidence,
                            bbox=track.bbox,
                            track_id=track.track_id,
                            metadata=rule_result.metadata
                        ))

        return events

Детектируемые события

Система зон безопасности

class SecurityZone:
    def __init__(self, name: str, polygon: list, rules: dict):
        self.name = name
        self.polygon = np.array(polygon, dtype=np.int32)
        self.rules = rules  # time-based rules, allowed persons, etc.

    def is_point_inside(self, point: tuple) -> bool:
        return cv2.pointPolygonTest(
            self.polygon,
            (float(point[0]), float(point[1])),
            False
        ) >= 0

    def is_allowed(self, track_id: int, timestamp: str) -> bool:
        """Проверка по расписанию и whitelist"""
        if not self.rules.get('time_restricted'):
            return True

        current_hour = int(timestamp.split(':')[0])
        allowed_hours = self.rules.get('allowed_hours', range(8, 18))

        if current_hour not in allowed_hours:
            return False

        if self.rules.get('whitelist_only'):
            return track_id in self.rules.get('whitelist', set())

        return True

Масштабирование: обработка 100+ камер

Одна A100 GPU обрабатывает ~32 потока 1080p@30fps с YOLOv8n. Для 100+ камер — GPU кластер с балансировкой нагрузки:

class CameraLoadBalancer:
    def __init__(self, gpu_workers: list):
        self.workers = gpu_workers
        self.camera_assignments = {}

    def assign_camera(self, camera_id: str) -> str:
        """Round-robin с учётом нагрузки"""
        least_loaded = min(self.workers, key=lambda w: w.load)
        self.camera_assignments[camera_id] = least_loaded
        return least_loaded.worker_id

Integration с физической охраной

• PSIM (Physical Security Information Management): централизованное управление всеми системами безопасности
• VMS (Video Management System): хранение видео с возможностью криминалистического поиска по атрибутам
• Access Control: автоматическая блокировка дверей при тревоге