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 _standing_kps(cx=150):
    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)  # shoulders above
    kps[11] = Keypoint(cx - 10, 200, 0.9); kps[12] = Keypoint(cx + 10, 200, 0.9)  # hips below
    kps[0] = Keypoint(cx, 60, 0.9)                                               # head above hips
    return kps

def _standing_track(track, t, cx=150):
    track.frames.append(TrackFrame(t, BBox(cx - 20, 60, cx + 20, 210), _standing_kps(cx)))
    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_two_simultaneous_tracks_have_independent_fall_state():
    # Per-track detector state must not leak between two people monitored at once: a fall on track A
    # must arm and emit for A only, never for a standing track B. All other engine/detector tests
    # exercise a single person, so this pins the multi-person isolation the fall/immobility/zone
    # timers rely on (they key every bit of state off track_id).
    eng = _engine()  # prone_confirm_s=0.5
    a = Track(1, deque(maxlen=90)); b = Track(2, deque(maxlen=90))
    _lying_track(a, 0.0); _standing_track(b, 0.0, cx=500)
    eng.process([a, b], [], now=0.0)
    _lying_track(a, 0.6); _standing_track(b, 0.6, cx=500)
    events = eng.process([a, b], [], now=0.6)
    falls = [e for e in events if e.event_type == EventType.FALL]
    assert len(falls) == 1                 # exactly one fall...
    assert falls[0].track_id == 1          # ...for the lying track A, not the standing track B


def test_fall_fires_at_exactly_prone_confirm_s_inclusive_boundary():
    # The confirm is `now - prone_since >= prone_confirm_s`, so a fall must fire at EXACTLY the
    # threshold, not one frame later. Pins the inclusive boundary against a >= -> > regression that
    # would silently delay every fall confirmation by a frame — up to +0.5s at a thermally throttled
    # 2 fps, on the single most safety-critical event. Existing tests only feed comfortably-past
    # values (0.6 vs 0.5), so none of them would catch that shift.
    eng = _engine()  # prone_confirm_s=0.5
    tr = Track(1, deque(maxlen=90))
    _lying_track(tr, 0.0); eng.process([tr], [], now=0.0)   # prone_since = 0.0
    _lying_track(tr, 0.5)                                     # exactly prone_confirm_s elapsed
    events = eng.process([tr], [], now=0.5)
    assert any(e.event_type == EventType.FALL for e in events)


def test_occurred_at_uses_wall_clock_while_timers_use_monotonic():
    # A fall is CONFIRMED by the monotonic timer (prone_confirm_s over the injected `now`), but the
    # event must be STAMPED with wall time so an NTP step does not mis-timestamp evidence (and the
    # server's wall-time retention purge does not delete it). Timer `now` and `wall_now` are
    # independent axes here to prove occurred_at follows wall time, not the monotonic clock.
    eng = _engine()  # prone_confirm_s=0.5
    tr = Track(1, deque(maxlen=90))
    _lying_track(tr, 0.0)
    eng.process([tr], [], now=0.0, wall_now=1_700_000_000.0)
    _lying_track(tr, 0.6)
    events = eng.process([tr], [], now=0.6, wall_now=1_700_000_000.6)
    fall = [e for e in events if e.event_type == EventType.FALL]
    assert len(fall) == 1
    assert fall[0].occurred_at == 1_700_000_000.6   # wall time, NOT the monotonic 0.6


def test_process_defaults_occurred_at_to_now_when_no_wall_clock():
    # Single-clock callers (unit tests, direct callers) omit wall_now — occurred_at must fall back
    # to the timer `now` so existing behavior is unchanged.
    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 = [e for e in events if e.event_type == EventType.FALL]
    assert len(fall) == 1 and fall[0].occurred_at == 0.6


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