feat(ergo): posture-load exposure scoring (REBA bands) with anonymous reports

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).

edge/app.py

@@ -8,6 +8,7 @@ from edge.detectors.zone import ZoneIntrusionDetector
 from edge.engine import EventEngine
 from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
 from edge.models import BBox
+from edge.events import EventType
 
 
 def _head_boxes(persons: list) -> list:
@@ -64,9 +65,13 @@ def run_once(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_f
     for tr in lost:
         pool.forget(tr.track_id)
     if events:
-        snaps = snapshotter.capture([frame], [_head_boxes(persons)])
+        # ERGO_REPORT is an anonymous aggregate (posture-exposure sums) — attaching a frame
+        # would turn a statistic into surveillance imagery. Capture only when a real incident
+        # needs evidence, and never attach snapshots to the ergo report itself.
+        evidential = [e for e in events if e.event_type != EventType.ERGO_REPORT]
+        snaps = snapshotter.capture([frame], [_head_boxes(persons)]) if evidential else []
         for e in events:
-            sink.try_send(e, snaps)
+            sink.try_send(e, [] if e.event_type == EventType.ERGO_REPORT else snaps)
     return events
 
 

edge/detectors/ergonomics.py

@@ -0,0 +1,70 @@
+from dataclasses import dataclass
+
+from edge.tracking import Track
+from edge.geometry import torso_angle_deg, arm_elevation_deg, knee_flexion_deg, core_visible
+
+
+@dataclass
+class ErgoConfig:
+    """자세 부담(인체공학) 노출 집계. REBA의 밴드 개념을 2D 단안 관절에 정직하게 축소한
+    것 — 정밀 REBA 점수가 아니라 '고위험 자세에 얼마나 오래 노출됐나'를 잰다."""
+    report_interval_s: float = 3600.0   # 카메라당 리포트 주기(노출이 있었던 경우만)
+    min_conf: float = 0.5
+    trunk_mod_deg: float = 20.0         # REBA trunk 밴드: 20~60° 중위험
+    trunk_high_deg: float = 60.0        # 60°+ 고위험
+    arm_high_deg: float = 90.0          # 어깨 위 거상
+    knee_flex_deg: float = 110.0        # 이보다 작은 내부각 = 깊은 웅크림
+    max_frame_gap_s: float = 2.0        # 검출 공백은 노출로 세지 않는다 (dt 상한)
+
+
+class ErgoMonitor:
+    """트랙별 프레임 dt로 자세 밴드 노출 시간을 누적하고, 주기마다 카메라 단위의 익명
+    합계를 리포트로 내보낸다. 개인 식별 없음: 트랙·좌표는 합산 과정에서 소거된다.
+    사고 상태(FALLING/FALLEN/NO_MOTION)의 트랙은 엔진이 걸러 넘긴다 — 쓰러진 사람의
+    90° 몸통은 인체공학 노출이 아니라 사고다."""
+
+    def __init__(self, cfg: ErgoConfig):
+        self.cfg = cfg
+        self._period_start: float | None = None
+        self._acc = self._zero()
+
+    @staticmethod
+    def _zero() -> dict:
+        return {"observed_s": 0.0, "trunk_mod_s": 0.0, "trunk_high_s": 0.0,
+                "arm_high_s": 0.0, "squat_s": 0.0}
+
+    def update(self, track: Track, now: float) -> None:
+        if self._period_start is None:
+            self._period_start = now
+        if len(track.frames) < 2:
+            return
+        dt = track.latest.t - track.frames[-2].t
+        if dt <= 0:
+            return
+        dt = min(dt, self.cfg.max_frame_gap_s)
+        kps = track.latest.keypoints
+        # 몸통 각도는 어깨·엉덩이 4점이 신뢰될 때만 — 저신뢰 프레임의 각도는 노이즈다.
+        if not core_visible(kps, self.cfg.min_conf):
+            return
+        self._acc["observed_s"] += dt
+        trunk = torso_angle_deg(kps)
+        if trunk >= self.cfg.trunk_high_deg:
+            self._acc["trunk_high_s"] += dt
+        elif trunk >= self.cfg.trunk_mod_deg:
+            self._acc["trunk_mod_s"] += dt
+        arm = arm_elevation_deg(kps, self.cfg.min_conf)
+        if arm is not None and arm >= self.cfg.arm_high_deg:
+            self._acc["arm_high_s"] += dt
+        knee = knee_flexion_deg(kps, self.cfg.min_conf)
+        if knee is not None and knee < self.cfg.knee_flex_deg:
+            self._acc["squat_s"] += dt
+
+    def report(self, now: float) -> dict | None:
+        """주기가 찼고 실제 노출 관측이 있었으면 합계를 내보내고 다음 주기를 시작한다."""
+        if self._period_start is None or now - self._period_start < self.cfg.report_interval_s:
+            return None
+        acc, self._acc = self._acc, self._zero()
+        self._period_start = now
+        if acc["observed_s"] <= 0:
+            return None   # 빈 주기: 사람이 없던 시간의 리포트는 소음이다
+        return {k: round(v, 1) for k, v in acc.items()}

edge/engine.py

@@ -7,6 +7,7 @@ from edge.detectors.immobility import ImmobilityWatchdog
 from edge.detectors.ghost import GhostIncidentDetector
 from edge.detectors.zone import ZoneIntrusionDetector
 from edge.detectors.occupancy import OccupancyWatchdog, OccupancyConfig
+from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
 
 
 @dataclass
@@ -23,13 +24,15 @@ _OCCUPANCY_TYPES = {"lone_worker": EventType.LONE_WORKER,
 class EventEngine:
     def __init__(self, fall: FallDetector, immobility: ImmobilityWatchdog,
                  ghost: GhostIncidentDetector, zone: ZoneIntrusionDetector,
-                 cfg: EngineConfig, occupancy: OccupancyWatchdog | None = None):
+                 cfg: EngineConfig, occupancy: OccupancyWatchdog | None = None,
+                 ergonomics: ErgoMonitor | None = None):
         self.fall = fall
         self.immobility = immobility
         self.ghost = ghost
         self.zone = zone
         # Keyword-with-default so the existing 5-arg constructors (runtime, tests) keep working.
         self.occupancy = occupancy if occupancy is not None else OccupancyWatchdog(OccupancyConfig())
+        self.ergonomics = ergonomics if ergonomics is not None else ErgoMonitor(ErgoConfig())
         self.cfg = cfg
         self._last_emit: dict[tuple, float] = {}
 
@@ -80,6 +83,10 @@ class EventEngine:
             sig, meta = self.fall.update(tr, now)
             state = self.fall._states.get(tr.track_id)
             self.ghost.note_state(tr.track_id, state.name if state else "STANDING")
+            # 자세 부담 노출은 사고 상태가 아닌 트랙만 — 쓰러진 사람의 90° 몸통은
+            # 인체공학 노출이 아니라 이미 사고로 처리 중인 상태다.
+            if state is None or state.name == "STANDING":
+                self.ergonomics.update(tr, now)
             if sig == "fall" and self._allow((EventType.FALL, tr.track_id), now):
                 events.append(self._mk(EventType.FALL, occurred_at, tr.track_id,
                                        metrics=meta, spatial=foot))
@@ -123,6 +130,12 @@ class EventEngine:
                 events.append(self._mk(etype, occurred_at, track_id=o["track_id"],
                                        zone_id=o["zone_id"], metrics=o["metrics"]))
 
+        # 자세 부담 주기 리포트 — 간격이 자체 게이트라 쿨다운(_allow)을 타지 않는다.
+        # 카메라 단위 익명 합계: track_id·spatial 없음 (개인 좌표를 남기지 않는다).
+        ergo = self.ergonomics.report(now)
+        if ergo is not None:
+            events.append(self._mk(EventType.ERGO_REPORT, occurred_at, metrics=ergo))
+
         for g in self.ghost.update(active_tracks, now):
             # No _allow cooldown for GHOST: ghost.update pops each incident from its candidates as it
             # emits, and track ids are never reused, so a given track fires GHOST at most once ever.

edge/events.py

@@ -17,6 +17,7 @@ class EventType(str, Enum):
     OVERCAPACITY = "overcapacity"      # 존 정원 초과 지속
     OVERSTAY = "overstay"              # 존 체류 상한 초과 (트랙별)
     NEAR_FALL = "near_fall"            # 니어미스: FALLING까지 갔다가 확정 전 회복
+    ERGO_REPORT = "ergo_report"        # 자세 부담 노출 주기 리포트 (카메라 단위, 익명)
 
 
 class Severity(str, Enum):
@@ -42,6 +43,7 @@ _SEVERITY = {
     EventType.OVERSTAY: Severity.WARNING,
     # 니어미스는 알람이 아니라 히트맵/통계 재료 — 소음이 되면 안 되므로 INFO.
     EventType.NEAR_FALL: Severity.INFO,
+    EventType.ERGO_REPORT: Severity.INFO,
 }
 
 

edge/geometry.py

@@ -1,7 +1,8 @@
 import math
 
 from edge.models import (
-    Keypoint, BBox, NOSE, L_SHOULDER, R_SHOULDER, L_HIP, R_HIP, CORE_JOINTS,
+    Keypoint, BBox, NOSE, L_SHOULDER, R_SHOULDER, L_ELBOW, R_ELBOW,
+    L_HIP, R_HIP, L_KNEE, R_KNEE, L_ANKLE, R_ANKLE, CORE_JOINTS,
 )
 
 
@@ -45,6 +46,44 @@ def head_below_hips(keypoints: list[Keypoint], nose_thr: float = 0.2) -> bool:
     return nose.y > hy
 
 
+def arm_elevation_deg(keypoints: list[Keypoint], min_conf: float = 0.5) -> float | None:
+    """어깨→팔꿈치 벡터가 '아래 수직'과 이루는 각의 좌우 최대값. 0=팔 내림, 90=수평 거상,
+    180=머리 위. 신뢰 가능한 쪽 팔만 사용; 양쪽 다 불가하면 None (측정 불가 ≠ 0°)."""
+    best = None
+    for s_i, e_i in ((L_SHOULDER, L_ELBOW), (R_SHOULDER, R_ELBOW)):
+        s, e = keypoints[s_i], keypoints[e_i]
+        if s.conf < min_conf or e.conf < min_conf:
+            continue
+        vx, vy = e.x - s.x, e.y - s.y
+        if not (vx or vy):
+            continue
+        # 이미지 y축은 아래로 증가: (0, +1)이 '팔 내림' 기준축.
+        ang = math.degrees(math.atan2(abs(vx), vy)) if vy >= 0 else \
+            180.0 - math.degrees(math.atan2(abs(vx), -vy))
+        best = ang if best is None else max(best, ang)
+    return best
+
+
+def knee_flexion_deg(keypoints: list[Keypoint], min_conf: float = 0.5) -> float | None:
+    """무릎 관절의 내부 각(엉덩이-무릎-발목)의 좌우 최소값. 180=곧게 섬, 작을수록 깊은
+    굴곡(웅크림). 신뢰 가능한 쪽 다리만 사용; 양쪽 다 불가하면 None."""
+    best = None
+    for h_i, k_i, a_i in ((L_HIP, L_KNEE, L_ANKLE), (R_HIP, R_KNEE, R_ANKLE)):
+        h, k, a = keypoints[h_i], keypoints[k_i], keypoints[a_i]
+        if h.conf < min_conf or k.conf < min_conf or a.conf < min_conf:
+            continue
+        v1 = (h.x - k.x, h.y - k.y)
+        v2 = (a.x - k.x, a.y - k.y)
+        n1 = math.hypot(*v1)
+        n2 = math.hypot(*v2)
+        if not n1 or not n2:
+            continue
+        cos = max(-1.0, min(1.0, (v1[0] * v2[0] + v1[1] * v2[1]) / (n1 * n2)))
+        ang = math.degrees(math.acos(cos))
+        best = ang if best is None else min(best, ang)
+    return best
+
+
 def visible_count(keypoints: list[Keypoint], thr: float = 0.2) -> int:
     return sum(1 for k in keypoints if k.conf >= thr)
 

server/migrations/2026-08-06-heatmap-spatial.sql

@@ -1,6 +1,7 @@
--- Heatmap raw material: normalized (0..1) spatial point per event, and the four new
--- event types (occupancy rules + near-miss). The chk_events_type constraint would REJECT
--- lone_worker/overcapacity/overstay/near_fall inserts on MySQL 8.0.16+ without this.
+-- Heatmap raw material: normalized (0..1) spatial point per event, and the five new
+-- event types (occupancy rules + near-miss + ergo report). The chk_events_type constraint
+-- would REJECT the new types' inserts on MySQL 8.0.16+ without this. (Never applied to any
+-- deployment before this revision, so the type list is edited in place.)
 ALTER TABLE move_events
     ADD COLUMN spatial_x DOUBLE NULL AFTER metrics_json,
     ADD COLUMN spatial_y DOUBLE NULL AFTER spatial_x;
