ilaandClaude Opus 4.8
8b33f61112
annotate_frame draws Chinese labels (e.g. '确认摔倒 P-0001') using a CJK font via Pillow when available (Windows uses Microsoft YaHei), and falls back to the ASCII cv2 label when Pillow or a CJK font is missing. 86 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
243 lines
8.4 KiB
Python
243 lines
8.4 KiB
Python
"""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)
|