48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
import numpy as np
|
|
from edge.models import Person, BBox, Keypoint
|
|
from edge.tracking import IouTracker
|
|
from edge.smoothing import KeypointSmoother
|
|
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()
|
|
smoother = KeypointSmoother()
|
|
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, smoother, engine, ring, snap, sink, now_fn)
|
|
run_once(src, pose, tracker, smoother, engine, ring, snap, sink, now_fn)
|
|
assert any(e.event_type == EventType.FALL for e in sink.events)
|