51eed139cf
Two operator-value features: evidence that shows HOW it happened, and the one-line answer to "what happened overnight". Evidence timeline (closes the P1-B "前中後 스냅샷 배선" follow-up): - The ring buffer now stores per-frame head boxes, and an event captures the frames nearest T-2.5s / T-1.0s / T-0 (server cap: 3 snapshots). A fall's evidentiary value is in the seconds BEFORE the person hit the floor - one end-state frame never showed how it happened. - Each past frame is blurred with the head positions of ITS OWN moment - blurring history with the current tick's boxes would leave a face exactly where it actually was (PIPA). - Right after startup the ring is short: fewer frames go out, never zero and never duplicated. Detail view captions the sequence 이전/직전/발생 시점 by position (offsets are edge config the server doesn't know). 24h catch-up brief: - The events console renders one line above the stream on load - "지난 24시간 - 감지 N건 (중대 n) · 니어미스 n · 평균 대응 t · 미확인 n건 남음" - reusing GET /api/report?days=1, with a link into /report. High-tinted when the window held critical/high events; hidden entirely until data arrives. Verified: pytest 235 pass (3 new: timeline selection, short-buffer dedup, 3-frame capture through run_once) / PHPUnit 138 pass + Chrome real-render (brief line with severe tint on the console; detail page showing the 3-frame timeline with captions; no console errors).
248 lines
11 KiB
Python
248 lines
11 KiB
Python
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
|