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).
70 lines
3.3 KiB
Python
70 lines
3.3 KiB
Python
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()}
|