silver_pose/v1/tests/test_alerts.py

import json
from datetime import datetime, timezone

import numpy as np
import pytest

import os

from v1.alerts import (
    AlertDispatcher,
    AlertSink,
    EventArtifactWriter,
    annotate_frame,
    annotation_label,
    screenshot_label,
    _find_font,
)
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_screenshot_label_is_ascii_only():
    label = screenshot_label("P-0001", "CONFIRMED")

    assert label == "P-0001 CONFIRMED"
    assert all(ord(ch) < 128 for ch in label)


def test_screenshot_label_strips_non_ascii_separator():
    # The live-view label uses a middle dot; the screenshot label must not.
    label = screenshot_label("P-0001", "CONFIRMED")

    assert "·" not in label


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


def test_annotation_label_is_chinese():
    assert annotation_label("P-0001", FallState.CONFIRMED) == "确认摔倒 P-0001"


def test_find_font_returns_none_when_missing_and_path_when_present(tmp_path):
    assert _find_font("/no/such/font.ttf") is None
    font = tmp_path / "x.ttf"
    font.write_bytes(b"stub")
    assert _find_font(str(font)) == str(font)


def test_annotate_falls_back_without_a_font():
    image = np.full((60, 80, 3), 30, dtype=np.uint8)

    out = annotate_frame(image, _confirmed_view(), font_path="/no/such/font.ttf")

    assert out.shape == image.shape  # ASCII fallback ran, no crash


def test_annotate_draws_labels_with_a_ttf_font():
    font = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"
    if not os.path.isfile(font):
        import pytest

        pytest.skip("no ttf available to exercise the PIL path")

    out = annotate_frame(np.full((60, 80, 3), 30, dtype=np.uint8), _confirmed_view(), font_path=font)

    assert out.shape == (60, 80, 3)