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" LONE_WORKER = "lone_worker" # 2인 1조 존에 단독 작업 지속 OVERCAPACITY = "overcapacity" # 존 정원 초과 지속 OVERSTAY = "overstay" # 존 체류 상한 초과 (트랙별) 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, # 단독 작업은 규정 위반이자 사고 시 구조 지연 요인 — 정원/체류보다 한 단계 높게. EventType.LONE_WORKER: Severity.HIGH, EventType.OVERCAPACITY: Severity.WARNING, EventType.OVERSTAY: 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, }