diff --git a/edge/app.py b/edge/app.py index b7c7f09..623957e 100644 --- a/edge/app.py +++ b/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 diff --git a/edge/detectors/ergonomics.py b/edge/detectors/ergonomics.py new file mode 100644 index 0000000..d28434b --- /dev/null +++ b/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()} diff --git a/edge/engine.py b/edge/engine.py index cb08d3b..92176ca 100644 --- a/edge/engine.py +++ b/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. diff --git a/edge/events.py b/edge/events.py index c58f9bb..d40c735 100644 --- a/edge/events.py +++ b/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, } diff --git a/edge/geometry.py b/edge/geometry.py index 7e4ba56..e48f83c 100644 --- a/edge/geometry.py +++ b/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) diff --git a/server/migrations/2026-08-06-heatmap-spatial.sql b/server/migrations/2026-08-06-heatmap-spatial.sql index 3323b2e..b916947 100644 --- a/server/migrations/2026-08-06-heatmap-spatial.sql +++ b/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' )); diff --git a/server/public/assets/app.js b/server/public/assets/app.js index 4434c3e..3496f15 100644 --- a/server/public/assets/app.js +++ b/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 = { diff --git a/server/public/assets/detail.js b/server/public/assets/detail.js index a3f853b..2cd9d12 100644 --- a/server/public/assets/detail.js +++ b/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 || []; diff --git a/server/schema.sql b/server/schema.sql index 3a4743a..cece199 100644 --- a/server/schema.sql +++ b/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; diff --git a/server/src/Ingest/EventValidator.php b/server/src/Ingest/EventValidator.php index a25a6fa..2aa3e90 100644 --- a/server/src/Ingest/EventValidator.php +++ b/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']; diff --git a/server/tests/EventValidatorTest.php b/server/tests/EventValidatorTest.php index 02303ff..9532ff0 100644 --- a/server/tests/EventValidatorTest.php +++ b/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"); } diff --git a/tests/test_app.py b/tests/test_app.py index 62862e2..da718a8 100644 --- a/tests/test_app.py +++ b/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) diff --git a/tests/test_engine.py b/tests/test_engine.py index 266af6b..db7e950 100644 --- a/tests/test_engine.py +++ b/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) diff --git a/tests/test_ergonomics.py b/tests/test_ergonomics.py new file mode 100644 index 0000000..b7f0b5a --- /dev/null +++ b/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