@@ -9,5 +10,5 @@ ALTER TABLE move_events DROP CHECK chk_events_type;
 ALTER TABLE move_events ADD CONSTRAINT chk_events_type CHECK (event_type IN (
     'fall','no_motion','immobility','ghost','zone_warning',
     'zone_restricted','pinch','monitoring_degraded','device_fault',
-    'lone_worker','overcapacity','overstay','near_fall'
+    'lone_worker','overcapacity','overstay','near_fall','ergo_report'
 ));

server/public/assets/app.js

@@ -22,7 +22,7 @@
     ghost: '위험 상태 소실', zone_warning: '경고구역 진입', zone_restricted: '제한구역 침입',
     pinch: '협착구역 침입', monitoring_degraded: '감시 성능 저하', device_fault: '장치 이상',
     lone_worker: '위험구역 단독 작업', overcapacity: '정원 초과', overstay: '체류 상한 초과',
-    near_fall: '넘어질 뻔함 (니어미스)'
+    near_fall: '넘어질 뻔함 (니어미스)', ergo_report: '작업자세 부담 리포트'
   };
 
   var SEVERITY_LABELS = {

server/public/assets/detail.js

@@ -28,9 +28,29 @@
     ghost: '위험 상태 소실', zone_warning: '경고구역 진입', zone_restricted: '제한구역 침입',
     pinch: '협착구역 침입', monitoring_degraded: '감시 성능 저하', device_fault: '장치 이상',
     lone_worker: '위험구역 단독 작업', overcapacity: '정원 초과', overstay: '체류 상한 초과',
-    near_fall: '넘어질 뻔함 (니어미스)'
+    near_fall: '넘어질 뻔함 (니어미스)', ergo_report: '작업자세 부담 리포트'
   };
 
