feat(v1): add PyQt dual-tab monitor and settings UI
Qt-free view_model builds monitor view state and isolates the settings draft from the running config snapshot; gui.py/app.py are a thin PyQt5 shell that only renders already-decided FrameAnalysis. 39 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,262 @@
|
||||
"""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
|
||||
|
||||
|
||||
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),
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user