silver_pose/v1/view_model.py

"""Pure-Python view state for the V1 monitor/settings UI (no Qt dependency).

The GUI layer only renders these structures; it never runs Pose, tracking or
event decisions. All event facts arrive already computed from ``FallPipeline``.
This keeps the acceptance-critical UI logic testable without PyQt5 or a display.
"""

import hashlib
import json
from dataclasses import dataclass
from enum import Enum
from typing import Dict, Optional, Sequence, Tuple

from v1.fall_state import FallState
from v1.video_source import SourceStatus


# COCO-17 skeleton edges (keypoint index pairs) used to draw the person overlay.
SKELETON_EDGES: Tuple[Tuple[int, int], ...] = (
    (5, 7),
    (7, 9),
    (6, 8),
    (8, 10),
    (5, 6),
    (5, 11),
    (6, 12),
    (11, 12),
    (11, 13),
    (13, 15),
    (12, 14),
    (14, 16),
    (0, 5),
    (0, 6),
)


class StatusColor(str, Enum):
    """Semantic color tokens from the UI spec. Red (CRITICAL) is CONFIRMED-only."""

    SUCCESS = "success"   # NORMAL / online   -> #15803D
    CAUTION = "caution"   # SUSPECT / notice  -> #B45309
    CRITICAL = "critical"  # CONFIRMED fall    -> #C62828
    OFFLINE = "offline"   # disconnected       -> #64748B


STATE_COLOR: Dict[FallState, StatusColor] = {
    FallState.NORMAL: StatusColor.SUCCESS,
    FallState.SUSPECT: StatusColor.CAUTION,
    FallState.CONFIRMED: StatusColor.CRITICAL,
    FallState.RECOVERING: StatusColor.CAUTION,
}

_STATE_SEVERITY: Dict[FallState, int] = {
    FallState.NORMAL: 0,
    FallState.RECOVERING: 1,
    FallState.SUSPECT: 2,
    FallState.CONFIRMED: 3,
}

CONNECTION_TEXT: Dict[SourceStatus, str] = {
    SourceStatus.CONNECTED: "在线",
    SourceStatus.RETRYING: "正在重连…",
    SourceStatus.ERROR: "连接错误",
    SourceStatus.EOF: "录像结束",
    SourceStatus.CLOSED: "已停止",
}


@dataclass(frozen=True)
class Point:
    x: float
    y: float


@dataclass(frozen=True)
class PersonOverlay:
    track_id: str
    state: FallState
    color: StatusColor
    box_xyxy: Tuple[float, float, float, float]
    keypoints: Tuple[Optional[Point], ...]
    skeleton_segments: Tuple[Tuple[Point, Point], ...]
    label: str


@dataclass(frozen=True)
class EventBadge:
    event_id: str
    track_id: str
    latency_seconds: float


@dataclass(frozen=True)
class MonitorViewState:
    connected: bool
    status_text: str
    status_color: StatusColor
    has_frame: bool
    people: Tuple[PersonOverlay, ...]
    events: Tuple[EventBadge, ...]
    highest_state: FallState
    diagnostics: Tuple[str, ...] = ()


def person_diagnostic_line(person) -> str:
    """Compact per-person view of the fall decision, to locate where it stops.

    Shows the quality gate (``acc``), horizontal geometry (``horiz``/``ang``),
    the rapid-drop signal (``rapid``), the policy candidate (``cand``) and the
    state. When the pose is rejected it shows the reason instead.
    """

    track_id = person.tracked_pose.track_id
    state = person.state.value
    pose_evidence = person.pose_evidence
    if pose_evidence is None or not pose_evidence.accepted:
        reason = getattr(pose_evidence, "reason", "no_evidence")
        return "{0} {1} acc=0 {2}".format(track_id, state, reason)
    angle = pose_evidence.horizontal_angle_degrees
    angle_text = "{0:.0f}deg".format(angle) if angle is not None else "-"
    candidate = int(getattr(person.evidence, "is_fall_candidate", False))
    return "{0} {1} acc=1 horiz={2} ang={3} rapid={4} cand={5}".format(
        track_id,
        state,
        int(pose_evidence.horizontal_pose),
        angle_text,
        int(pose_evidence.rapid_vertical_change),
        candidate,
    )