+  // Metric keys -> Korean, so the detail view can print what the edge measured (the ergo
+  // report's whole value IS its metrics). Unknown keys fall through as the raw key.
+  var METRIC_LABELS = {
+    observed_s: '관측 시간', trunk_high_s: '허리 고위험(60°+) 노출',
+    trunk_mod_s: '허리 중위험(20~60°) 노출', arm_high_s: '팔 거상(90°+) 노출',
+    squat_s: '웅크림 노출', prone_s: '엎드림 지속', immobility_s: '부동 지속',
+    waited_s: '재출현 대기', dwell_s: '체류 시간', limit_s: '체류 상한',
+    alone_s: '단독 작업 지속', workers: '감지 인원', required: '최소 인원', limit: '정원',
+    torso_angle: '몸통 각도(°)', aspect_ratio: '박스 종횡비', height_remaining: '잔여 높이 비율',
+    last_state: '소실 시 상태'
+  };
+
+  function metricValue(key, v) {
+    if (/_s$/.test(key) && typeof v === 'number') {
+      if (v >= 90) return Math.round(v / 60) + '분 ' + Math.round(v % 60) + '초';
+      return v + '초';
+    }
+    return String(v);
+  }
+
   var meta = document.getElementById('event-detail-meta');
   var snapBox = document.getElementById('event-detail-snapshots');
   if (!meta) return;
