9e4225a338
Heinrich's-law play: the signals we already compute but used to throw away
become a spatial risk map of the workplace, drawn over the camera's own frame.
Edge:
- near_fall (INFO): a track that reached FALLING (sustained prone) but recovered
before the FALLEN confirm. Treated as a full event per product decision -
same ingest, snapshots, dedup, ack flow as incidents. Fix en route: prone_s
used a truthiness check that turned prone_since=0.0 into "no timer".
- Every per-track event (fall/no_motion/near_fall/immobility/zone hits) now
carries the foot point in the zones' normalized 0..1 space (reusing the zone
detector's own coordinate math); ghost's pixel-space loss point is normalized
at emit. to_payload omits the key when absent, so an old server ignores it.
Server:
- move_events.spatial_x/spatial_y (nullable DOUBLE) + validator: optional
[x, y] finite-number pair; range deliberately NOT enforced (an off-by-a-hair
foot must never reject a FALL alert - renderers clamp).
- MIGRATION FIX: chk_events_type would have REJECTED the previous commit's
lone_worker/overcapacity/overstay inserts on MySQL 8.0.16+ - the constraint
is rebuilt with all 13 types (incl. near_fall).
- GET /api/heatmap/{camera}?days=N (1..90, default 30, 5000-point cap):
spatial points with the same fail-closed site scoping as the event list.
UI (zone editor): "위험 히트맵" toggle + day window. Additive radial blots in
an offscreen layer under the zone polygons; severity weighs the alpha so a
cluster of near-misses reads lighter than one confirmed fall at the same spot.
Events console/detail label near_fall in Korean.
Verified: pytest 222 pass / PHPUnit 130 pass + Chrome real-render (blue
near-miss cluster + red incident hotspot inside the restricted zone, amber
cluster in the warning zone, zones drawn on top; no console errors).
167 lines
7.4 KiB
Python
167 lines
7.4 KiB
Python
from collections import deque
|
|
from edge.models import Keypoint, BBox
|
|
from edge.tracking import Track, TrackFrame
|
|
from edge.detectors.fall import FallDetector, FallConfig
|
|
|
|
def _standing_kps():
|
|
kps = [Keypoint(50, 150, 0.9) for _ in range(17)]
|
|
kps[5] = Keypoint(40, 100, 0.9); kps[6] = Keypoint(60, 100, 0.9)
|
|
kps[11] = Keypoint(40, 200, 0.9); kps[12] = Keypoint(60, 200, 0.9)
|
|
kps[0] = Keypoint(50, 60, 0.9) # nose above hips
|
|
return kps
|
|
|
|
def _lying_kps():
|
|
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) # nose below hips
|
|
return kps
|
|
|
|
def _track():
|
|
return Track(1, deque(maxlen=90))
|
|
|
|
def _feed(det, track, kps, bbox, t):
|
|
track.frames.append(TrackFrame(t, bbox, kps))
|
|
return det.update(track, now=t)
|
|
|
|
def test_standing_produces_no_event():
|
|
det = FallDetector(FallConfig())
|
|
tr = _track()
|
|
sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=0.0)
|
|
assert sig is None
|
|
|
|
|
|
def test_fall_detected_from_bbox_when_keypoints_low_confidence():
|
|
# A fallen person is exactly when the pose model loses confidence (occluded, on the
|
|
# floor). The old blanket confidence gate returned None -> fall/no_motion/ghost all
|
|
# DISABLED for the person on the ground. The bbox cues (wide-and-short) don't need
|
|
# keypoints and must still fire.
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5, min_conf=0.5, min_visible=10))
|
|
tr = _track()
|
|
low_conf = [Keypoint(100, 100, 0.1) for _ in range(17)] # all below min_conf
|
|
wide_short = BBox(0, 0, 200, 100) # aspect 2.0 -> prone
|
|
_feed(det, tr, low_conf, wide_short, t=0.0) # FALLING
|
|
sig, _ = _feed(det, tr, low_conf, wide_short, t=0.6) # prone 0.6s > 0.5 -> FALLEN
|
|
assert sig == "fall"
|
|
|
|
|
|
def test_low_confidence_standing_is_not_a_false_fall():
|
|
# ...but a low-confidence STANDING person (tall, narrow bbox, no collapse) must NOT be
|
|
# flagged — the bbox cues alone stay conservative.
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5, min_conf=0.5, min_visible=10))
|
|
tr = _track()
|
|
low_conf = [Keypoint(50, 100, 0.1) for _ in range(17)]
|
|
tall_narrow = BBox(40, 60, 60, 210) # aspect ~0.13 -> not prone
|
|
_feed(det, tr, low_conf, tall_narrow, t=0.0)
|
|
sig, _ = _feed(det, tr, low_conf, tall_narrow, t=0.6)
|
|
assert sig is None
|
|
|
|
def test_lying_confirmed_after_prone_seconds_emits_fall():
|
|
det = FallDetector(FallConfig(prone_confirm_s=1.0))
|
|
tr = _track()
|
|
lying_box = BBox(90, 290, 210, 320) # 넓고 낮음 (aspect>1.25)
|
|
_feed(det, tr, _lying_kps(), lying_box, t=0.0)
|
|
sig, meta = _feed(det, tr, _lying_kps(), lying_box, t=1.5) # prone 1.5s > 1.0
|
|
assert sig == "fall"
|
|
assert meta["torso_angle"] > 80
|
|
|
|
def test_front_fall_height_collapse_triggers_without_angle():
|
|
# 정면 낙상: 각도/종횡비 안 변해도 bbox 높이 급붕괴
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5))
|
|
tr = _track()
|
|
tall = BBox(40, 60, 60, 210) # height 150
|
|
_feed(det, tr, _standing_kps(), tall, t=0.0)
|
|
short = BBox(40, 180, 60, 210) # height 30 (붕괴 0.8)
|
|
_feed(det, tr, _standing_kps(), short, t=0.2)
|
|
sig, _ = _feed(det, tr, _standing_kps(), short, t=0.9)
|
|
assert sig == "fall"
|
|
|
|
def test_no_motion_after_fall_escalates():
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5, no_motion_s=1.0, motion_thr=0.05))
|
|
tr = _track()
|
|
box = BBox(90, 290, 210, 320)
|
|
_feed(det, tr, _lying_kps(), box, t=0.0)
|
|
_feed(det, tr, _lying_kps(), box, t=0.6) # -> fall
|
|
_feed(det, tr, _lying_kps(), box, t=1.0) # 미동 지속
|
|
sig, _ = _feed(det, tr, _lying_kps(), box, t=2.0) # 부동 >1.0s
|
|
assert sig == "no_motion"
|
|
|
|
def test_recovery_clears_state():
|
|
# Recovery requires SUSTAINED non-prone for recover_grace_s, not a single frame.
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5, recover_grace_s=1.0))
|
|
tr = _track()
|
|
box = BBox(90, 290, 210, 320)
|
|
_feed(det, tr, _lying_kps(), box, t=0.0)
|
|
_feed(det, tr, _lying_kps(), box, t=0.6) # fall
|
|
standing_box = BBox(40, 60, 60, 210)
|
|
# a single standing frame right after a fall must NOT recover yet
|
|
sig, _ = _feed(det, tr, _standing_kps(), standing_box, t=1.0)
|
|
assert sig is None
|
|
# sustained standing through the grace window -> recover fires
|
|
sig, _ = _feed(det, tr, _standing_kps(), standing_box, t=2.5)
|
|
assert sig == "recover"
|
|
|
|
|
|
def test_single_noisy_nonprone_frame_does_not_cancel_no_motion():
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5, no_motion_s=1.0, motion_thr=0.05,
|
|
recover_grace_s=1.0))
|
|
tr = _track()
|
|
box = BBox(90, 290, 210, 320)
|
|
_feed(det, tr, _lying_kps(), box, t=0.0)
|
|
_feed(det, tr, _lying_kps(), box, t=0.6) # -> fall
|
|
_feed(det, tr, _lying_kps(), box, t=1.0) # still
|
|
sig, _ = _feed(det, tr, _lying_kps(), box, t=2.0) # immobile >1.0s
|
|
assert sig == "no_motion"
|
|
# a single noisy non-prone frame within the grace window must NOT cancel
|
|
# the in-progress CRITICAL no_motion incident.
|
|
sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=2.3)
|
|
assert sig is None
|
|
st = det._states[tr.track_id]
|
|
assert st.name == "NO_MOTION"
|
|
|
|
def test_single_prone_frame_does_not_enter_falling():
|
|
# FALLING is a DANGEROUS state: if the track vanishes while in it, the ghost detector
|
|
# arms a CRITICAL incident. A single prone-looking frame (a bend, a bbox glitch) must
|
|
# not be enough — FALLING requires sustained prone (falling_confirm_s).
|
|
det = FallDetector(FallConfig()) # falling_confirm_s=1.0
|
|
tr = _track()
|
|
box = BBox(90, 290, 210, 320)
|
|
_feed(det, tr, _lying_kps(), box, t=0.0)
|
|
assert det._states[1].name == "STANDING"
|
|
_feed(det, tr, _lying_kps(), box, t=1.2) # sustained prone -> now FALLING
|
|
assert det._states[1].name == "FALLING"
|
|
|
|
|
|
def test_recovery_before_confirm_signals_near_fall():
|
|
# FALLING(지속 prone)까지 갔다가 FALLEN 확정 전에 회복 = 니어미스. 사고 알람이 아니라
|
|
# 히트맵/통계 재료로 구분된 신호("near_fall")여야 한다.
|
|
det = FallDetector(FallConfig(prone_confirm_s=5.0, falling_confirm_s=0.5, recover_grace_s=1.0))
|
|
tr = _track()
|
|
box = BBox(90, 290, 210, 320)
|
|
_feed(det, tr, _lying_kps(), box, t=0.0)
|
|
_feed(det, tr, _lying_kps(), box, t=1.0) # FALLING (확정 전)
|
|
standing_box = BBox(40, 60, 60, 210)
|
|
_feed(det, tr, _standing_kps(), standing_box, t=2.0) # 회복 시작
|
|
sig, meta = _feed(det, tr, _standing_kps(), standing_box, t=3.5) # grace 경과
|
|
assert sig == "near_fall"
|
|
assert meta["prone_s"] == 2.0 # prone 0.0 → 비-prone 2.0
|
|
|
|
|
|
def test_recovery_after_confirmed_fall_is_recover_not_near_fall():
|
|
det = FallDetector(FallConfig(prone_confirm_s=0.5, recover_grace_s=1.0))
|
|
tr = _track()
|
|
box = BBox(90, 290, 210, 320)
|
|
_feed(det, tr, _lying_kps(), box, t=0.0)
|
|
_feed(det, tr, _lying_kps(), box, t=0.6) # fall 확정
|
|
standing_box = BBox(40, 60, 60, 210)
|
|
_feed(det, tr, _standing_kps(), standing_box, t=1.0)
|
|
sig, _ = _feed(det, tr, _standing_kps(), standing_box, t=2.5)
|
|
assert sig == "recover"
|
|
|
|
|
|
def test_low_confidence_is_ignored():
|
|
det = FallDetector(FallConfig())
|
|
tr = _track()
|
|
kps = [Keypoint(0, 0, 0.1) for _ in range(17)] # 전부 저신뢰
|
|
sig, _ = _feed(det, tr, kps, BBox(0, 0, 100, 30), t=0.0)
|
|
assert sig is None
|