move_detect/edge/events.py

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"              # 존 체류 상한 초과 (트랙별)
    NEAR_FALL = "near_fall"            # 니어미스: FALLING까지 갔다가 확정 전 회복
    ERGO_REPORT = "ergo_report"        # 자세 부담 노출 주기 리포트 (카메라 단위, 익명)
    # --- 감시 자체의 건전성 ---------------------------------------------------------
    # 존 판정은 카메라 화각이 고정이라는 전제 위에 서 있다. 청소하다 삼각대를 건드리면
    # 제한구역이 엉뚱한 곳을 가리키는데, 시스템은 계속 "정상"이라고 보고한다. 이 조용한
    # 실패를 드러내는 것이 아래 세 유형이다.
    CAMERA_MOVED = "camera_moved"          # 배경 기준 프레임 대비 화각 이동
    CAMERA_OBSTRUCTED = "camera_obstructed"  # 렌즈 가림 (분산 급락)
    LOW_LIGHT = "low_light"                # 조도 저하로 검출력 하락
    # --- 모델 확보 후 활성화되는 자리 -----------------------------------------------
    # 감지기 배선과 이벤트 계약은 준비되어 있고, 추론 모델만 비어 있다. AGPL 아닌
    # 가중치를 확보하기 전까지 엣지는 이 유형을 발화하지 않는다 (edge/detectors/pending.py).
    PPE_MISSING = "ppe_missing"        # 보호구(안전모·안전대) 미착용
    PROXIMITY = "proximity"            # 중장비-사람 근접
    FIRE_SMOKE = "fire_smoke"          # 화재·연기


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,
    # 니어미스는 알람이 아니라 히트맵/통계 재료 — 소음이 되면 안 되므로 INFO.
    EventType.NEAR_FALL: Severity.INFO,
    EventType.ERGO_REPORT: Severity.INFO,
    # 화각이 틀어지면 그 카메라의 존 판정 전체가 무효다. 사람이 다친 사건은 아니지만
    # "지금부터 못 보고 있다"는 뜻이라 HIGH — 정비 대기열이 아니라 당일 처리 대상이다.
    EventType.CAMERA_MOVED: Severity.HIGH,
    EventType.CAMERA_OBSTRUCTED: Severity.HIGH,
    # 조도 저하는 검출력이 떨어졌다는 신호이지 실패는 아니다 — monitoring_degraded 와 같은 급.
    EventType.LOW_LIGHT: Severity.WARNING,
    EventType.PPE_MISSING: Severity.WARNING,
    EventType.PROXIMITY: Severity.HIGH,
    EventType.FIRE_SMOKE: Severity.CRITICAL,
}


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:
        payload = {
            "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,
        }
        # 정규화 발/기준 좌표(0..1) — 서버 히트맵의 원료. None이면 키 자체를 생략해
        # 구서버(스키마에 spatial 없음)와의 하위호환을 지킨다.
        if self.spatial is not None:
            payload["spatial"] = [round(float(self.spatial[0]), 4), round(float(self.spatial[1]), 4)]
        return payload