move_detect/tests/test_app.py

import numpy as np
from edge.models import Person, BBox, Keypoint
from edge.tracking import IouTracker
from edge.smoothing import TrackSmootherPool
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
from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
from edge.events import EventType
from edge.app import run_once

def _lying_person():
    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 Person(BBox(90, 290, 210, 320), kps)

class _Source:
    def read(self): return True, np.full((360, 640, 3), 128, np.uint8)

class _Pose:
    def infer(self, frame): return [_lying_person()]

class _Sink:
    def __init__(self): self.events = []
    def try_send(self, event, snapshots):
        self.events.append(event); return True

def test_run_once_pipeline_emits_fall_after_prone():
    src, pose = _Source(), _Pose()
    tracker = IouTracker()
    pool = TrackSmootherPool()
    engine = EventEngine(
        FallDetector(FallConfig(prone_confirm_s=0.5)),
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        ZoneIntrusionDetector([], ZoneConfig()),
        EngineConfig(camera_id="cam-01"),
    )
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
    times = iter([0.0, 0.6])
    def now_fn(): return next(times)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    assert any(e.event_type == EventType.FALL for e in sink.events)


def test_run_once_stamps_events_with_wall_clock_not_monotonic():
    # run_once must thread a separate wall clock through to occurred_at: the daemon runs the loop on
    # a monotonic clock (NTP-step-immune timers) but events have to carry real wall time.
    src, pose = _Source(), _Pose()
    tracker = IouTracker()
    pool = TrackSmootherPool()
    engine = EventEngine(
        FallDetector(FallConfig(prone_confirm_s=0.5)),
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        ZoneIntrusionDetector([], ZoneConfig()),
        EngineConfig(camera_id="cam-01"),
    )
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
    mono = iter([0.0, 0.6])                       # monotonic loop clock (drives prone_confirm_s)
    wallt = iter([1_700_000_000.0, 1_700_000_000.6])  # wall clock (stamps occurred_at)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink,
             now_fn=lambda: next(mono), wall_fn=lambda: next(wallt))
    run_once(src, pose, tracker, pool, engine, ring, snap, sink,
             now_fn=lambda: next(mono), wall_fn=lambda: next(wallt))
    falls = [e for e in sink.events if e.event_type == EventType.FALL]
    assert falls and falls[0].occurred_at == 1_700_000_000.6


def _engine(**fall_kw):
    return EventEngine(
        FallDetector(FallConfig(**fall_kw)),
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        ZoneIntrusionDetector([], ZoneConfig()),
        EngineConfig(camera_id="cam-01"),
    )


class _ScriptedPose:
    def __init__(self, frames): self.frames = list(frames)
    def infer(self, frame): return self.frames.pop(0)


def test_dropout_tick_does_not_feed_detectors_stale_frame():
    # A track not re-detected this tick still sits in the tracker (< max_age) with a frozen
    # latest frame. Passing it to the engine lets prone_confirm accrue on stale data: here the
    # dropout tick alone would confirm the fall. A stale track must be treated as no-data.
    src = _Source()
    pose = _ScriptedPose([[_lying_person()], []])
    tracker = IouTracker()   # max_age=1.0 > dropout gap, so the track stays active
    pool = TrackSmootherPool()
    engine = _engine(prone_confirm_s=0.5, falling_confirm_s=0.1)
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
    times = iter([0.0, 0.6])
    def now_fn(): return next(times)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)  # dropout tick
    assert not any(e.event_type == EventType.FALL for e in sink.events)


def test_dropout_then_redetection_still_confirms_fall():
    # ...but the person re-detected still prone MUST confirm — the dropout pauses evidence,
    # it does not erase the incident.
    src = _Source()
    pose = _ScriptedPose([[_lying_person()], [], [_lying_person()]])
    tracker = IouTracker()
    pool = TrackSmootherPool()
    engine = _engine(prone_confirm_s=0.5, falling_confirm_s=0.1)
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
    times = iter([0.0, 0.3, 0.7])
    def now_fn(): return next(times)
    for _ in range(3):
        run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    assert any(e.event_type == EventType.FALL for e in sink.events)


