"""Unified Qt dashboard for the finger tactile example.""" from __future__ import annotations import sys import time from PyQt5.QtCore import Qt, QTimer from PyQt5.QtGui import QIcon from PyQt5.QtWidgets import ( QApplication, QFileDialog, QMainWindow, QSplitter, QVBoxLayout, QWidget, ) from pipeline.config import PipelineConfig from pipeline.service import FingerPipeline from app_paths import resource_path from .layout import ( DashboardLayout, LAYOUT, PANEL_SETUP, PanelSetup, top_row_splitter_sizes, ) from .mesh_view_state import MeshViewState from .styles import STYLE from .widgets import ( ArrowPanelWidget, ForceChartWidget, MeshViewportWidget, SettingsPanelWidget, TeeStream, WidgetLogHandler, make_log_widget, ) class FingerDashboardWindow(QMainWindow): """Single-window finger demo: 3D mesh | UV arrows | force charts.""" def __init__( self, config: PipelineConfig, *, layout: DashboardLayout = LAYOUT, log_file=None, ): super().__init__() self.config = config self.layout = layout self.pipeline = FingerPipeline(config) self._running = False self._frame_index = 0 self._wall_fps = 0.0 self._wall_fps_alpha = 0.15 self.setWindowTitle(layout.window_title) self.setStyleSheet(STYLE) logo_path = resource_path("logo.png") if logo_path.is_file(): self.setWindowIcon(QIcon(str(logo_path))) self._setup_ui() self._wire_logging(log_file) self.timer = QTimer() self.timer.timeout.connect(self._on_timer) def _setup_ui(self) -> None: layout = self.layout central = QWidget() self.setCentralWidget(central) root_layout = QVBoxLayout(central) root_layout.setContentsMargins(layout.root_margin_px, layout.root_margin_px, layout.root_margin_px, layout.root_margin_px) root_layout.setSpacing(layout.root_spacing_px) setup = layout.panel_setup self._log_panel_visible = setup.log_panel.enabled self.mesh_state = MeshViewState.from_layout(layout) self.mesh_viewport = MeshViewportWidget(layout=layout, mesh_state=self.mesh_state) arrows_slot = layout.slot("arrows") self.arrow_panel: ArrowPanelWidget | None = None if arrows_slot is not None and arrows_slot.enabled: self.arrow_panel = ArrowPanelWidget( flow_threshold=layout.arrow_flow_threshold, background=layout.arrow_background, layout=layout, ) self.force_charts = ForceChartWidget(layout=layout, lowpass_cutoff_hz=self.config.lowpass_cutoff_hz) self.log_widget = make_log_widget(layout.log_config) self.settings_panel = SettingsPanelWidget( layout=layout, mesh_state=self.mesh_state, show_log=self._log_panel_visible, ) self.settings_panel.set_video_source(str(self.config.video)) self.settings_panel.set_config_path(self.config.config) self.settings_panel.set_calibration_path(self.config.calib or "") self.settings_panel.set_finger_export_dir(self.config.export_dir) self.settings_panel.log_visibility_changed.connect(self._set_log_panel_visible) self.settings_panel.free_rotation_changed.connect(self.mesh_viewport.set_free_rotation) if self.arrow_panel is not None: self.settings_panel.arrow_background_changed.connect(self.arrow_panel.set_background) self.settings_panel.view_reset_requested.connect(self.mesh_viewport.reset_view) self.settings_panel.browse_video_requested.connect(self._select_camera) self.settings_panel.browse_config_requested.connect(self._select_config) self.settings_panel.browse_finger_assets_requested.connect(self._select_finger_assets) self.settings_panel.btn_play_pause.clicked.connect(self._on_play_pause_clicked) self.settings_panel.btn_reset_tracker.clicked.connect(self._on_reset) self.settings_panel.set_play_pause_state("run") self._panel_widgets: dict[str, QWidget] = {} self._panel_widgets["mesh"] = self.mesh_viewport if self.arrow_panel is not None: self._panel_widgets["arrows"] = self.arrow_panel self._panel_widgets["charts"] = self.force_charts.widget self._panel_widgets["settings"] = self.settings_panel self._panel_widgets["log"] = self.log_widget self.charts_column = QSplitter(Qt.Vertical) self.charts_column.addWidget(self.force_charts.widget) if setup.log_panel.enabled: self.charts_column.addWidget(self.log_widget) self.log_widget.setVisible(self._log_panel_visible) self.charts_column.setStretchFactor(0, setup.charts_log_stretch) self.charts_column.setStretchFactor(1, setup.log_stretch) self.charts_column.setChildrenCollapsible(False) h_splitter = QSplitter(Qt.Horizontal) for slot in setup.top_row.slots: if not slot.enabled: continue if slot.id == "charts": h_splitter.addWidget(self.charts_column) else: h_splitter.addWidget(self._panel_widgets[slot.id]) h_splitter.setStretchFactor(h_splitter.count() - 1, slot.stretch) h_splitter.setChildrenCollapsible(setup.top_row.collapsible) self.h_splitter = h_splitter root_layout.addWidget(h_splitter) self.resize(layout.window_width, layout.window_height) QTimer.singleShot(0, self._apply_panel_layout) def _set_log_panel_visible(self, visible: bool) -> None: self._log_panel_visible = bool(visible) self.log_widget.setVisible(self._log_panel_visible) QTimer.singleShot(0, self._apply_panel_layout) def _apply_panel_layout(self) -> None: setup = self.layout.panel_setup column_height = self.layout.charts_column_height_px() charts_slot = self.layout.top_slot("charts") log_slot = setup.log_panel for slot in setup.top_row.slots: if not slot.enabled: continue if slot.id in ("mesh", "arrows"): self._panel_widgets[slot.id].set_fixed_size(slot.width_px, column_height) elif slot.id == "settings": settings_height = slot.height_px if slot.height_px > 0 else column_height self.settings_panel.set_panel_size(slot.width_px, settings_height) if ( self._log_panel_visible and log_slot.enabled and log_slot.height_px > 0 ): self.log_widget.show() self.log_widget.setFixedHeight(log_slot.height_px) self.force_charts.set_panel_size(charts_slot.width_px, charts_slot.height_px) self.charts_column.setSizes([charts_slot.height_px, log_slot.height_px]) else: self.log_widget.hide() self.force_charts.set_panel_size(charts_slot.width_px, column_height) self.charts_column.setSizes([column_height, 0]) self.h_splitter.setSizes(list(top_row_splitter_sizes(self.layout))) def resizeEvent(self, event): super().resizeEvent(event) QTimer.singleShot(0, self._apply_panel_layout) def showEvent(self, event): super().showEvent(event) QTimer.singleShot(0, self._apply_panel_layout) def _wire_logging(self, log_file) -> None: streams = [] if log_file is not None: streams.append(log_file) self.log_handler = WidgetLogHandler(self.log_widget) streams.append(self.log_handler) sys.stdout = TeeStream(*streams) sys.stderr = sys.stdout def _set_play_pause_button(self, state: str) -> None: self.settings_panel.set_play_pause_state(state) def _on_play_pause_clicked(self) -> None: if self._running: self._on_stop() else: self._on_start() def _on_start(self) -> None: if self._running: return self._set_play_pause_button("starting") self.settings_panel.set_finger_export_dir_enabled(False) QApplication.processEvents() self.config.video = self.settings_panel.video_source() self.config.config = self.settings_panel.config_path() cal = self.settings_panel.calibration_path() self.config.calib = cal or None self.config.export_dir = self.settings_panel.finger_export_dir() try: self.pipeline = FingerPipeline(self.config) self.pipeline.open() self.mesh_viewport.configure(self.pipeline) self.force_charts.reset() QTimer.singleShot(0, self._apply_panel_layout) except Exception as exc: self.mesh_viewport.close_renderer() self._set_play_pause_button("run") self.settings_panel.set_finger_export_dir_enabled(True) print(f"启动失败: {exc}") return self._running = True self._frame_index = 0 self._wall_fps = 0.0 self.timer.start(self.layout.timer_interval_ms) self._set_play_pause_button("pause") self.settings_panel.btn_reset_tracker.setEnabled(True) print( f"▶ 启动 — 摄像机={self.config.video}, 配置={self.config.config}, " f"标定={self.config.calib or '(none)'}, 资产={self.config.export_dir}" ) def _on_stop(self) -> None: if not self._running: return self._running = False self.timer.stop() self.log_handler.clear_status_line() self._set_play_pause_button("starting") QApplication.processEvents() self.mesh_viewport.close_renderer() self.pipeline.close() self._set_play_pause_button("run") self.settings_panel.btn_reset_tracker.setEnabled(False) self.settings_panel.set_finger_export_dir_enabled(True) def _on_reset(self) -> None: if not self._running: print("请先启动传感器后再重置追踪器。") return self.pipeline.reset() print("追踪器已重置。") def _on_timer(self) -> None: if not self._running: return frame_t0 = time.perf_counter() frame, timings = self.pipeline.tick_timed() if frame is None: print("采集结束或读取失败。") self._on_stop() return self._frame_index += 1 layout = self.layout profile = self.pipeline.profiler mesh_ms = 0.0 if layout.mesh_render_stride <= 1 or self._frame_index % layout.mesh_render_stride == 0: mesh_ms = self.mesh_viewport.update_frame(frame) if profile is not None and mesh_ms > 0: profile.record("mesh", mesh_ms) arrows_ms = 0.0 if self.arrow_panel is not None: t0 = time.perf_counter() self.arrow_panel.update_frame(frame) arrows_ms = (time.perf_counter() - t0) * 1000.0 if profile is not None: profile.record("arrows", arrows_ms) charts_ms = 0.0 if layout.chart_update_stride <= 1 or self._frame_index % layout.chart_update_stride == 0: t0 = time.perf_counter() self.force_charts.update_frame(frame) charts_ms = (time.perf_counter() - t0) * 1000.0 if profile is not None: profile.record("charts", charts_ms) total_ms = (time.perf_counter() - frame_t0) * 1000.0 instant_fps = 1000.0 / total_ms if total_ms > 0 else 0.0 if self._wall_fps <= 0: self._wall_fps = instant_fps else: self._wall_fps = ( self._wall_fps_alpha * instant_fps + (1.0 - self._wall_fps_alpha) * self._wall_fps ) if profile is not None: profile.record("total", total_ms) self.pipeline.record_total_frame(total_ms) flow_mag = 0.0 if frame.flow.size: flow_mag = float((frame.flow[:, :, 0] ** 2 + frame.flow[:, :, 1] ** 2).mean() ** 0.5) log_cfg = layout.log_config if log_cfg.status_enabled: self.log_handler.set_status_line( log_cfg.format_status( fps=frame.fps, fnormal=frame.fnormal, fshearx=frame.fshearx, fsheary=frame.fsheary, flow_rms=flow_mag, mask_coverage=frame.mask_coverage, wall_fps=self._wall_fps, ) ) def _select_camera(self) -> None: path, _ = QFileDialog.getOpenFileName( self, "选择视频文件", "data", "视频 (*.mp4 *.avi *.mov);;所有文件 (*)", ) if path: self.settings_panel.set_video_source(path) def _select_config(self) -> None: path, _ = QFileDialog.getOpenFileName( self, "选择配置文件", "config", "JSON (*.json);;所有文件 (*)", ) if path: self.settings_panel.set_config_path(path) def _select_finger_assets(self) -> None: path = QFileDialog.getExistingDirectory( self, "选择 finger 资产目录", self.settings_panel.finger_export_dir() or "assets/finger", ) if path: self.settings_panel.set_finger_export_dir(path) def keyPressEvent(self, event): key = event.key() if key == Qt.Key_Escape: if self.isFullScreen(): self.showNormal() else: self.showFullScreen() elif key == Qt.Key_Q: print("\n正在退出...") self.close() elif key == Qt.Key_S: if self.pipeline.start_recording(): print("开始录像。") elif key == Qt.Key_E: self.pipeline.stop_recording() print("录像结束。") elif key == Qt.Key_R: self._on_reset() else: super().keyPressEvent(event) def closeEvent(self, event): self.timer.stop() self.mesh_viewport.close_renderer() self.pipeline.close() event.accept() def run_dashboard(config: PipelineConfig, *, layout: DashboardLayout = LAYOUT) -> int: """Launch the unified finger dashboard and block until the window closes.""" from .app import get_or_create_qapp print("\n按键说明：Q 退出 | R 重置追踪器 | S 开始录像 | E 结束录像 | ESC 全屏") if config.profile: print(f"Profiling enabled (report every {config.profile_every} frames).") print("Log FPS= is end-to-end dashboard rate; sdk= is Sensor internal rate.") app = get_or_create_qapp() window = FingerDashboardWindow(config, layout=layout) window.show() QTimer.singleShot(0, window._apply_panel_layout) return app.exec_()