feat(edge): Event model + severity mapping (§7.3 payload)

edge/events.py

@@ -0,0 +1,64 @@
+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,
+        }

tests/test_events.py

@@ -0,0 +1,22 @@
+from edge.events import EventType, Severity, Event, severity_for
+
+
+def test_severity_mapping():
+    assert severity_for(EventType.NO_MOTION) == Severity.CRITICAL
+    assert severity_for(EventType.ZONE_RESTRICTED) == Severity.HIGH
+    assert severity_for(EventType.FALL) == Severity.HIGH
+    assert severity_for(EventType.IMMOBILITY) == Severity.WARNING
+    assert severity_for(EventType.ZONE_WARNING) == Severity.WARNING
+
+
+def test_event_to_payload_shape():
+    e = Event(EventType.FALL, Severity.HIGH, camera_id="cam-01", track_id=7,
+              zone_id=None, occurred_at=1000.0, duration_s=3.2,
+              metrics={"torso_angle": 78.0}, spatial=None)
+    p = e.to_payload(device_id="pi-01")
+    assert p["device_id"] == "pi-01"
+    assert p["event_type"] == "fall"
+    assert p["severity"] == "high"
+    assert p["track_id"] == 7
+    assert p["metrics"]["torso_angle"] == 78.0
+    assert "snapshots" not in p  # snapshots는 Alerter가 붙임