import uuid as _uuid from dataclasses import dataclass, field 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) 유령사고 등 공간 기준 # Stable idempotency key: generated ONCE at event creation and carried unchanged through the # offline queue and every retry, so a POST that succeeded server-side but timed out # client-side is deduplicated by the server (see server IngestController) instead of stored twice. event_uuid: str = field(default_factory=lambda: _uuid.uuid4().hex) def to_payload(self, device_id: str) -> dict: return { "device_id": device_id, "event_uuid": self.event_uuid, "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, }