b358822247
Same pose stream, new safety axis: musculoskeletal-load prevention. This is an honest 2D reduction of REBA - band exposure TIME, not a precise REBA score. Edge: - geometry: arm_elevation_deg (shoulder->elbow vs vertical-down, best of both arms) and knee_flexion_deg (hip-knee-ankle interior angle, worst of both legs). Both return None when no confident side exists - "unmeasurable" must never read as "0 deg / straight". - detectors/ergonomics.py: per-frame dt accumulation into bands (trunk 20-60 and 60+ deg via the existing torso angle, arm 90+ deg, knee <110 deg), gated on confident core keypoints; detection gaps are dt-capped (2s) so a dropout never counts as exposure. Hourly per-camera report, skipped when nothing was observed (an empty-workshop report is noise). - Engine: tracks in an incident state (FALLING/FALLEN/NO_MOTION) are excluded - a fallen worker's 90-deg trunk is an incident, not an ergonomic exposure. The report event (ergo_report/INFO) carries no track_id and no spatial: a camera-level anonymous aggregate, per the PIPA-first product stance. - run_once: NO snapshots on ergo_report (a frame would turn a statistic into surveillance imagery); capture is skipped entirely when only ergo fired. Server/UI: ergo_report accepted (validator + chk_events_type, edited in the still-undeployed heatmap migration); detail view now renders event metrics with Korean labels and seconds humanized - the ergo report's whole value IS its metrics. Korean type label in list + detail. Verified: pytest 232 pass (10 new) / PHPUnit 135 pass. NOTE: the detail-view visual check is pending - the Chrome extension disconnected mid-verification; harness is staged under the session scratchpad (events/1/view).
87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
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" # 자세 부담 노출 주기 리포트 (카메라 단위, 익명)
|
|
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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
|