def test_event_snapshots_are_a_timeline_not_one_frame():
    # 낙상의 증거 가치는 바닥에 닿기 전 몇 초에 있다: 이벤트에는 현재 1장이 아니라
    # 링버퍼에서 고른 과거→현재 타임라인(최대 3장)이 붙어야 한다.
    class _SnapSink:
        def __init__(self): self.sent = []
        def try_send(self, event, snapshots):
            self.sent.append((event, snapshots)); return True

    src, pose = _Source(), _Pose()
    tracker = IouTracker()
    pool = TrackSmootherPool()
    engine = _engine(prone_confirm_s=3.0, falling_confirm_s=0.1)
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _SnapSink()
    times = iter([0.0, 1.0, 2.0, 3.5])
    def now_fn(): return next(times)
    for _ in range(4):                       # 3.5초에 prone_confirm(3.0) 충족 → fall
        run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    falls = [(e, s) for e, s in sink.sent if e.event_type == EventType.FALL]
    assert len(falls) == 1
    assert len(falls[0][1]) == 3             # 과거 2장 + 발생 시점 1장
    for jpg in falls[0][1]:
        assert jpg[:2] == b"\xff\xd8"


def test_ergo_report_carries_no_snapshots():
    # ERGO_REPORT는 익명 통계다 — 프레임이 붙는 순간 통계가 감시 이미지가 된다.
    from edge.detectors.ergonomics import ErgoMonitor, ErgoConfig
    from edge.detectors.fall import FallDetector, FallConfig

    class _SnapSink:
        def __init__(self): self.sent = []
        def try_send(self, event, snapshots):
            self.sent.append((event, snapshots)); return True

    src = _Source()
    pose = _ScriptedPose([[_person_at(500)], [_person_at(500)]])
    tracker = IouTracker()
    pool = TrackSmootherPool()
    engine = EventEngine(
        FallDetector(FallConfig(prone_confirm_s=999)),
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        ZoneIntrusionDetector([], ZoneConfig()),
        EngineConfig(camera_id="cam-01"),
        ergonomics=ErgoMonitor(ErgoConfig(report_interval_s=0.5)))
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _SnapSink()
    times = iter([0.0, 0.6])
    def now_fn(): return next(times)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    ergo = [(e, s) for e, s in sink.sent if e.event_type == EventType.ERGO_REPORT]
    assert len(ergo) == 1
    assert ergo[0][1] == []   # 스냅샷 미부착


def _lying_person_at(x1, x2):
    kps = [Keypoint((x1 + x2) / 2, 15, 0.9) for _ in range(17)]
    return Person(BBox(x1, 0, x2, 20), kps)


def test_contended_crossing_resets_fall_state():
    # Two prone tracks; then one box overlapping BOTH (a crossing). The surviving match may
    # be a swap, so the contended track's fall state must reset — no FALL may fire off the
    # pre-crossing prone timer.
    src = _Source()
    pose = _ScriptedPose([
        [_lying_person_at(0, 40), _lying_person_at(30, 70)],
        [_lying_person_at(15, 55)],
    ])
    tracker = IouTracker(iou_thr=0.3)
    pool = TrackSmootherPool()
    engine = _engine(prone_confirm_s=0.05, falling_confirm_s=0.01)
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
    times = iter([0.0, 0.1])
    def now_fn(): return next(times)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    assert not any(e.event_type == EventType.FALL for e in sink.events)


def _person_at(cx):
    # standing person centered at cx; nose at (cx, 60)
    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)
    kps[11] = Keypoint(cx - 10, 200, 0.9); kps[12] = Keypoint(cx + 10, 200, 0.9)
    kps[0] = Keypoint(cx, 60, 0.9)
    return Person(BBox(cx - 20, 60, cx + 20, 210), kps)


class _TwoPersonPose:
    def __init__(self, cx_a, cx_b):
        self.cx_a, self.cx_b = cx_a, cx_b

    def infer(self, frame):
        return [_person_at(self.cx_a), _person_at(self.cx_b)]


