From 3240428380e14bcdac506a8cbabed3bbba41be32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sat, 11 Jul 2026 22:15:47 +0900 Subject: [PATCH] fix(edge): require sustained non-prone before FallDetector recover (FIX 2) edge/detectors/fall.py update(): a single frame where prone was false while state was in {FALLING, FALLEN, NO_MOTION} immediately emitted "recover" and reset _State. One jittery frame (a momentary pose-estimation miss) could wipe an in-progress CRITICAL no_motion incident. Add FallConfig.recover_grace_s (default 1.0s) and _State.not_prone_since. On the first non-prone frame while in an incident state, start the grace timer instead of recovering immediately; only reset to STANDING once recover_grace_s of sustained non-prone has elapsed. A re-prone frame while in an incident state clears the pending grace timer (cancels the recover attempt). The non-incident STANDING path is unchanged. BEHAVIOR CHANGE: recover now requires recover_grace_s of sustained non-prone instead of firing on a single frame. Tests: test_recovery_clears_state (tests/test_fall.py) strengthened per review brief to assert a single standing frame does NOT recover but sustained standing across the grace window does. Added test_single_noisy_nonprone_frame_does_not_cancel_no_motion covering the life-safety scenario: one noisy non-prone frame during NO_MOTION must not cancel the incident. --- edge/detectors/fall.py | 16 +++++++++++++--- tests/test_fall.py | 28 ++++++++++++++++++++++++++-- 2 files changed, 39 insertions(+), 5 deletions(-) diff --git a/edge/detectors/fall.py b/edge/detectors/fall.py index 4aa32a2..90202fc 100644 --- a/edge/detectors/fall.py +++ b/edge/detectors/fall.py @@ -17,6 +17,7 @@ class FallConfig: motion_thr: float = 0.02 # 정규화 관절변위 임계 min_conf: float = 0.5 min_visible: int = 10 + recover_grace_s: float = 1.0 # 사고 상태에서 회복 판정까지 요구되는 지속 비-prone 시간 @dataclass @@ -27,6 +28,7 @@ class _State: still_since: float | None = None fall_emitted: bool = False no_motion_emitted: bool = False + not_prone_since: float | None = None class FallDetector: @@ -76,12 +78,20 @@ class FallDetector: if not prone: if st.name in ("FALLING", "FALLEN", "NO_MOTION"): - self._states[track.track_id] = _State(baseline_height=bbox.height) - return "recover", {} + # 단일 튐(noisy) 프레임만으로 CRITICAL 사고를 즉시 취소하지 않도록 + # recover_grace_s만큼 지속된 비-prone을 요구한다. + if st.not_prone_since is None: + st.not_prone_since = now + return None, {} + if now - st.not_prone_since >= self.cfg.recover_grace_s: + self._states[track.track_id] = _State(baseline_height=bbox.height) + return "recover", {} + return None, {} st.prone_since = None return None, {} - # prone 상태 + # prone 상태 (재-prone 프레임은 대기 중인 recover 유예를 취소) + st.not_prone_since = None if st.prone_since is None: st.prone_since = now st.name = "FALLING" diff --git a/tests/test_fall.py b/tests/test_fall.py index e249d52..9d2f02f 100644 --- a/tests/test_fall.py +++ b/tests/test_fall.py @@ -61,14 +61,38 @@ def test_no_motion_after_fall_escalates(): assert sig == "no_motion" def test_recovery_clears_state(): - det = FallDetector(FallConfig(prone_confirm_s=0.5)) + # 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 - sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=1.0) + 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_low_confidence_is_ignored(): det = FallDetector(FallConfig()) tr = _track()