silver_pose/v1/alerts.py

"""Confirmed-event side effects: annotated screenshot, JSONL log, sound, popup.

Side effects run once per ``event_id``. Screenshot annotation and JSONL writing
are Qt-free and unit-tested; sound and popup are injected sinks so the desktop
adapters (Windows) stay thin. Artifact names use only the event ID and a date
folder — never an RTSP address, credential, or client name.
"""

import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Dict, Optional, Sequence, Tuple

import cv2
import numpy as np

from v1.fall_state import FallEvent, FallState
from v1.view_model import MonitorViewState, StatusColor


# Semantic tokens as OpenCV BGR tuples (see docs/ui/silver-pose-ui-ux-spec.md).
_COLOR_BGR: Dict[StatusColor, Tuple[int, int, int]] = {
    StatusColor.SUCCESS: (61, 128, 21),
    StatusColor.CAUTION: (9, 83, 180),
    StatusColor.CRITICAL: (40, 40, 198),
    StatusColor.OFFLINE: (139, 116, 100),
}


@dataclass(frozen=True)
class AlertRecord:
    event: FallEvent
    screenshot_path: Path
    log_path: Path
    created_at_utc: str


class AlertSink:
    """Injectable output for sound and popup. Default implementation is silent."""

    def play_sound(self, event: FallEvent) -> None:  # pragma: no cover - adapter
        pass

    def show_popup(self, record: AlertRecord) -> None:  # pragma: no cover - adapter
        pass


def screenshot_label(track_id: str, state_value: str) -> str:
    """ASCII-only label for ``cv2.putText``.

    OpenCV's Hershey fonts render only ASCII; non-ASCII characters (such as the
    middle dot used in the live-view label) come out garbled, so the screenshot
    label is built from the id and state and stripped to ASCII.
    """

    text = "{0} {1}".format(track_id, state_value)
    return text.encode("ascii", "ignore").decode("ascii")


def annotate_frame(image: np.ndarray, view: MonitorViewState) -> np.ndarray:
    """Return a copy of ``image`` with boxes, skeleton, labels and a state border."""

    canvas = image.copy()
    for overlay in view.people:
        color = _COLOR_BGR[overlay.color]
        left, top, right, bottom = (int(round(v)) for v in overlay.box_xyxy)
        cv2.rectangle(canvas, (left, top), (right, bottom), color, 2)
        for start, end in overlay.skeleton_segments:
            cv2.line(
                canvas,
                (int(round(start.x)), int(round(start.y))),
                (int(round(end.x)), int(round(end.y))),
                color,
                2,
            )
        for point in overlay.keypoints:
            if point is not None:
                cv2.circle(canvas, (int(round(point.x)), int(round(point.y))), 3, color, -1)
        cv2.putText(
            canvas,
            screenshot_label(overlay.track_id, overlay.state.value),
            (left, max(0, top - 6)),
            cv2.FONT_HERSHEY_SIMPLEX,
            0.5,
            color,
            1,
            cv2.LINE_AA,
        )
    if view.highest_state is FallState.CONFIRMED:
        height, width = canvas.shape[:2]
        cv2.rectangle(
            canvas, (0, 0), (width - 1, height - 1), _COLOR_BGR[StatusColor.CRITICAL], 6
        )
    return canvas


class EventArtifactWriter:
    """Persist one annotated screenshot and one JSONL line per confirmed event."""

    def __init__(
        self,
        event_dir: Path,
        source_id: str,
        clock: Optional[Callable[[], datetime]] = None,
    ) -> None:
        self._event_dir = Path(event_dir)
        self._source_id = source_id
        self._clock = clock or (lambda: datetime.now(timezone.utc))

    def write(
        self, event: FallEvent, image: np.ndarray, view: MonitorViewState
    ) -> AlertRecord:
        created = self._clock().astimezone(timezone.utc)
        day_dir = self._event_dir / created.strftime("%Y%m%d")
        day_dir.mkdir(parents=True, exist_ok=True)
        screenshot_path = day_dir / "{0}.png".format(event.event_id)
        annotated = annotate_frame(image, view)
        if not cv2.imwrite(str(screenshot_path), annotated):
            raise IOError("failed to save screenshot: {0}".format(screenshot_path))
        log_path = day_dir / "events.jsonl"
        record = {
            "event_id": event.event_id,
            "track_id": event.track_id,
            "config_version": event.config_version,
            "source_id": self._source_id,
            "state": event.state.value,
            "confirmed_at_utc": created.isoformat(),
            "suspected_at_monotonic": event.suspected_at_monotonic,
            "confirmed_at_monotonic": event.confirmed_at_monotonic,
            "latency_seconds": event.latency_seconds,
            "screenshot": screenshot_path.relative_to(self._event_dir).as_posix(),
        }
        with log_path.open("a", encoding="utf-8") as stream:
            stream.write(json.dumps(record, ensure_ascii=False) + "\n")
        return AlertRecord(
            event=event,
            screenshot_path=screenshot_path,
            log_path=log_path,
            created_at_utc=record["confirmed_at_utc"],
        )


class AlertDispatcher:
    """De-duplicate confirmed events and fire each side effect exactly once."""

    def __init__(self, writer: EventArtifactWriter, sink: Optional[AlertSink] = None) -> None:
        self._writer = writer
        self._sink = sink or AlertSink()
        self._seen_event_ids: set = set()

    def dispatch(
        self,
        events: Sequence[FallEvent],
        image: Optional[np.ndarray],
        view: MonitorViewState,
    ) -> Sequence[AlertRecord]:
        records = []
        for event in events:
            if event.event_id in self._seen_event_ids or image is None:
                continue
            self._seen_event_ids.add(event.event_id)
            record = self._writer.write(event, image, view)
            self._sink.play_sound(event)
            self._sink.show_popup(record)
            records.append(record)
        return tuple(records)