"""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 try: from PIL import Image, ImageDraw, ImageFont _PIL_AVAILABLE = True except ImportError: # pragma: no cover - optional dependency _PIL_AVAILABLE = False 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), } _STATE_LABEL_ZH: Dict[FallState, str] = { FallState.NORMAL: "正常", FallState.SUSPECT: "疑似", FallState.CONFIRMED: "确认摔倒", FallState.RECOVERING: "恢复中", } # Only CJK-capable fonts, so auto-discovery never renders tofu for Chinese; # absent one, annotation falls back to ASCII. Windows finds Microsoft YaHei. _FONT_CANDIDATES: Tuple[str, ...] = ( "C:/Windows/Fonts/msyh.ttc", "C:/Windows/Fonts/msyhbd.ttc", "C:/Windows/Fonts/simhei.ttf", "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", ) def annotation_label(track_id: str, state: FallState) -> str: """Chinese screenshot label, e.g. '确认摔倒 P-0001'.""" return "{0} {1}".format(_STATE_LABEL_ZH.get(state, getattr(state, "value", state)), track_id) def _find_font(font_path: Optional[str] = None) -> Optional[str]: candidates = [font_path] if font_path else list(_FONT_CANDIDATES) for path in candidates: if path and Path(path).is_file(): return path return None @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, font_path: Optional[str] = None ) -> np.ndarray: """Return a copy of ``image`` with boxes, skeleton, a state border and labels. Labels are Chinese via Pillow when a CJK font is available, otherwise an ASCII fallback via ``cv2.putText`` (which cannot render non-ASCII glyphs). """ 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) 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 _draw_labels(canvas, view, font_path) def _draw_labels( canvas: np.ndarray, view: MonitorViewState, font_path: Optional[str] ) -> np.ndarray: resolved = _find_font(font_path) if _PIL_AVAILABLE and resolved: return _draw_labels_pil(canvas, view, resolved) for overlay in view.people: color = _COLOR_BGR[overlay.color] left = int(round(overlay.box_xyxy[0])) top = int(round(overlay.box_xyxy[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, ) return canvas def _draw_labels_pil( canvas: np.ndarray, view: MonitorViewState, font_path: str ) -> np.ndarray: rgb = Image.fromarray(np.ascontiguousarray(canvas[:, :, ::-1])) draw = ImageDraw.Draw(rgb) font = ImageFont.truetype(font_path, 18) for overlay in view.people: blue, green, red = _COLOR_BGR[overlay.color] left = int(round(overlay.box_xyxy[0])) top = int(round(overlay.box_xyxy[1])) draw.text( (left, max(0, top - 24)), annotation_label(overlay.track_id, overlay.state), fill=(red, green, blue), font=font, ) return np.ascontiguousarray(np.array(rgb)[:, :, ::-1]) 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)