def test_two_persons_get_independent_smoothing():
    # person A far on the left, person B far on the right; small per-frame
    # jitter so IoU tracking stays continuous across frames.
    src = _Source()
    pose = _TwoPersonPose(cx_a=100, cx_b=900)
    tracker = IouTracker()
    pool = TrackSmootherPool()
    engine = EventEngine(
        FallDetector(FallConfig(prone_confirm_s=999)),
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        ZoneIntrusionDetector([], ZoneConfig()),
        EngineConfig(camera_id="cam-01"),
    )
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink()
    times = iter([0.0, 0.1])
    def now_fn(): return next(times)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)

    tracks = {tr.track_id: tr for tr in tracker._tracks.values()}
    assert len(tracks) == 2
    xs = sorted(tr.latest.keypoints[0].x for tr in tracks.values())
    left_nose_x, right_nose_x = xs
    # each track's smoothed nose must stay on its own side, nowhere near
    # the midpoint between the two persons (500) — proves no shared filter
    # state leaked between tracks.
    assert left_nose_x < 300
    assert right_nose_x > 700


def test_overstay_goes_out_without_evidence_while_the_same_batch_keeps_it():
    """체류 상한 초과에는 사진이 붙지 않는다 — 같은 배치의 구역 침입에는 붙는 채로.

    상수만 검사하면 배선이 끊겨도 초록이 뜬다(이 프로젝트가 감사마다 잡아온 모양이다).
    그래서 실제 run_once 를 태워, 한 번의 전송에서 한쪽은 증거를 들고 다른 쪽은 안 드는지를
    본다. 체류 초과의 발화 조건은 시간 경과 하나뿐이라, 여기 사진이 붙으면 정상 작업자가
    규정대로 일하는 동안 주기적으로 본인 사진이 서버로 가는 시스템이 된다.
    """
    from edge.detectors.zone import Zone
    from edge.detectors.occupancy import OccupancyWatchdog, OccupancyConfig

    class _SnapSink:
        def __init__(self): self.sent = []
        def try_send(self, event, snapshots):
            self.sent.append((event, snapshots)); return True

    poly = [(0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0)]     # 프레임 전체
    zone = Zone("z1", "restricted", poly, {"max_dwell_s": 1.0})

    src = _Source()
    pose = _ScriptedPose([[_person_at(320)], [_person_at(320)]])
    engine = EventEngine(
        FallDetector(FallConfig(prone_confirm_s=999)),
        ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)),
        GhostIncidentDetector(GhostConfig()),
        ZoneIntrusionDetector([zone], ZoneConfig()),
        EngineConfig(camera_id="cam-01"),
        # 기본 설정 그대로. 바꾸는 것은 존 규칙(현장 설정)인 max_dwell_s 뿐이다.
        occupancy=OccupancyWatchdog(OccupancyConfig()))
    ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _SnapSink()

    # 트래커·스무더는 프레임 간에 공유한다 — 체류는 "같은 트랙이 계속 있었는가"라서,
    # 매 호출 새 트래커를 주면 실제와 다른 조건에서 통과할 수 있다.
    tracker, pool = IouTracker(), TrackSmootherPool()
    times = iter([0.0, 1.5])
    def now_fn(): return next(times)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)
    run_once(src, pose, tracker, pool, engine, ring, snap, sink, now_fn)

    by_type = {e.event_type: s for e, s in sink.sent}
    assert EventType.OVERSTAY in by_type, "체류 상한 초과가 발화하지 않았습니다 — 테스트가 무의미합니다"
    assert by_type[EventType.OVERSTAY] == [], \
        "체류 상한 초과에 사진이 붙었습니다. 오래 있었다는 이유로 작업자 사진이 남습니다."

    # 대조군: 같은 파이프라인에서 진짜 사건에는 증거가 붙어 있어야 한다. 이게 없으면
    # 위 단언은 "증거 기능이 통째로 죽었다"로도 통과한다.
    assert EventType.ZONE_RESTRICTED in by_type, "구역 침입이 발화하지 않았습니다"
    assert by_type[EventType.ZONE_RESTRICTED], \
        "구역 침입에서 증거가 사라졌습니다 — 사고 조사에서 볼 것이 없어집니다."