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).
224 lines
9.5 KiB
Python
224 lines
9.5 KiB
Python
import numpy as np
|
|
from edge.models import Person, BBox, Keypoint
|
|
from edge.tracking import IouTracker
|
|
from edge.smoothing import TrackSmootherPool
|
|
from edge.detectors.fall import FallDetector, FallConfig
|
|
from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig
|
|
from edge.detectors.ghost import GhostIncidentDetector, GhostConfig
|
|
from edge.detectors.zone import ZoneIntrusionDetector, ZoneConfig
|
|
from edge.engine import EventEngine, EngineConfig
|
|
from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
|
|
from edge.events import EventType
|
|
from edge.app import run_once
|
|
|
|
def _lying_person():
|
|
kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
|
|
kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
|
|
kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
|
|
kps[0] = Keypoint(90, 320, 0.9)
|
|
return Person(BBox(90, 290, 210, 320), kps)
|
|
|
|
class _Source:
|
|
def read(self): return True, np.full((360, 640, 3), 128, np.uint8)
|
|
|
|
class _Pose:
|
|
def infer(self, frame): return [_lying_person()]
|
|
|
|
class _Sink:
|
|
def __init__(self): self.events = []
|
|
def try_send(self, event, snapshots):
|
|
self.events.append(event); return True
|
|
|
|
def test_run_once_pipeline_emits_fall_after_prone():
|
|
src, pose = _Source(), _Pose()
|
|
tracker = IouTracker()
|
|
pool = TrackSmootherPool()
|
|
engine = EventEngine(
|
|
FallDetector(FallConfig(prone_confirm_s=0.5)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01"),
|
|
)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
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)
|
|
assert any(e.event_type == EventType.FALL for e in sink.events)
|
|
|
|
|
|
def test_run_once_stamps_events_with_wall_clock_not_monotonic():
|
|
# run_once must thread a separate wall clock through to occurred_at: the daemon runs the loop on
|
|
# a monotonic clock (NTP-step-immune timers) but events have to carry real wall time.
|
|
src, pose = _Source(), _Pose()
|
|
tracker = IouTracker()
|
|
pool = TrackSmootherPool()
|
|
engine = EventEngine(
|
|
FallDetector(FallConfig(prone_confirm_s=0.5)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01"),
|
|
)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
mono = iter([0.0, 0.6]) # monotonic loop clock (drives prone_confirm_s)
|
|
wallt = iter([1_700_000_000.0, 1_700_000_000.6]) # wall clock (stamps occurred_at)
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink,
|
|
now_fn=lambda: next(mono), wall_fn=lambda: next(wallt))
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink,
|
|
now_fn=lambda: next(mono), wall_fn=lambda: next(wallt))
|
|
falls = [e for e in sink.events if e.event_type == EventType.FALL]
|
|
assert falls and falls[0].occurred_at == 1_700_000_000.6
|
|
|
|
|
|
def _engine(**fall_kw):
|
|
return EventEngine(
|
|
FallDetector(FallConfig(**fall_kw)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01"),
|
|
)
|
|
|
|
|
|
class _ScriptedPose:
|
|
def __init__(self, frames): self.frames = list(frames)
|
|
def infer(self, frame): return self.frames.pop(0)
|
|
|
|
|
|
def test_dropout_tick_does_not_feed_detectors_stale_frame():
|
|
# A track not re-detected this tick still sits in the tracker (< max_age) with a frozen
|
|
# latest frame. Passing it to the engine lets prone_confirm accrue on stale data: here the
|
|
# dropout tick alone would confirm the fall. A stale track must be treated as no-data.
|
|
src = _Source()
|
|
pose = _ScriptedPose([[_lying_person()], []])
|
|
tracker = IouTracker() # max_age=1.0 > dropout gap, so the track stays active
|
|
pool = TrackSmootherPool()
|
|
engine = _engine(prone_confirm_s=0.5, falling_confirm_s=0.1)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
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) # dropout tick
|
|
assert not any(e.event_type == EventType.FALL for e in sink.events)
|
|
|
|
|
|
def test_dropout_then_redetection_still_confirms_fall():
|
|
# ...but the person re-detected still prone MUST confirm — the dropout pauses evidence,
|
|
# it does not erase the incident.
|
|
src = _Source()
|
|
pose = _ScriptedPose([[_lying_person()], [], [_lying_person()]])
|
|
tracker = IouTracker()
|
|
pool = TrackSmootherPool()
|
|
engine = _engine(prone_confirm_s=0.5, falling_confirm_s=0.1)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
times = iter([0.0, 0.3, 0.7])
|
|
def now_fn(): return next(times)
|
|
for _ in range(3):
|
|
run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
|
|
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)
|
|
|
|
|
|
def test_contended_crossing_resets_fall_state():
|
|
# Two prone tracks; then one box overlapping BOTH (a crossing). The surviving match may
|
|
# be a swap, so the contended track's fall state must reset — no FALL may fire off the
|
|
# pre-crossing prone timer.
|
|
src = _Source()
|
|
pose = _ScriptedPose([
|
|
[_lying_person_at(0, 40), _lying_person_at(30, 70)],
|
|
[_lying_person_at(15, 55)],
|
|
])
|
|
tracker = IouTracker(iou_thr=0.3)
|
|
pool = TrackSmootherPool()
|
|
engine = _engine(prone_confirm_s=0.05, falling_confirm_s=0.01)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
times = iter([0.0, 0.1])
|
|
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)
|
|
assert not any(e.event_type == EventType.FALL for e in sink.events)
|
|
|
|
|
|
def _person_at(cx):
|
|
# standing person centered at cx; nose at (cx, 60)
|
|
kps = [Keypoint(cx, 150, 0.9) for _ in range(17)]
|
|
kps[5] = Keypoint(cx - 10, 100, 0.9); kps[6] = Keypoint(cx + 10, 100, 0.9)
|
|
kps[11] = Keypoint(cx - 10, 200, 0.9); kps[12] = Keypoint(cx + 10, 200, 0.9)
|
|
kps[0] = Keypoint(cx, 60, 0.9)
|
|
return Person(BBox(cx - 20, 60, cx + 20, 210), kps)
|
|
|
|
|
|
class _TwoPersonPose:
|
|
def __init__(self, cx_a, cx_b):
|
|
self.cx_a, self.cx_b = cx_a, cx_b
|
|
|
|
def infer(self, frame):
|
|
return [_person_at(self.cx_a), _person_at(self.cx_b)]
|
|
|
|
|
|
def test_two_persons_get_independent_smoothing():
|
|
# person A far on the left, person B far on the right; small per-frame
|
|
# jitter so IoU tracking stays continuous across frames.
|
|
src = _Source()
|
|
pose = _TwoPersonPose(cx_a=100, cx_b=900)
|
|
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"),
|
|
)
|
|
ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
|
|
times = iter([0.0, 0.1])
|
|
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)
|
|
|
|
tracks = {tr.track_id: tr for tr in tracker._tracks.values()}
|
|
assert len(tracks) == 2
|
|
xs = sorted(tr.latest.keypoints[0].x for tr in tracks.values())
|
|
left_nose_x, right_nose_x = xs
|
|
# each track's smoothed nose must stay on its own side, nowhere near
|
|
# the midpoint between the two persons (500) — proves no shared filter
|
|
# state leaked between tracks.
|
|
assert left_nose_x < 300
|
|
assert right_nose_x > 700
|