def build_monitor_view(analysis, keypoint_min_confidence: float = 0.4) -> MonitorViewState:
    """Turn one ``FrameAnalysis`` into render-only instructions.

    Non-connected frames never carry people or a fall label, matching the rule
    that connection problems must not surface as fall alarms.
    """

    if not 0.0 <= keypoint_min_confidence <= 1.0:
        raise ValueError("keypoint_min_confidence must be between 0 and 1")
    packet = analysis.packet
    status = packet.status
    connected = status is SourceStatus.CONNECTED
    overlays = tuple(
        _person_overlay(person, keypoint_min_confidence) for person in analysis.people
    )
    events = tuple(
        EventBadge(event.event_id, event.track_id, event.latency_seconds)
        for event in analysis.events
    )
    return MonitorViewState(
        connected=connected,
        status_text=CONNECTION_TEXT.get(status, str(getattr(status, "value", status))),
        status_color=StatusColor.SUCCESS if connected else StatusColor.OFFLINE,
        has_frame=packet.image is not None,
        people=overlays,
        events=events,
        highest_state=_highest_state(overlays),
        diagnostics=tuple(person_diagnostic_line(person) for person in analysis.people),
    )


def _person_overlay(person, min_confidence: float) -> PersonOverlay:
    pose = person.tracked_pose.pose
    state = person.state
    points = tuple(
        Point(keypoint.x, keypoint.y) if keypoint.confidence >= min_confidence else None
        for keypoint in pose.keypoints
    )
    segments = tuple(
        (points[start], points[end])
        for start, end in SKELETON_EDGES
        if points[start] is not None and points[end] is not None
    )
    return PersonOverlay(
        track_id=person.tracked_pose.track_id,
        state=state,
        color=STATE_COLOR[state],
        box_xyxy=pose.box_xyxy,
        keypoints=points,
        skeleton_segments=segments,
        label="{0} · {1}".format(person.tracked_pose.track_id, state.value),
    )


def _highest_state(overlays: Sequence[PersonOverlay]) -> FallState:
    highest = FallState.NORMAL
    for overlay in overlays:
        if _STATE_SEVERITY[overlay.state] > _STATE_SEVERITY[highest]:
            highest = overlay.state
    return highest


# --- Settings draft with an explicit next-start apply lifecycle --------------

FIELD_BOUNDS: Dict[str, Tuple[float, float]] = {
    "keypoint_confidence_threshold": (0.0, 1.0),
    "suspect_window_seconds": (0.0, 30.0),
    "confirm_window_seconds": (1.0, 3.0),
    "recovery_window_seconds": (0.0, 300.0),
    "cooldown_seconds": (0.0, 3600.0),
    "model_confidence_threshold": (0.0, 1.0),
}


class DraftValidationError(ValueError):
    """Raised when a settings draft value is outside its allowed range."""


def config_version(values: Dict[str, float]) -> str:
    canonical = json.dumps(
        {key: float(values[key]) for key in FIELD_BOUNDS},
        sort_keys=True,
        separators=(",", ":"),
    )
    return "cfg-" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12]


class SettingsDraft:
    """Hold three isolated copies: running snapshot, saved, and editable draft.

    Editing changes only the draft. Saving copies the draft into ``saved`` but
    does not touch the running snapshot. ``start_monitoring`` promotes the saved
    values into a new immutable running snapshot; this is the only moment the
    running configuration version changes.
    """

    def __init__(self, initial: Dict[str, float]) -> None:
        missing = set(FIELD_BOUNDS) - set(initial)
        if missing:
            raise ValueError("missing settings fields: {0}".format(sorted(missing)))
        self._running = {key: float(initial[key]) for key in FIELD_BOUNDS}
        for key, value in self._running.items():
            low, high = FIELD_BOUNDS[key]
            if not low <= value <= high:
                raise DraftValidationError(
                    "{0} must be between {1} and {2}".format(key, low, high)
                )
        self._saved = dict(self._running)
        self._draft = dict(self._running)

    def edit(self, key: str, value) -> None:
        if key not in FIELD_BOUNDS:
            raise DraftValidationError("unknown settings field: {0}".format(key))
        try:
            parsed = float(value)
        except (TypeError, ValueError):
            raise DraftValidationError("{0} must be numeric".format(key))
        low, high = FIELD_BOUNDS[key]
        if not low <= parsed <= high:
            raise DraftValidationError(
                "{0} must be between {1} and {2}".format(key, low, high)
            )
        self._draft[key] = parsed

    @property
    def draft_values(self) -> Dict[str, float]:
        return dict(self._draft)

    @property
    def saved_values(self) -> Dict[str, float]:
        return dict(self._saved)

    @property
    def running_values(self) -> Dict[str, float]:
        return dict(self._running)

    @property
    def is_dirty(self) -> bool:
        return self._draft != self._saved

    @property
    def has_pending_for_next_start(self) -> bool:
        return self._saved != self._running

    @property
    def running_version(self) -> str:
        return config_version(self._running)

    def save(self) -> str:
        self._saved = dict(self._draft)
        if self._saved == self._running:
            return "已保存，与当前运行配置一致"
        return "已保存，将在下次开始监控时生效"

    def discard(self) -> None:
        self._draft = dict(self._saved)

    def start_monitoring(self) -> str:
        self._running = dict(self._saved)
        return self.running_version