move_detect/edge/events.py

64 lines
1.7 KiB
Python

from dataclasses import dataclass
from enum import Enum
class EventType(str, Enum):
FALL = "fall"
NO_MOTION = "no_motion"
IMMOBILITY = "immobility"
GHOST = "ghost"
ZONE_WARNING = "zone_warning"
ZONE_RESTRICTED = "zone_restricted"
PINCH = "pinch"
MONITORING_DEGRADED = "monitoring_degraded"
DEVICE_FAULT = "device_fault"
class Severity(str, Enum):
CRITICAL = "critical"
HIGH = "high"
WARNING = "warning"
INFO = "info"
_SEVERITY = {
EventType.NO_MOTION: Severity.CRITICAL,
EventType.GHOST: Severity.CRITICAL,
EventType.FALL: Severity.HIGH,
EventType.ZONE_RESTRICTED: Severity.HIGH,
EventType.PINCH: Severity.HIGH,
EventType.IMMOBILITY: Severity.WARNING,
EventType.ZONE_WARNING: Severity.WARNING,
EventType.MONITORING_DEGRADED: Severity.WARNING,
EventType.DEVICE_FAULT: Severity.WARNING,
}
def severity_for(event_type: EventType) -> Severity:
return _SEVERITY.get(event_type, Severity.INFO)
@dataclass(frozen=True)
class Event:
event_type: EventType
severity: Severity
camera_id: str
track_id: int | None
zone_id: str | None
occurred_at: float
duration_s: float
metrics: dict
spatial: tuple | None # (x, y) 유령사고 등 공간 기준
def to_payload(self, device_id: str) -> dict:
return {
"device_id": device_id,
"camera_id": self.camera_id,
"event_type": self.event_type.value,
"severity": self.severity.value,
"track_id": self.track_id,
"zone_id": self.zone_id,
"occurred_at": self.occurred_at,
"duration_s": round(self.duration_s, 2),
"metrics": self.metrics,
}