move_detect/tests/test_engine.py

from collections import deque
from edge.models import Keypoint, BBox
from edge.tracking import Track, TrackFrame
from edge.events import EventType, Severity
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, Zone
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_pinch_zone_emits_pinch_event_not_warning():
    # A 협착(pinch) zone is the highest-danger zone type (machinery entrapment). It must fire
    # as PINCH/HIGH, not be silently downgraded to a generic ZONE_WARNING.
    poly = [(100, 100), (200, 100), (200, 200), (100, 200)]
    zdet = ZoneIntrusionDetector([Zone("z1", "pinch", poly)], ZoneConfig(debounce_s=0.0))
    eng = EventEngine(
        FallDetector(FallConfig(prone_confirm_s=999)),     # keep fall/no_motion quiet
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        zdet, EngineConfig(camera_id="cam-01", cooldown_s=30.0))
    kps = [Keypoint(150, 150, 0.9) for _ in range(17)]     # both ankles inside the zone
    tr = Track(2, deque(maxlen=90))
    tr.frames.append(TrackFrame(0.0, BBox(140, 140, 160, 160), kps))
    events = eng.process([tr], [], now=0.0)
    pinch = [e for e in events if e.event_type == EventType.PINCH]
    assert len(pinch) == 1
    assert pinch[0].severity == Severity.HIGH
    assert pinch[0].zone_id == "z1"
    assert not any(e.event_type == EventType.ZONE_WARNING for e in events)


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