"""Exhibition-friendly Sensor Bench window (qt_viewer_2-style layout).""" from __future__ import annotations import time from importlib.metadata import PackageNotFoundError, version as pkg_version from pathlib import Path from typing import Optional, Protocol import cv2 import numpy as np from PyQt5.QtCore import Qt, QObject, QTimer, pyqtSignal from PyQt5.QtGui import QImage, QPixmap from PyQt5.QtWidgets import ( QAction, QCheckBox, QComboBox, QFileDialog, QHBoxLayout, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QPushButton, QSizePolicy, QSplitter, QVBoxLayout, QWidget, ) try: import pyqtgraph as pg except Exception as exc: # pragma: no cover raise ImportError( "PyQtGraph is required. Install with: pip install pyqt5 pyqtgraph" ) from exc from .acquisition import ( FixedCameraMode, SdkAcquisitionWorker, SdkPacket, draw_arrow_display, load_fixed_mode, orisys_importable, ) from .camera_discovery import list_cameras, pick_default_camera from .dd06_acquisition import Dd06AcquisitionWorker from .force_calibration import ( FORCE_SAMPLE_RATE_HZ, PLOT_BUFFER_POINTS, PLOT_WINDOW_SEC, Y_AXIS_MIN_SPAN, Y_RANGE_ALPHA, calibrate_normal_force, ) from .i18n import DEFAULT_LOCALE, Translator from .paths import ( config_display_name, default_dd06_config, default_ddjx01_config, default_locales_dir, default_recordings_dir, list_orisys_configs, orisys_configs_dir, project_root, ) from .plot_history import MetricHistory from .recorder import RawRecorder CURVE_COLORS = ("#8ab4f8", "#34a853", "#fbbc04") APP_VERSION_FALLBACK = "0.1.0" STYLE = """ QMainWindow, QWidget { background-color: #1e1f22; color: #e8eaed; font-family: "Segoe UI", "Microsoft YaHei", "SimHei", Arial, sans-serif; font-size: 13px; } QMenuBar { background-color: #1e1f22; color: #e8eaed; } QMenuBar::item:selected { background-color: #2d3139; } QMenu { background-color: #2a2d33; color: #e8eaed; border: 1px solid #4a505a; } QMenu::item:selected { background-color: #243447; } QPushButton { background-color: #2d3139; border: 1px solid #4a505a; border-radius: 4px; padding: 6px 14px; min-height: 28px; color: #e8eaed; } QPushButton:hover { background-color: #3a404c; } QPushButton:pressed { background-color: #252930; } QPushButton:disabled { color: #6b7280; background-color: #25262a; } QPushButton#btnPrimary { background-color: #1a73e8; border-color: #1a73e8; color: white; font-weight: 600; } QPushButton#btnPrimary:hover { background-color: #1967d2; } QPushButton#btnDanger { background-color: #c5221f; border-color: #c5221f; color: white; font-weight: 600; } QPushButton#btnDanger:hover { background-color: #a50e0e; } QLineEdit, QComboBox { background-color: #2a2d33; border: 1px solid #4a505a; border-radius: 4px; padding: 5px 8px; color: #e8eaed; } QComboBox QAbstractItemView { background-color: #2a2d33; color: #e8eaed; selection-background-color: #243447; } QLabel#toolbarCaption { color: #9aa0a6; } QLabel#imagePanel { color: #6b7280; background-color: #121316; border: 1px solid #3c4048; border-radius: 6px; } QPlainTextEdit { background-color: #121316; color: #9aa0a6; border: 1px solid #3c4048; border-radius: 6px; font-family: "Cascadia Code", "Consolas", monospace; font-size: 12px; padding: 6px; } QSplitter::handle { background-color: #2a2d33; } """ class AcquisitionPipeline(Protocol): is_running: bool def latest(self) -> Optional[SdkPacket]: ... def stop(self) -> None: ... def reset(self) -> None: ... def ndarray_to_pixmap(frame: np.ndarray) -> QPixmap: if frame.ndim == 2 or (frame.ndim == 3 and frame.shape[2] == 1): gray = frame if frame.ndim == 2 else frame[:, :, 0] h, w = gray.shape[:2] image = QImage(gray.data, w, h, w, QImage.Format_Grayscale8) return QPixmap.fromImage(image.copy()) if frame.ndim == 3 and frame.shape[2] == 4: rgba = cv2.cvtColor(frame, cv2.COLOR_BGRA2RGBA) h, w = rgba.shape[:2] image = QImage(rgba.data, w, h, w * 4, QImage.Format_RGBA8888) return QPixmap.fromImage(image.copy()) rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) h, w = rgb.shape[:2] image = QImage(rgb.data, w, h, w * 3, QImage.Format_RGB888) return QPixmap.fromImage(image.copy()) def app_version() -> str: try: return pkg_version("sensor-bench") except PackageNotFoundError: return APP_VERSION_FALLBACK class _AcquisitionBridge(QObject): error = pyqtSignal(str) status = pyqtSignal(str) class ExhibitionWindow(QMainWindow): """External exhibition UI: toolbar + image + FN/FX/FY plots + log.""" def __init__(self, *, locales_dir: Optional[Path] = None, locale: str = DEFAULT_LOCALE): super().__init__() self.resize(1480, 900) self.setStyleSheet(STYLE) self.tr = Translator(locales_dir or default_locales_dir(), locale=locale) self.recordings_dir = default_recordings_dir() self.force_history = MetricHistory( npoints=PLOT_BUFFER_POINTS, channels=3, window_sec=PLOT_WINDOW_SEC, sample_rate_hz=FORCE_SAMPLE_RATE_HZ, y_axis_min_span=Y_AXIS_MIN_SPAN, y_range_alpha=Y_RANGE_ALPHA, ) self._acq_bridge = _AcquisitionBridge() self._acq_bridge.error.connect(self._on_acquisition_error) self._acq_bridge.status.connect(self._on_acquisition_status) self.stitch_worker = SdkAcquisitionWorker( on_error=lambda m: self._acq_bridge.error.emit(m), on_status=lambda m: self._acq_bridge.status.emit(m), ) self.dd06_worker = Dd06AcquisitionWorker( on_error=lambda m: self._acq_bridge.error.emit(m), on_status=lambda m: self._acq_bridge.status.emit(m), ) self.acquisition: AcquisitionPipeline = self.stitch_worker self.recorder = RawRecorder() self._source_pixmap: Optional[QPixmap] = None self._last_packet: Optional[SdkPacket] = None self._last_packet_id = 0 self._awaiting_first_frame = False self._display_aspect = 1.0 self._view_mode = "pure_color" self._sensor_type = "orisys" self._log_panel_visible = False self._fixed_mode: Optional[FixedCameraMode] = None self._orisys_config = str(default_ddjx01_config()) self._dd06_config = str(default_dd06_config()) self._build_ui() self._create_plots() self._apply_language() self._refresh_orisys_status() self._log(self.t("exhibition.hints")) self.timer = QTimer(self) self.timer.timeout.connect(self._on_tick) self.timer.start(16) def t(self, key: str, **kwargs) -> str: return self.tr.t(key, **kwargs) def _log(self, message: str) -> None: stamp = time.strftime("%H:%M:%S") self.log_widget.appendPlainText(f"[{stamp}] {message}") def _pipeline_running(self) -> bool: return bool(self.stitch_worker.is_running or self.dd06_worker.is_running) def _build_ui(self) -> None: self._build_menu() central = QWidget() self.setCentralWidget(central) root = QVBoxLayout(central) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) bar = QHBoxLayout() bar.setSpacing(8) self.sensor_caption = QLabel() self.sensor_caption.setObjectName("toolbarCaption") bar.addWidget(self.sensor_caption) self.sensor_combo = QComboBox() self.sensor_combo.addItem("", "orisys") self.sensor_combo.addItem("", "dd06") self.sensor_combo.currentIndexChanged.connect(self._on_sensor_type_changed) bar.addWidget(self.sensor_combo) self.btn_search = QPushButton() self.btn_search.clicked.connect(self._on_search_cameras) bar.addWidget(self.btn_search) self.btn_camera = QPushButton() self.btn_camera.clicked.connect(self._on_browse_camera) bar.addWidget(self.btn_camera) self.camera_edit = QLineEdit("0") bar.addWidget(self.camera_edit, 1) self.btn_config = QPushButton() self.btn_config.clicked.connect(self._on_browse_config) bar.addWidget(self.btn_config) self.config_combo = QComboBox() self.config_combo.setMinimumWidth(160) self.config_combo.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed) bar.addWidget(self.config_combo, 1) self._populate_config_combo(preferred=self._orisys_config) self.btn_start = QPushButton() self.btn_start.setObjectName("btnPrimary") self.btn_start.clicked.connect(self._on_start) bar.addWidget(self.btn_start) self.btn_stop = QPushButton() self.btn_stop.setEnabled(False) self.btn_stop.clicked.connect(self._on_stop) bar.addWidget(self.btn_stop) self.btn_reset = QPushButton() self.btn_reset.setEnabled(False) self.btn_reset.clicked.connect(self._on_reset) bar.addWidget(self.btn_reset) self.btn_record = QPushButton() self.btn_record.setObjectName("btnDanger") self.btn_record.setEnabled(False) self.btn_record.clicked.connect(self._on_toggle_record) bar.addWidget(self.btn_record) self.show_log_check = QCheckBox() self.show_log_check.setChecked(False) self.show_log_check.toggled.connect(self._on_show_log_changed) bar.addWidget(self.show_log_check) self.view_caption = QLabel() self.view_caption.setObjectName("toolbarCaption") bar.addWidget(self.view_caption) self.view_combo = QComboBox() self.view_combo.addItem("", "pure_color") self.view_combo.addItem("", "captured") self.view_combo.addItem("", "raw") self.view_combo.setCurrentIndex(0) self.view_combo.currentIndexChanged.connect(self._on_view_changed) bar.addWidget(self.view_combo) self.language_caption = QLabel() self.language_caption.setObjectName("toolbarCaption") bar.addWidget(self.language_caption) self.language_combo = QComboBox() self.language_combo.addItem("", "zh_CN") self.language_combo.addItem("", "en_US") self.language_combo.setCurrentIndex(0 if self.tr.locale == "zh_CN" else 1) self.language_combo.currentIndexChanged.connect(self._on_language_changed) bar.addWidget(self.language_combo) root.addLayout(bar) v_splitter = QSplitter(Qt.Vertical) h_splitter = QSplitter(Qt.Horizontal) left = QWidget() left_layout = QVBoxLayout(left) left_layout.setContentsMargins(0, 0, 0, 0) self.image_label = QLabel() self.image_label.setObjectName("imagePanel") self.image_label.setMinimumSize(240, 240) self.image_label.setAlignment(Qt.AlignCenter) self.image_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) left_layout.addWidget(self.image_label) h_splitter.addWidget(left) pg.setConfigOptions(antialias=True, foreground="#e8eaed", background="#121316") self.graph_widget = pg.GraphicsLayoutWidget() self.graph_widget.setMinimumWidth(320) h_splitter.addWidget(self.graph_widget) self.h_splitter = h_splitter v_splitter.addWidget(h_splitter) self.log_widget = QPlainTextEdit() self.log_widget.setReadOnly(True) self.log_widget.setMaximumBlockCount(500) self.log_widget.setMinimumHeight(110) self.log_widget.setVisible(self._log_panel_visible) v_splitter.addWidget(self.log_widget) v_splitter.setStretchFactor(0, 5) v_splitter.setStretchFactor(1, 1) root.addWidget(v_splitter, 1) self.status_label = QLabel() self.status_label.setStyleSheet("color: #9aa0a6;") root.addWidget(self.status_label) def _build_menu(self) -> None: menu = self.menuBar() self.help_menu = menu.addMenu("") self.about_action = QAction("", self) self.about_action.triggered.connect(self._on_about) self.help_menu.addAction(self.about_action) def _create_plots(self) -> None: self.plots = [] self.curves = [] self.dot_cores = [] self.plot_titles = [ "exhibition.plot_fn", "exhibition.plot_fx", "exhibition.plot_fy", ] axis_pen = pg.mkPen(color="#3c4048", width=1) for i, title_key in enumerate(self.plot_titles): plot = self.graph_widget.addPlot(title=self.t(title_key)) plot.showGrid(x=False, y=True, alpha=0.2) plot.getAxis("left").setPen(axis_pen) plot.getAxis("bottom").setPen(axis_pen) plot.setLabel("bottom", "s") curve = plot.plot(pen=pg.mkPen(CURVE_COLORS[i], width=2)) dot = plot.plot( [], [], pen=None, symbol="o", symbolSize=6, symbolBrush=CURVE_COLORS[i], ) plot.setXRange(0, PLOT_WINDOW_SEC, padding=0) plot.enableAutoRange(x=False, y=False) self.plots.append(plot) self.curves.append(curve) self.dot_cores.append(dot) if i < 2: self.graph_widget.nextRow() def _apply_language(self) -> None: self.setWindowTitle(self.t("exhibition.title")) self.help_menu.setTitle(self.t("exhibition.menu_help")) self.about_action.setText(self.t("exhibition.menu_about")) self.sensor_caption.setText(self.t("exhibition.sensor_type")) self.btn_search.setText(self.t("exhibition.btn_search")) self.btn_camera.setText(self.t("exhibition.btn_camera")) self.btn_config.setText(self.t("exhibition.btn_config")) self.btn_start.setText(self.t("exhibition.btn_start")) self.btn_stop.setText(self.t("exhibition.btn_stop")) self.btn_reset.setText(self.t("exhibition.btn_reset")) self._refresh_record_button() self.view_caption.setText(self.t("exhibition.view")) self.show_log_check.setText(self.t("exhibition.show_log")) self.language_caption.setText(self.t("app.language")) self.camera_edit.setPlaceholderText(self.t("exhibition.placeholder_camera")) self._update_config_tooltip() self.language_combo.blockSignals(True) self.language_combo.setItemText(0, self.t("app.lang_zh")) self.language_combo.setItemText(1, self.t("app.lang_en")) self.language_combo.blockSignals(False) self.sensor_combo.blockSignals(True) self.sensor_combo.setItemText(0, self.t("exhibition.sensor_orisys")) self.sensor_combo.setItemText(1, self.t("exhibition.sensor_dd06")) self.sensor_combo.blockSignals(False) self.view_combo.blockSignals(True) self.view_combo.setItemText(0, self.t("exhibition.view_pure_color")) self.view_combo.setItemText(1, self.t("exhibition.view_captured")) self.view_combo.setItemText(2, self.t("exhibition.view_raw")) self.view_combo.blockSignals(False) for plot, title_key in zip(self.plots, self.plot_titles): plot.setTitle(self.t(title_key), color="#9aa0a6", size="12pt") if self._source_pixmap is None: self.image_label.setText(self.t("labels.no_preview")) self.status_label.setText(self.t("status.ready")) def _selected_config(self) -> str: data = self.config_combo.currentData() if data: return str(data) return self.config_combo.currentText().strip() def _set_config_selection(self, value: str) -> None: value = str(value or "").strip() if not value: return for index in range(self.config_combo.count()): if str(self.config_combo.itemData(index) or "") == value: self.config_combo.setCurrentIndex(index) return # Custom path not yet in the list. if Path(value).stem.lower() == "options" and Path(value).suffix.lower() == ".json": return self.config_combo.addItem(config_display_name(value), value) self.config_combo.setCurrentIndex(self.config_combo.count() - 1) def _populate_config_combo(self, *, preferred: str | None = None) -> None: current = preferred or self._selected_config() or default_ddjx01_config() custom_paths = [] for index in range(self.config_combo.count()): data = str(self.config_combo.itemData(index) or "") if data and (Path(data).suffix.lower() == ".json" or "/" in data or "\\" in data): if data not in custom_paths: custom_paths.append(data) self.config_combo.blockSignals(True) self.config_combo.clear() for entry in list_orisys_configs(): self.config_combo.addItem(config_display_name(entry), entry) for path in custom_paths: # Avoid duplicating local entries already listed above. if any(str(self.config_combo.itemData(i) or "") == path for i in range(self.config_combo.count())): continue if Path(path).stem.lower() == "options": continue self.config_combo.addItem(config_display_name(path), path) if self.config_combo.count() == 0 and current: self.config_combo.addItem(config_display_name(current), current) self.config_combo.blockSignals(False) self._set_config_selection(current) self._update_config_tooltip() def _update_config_tooltip(self) -> None: cfg_dir = orisys_configs_dir() selected = self._selected_config() if cfg_dir is not None: tip = self.t("exhibition.config_dir_tip", path=cfg_dir, selected=selected) else: tip = self.t("exhibition.config_dir_missing") self.config_combo.setToolTip(tip) self.btn_config.setToolTip(tip) def _refresh_record_button(self) -> None: if self.recorder.is_recording: self.btn_record.setText(self.t("exhibition.btn_stop_record")) else: self.btn_record.setText(self.t("exhibition.btn_start_record")) def _on_about(self) -> None: ok, orisys_ver = orisys_importable() orisys_text = orisys_ver if ok else self.t("status.orisys_missing") QMessageBox.about( self, self.t("exhibition.about_title"), self.t( "exhibition.about_body", app_version=app_version(), orisys_version=orisys_text, ), ) def _on_language_changed(self, *_args) -> None: locale = self.language_combo.currentData() if not locale: return self.tr.set_locale(str(locale)) self._apply_language() self._refresh_orisys_status() def _on_sensor_type_changed(self, *_args) -> None: if self._pipeline_running(): return kind = str(self.sensor_combo.currentData() or "orisys") if kind == self._sensor_type: return # Remember current config for the leaving type. if self._sensor_type == "orisys": self._orisys_config = self._selected_config() or self._orisys_config else: self._dd06_config = self._selected_config() or self._dd06_config self._sensor_type = kind self.acquisition = self.dd06_worker if kind == "dd06" else self.stitch_worker preferred = self._dd06_config if kind == "dd06" else self._orisys_config self._populate_config_combo(preferred=preferred) self._log(self.t("exhibition.sensor_switched", sensor=kind)) def _on_view_changed(self, *_args) -> None: mode = self.view_combo.currentData() if mode in ("pure_color", "captured", "raw"): self._view_mode = str(mode) if self._last_packet is not None: self._update_image(self._last_packet) def _on_show_log_changed(self, visible: bool) -> None: self._log_panel_visible = bool(visible) self.log_widget.setVisible(self._log_panel_visible) def _refresh_orisys_status(self) -> None: ok, detail = orisys_importable() if ok: self.status_label.setText(self.t("status.orisys_ok", version=detail)) self.btn_start.setEnabled(not self._pipeline_running()) self._populate_config_combo(preferred=self._selected_config() or default_ddjx01_config()) else: self.status_label.setText(self.t("status.orisys_missing")) self.btn_start.setEnabled(False) self._update_config_tooltip() def _parse_camera_source(self) -> str | int: text = self.camera_edit.text().strip() or "0" if text.isdigit(): return int(text) return text def _on_search_cameras(self) -> None: if self._pipeline_running(): return preferred = self.camera_edit.text().strip() cameras = list_cameras() if not cameras: self._log(self.t("exhibition.camera_none")) return names = ", ".join(cam.label() for cam in cameras) self._log(self.t("exhibition.camera_found", count=len(cameras), names=names)) selected = pick_default_camera( cameras, preferred if preferred.isdigit() else None ) if selected is not None: self.camera_edit.setText(selected) def _on_browse_camera(self) -> None: path, _ = QFileDialog.getOpenFileName( self, self.t("exhibition.dialog_video"), str(project_root()), "Video (*.mp4 *.avi *.mov);;All (*)", ) if path: self.camera_edit.setText(path) def _on_browse_config(self) -> None: start_dir = orisys_configs_dir() or project_root() path, _ = QFileDialog.getOpenFileName( self, self.t("dialogs.select_config"), str(start_dir), "JSON (*.json);;All (*)", ) if path: self._set_config_selection(path) self._update_config_tooltip() def _on_start(self) -> None: ok, detail = orisys_importable() if not ok: QMessageBox.warning( self, self.t("dialogs.orisys_title"), self.t("dialogs.orisys_required", detail=detail), ) return config = self._selected_config() if not config: QMessageBox.warning( self, self.t("dialogs.config_title"), self.t("dialogs.config_required") ) return try: mode = load_fixed_mode(config) except Exception as exc: # noqa: BLE001 QMessageBox.warning( self, self.t("dialogs.config_title"), self.t("labels.fixed_mode_invalid", error=exc), ) return self.force_history.reset() self._awaiting_first_frame = True self._source_pixmap = None self.image_label.clear() self.image_label.setText(self.t("labels.no_preview")) self.btn_start.setEnabled(False) self.btn_start.setText(self.t("exhibition.btn_starting")) self.sensor_combo.setEnabled(False) try: if self._sensor_type == "dd06": self.acquisition = self.dd06_worker mode = self.dd06_worker.start( self._parse_camera_source(), config_path=config, ) else: self.acquisition = self.stitch_worker mode = self.stitch_worker.start( self._parse_camera_source(), config_path=config, cal_path=None, backend="auto", ) except Exception as exc: # noqa: BLE001 self._awaiting_first_frame = False self.btn_start.setEnabled(True) self.btn_start.setText(self.t("exhibition.btn_start")) self.sensor_combo.setEnabled(True) QMessageBox.critical(self, self.t("dialogs.capture_title"), str(exc)) self._log(self.t("status.capture_error", message=exc)) return self._fixed_mode = mode self.btn_start.setText(self.t("exhibition.btn_start")) self.btn_stop.setEnabled(True) self.btn_reset.setEnabled(True) self.btn_record.setEnabled(False) self.btn_search.setEnabled(False) self.btn_camera.setEnabled(False) self.camera_edit.setEnabled(False) self.config_combo.setEnabled(False) self.status_label.setText(self.t("status.opening", mode=mode.label())) self._log(self.t("status.opening", mode=mode.label())) def _on_stop(self) -> None: if self.recorder.is_recording: self._stop_record() self.stitch_worker.stop() self.dd06_worker.stop() self._awaiting_first_frame = False self._last_packet = None self._source_pixmap = None self.image_label.clear() self.image_label.setText(self.t("labels.no_preview")) self.btn_start.setEnabled(orisys_importable()[0]) self.btn_stop.setEnabled(False) self.btn_reset.setEnabled(False) self.btn_record.setEnabled(False) self.btn_search.setEnabled(True) self.btn_camera.setEnabled(True) self.camera_edit.setEnabled(True) self.config_combo.setEnabled(True) self.sensor_combo.setEnabled(True) self.status_label.setText(self.t("status.preview_stopped")) self._log(self.t("status.preview_stopped")) def _on_reset(self) -> None: if not self._pipeline_running(): self._log(self.t("exhibition.reset_need_start")) return try: self.acquisition.reset() except Exception as exc: # noqa: BLE001 self._log(self.t("exhibition.reset_failed", message=exc)) return self._log(self.t("exhibition.reset_done")) def _on_toggle_record(self) -> None: if self.recorder.is_recording: self._stop_record() else: self._start_record() def _start_record(self) -> None: packet = self._last_packet or self.acquisition.latest() if packet is None: QMessageBox.warning( self, self.t("dialogs.recording_title"), self.t("dialogs.no_frame") ) return record_fps = packet.fps or (packet.mode.fps if packet.mode else 30.0) try: path = self.recorder.start( self.recordings_dir, fps=float(record_fps), frame=packet.raw_frame ) except Exception as exc: # noqa: BLE001 QMessageBox.critical(self, self.t("dialogs.recording_title"), str(exc)) return self._refresh_record_button() self._log(self.t("exhibition.recording_started", name=path.name)) def _stop_record(self) -> None: path, count = self.recorder.stop() self._refresh_record_button() if path is None: return self._log(self.t("status.saved", count=count, path=path)) def _on_acquisition_error(self, message: str) -> None: self._awaiting_first_frame = False self.btn_start.setEnabled(orisys_importable()[0]) self.btn_stop.setEnabled(False) self.btn_reset.setEnabled(False) self.btn_record.setEnabled(False) self.btn_search.setEnabled(True) self.btn_camera.setEnabled(True) self.camera_edit.setEnabled(True) self.config_combo.setEnabled(True) self.sensor_combo.setEnabled(True) self.status_label.setText(self.t("status.capture_error", message=message)) self._log(self.t("status.capture_error", message=message)) QMessageBox.warning(self, self.t("dialogs.capture_title"), message) def _on_acquisition_status(self, message: str) -> None: self.status_label.setText(message) self._log(message) def _on_tick(self) -> None: packet = self.acquisition.latest() if packet is None: return packet_id = id(packet) if packet_id == self._last_packet_id and not self._awaiting_first_frame: return if self._awaiting_first_frame: self._awaiting_first_frame = False self.btn_record.setEnabled(True) info = self.t( "status.frame_info", raw_w=packet.width, raw_h=packet.height, proc_w=packet.processed_width, proc_h=packet.processed_height, fps=packet.fps, ) self.status_label.setText(info) self._log(info) self._log( self.t( "exhibition.force_fn_n", value=calibrate_normal_force(packet.normal_force), ) ) self._last_packet = packet self._last_packet_id = packet_id self._update_image(packet) self.force_history.push( packet.timestamp, packet.normal_force, packet.shear_x, packet.shear_y, ) self._update_plots() if self.recorder.is_recording: self.recorder.write(packet.raw_frame) self.status_label.setText( self.t( "status.recording_frames", name=self.recorder.path.name if self.recorder.path else "", count=self.recorder.frame_count, ) ) def _update_image(self, packet: SdkPacket) -> None: if self._view_mode == "raw": frame = packet.raw_frame elif self._view_mode == "captured": frame = draw_arrow_display( packet.processed_frame, packet.flow, background="captured" ) else: frame = draw_arrow_display( packet.processed_frame, packet.flow, background="pure_color" ) h, w = frame.shape[:2] aspect = float(w) / float(h) if h else 1.0 first = self._source_pixmap is None aspect_changed = abs(aspect - self._display_aspect) > 1e-3 self._display_aspect = aspect self._source_pixmap = ndarray_to_pixmap(frame) if first or aspect_changed: self._adjust_image_panel() else: self._render_image() def _adjust_image_panel(self) -> None: available = max(self.height() - 220, 200) image_w = int(available * self._display_aspect) image_w = max(min(image_w, max(self.width() // 2, 240)), 240) self.h_splitter.setSizes([image_w, max(self.width() - image_w, 320)]) self._render_image() def _render_image(self) -> None: if self._source_pixmap is None or self._source_pixmap.isNull(): return target = self.image_label.size() if target.width() < 2 or target.height() < 2: return scaled = self._source_pixmap.scaled( target, Qt.KeepAspectRatio, Qt.SmoothTransformation ) self.image_label.setPixmap(scaled) def _update_plots(self) -> None: state = self.force_history.state() if state is None: return for i, curve in enumerate(self.curves): x = state.times y = state.values[:, i] curve.setData(x, y) ymin, ymax = state.y_ranges[i] self.plots[i].setYRange(ymin, ymax) self.dot_cores[i].setData(x[-1:], y[-1:]) x0, x1 = state.x_range for plot in self.plots: plot.setXRange(x0, x1, padding=0) def resizeEvent(self, event) -> None: # noqa: N802 super().resizeEvent(event) self._render_image() def keyPressEvent(self, event) -> None: # noqa: N802 key = event.key() if key == Qt.Key_Escape: if self.isFullScreen(): self.showNormal() else: self.showFullScreen() elif key == Qt.Key_Q: self.close() elif key == Qt.Key_S: if not self.recorder.is_recording and self._pipeline_running(): self._start_record() elif key == Qt.Key_E: if self.recorder.is_recording: self._stop_record() elif key == Qt.Key_R: self._on_reset() else: super().keyPressEvent(event) def closeEvent(self, event) -> None: # noqa: N802 if self.recorder.is_recording: self.recorder.stop() self.stitch_worker.stop() self.dd06_worker.stop() self.timer.stop() super().closeEvent(event)