@@ -100,6 +120,12 @@
     addRow('확인', textNode(ev.acknowledged_by || '미확인'));
     if (ev.false_positive_by) addRow('오탐', textNode(ev.false_positive_by + ' 태그'));
 
+    // 측정 지표: 엣지가 이벤트에 실은 수치 전부. ergo_report는 이 블록이 본문이다.
+    var metrics = ev.metrics || {};
+    Object.keys(metrics).forEach(function (key) {
+      addRow(METRIC_LABELS[key] || key, textNode(metricValue(key, metrics[key])));
+    });
+
     if (!snapBox) return;
     snapBox.textContent = '';
     var snaps = data.snapshots || [];

server/schema.sql

@@ -146,7 +146,7 @@ CREATE TABLE move_events (
     CONSTRAINT chk_events_type CHECK (event_type IN (
         'fall','no_motion','immobility','ghost','zone_warning',
         'zone_restricted','pinch','monitoring_degraded','device_fault',
-        'lone_worker','overcapacity','overstay','near_fall'
+        'lone_worker','overcapacity','overstay','near_fall','ergo_report'
     )),
     CONSTRAINT chk_events_severity CHECK (severity IN ('critical','high','warning','info'))
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

server/src/Ingest/EventValidator.php

@@ -6,7 +6,7 @@ final class EventValidator
 {
     private const TYPES = ['fall','no_motion','immobility','ghost','zone_warning',
         'zone_restricted','pinch','monitoring_degraded','device_fault',
-        'lone_worker','overcapacity','overstay','near_fall'];
+        'lone_worker','overcapacity','overstay','near_fall','ergo_report'];
     private const SEVERITIES = ['critical','high','warning','info'];
     private const REQUIRED = ['device_id','camera_id','event_type','severity','occurred_at'];
 

server/tests/EventValidatorTest.php

@@ -49,7 +49,7 @@ final class EventValidatorTest extends TestCase
         $this->assertFalse(EventValidator::validate($p)['ok']);
     }
     public function testAcceptsOccupancyRuleEventTypes(): void {
-        foreach (['lone_worker', 'overcapacity', 'overstay', 'near_fall'] as $type) {
+        foreach (['lone_worker', 'overcapacity', 'overstay', 'near_fall', 'ergo_report'] as $type) {
             $p = $this->valid(); $p['event_type'] = $type;
             $this->assertTrue(EventValidator::validate($p)['ok'], "$type must be a known type");
         }

tests/test_app.py

@@ -120,6 +120,37 @@ def test_dropout_then_redetection_still_confirms_fall():
     assert any(e.event_type == EventType.FALL for e in sink.events)
 
 
+def test_ergo_report_carries_no_snapshots():
+    # ERGO_REPORT는 익명 통계다 — 프레임이 붙는 순간 통계가 감시 이미지가 된다.
+    from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
+    from edge.detectors.fall import FallDetector, FallConfig
+
+    class _SnapSink:
+        def __init__(self): self.sent = []
+        def try_send(self, event, snapshots):
+            self.sent.append((event, snapshots)); return True
+
+    src = _Source()
+    pose = _ScriptedPose([[_person_at(500)], [_person_at(500)]])
+    tracker = IouTracker()
+    pool = TrackSmootherPool()
+    engine = EventEngine(
+        FallDetector(FallConfig(prone_confirm_s=999)),
+        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
+        GhostIncidentDetector(GhostConfig()),
+        ZoneIntrusionDetector([], ZoneConfig()),
+        EngineConfig(camera_id="cam-01"),
+        ergonomics=ErgoMonitor(ErgoConfig(report_interval_s=0.5)))
+    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _SnapSink()
+    times = iter([0.0, 0.6])
+    def now_fn(): return next(times)
+    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
+    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
+    ergo = [(e, s) for e, s in sink.sent if e.event_type == EventType.ERGO_REPORT]
+    assert len(ergo) == 1
+    assert ergo[0][1] == []   # 스냅샷 미부착
+
+
 def _lying_person_at(x1, x2):
     kps = [Keypoint((x1 + x2) / 2, 15, 0.9) for _ in range(17)]
     return Person(BBox(x1, 0, x2, 20), kps)

tests/test_engine.py

@@ -261,6 +261,43 @@ def test_ghost_spatial_is_normalized_with_frame_size():
     assert 0.0 <= x <= 1.0 and 0.0 <= y <= 1.0                 # not raw pixels (150, 305)
 
 
+def test_ergo_report_emitted_anonymous_and_info():
+    from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
+    eng = EventEngine(
+        FallDetector(FallConfig(prone_confirm_s=999)),
+        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
+        GhostIncidentDetector(GhostConfig()),
+        ZoneIntrusionDetector([], ZoneConfig()),
+        EngineConfig(camera_id="cam-01"),
+        ergonomics=ErgoMonitor(ErgoConfig(report_interval_s=0.5)))
+    tr = Track(1, deque(maxlen=90))
+    _standing_track(tr, 0.0); eng.process([tr], [], now=0.0)
+    _standing_track(tr, 0.6)
+    events = eng.process([tr], [], now=0.6)
+    rep = [e for e in events if e.event_type == EventType.ERGO_REPORT]
+    assert len(rep) == 1
+    assert rep[0].severity == Severity.INFO
+    assert rep[0].track_id is None and rep[0].spatial is None   # 익명 합계: 개인 좌표 없음
+    assert rep[0].metrics["observed_s"] == 0.6
+
+
+def test_incident_state_track_does_not_accrue_ergo_exposure():
+    # 쓰러진 사람의 90° 몸통은 인체공학 노출이 아니라 사고다 — FALLING 이후는 집계 제외.
+    from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
+    eng = EventEngine(
+        FallDetector(FallConfig(prone_confirm_s=999, falling_confirm_s=0.05)),
+        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
+        GhostIncidentDetector(GhostConfig()),
+        ZoneIntrusionDetector([], ZoneConfig()),
+        EngineConfig(camera_id="cam-01"),
+        ergonomics=ErgoMonitor(ErgoConfig(report_interval_s=0.2)))
+    tr = Track(1, deque(maxlen=90))
+    for t in (0.0, 0.1, 0.2, 0.3):
+        _lying_track(tr, t)
+        events = eng.process([tr], [], now=t)
+        assert not any(e.event_type == EventType.ERGO_REPORT for e in events)
+
+
 def test_allow_cooldown_primitive():
     eng = _engine()  # EngineConfig(camera_id="cam-01", cooldown_s=30.0)
     key = ("FALL", 1)

tests/test_ergonomics.py

@@ -0,0 +1,94 @@
+from collections import deque
+
+from edge.models import Keypoint, BBox
+from edge.tracking import Track, TrackFrame
+from edge.geometry import arm_elevation_deg, knee_flexion_deg
+from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
+
+
+def _kps(overrides=None, conf=0.9):
+    """직립 자세 기본형: 몸통 수직, 팔 내림, 다리 곧게. overrides로 관절만 바꾼다."""
+    base = {
+        0: (50, 40),                       # nose
+        5: (40, 100), 6: (60, 100),        # shoulders
+        7: (40, 140), 8: (60, 140),        # elbows (아래로 = 팔 내림)
+        11: (40, 200), 12: (60, 200),      # hips
+        13: (40, 260), 14: (60, 260),      # knees
+        15: (40, 320), 16: (60, 320),      # ankles
+    }
+    if overrides:
+        base.update(overrides)
+    kps = []
+    for i in range(17):
+        x, y = base.get(i, (50, 150))
+        kps.append(Keypoint(x, y, conf))
+    return kps
+
+
+def _feed(mon, track, kps, t):
+    track.frames.append(TrackFrame(t, BBox(20, 30, 80, 330), kps))
+    mon.update(track, now=t)
+
+
+def test_arm_elevation_down_vs_raised():
+    assert arm_elevation_deg(_kps()) < 20.0                                  # 팔 내림
+    raised = _kps({7: (40, 60), 8: (60, 60)})                                # 팔꿈치가 어깨 위
+    assert arm_elevation_deg(raised) > 90.0
+    low_conf = [Keypoint(0, 0, 0.1) for _ in range(17)]
+    assert arm_elevation_deg(low_conf) is None                               # 측정 불가 ≠ 0°
+
+
+def test_knee_flexion_straight_vs_squat():
+    assert knee_flexion_deg(_kps()) > 160.0                                  # 곧게 섬
+    squat = _kps({13: (70, 220), 14: (90, 220), 15: (40, 230), 16: (60, 230)})
+    assert knee_flexion_deg(squat) < 110.0
+
+
+def test_upright_posture_accrues_no_risk_exposure():
+    mon = ErgoMonitor(ErgoConfig(report_interval_s=10.0))
+    tr = Track(1, deque(maxlen=90))
+    for i in range(11):
+        _feed(mon, tr, _kps(), t=float(i))
+    rep = mon.report(now=11.0)
+    assert rep is not None
+    assert rep["observed_s"] == 10.0
+    assert rep["trunk_high_s"] == 0.0 and rep["trunk_mod_s"] == 0.0
+    assert rep["arm_high_s"] == 0.0 and rep["squat_s"] == 0.0
+
+
+def test_bent_trunk_accrues_high_band():
+    mon = ErgoMonitor(ErgoConfig(report_interval_s=10.0))
+    tr = Track(1, deque(maxlen=90))
+    # 어깨가 엉덩이 높이 근처까지 숙임: 몸통이 수평에 가깝다 (>60°)
+    bent = _kps({5: (140, 190), 6: (160, 190)})
+    for i in range(11):
+        _feed(mon, tr, bent, t=float(i))
+    rep = mon.report(now=11.0)
+    assert rep["trunk_high_s"] == 10.0
+
+
+def test_detection_gap_does_not_count_as_exposure():
+    mon = ErgoMonitor(ErgoConfig(report_interval_s=100.0, max_frame_gap_s=2.0))
+    tr = Track(1, deque(maxlen=90))
+    _feed(mon, tr, _kps(), t=0.0)
+    _feed(mon, tr, _kps(), t=60.0)   # 60초 공백 — 2초 상한으로만 집계
+    mon._period_start = -100.0       # 주기 강제 만료
+    rep = mon.report(now=61.0)
+    assert rep["observed_s"] == 2.0
+
+
+def test_empty_period_reports_nothing():
+    mon = ErgoMonitor(ErgoConfig(report_interval_s=5.0))
+    tr = Track(1, deque(maxlen=90))
+    _feed(mon, tr, [Keypoint(0, 0, 0.1) for _ in range(17)], t=0.0)   # 저신뢰만
+    _feed(mon, tr, [Keypoint(0, 0, 0.1) for _ in range(17)], t=1.0)
+    assert mon.report(now=6.0) is None                                 # 관측 0 → 리포트 없음
+    assert mon.report(now=7.0) is None                                 # 주기는 리셋됨
+
+
+def test_low_confidence_core_is_not_measured():
+    mon = ErgoMonitor(ErgoConfig(report_interval_s=5.0))
+    tr = Track(1, deque(maxlen=90))
+    _feed(mon, tr, _kps(conf=0.2), t=0.0)
+    _feed(mon, tr, _kps(conf=0.2), t=1.0)
+    assert mon._acc["observed_s"] == 0.0