752aabaff1
fall._states, immobility._still_since/_fired, zone._inside_since/_fired, and engine._last_emit were never pruned when a track was lost (ghost already pruned via on_lost). On a 24/7 edge device this is unbounded growth as track IDs keep incrementing. Add forget(track_id) to FallDetector, ImmobilityWatchdog, and ZoneIntrusionDetector (drops that track's per-track keys/sets). EventEngine.forget(track_id) calls each sub-detector's forget and removes _last_emit cooldown keys whose tuple's second element is that track_id (covers FALL/NO_MOTION/IMMOBILITY/zone/GHOST — GHOST keys are (GHOST, track_id) after FIX 4). EventEngine.process now calls self.forget(tr.track_id) for every track in lost_tracks after emitting, since ghost.on_lost already consumed that track's ghost-relevant state at the top of process(). app.py run_once already calls pool.forget(tr.track_id) for lost tracks as part of FIX 1, completing the cleanup chain (smoother pool + detector state + dedup cache). Tests: tests/test_engine.py adds test_forget_prunes_state_on_lost — fires a FALL event, then passes the track as lost, and asserts engine._last_emit no longer has that track's keys and fall._states no longer has the track id.
95 lines
3.8 KiB
Python
95 lines
3.8 KiB
Python
from collections import deque
|
|
from edge.models import Keypoint, BBox
|
|
from edge.tracking import Track, TrackFrame
|
|
from edge.events import EventType
|
|
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
|
|
|
|
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)
|
|
return kps
|
|
|
|
def _lying_track(track, t):
|
|
track.frames.append(TrackFrame(t, BBox(90, 290, 210, 320), _lying_kps()))
|
|
return track
|
|
|
|
def _engine():
|
|
return EventEngine(
|
|
FallDetector(FallConfig(prone_confirm_s=0.5)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01", cooldown_s=30.0),
|
|
)
|
|
|
|
def test_fall_produces_fall_event():
|
|
eng = _engine()
|
|
tr = Track(1, deque(maxlen=90))
|
|
_lying_track(tr, 0.0)
|
|
eng.process([tr], [], now=0.0)
|
|
_lying_track(tr, 0.6)
|
|
events = eng.process([tr], [], now=0.6)
|
|
types = [e.event_type for e in events]
|
|
assert EventType.FALL in types
|
|
assert events[0].camera_id == "cam-01"
|
|
|
|
def test_cooldown_suppresses_duplicate():
|
|
eng = _engine()
|
|
tr = Track(1, deque(maxlen=90))
|
|
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
|
|
_lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # FALL 1회
|
|
_lying_track(tr, 0.7)
|
|
again = eng.process([tr], [], now=0.7)
|
|
assert all(e.event_type != EventType.FALL for e in again)
|
|
|
|
def test_no_motion_event_carries_duration():
|
|
eng = EventEngine(
|
|
FallDetector(FallConfig(prone_confirm_s=0.5, no_motion_s=1.0, motion_thr=0.05)),
|
|
ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
|
|
GhostIncidentDetector(GhostConfig()),
|
|
ZoneIntrusionDetector([], ZoneConfig()),
|
|
EngineConfig(camera_id="cam-01", cooldown_s=30.0),
|
|
)
|
|
tr = Track(1, deque(maxlen=90))
|
|
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
|
|
_lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # -> fall
|
|
_lying_track(tr, 1.0); eng.process([tr], [], now=1.0) # still
|
|
_lying_track(tr, 2.0)
|
|
events = eng.process([tr], [], now=2.0) # immobile >1.0s -> no_motion
|
|
no_motion_events = [e for e in events if e.event_type == EventType.NO_MOTION]
|
|
assert len(no_motion_events) == 1
|
|
e = no_motion_events[0]
|
|
assert e.duration_s > 0
|
|
assert e.duration_s == e.metrics["immobility_s"]
|
|
|
|
|
|
def test_forget_prunes_state_on_lost():
|
|
eng = _engine()
|
|
tr = Track(1, deque(maxlen=90))
|
|
_lying_track(tr, 0.0); eng.process([tr], [], now=0.0)
|
|
_lying_track(tr, 0.6)
|
|
events = eng.process([tr], [], now=0.6) # FALL fires
|
|
assert any(e.event_type == EventType.FALL for e in events)
|
|
assert (EventType.FALL, 1) in eng._last_emit
|
|
assert 1 in eng.fall._states
|
|
|
|
# track is now lost -> engine must prune its per-track state
|
|
eng.process([], [tr], now=1.0)
|
|
assert not any(len(k) >= 2 and k[1] == 1 for k in eng._last_emit)
|
|
assert 1 not in eng.fall._states
|
|
|
|
|
|
def test_allow_cooldown_primitive():
|
|
eng = _engine() # EngineConfig(camera_id="cam-01", cooldown_s=30.0)
|
|
key = ("FALL", 1)
|
|
assert eng._allow(key, now=0.0) is True # first time allowed
|
|
assert eng._allow(key, now=5.0) is False # within cooldown -> suppressed
|
|
assert eng._allow(key, now=40.0) is True # after cooldown_s elapsed -> allowed again
|
|
# a different key is independent
|
|
assert eng._allow(("FALL", 2), now=5.0) is True
|