feat(v1): add local alerts, screenshot and JSONL event log
AlertDispatcher de-duplicates by event_id and fires each side effect once per CONFIRMED: annotated screenshot, one JSONL line (config_version, source_id, no rtsp/credentials), sound and popup via injectable sink. 44 tests pass; real-frame artifact smoke verified. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,123 @@
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from v1.alerts import AlertDispatcher, AlertSink, EventArtifactWriter, annotate_frame
|
||||
from v1.fall_state import FallEvent, FallState
|
||||
from v1.view_model import MonitorViewState, PersonOverlay, Point, StatusColor
|
||||
|
||||
|
||||
class RecordingSink(AlertSink):
|
||||
def __init__(self):
|
||||
self.sounds = []
|
||||
self.popups = []
|
||||
|
||||
def play_sound(self, event):
|
||||
self.sounds.append(event.event_id)
|
||||
|
||||
def show_popup(self, record):
|
||||
self.popups.append(record.event.event_id)
|
||||
|
||||
|
||||
def _event(event_id="FALL-000001", track_id="P-0003"):
|
||||
return FallEvent(
|
||||
event_id=event_id,
|
||||
track_id=track_id,
|
||||
config_version="cfg-abc123",
|
||||
suspected_at_monotonic=0.0,
|
||||
confirmed_at_monotonic=1.8,
|
||||
latency_seconds=1.8,
|
||||
)
|
||||
|
||||
|
||||
def _confirmed_view():
|
||||
overlay = PersonOverlay(
|
||||
track_id="P-0003",
|
||||
state=FallState.CONFIRMED,
|
||||
color=StatusColor.CRITICAL,
|
||||
box_xyxy=(20.0, 20.0, 80.0, 180.0),
|
||||
keypoints=(Point(30.0, 40.0), Point(40.0, 40.0)),
|
||||
skeleton_segments=((Point(30.0, 40.0), Point(40.0, 40.0)),),
|
||||
label="P-0003 · CONFIRMED",
|
||||
)
|
||||
return MonitorViewState(
|
||||
connected=True,
|
||||
status_text="在线",
|
||||
status_color=StatusColor.SUCCESS,
|
||||
has_frame=True,
|
||||
people=(overlay,),
|
||||
events=(),
|
||||
highest_state=FallState.CONFIRMED,
|
||||
)
|
||||
|
||||
|
||||
def _writer(tmp_path):
|
||||
fixed = datetime(2026, 7, 21, 8, 30, 0, tzinfo=timezone.utc)
|
||||
return EventArtifactWriter(tmp_path, source_id="lobby-camera-01", clock=lambda: fixed)
|
||||
|
||||
|
||||
def _image():
|
||||
return np.full((200, 240, 3), 30, dtype=np.uint8)
|
||||
|
||||
|
||||
def test_confirmed_event_fires_each_side_effect_once(tmp_path):
|
||||
sink = RecordingSink()
|
||||
dispatcher = AlertDispatcher(_writer(tmp_path), sink)
|
||||
event = _event()
|
||||
|
||||
first = dispatcher.dispatch([event], _image(), _confirmed_view())
|
||||
second = dispatcher.dispatch([event], _image(), _confirmed_view())
|
||||
|
||||
assert len(first) == 1
|
||||
assert second == ()
|
||||
assert sink.sounds == ["FALL-000001"]
|
||||
assert sink.popups == ["FALL-000001"]
|
||||
log_path = tmp_path / "20260721" / "events.jsonl"
|
||||
assert log_path.read_text(encoding="utf-8").count("\n") == 1
|
||||
assert (tmp_path / "20260721" / "FALL-000001.png").is_file()
|
||||
|
||||
|
||||
def test_jsonl_record_has_fields_and_no_source_address(tmp_path):
|
||||
dispatcher = AlertDispatcher(_writer(tmp_path))
|
||||
dispatcher.dispatch([_event()], _image(), _confirmed_view())
|
||||
|
||||
line = (tmp_path / "20260721" / "events.jsonl").read_text(encoding="utf-8").strip()
|
||||
record = json.loads(line)
|
||||
assert record["event_id"] == "FALL-000001"
|
||||
assert record["config_version"] == "cfg-abc123"
|
||||
assert record["source_id"] == "lobby-camera-01"
|
||||
assert record["state"] == "CONFIRMED"
|
||||
assert record["confirmed_at_utc"] == "2026-07-21T08:30:00+00:00"
|
||||
assert record["screenshot"] == "20260721/FALL-000001.png"
|
||||
assert "rtsp" not in line.lower()
|
||||
assert "password" not in line.lower()
|
||||
|
||||
|
||||
def test_missing_image_produces_no_side_effect(tmp_path):
|
||||
sink = RecordingSink()
|
||||
dispatcher = AlertDispatcher(_writer(tmp_path), sink)
|
||||
|
||||
records = dispatcher.dispatch([_event()], None, _confirmed_view())
|
||||
|
||||
assert records == ()
|
||||
assert sink.sounds == []
|
||||
assert not (tmp_path / "20260721").exists()
|
||||
|
||||
|
||||
def test_distinct_events_produce_two_log_lines(tmp_path):
|
||||
dispatcher = AlertDispatcher(_writer(tmp_path))
|
||||
dispatcher.dispatch([_event("FALL-000001")], _image(), _confirmed_view())
|
||||
dispatcher.dispatch([_event("FALL-000002", "P-0009")], _image(), _confirmed_view())
|
||||
|
||||
lines = (tmp_path / "20260721" / "events.jsonl").read_text(encoding="utf-8").strip().split("\n")
|
||||
assert len(lines) == 2
|
||||
|
||||
|
||||
def test_annotate_draws_red_border_only_for_confirmed():
|
||||
image = np.full((60, 80, 3), 30, dtype=np.uint8)
|
||||
confirmed = annotate_frame(image, _confirmed_view())
|
||||
corner = confirmed[0, 0].tolist()
|
||||
assert corner == [40, 40, 198] # critical red in BGR
|
||||
assert confirmed.shape == image.shape
|
||||
Reference in New Issue
Block a user