"""Thin PyQt5 rendering shell for the V1 monitor and settings tabs. This module only renders ``MonitorViewState`` and binds ``SettingsDraft``; it never runs Pose, tracking or fall decisions. All event facts arrive from the ``FallPipeline`` worker in ``v1/app.py``. Because acceptance-critical logic lives in ``v1/view_model.py`` (Qt-free and unit-tested), this file is the only surface that requires a Windows + PyQt5 + display smoke test. """ from typing import Dict, Optional import numpy as np from PyQt5 import QtCore, QtGui, QtWidgets from v1.alerts import AlertRecord, AlertSink from v1.view_model import ( FIELD_BOUNDS, DraftValidationError, MonitorViewState, SettingsDraft, StatusColor, ) COLOR_HEX: Dict[StatusColor, str] = { StatusColor.SUCCESS: "#15803D", StatusColor.CAUTION: "#B45309", StatusColor.CRITICAL: "#C62828", StatusColor.OFFLINE: "#64748B", } # Light Windows theme tokens (see docs/ui/silver-pose-ui-ux-spec.md). _QSS = """ QWidget { background: #EAF1F8; color: #1E293B; font-family: 'Segoe UI Variable','Microsoft YaHei UI',sans-serif; } QTabWidget::pane { background: #FFFFFF; border: 1px solid #C8D4E3; } QTabBar::tab { padding: 8px 20px; background: #FFFFFF; color: #52657C; } QTabBar::tab:selected { color: #1E293B; border-bottom: 2px solid #2563EB; } QGroupBox { background: #FFFFFF; border: 1px solid #C8D4E3; border-radius: 8px; margin-top: 12px; } QPushButton { background: #FFFFFF; border: 1px solid #C8D4E3; border-radius: 8px; padding: 8px 14px; min-height: 40px; } QPushButton:focus { border: 2px solid #2563EB; } QLabel[role="primary-action"] { } """ _FIELD_LABELS = { "keypoint_confidence_threshold": "关键点置信度阈值", "suspect_window_seconds": "快速下移到水平的最大间隔（秒）", "confirm_window_seconds": "确认窗口（秒，1–3）", "recovery_window_seconds": "恢复稳定窗口（秒）", "cooldown_seconds": "确认后恢复判断冷却（秒）", "model_confidence_threshold": "模型检测置信度阈值", } _FIELD_STEP = { "keypoint_confidence_threshold": 0.05, "suspect_window_seconds": 0.1, "confirm_window_seconds": 0.1, "recovery_window_seconds": 0.5, "cooldown_seconds": 1.0, "model_confidence_threshold": 0.05, } def bgr_to_qimage(image: np.ndarray) -> QtGui.QImage: """Convert an OpenCV BGR frame to a QImage owning a contiguous copy.""" rgb = np.ascontiguousarray(image[:, :, ::-1]) height, width, _ = rgb.shape return QtGui.QImage( rgb.data, width, height, 3 * width, QtGui.QImage.Format_RGB888 ).copy() class QtAlertSink(AlertSink): """Desktop sound + one non-blocking popup per confirmed event (Windows smoke).""" def __init__(self, parent: Optional[QtWidgets.QWidget] = None) -> None: self._parent = parent def play_sound(self, event) -> None: QtWidgets.QApplication.beep() def show_popup(self, record: AlertRecord) -> None: box = QtWidgets.QMessageBox(self._parent) box.setIcon(QtWidgets.QMessageBox.Warning) box.setWindowTitle("确认摔倒") box.setText( "人员 {0} 确认摔倒\n确认延迟 {1:.2f} 秒\n截图：{2}".format( record.event.track_id, record.event.latency_seconds, record.screenshot_path.name, ) ) box.setStandardButtons(QtWidgets.QMessageBox.Ok) box.button(QtWidgets.QMessageBox.Ok).setText("我已知晓") box.setModal(False) box.show() class VideoView(QtWidgets.QWidget): """Paint the latest frame plus box/skeleton/id/state overlays.""" def __init__(self, parent: Optional[QtWidgets.QWidget] = None) -> None: super().__init__(parent) self.setMinimumSize(480, 270) self._image: Optional[QtGui.QImage] = None self._view: Optional[MonitorViewState] = None def update_frame(self, image: Optional[np.ndarray], view: MonitorViewState) -> None: self._image = bgr_to_qimage(image) if image is not None else None self._view = view self.update() def paintEvent(self, _event: QtGui.QPaintEvent) -> None: painter = QtGui.QPainter(self) painter.fillRect(self.rect(), QtGui.QColor("#0B1220")) if self._image is None: painter.setPen(QtGui.QColor("#EAF1F8")) text = self._view.status_text if self._view else "无画面" painter.drawText(self.rect(), QtCore.Qt.AlignCenter, text) painter.end() return target = self._fitted_rect(self._image) painter.drawImage(target, self._image) if self._view is not None: self._draw_overlays(painter, target, self._image) self._draw_diagnostics(painter) painter.end() def _draw_diagnostics(self, painter: QtGui.QPainter) -> None: lines = list(self._view.diagnostics) if not lines: return painter.setFont(QtGui.QFont("Consolas", 9)) metrics = painter.fontMetrics() line_h = metrics.height() box_w = max(metrics.width(line) for line in lines) + 12 box_h = line_h * len(lines) + 8 painter.fillRect(QtCore.QRect(6, 6, box_w, box_h), QtGui.QColor(0, 0, 0, 160)) painter.setPen(QtGui.QColor("#9BE9A8")) y = 6 + line_h for line in lines: painter.drawText(12, y, line) y += line_h def _fitted_rect(self, image: QtGui.QImage) -> QtCore.QRect: widget = self.rect() scale = min(widget.width() / image.width(), widget.height() / image.height()) width = int(image.width() * scale) height = int(image.height() * scale) left = widget.left() + (widget.width() - width) // 2 top = widget.top() + (widget.height() - height) // 2 return QtCore.QRect(left, top, width, height) def _draw_overlays( self, painter: QtGui.QPainter, target: QtCore.QRect, image: QtGui.QImage ) -> None: scale_x = target.width() / image.width() scale_y = target.height() / image.height() def to_widget(x: float, y: float) -> QtCore.QPointF: return QtCore.QPointF(target.left() + x * scale_x, target.top() + y * scale_y) for overlay in self._view.people: color = QtGui.QColor(COLOR_HEX[overlay.color]) painter.setPen(QtGui.QPen(color, 3)) left, top, right, bottom = overlay.box_xyxy painter.drawRect( QtCore.QRectF(to_widget(left, top), to_widget(right, bottom)) ) painter.setPen(QtGui.QPen(color, 2)) for start, end in overlay.skeleton_segments: painter.drawLine(to_widget(start.x, start.y), to_widget(end.x, end.y)) for point in overlay.keypoints: if point is not None: painter.drawEllipse(to_widget(point.x, point.y), 2.5, 2.5) painter.setPen(QtGui.QColor("#FFFFFF")) painter.drawText(to_widget(left, top - 6), overlay.label) class MonitorTab(QtWidgets.QWidget): """Live view: video, connection status, recent events, start/stop.""" start_requested = QtCore.pyqtSignal() stop_requested = QtCore.pyqtSignal() def __init__(self, parent: Optional[QtWidgets.QWidget] = None) -> None: super().__init__(parent) layout = QtWidgets.QVBoxLayout(self) self._status = QtWidgets.QLabel("未开始") self._status.setMinimumHeight(24) self.video = VideoView() self.events = QtWidgets.QListWidget() self.events.setMaximumHeight(140) self.start_button = QtWidgets.QPushButton("开始监控") self.stop_button = QtWidgets.QPushButton("停止监控") self.stop_button.setEnabled(False) self.start_button.clicked.connect(self.start_requested.emit) self.stop_button.clicked.connect(self.stop_requested.emit) controls = QtWidgets.QHBoxLayout() controls.addWidget(self.start_button) controls.addWidget(self.stop_button) controls.addStretch(1) layout.addWidget(self._status) layout.addWidget(self.video, 1) layout.addWidget(QtWidgets.QLabel("最近事件")) layout.addWidget(self.events) layout.addLayout(controls) def render_view(self, image, view: MonitorViewState) -> None: self.video.update_frame(image, view) color = COLOR_HEX[view.status_color] self._status.setText("连接状态：{0}".format(view.status_text)) self._status.setStyleSheet("color: {0}; font-weight: 600;".format(color)) for event in view.events: self.events.insertItem( 0, "{0} · {1} · 延迟 {2:.2f}s".format( event.event_id, event.track_id, event.latency_seconds ), ) def set_running(self, running: bool) -> None: self.start_button.setEnabled(not running) self.stop_button.setEnabled(running) class ConnectionTestWorker(QtCore.QThread): """Build the RTSP URL and probe one frame off the GUI thread.""" probed = QtCore.pyqtSignal(bool, str, object) def __init__(self, host, port, channel, username, password) -> None: super().__init__() self._args = (host, port, channel, username, password) def run(self) -> None: from v1.camera import build_rtsp_url, probe_stream host, port, channel, username, password = self._args try: url = build_rtsp_url(host, port, username, password, channel) except Exception as exc: # noqa: BLE001 - report any build error to the UI self.probed.emit(False, "参数错误：{0}".format(exc), None) return result = probe_stream(url, attempts=40) self.probed.emit(result.ok, result.message, result.frame) class SettingsTab(QtWidgets.QWidget): """Edit a non-sensitive draft; changes apply only on the next start.""" def __init__( self, draft: SettingsDraft, env_ready: bool, model_summary: str, config_path: Optional[str] = None, camera: Optional[Dict] = None, parent: Optional[QtWidgets.QWidget] = None, ) -> None: super().__init__(parent) self._draft = draft self._config_path = config_path self._spins: Dict[str, QtWidgets.QDoubleSpinBox] = {} self._probe_worker: Optional[ConnectionTestWorker] = None layout = QtWidgets.QVBoxLayout(self) source_box = QtWidgets.QGroupBox("来源与模型（只读）") source_form = QtWidgets.QFormLayout(source_box) source_form.addRow("RTSP 环境变量", QtWidgets.QLabel("已就绪" if env_ready else "未就绪")) source_form.addRow("模型", QtWidgets.QLabel(model_summary)) layout.addWidget(source_box) layout.addWidget(self._build_camera_box(camera or {})) params_box = QtWidgets.QGroupBox("事件参数草稿") form = QtWidgets.QFormLayout(params_box) values = draft.draft_values for key, (low, high) in FIELD_BOUNDS.items(): spin = QtWidgets.QDoubleSpinBox() spin.setRange(low, high) spin.setSingleStep(_FIELD_STEP[key]) spin.setDecimals(2) spin.setValue(values[key]) spin.valueChanged.connect(lambda value, name=key: self._on_edit(name, value)) self._spins[key] = spin form.addRow(_FIELD_LABELS[key], spin) layout.addWidget(params_box) self.status = QtWidgets.QLabel("") self.save_button = QtWidgets.QPushButton("保存（下次启动生效）") self.reset_button = QtWidgets.QPushButton("重置") self.save_button.clicked.connect(self._on_save) self.reset_button.clicked.connect(self._on_reset) actions = QtWidgets.QHBoxLayout() actions.addWidget(self.save_button) actions.addWidget(self.reset_button) actions.addStretch(1) layout.addWidget(self.status) layout.addLayout(actions) layout.addStretch(1) def _on_edit(self, key: str, value: float) -> None: try: self._draft.edit(key, value) except DraftValidationError as exc: self.status.setText("校验失败：{0}".format(exc)) return self.status.setText("已修改，未保存" if self._draft.is_dirty else "") def _on_save(self) -> None: self.status.setText(self._draft.save()) def _on_reset(self) -> None: self._draft.discard() for key, spin in self._spins.items(): spin.blockSignals(True) spin.setValue(self._draft.draft_values[key]) spin.blockSignals(False) self.status.setText("已重置为已保存值") def _build_camera_box(self, camera: Dict) -> QtWidgets.QGroupBox: box = QtWidgets.QGroupBox("摄像头连接（保存至本地未跟踪配置）") outer = QtWidgets.QHBoxLayout(box) form = QtWidgets.QFormLayout() self.host_edit = QtWidgets.QLineEdit(str(camera.get("host", ""))) self.port_spin = QtWidgets.QSpinBox() self.port_spin.setRange(1, 65535) self.port_spin.setValue(int(camera.get("port", 554))) self.channel_combo = QtWidgets.QComboBox() self.channel_combo.addItem("101 主码流", "101") self.channel_combo.addItem("102 子码流", "102") self.channel_combo.setCurrentIndex(1 if str(camera.get("channel", "102")) == "102" else 0) self.user_edit = QtWidgets.QLineEdit(str(camera.get("username", ""))) self.password_edit = QtWidgets.QLineEdit(str(camera.get("password", ""))) self.password_edit.setEchoMode(QtWidgets.QLineEdit.Password) self.transport_combo = QtWidgets.QComboBox() self.transport_combo.addItem("TCP", "tcp") self.transport_combo.addItem("UDP", "udp") self.transport_combo.setCurrentIndex(1 if str(camera.get("transport", "tcp")) == "udp" else 0) self.timeout_spin = QtWidgets.QDoubleSpinBox() self.timeout_spin.setRange(0.0, 60.0) self.timeout_spin.setSingleStep(1.0) self.timeout_spin.setDecimals(1) self.timeout_spin.setValue(float(camera.get("timeout_seconds", 5.0))) self.low_latency_check = QtWidgets.QCheckBox("低延迟（丢弃缓冲）") self.low_latency_check.setChecked(bool(camera.get("low_latency", True))) form.addRow("IP / 主机", self.host_edit) form.addRow("端口", self.port_spin) form.addRow("通道", self.channel_combo) form.addRow("账号", self.user_edit) form.addRow("密码", self.password_edit) form.addRow("传输协议", self.transport_combo) form.addRow("连接超时（秒）", self.timeout_spin) form.addRow("", self.low_latency_check) self.test_button = QtWidgets.QPushButton("测试连接") self.save_camera_button = QtWidgets.QPushButton("保存连接到本地配置") self.test_button.clicked.connect(self._on_test_connection) self.save_camera_button.clicked.connect(self._on_save_camera) self.camera_status = QtWidgets.QLabel("") buttons = QtWidgets.QHBoxLayout() buttons.addWidget(self.test_button) buttons.addWidget(self.save_camera_button) buttons.addStretch(1) form.addRow(buttons) form.addRow(self.camera_status) self.preview = QtWidgets.QLabel("预览区域") self.preview.setMinimumSize(320, 180) self.preview.setAlignment(QtCore.Qt.AlignCenter) self.preview.setStyleSheet("background:#0B1220;color:#EAF1F8;") outer.addLayout(form, 1) outer.addWidget(self.preview, 1) return box def _on_test_connection(self) -> None: if self._probe_worker is not None: return self.camera_status.setText("正在测试连接…") self.test_button.setEnabled(False) worker = ConnectionTestWorker( self.host_edit.text().strip(), self.port_spin.value(), self.channel_combo.currentData(), self.user_edit.text().strip(), self.password_edit.text(), ) worker.probed.connect(self._on_probe_done) worker.finished.connect(self._on_probe_finished) self._probe_worker = worker worker.start() def _on_probe_done(self, ok: bool, message: str, frame) -> None: self.camera_status.setText(message) if ok and frame is not None: image = bgr_to_qimage(frame) pixmap = QtGui.QPixmap.fromImage(image).scaled( self.preview.size(), QtCore.Qt.KeepAspectRatio, QtCore.Qt.SmoothTransformation ) self.preview.setPixmap(pixmap) def _on_probe_finished(self) -> None: self._probe_worker = None self.test_button.setEnabled(True) def _on_save_camera(self) -> None: if not self._config_path: self.camera_status.setText("未指定本地配置路径，无法保存") return from v1.config import write_local_camera_source try: write_local_camera_source( self._config_path, source_id=self.host_edit.text().strip() or "ip-camera", host=self.host_edit.text().strip(), port=self.port_spin.value(), channel=self.channel_combo.currentData(), username=self.user_edit.text().strip(), password=self.password_edit.text(), transport=self.transport_combo.currentData(), timeout_seconds=self.timeout_spin.value(), low_latency=self.low_latency_check.isChecked(), ) except Exception as exc: # noqa: BLE001 - surface any write error to the UI self.camera_status.setText("保存失败：{0}".format(exc)) return self.camera_status.setText("已保存到本地配置，将在下次开始监控时生效") class MainWindow(QtWidgets.QMainWindow): """Top dual-tab window: 实时监控 / 设置.""" def __init__( self, draft: SettingsDraft, env_ready: bool, model_summary: str, source_name: str, config_path: Optional[str] = None, camera: Optional[Dict] = None, ) -> None: super().__init__() self.setWindowTitle("Silver Pose · {0}".format(source_name)) self.resize(1280, 800) self.setStyleSheet(_QSS) self.monitor = MonitorTab() self.settings = SettingsTab(draft, env_ready, model_summary, config_path, camera) tabs = QtWidgets.QTabWidget() tabs.setTabPosition(QtWidgets.QTabWidget.North) tabs.addTab(self.monitor, "实时监控") tabs.addTab(self.settings, "设置") self.setCentralWidget(tabs)