"""Main window shell for the customer app.""" from __future__ import annotations import logging import subprocess import sys from pathlib import Path from typing import Any import numpy as np from PyQt5.QtCore import Qt, QTimer from PyQt5.QtGui import QIcon, QImage from PyQt5.QtWidgets import ( QAction, QActionGroup, QApplication, QFileDialog, QFrame, QHBoxLayout, QLabel, QMainWindow, QMessageBox, QSplitter, QStatusBar, QVBoxLayout, QWidget, ) from orisys_app_shared.force import select_display_forces from orisys_app_shared.i18n import set_locale, t from orisys_app_shared.logging import report_error from orisys_app_shared.paths import logo_path, sensor_json_path from orisys_app_shared.prefs import save_preferences from orisys_app_shared.theme import build_qss, clear_theme_cache, color from orisys_customer.ui.about_dialog import AboutDialog from orisys_customer.ui.arrow_panel import ArrowPanel from orisys_customer.ui.brand_bar import BrandBar from orisys_customer.ui.cal_panel import CalPanel from orisys_customer.ui.force_plots import ForcePlotsPanel from orisys_customer.ui.mesh3d_panel import Mesh3dPanel from orisys_customer.ui.plot_settings_dialog import ( DEFAULT_PLOT_FN_MAX, DEFAULT_PLOT_SHEAR_MAX, DEFAULT_PLOT_WINDOW_SEC, PlotSettingsDialog, ) from orisys_customer.ui.sensor_picker import SensorPickerWidget from orisys_customer.ui.toolbar import ToolbarWidget logger = logging.getLogger(__name__) RUNNABLE_TYPES = frozenset({"box", "mini", "finger"}) ARROW_ROTATE_TYPES = frozenset({"box", "mini"}) DEFAULT_SPLITTER_SIZES = (340, 480, 400) MIN_PANEL_WIDTHS = { "mesh3d": 240, "arrow": 300, "plots": 340, "cal": 300, } SESSION_PHASES = ("idle", "connecting", "running", "error") def _bgr_to_qimage(frame: np.ndarray) -> QImage: rgb = frame[:, :, ::-1].copy() h, w, ch = rgb.shape bytes_per_line = ch * w return QImage(rgb.data, w, h, bytes_per_line, QImage.Format_RGB888).copy() def _reveal_in_explorer(path: Path) -> None: path = path.resolve() if sys.platform == "win32": subprocess.run(["explorer", "/select,", str(path)], check=False) return if sys.platform == "darwin": subprocess.run(["open", "-R", str(path)], check=False) return folder = path.parent if path.is_file() else path subprocess.run(["xdg-open", str(folder)], check=False) class CustomerMainWindow(QMainWindow): def __init__(self, prefs: dict[str, Any] | None = None) -> None: super().__init__() self.prefs: dict[str, Any] = dict(prefs or {}) self._theme = str(self.prefs.get("theme") or "dark") self._language = str(self.prefs.get("language") or "zh_CN") self._show_cal = bool(self.prefs.get("show_cal_panel", False)) self._show_finger_3d = bool(self.prefs.get("show_finger_3d", True)) self._show_fps = bool(self.prefs.get("show_fps", False)) self._session_phase = "idle" self._running = False self._preview = None self._preview_timer = QTimer(self) self._preview_timer.setInterval(33) self._preview_timer.timeout.connect(self._on_preview_tick) self._run_timer = QTimer(self) self._run_timer.setInterval(16) self._run_timer.timeout.connect(self._on_run_tick) self._box_worker = None self._finger_worker = None self._active_worker = None self._active_sensor_type = "box" self._last_session_error = "" self._last_plot_packet_key: tuple[int, float] | None = None self._last_rendered_view_mode: str | None = None self._run_source_label = "" icon = QIcon(str(logo_path())) if not icon.isNull(): self.setWindowIcon(icon) central = QWidget() self.setCentralWidget(central) root = QVBoxLayout(central) root.setContentsMargins(10, 10, 10, 10) root.setSpacing(8) self.brand_bar = BrandBar(theme=self._theme) root.addWidget(self.brand_bar) top = QHBoxLayout() top.setSpacing(10) self.toolbar = ToolbarWidget() self.sensor_picker = SensorPickerWidget() sensor_card = QFrame() sensor_card.setObjectName("contentPanel") sensor_card_layout = QVBoxLayout(sensor_card) sensor_card_layout.setContentsMargins(0, 0, 0, 0) sensor_card_layout.setSpacing(0) sensor_card_layout.addWidget(self.sensor_picker) top.addWidget(self.toolbar, 2) top.addWidget(sensor_card) root.addLayout(top) self.content_splitter = QSplitter(Qt.Horizontal) self.mesh3d_panel = Mesh3dPanel() self.arrow_panel = ArrowPanel() self.force_plots = ForcePlotsPanel(theme=self._theme) self.cal_panel = CalPanel(theme=self._theme) self.content_splitter.addWidget(self.mesh3d_panel) self.content_splitter.addWidget(self.arrow_panel) self.content_splitter.addWidget(self.force_plots) self.content_splitter.addWidget(self.cal_panel) self.content_splitter.setChildrenCollapsible(False) root.addWidget(self.content_splitter, 1) self.setStatusBar(QStatusBar()) self.status_chip = QLabel() self.status_chip.setObjectName("statusChip") self.statusBar().addPermanentWidget(self.status_chip) self.cal_panel.status_message.connect(self.statusBar().showMessage) self.cal_panel.calibration_exported.connect(self._on_calibration_exported) self._set_phase("idle") self._build_menus() self._wire_toolbar() self._wire_sensor_picker() self._restore_prefs_into_ui() self._apply_panel_visibility() self._restore_splitter_sizes() self._update_run_enabled() self.cal_panel.set_meta_context( sensor_type=self.sensor_picker.sensor_type(), config_ref=self._resolve_config_ref(), ) self.retranslate() self._restore_geometry() self._apply_default_size() @property def _preview_open(self) -> bool: return self._preview is not None and bool(self._preview.is_open) def _get_preview(self): if self._preview is None: from orisys_app_shared.camera import RawPreviewSession self._preview = RawPreviewSession() return self._preview def _ensure_workers(self) -> None: if self._box_worker is not None and self._finger_worker is not None: return from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker if self._box_worker is None: self._box_worker = SdkAcquisitionWorker( on_error=self._on_worker_error, on_status=self._on_worker_status, ) if self._finger_worker is None: self._finger_worker = FingerAcquisitionWorker( on_error=self._on_worker_error, on_status=self._on_worker_status, ) def _wire_toolbar(self) -> None: self.toolbar.search_clicked.connect(self._on_search_cameras) self.toolbar.preview_toggled.connect(self._on_preview_toggled) self.toolbar.camera_id_changed.connect(self._on_camera_id_changed) self.toolbar.source_mode_changed.connect(self._on_source_mode_changed) self.toolbar.browse_video_clicked.connect(self._on_browse_video) self.toolbar.run_toggled.connect(self._on_run_toggled) self.toolbar.reset_clicked.connect(self._on_reset) self.toolbar.units_changed.connect(self._on_units_changed) def _wire_sensor_picker(self) -> None: self.sensor_picker.sensor_type_changed.connect(self._on_sensor_type_changed) self.sensor_picker.config_changed.connect(self._on_config_changed) self.arrow_panel.view_mode_changed.connect(self._on_image_view_mode_changed) def _restore_prefs_into_ui(self) -> None: raw_id = str(self.prefs.get("camera_id") or "0").strip() try: cam_id = int(raw_id) except ValueError: cam_id = 0 self.toolbar.set_camera_id(cam_id) self.toolbar.set_source_mode(str(self.prefs.get("source_mode") or "camera")) self.toolbar.set_video_path(str(self.prefs.get("video_path") or "")) self.toolbar.set_units(str(self.prefs.get("units") or "newton")) self.arrow_panel.set_view_mode(str(self.prefs.get("image_view_mode") or "arrows_pure")) self.arrow_panel.set_fps_visible(self._show_fps) self.arrow_panel.set_rotation_deg(self.prefs.get("arrow_rotation_deg", 90)) self._apply_plot_prefs_to_panel() sensor_type = str(self.prefs.get("sensor_type") or "box") config_name = str(self.prefs.get("config_name") or "") self.sensor_picker.set_sensor_type(sensor_type) if config_name: self.sensor_picker.set_config_name(config_name) else: self.prefs["config_name"] = self.sensor_picker.config_name() self._apply_arrow_rotation_for_sensor() def _apply_default_size(self) -> None: """Give the first launch a usable size for the multi-panel layout.""" default = self.prefs.get("window_default_size") or [1440, 880] try: width, height = int(default[0]), int(default[1]) except (TypeError, ValueError, IndexError): width, height = 1440, 880 screen = QApplication.primaryScreen() if screen is not None: avail = screen.availableGeometry() width = min(width, max(1024, avail.width() - 80)) height = min(height, max(680, avail.height() - 80)) if not self.prefs.get("window_geometry"): self.resize(width, height) def _restore_splitter_sizes(self) -> None: stored = self.prefs.get("splitter_sizes") if isinstance(stored, list) and len(stored) == self.content_splitter.count(): try: self.content_splitter.setSizes([int(v) for v in stored]) return except (TypeError, ValueError): pass sizes = list(DEFAULT_SPLITTER_SIZES) if not self._show_finger_3d or self.sensor_picker.sensor_type() != "finger": sizes[0] = 0 self.content_splitter.setSizes(sizes) def _build_menus(self) -> None: menu_bar = self.menuBar() self.file_menu = menu_bar.addMenu("") self.act_open_video = QAction(self) self.act_open_video.triggered.connect(self._on_browse_video) self.file_menu.addAction(self.act_open_video) self.act_reveal_config = QAction(self) self.act_reveal_config.triggered.connect(self._on_reveal_config) self.file_menu.addAction(self.act_reveal_config) self.view_menu = menu_bar.addMenu("") self.act_cal_panel = QAction(self) self.act_cal_panel.setCheckable(True) self.act_cal_panel.setChecked(self._show_cal) self.act_cal_panel.toggled.connect(self._on_cal_panel_toggled) self.view_menu.addAction(self.act_cal_panel) self.act_finger_3d = QAction(self) self.act_finger_3d.setCheckable(True) self.act_finger_3d.setChecked(self._show_finger_3d) self.act_finger_3d.toggled.connect(self._on_finger_3d_toggled) self.view_menu.addAction(self.act_finger_3d) self.act_show_fps = QAction(self) self.act_show_fps.setCheckable(True) self.act_show_fps.setChecked(self._show_fps) self.act_show_fps.toggled.connect(self._on_show_fps_toggled) self.view_menu.addAction(self.act_show_fps) self.act_plot_settings = QAction(self) self.act_plot_settings.triggered.connect(self._on_plot_settings) self.view_menu.addAction(self.act_plot_settings) self.act_rotate_arrows = QAction(self) self.act_rotate_arrows.triggered.connect(self._on_rotate_arrows) self.view_menu.addAction(self.act_rotate_arrows) self.view_menu.addSeparator() self.act_run = QAction(self) self.act_run.setShortcut("F5") self.act_run.triggered.connect(self._on_run_shortcut) self.view_menu.addAction(self.act_run) self.act_preview = QAction(self) self.act_preview.setShortcut("F6") self.act_preview.triggered.connect(self._on_preview_shortcut) self.view_menu.addAction(self.act_preview) self.act_reset = QAction(self) self.act_reset.setShortcut("Ctrl+R") self.act_reset.triggered.connect(self._on_reset) self.view_menu.addAction(self.act_reset) self.view_menu.addSeparator() self.theme_menu = self.view_menu.addMenu("") self.theme_group = QActionGroup(self) self.theme_group.setExclusive(True) self.act_theme_dark = QAction(self) self.act_theme_dark.setCheckable(True) self.act_theme_light = QAction(self) self.act_theme_light.setCheckable(True) self.theme_group.addAction(self.act_theme_dark) self.theme_group.addAction(self.act_theme_light) self.theme_menu.addAction(self.act_theme_dark) self.theme_menu.addAction(self.act_theme_light) self.act_theme_dark.triggered.connect(lambda: self._set_theme("dark")) self.act_theme_light.triggered.connect(lambda: self._set_theme("light")) self.lang_menu = self.view_menu.addMenu("") self.lang_group = QActionGroup(self) self.lang_group.setExclusive(True) self.act_lang_zh = QAction(self) self.act_lang_zh.setCheckable(True) self.act_lang_en = QAction(self) self.act_lang_en.setCheckable(True) self.lang_group.addAction(self.act_lang_zh) self.lang_group.addAction(self.act_lang_en) self.lang_menu.addAction(self.act_lang_zh) self.lang_menu.addAction(self.act_lang_en) self.act_lang_zh.triggered.connect(lambda: self._set_language("zh_CN")) self.act_lang_en.triggered.connect(lambda: self._set_language("en_US")) self.help_menu = menu_bar.addMenu("") self.act_about = QAction(self) self.act_about.triggered.connect(self._show_about) self.help_menu.addAction(self.act_about) self._sync_menu_checks() def _sync_menu_checks(self) -> None: self.act_theme_dark.setChecked(self._theme != "light") self.act_theme_light.setChecked(self._theme == "light") self.act_lang_zh.setChecked(self._language.startswith("zh")) self.act_lang_en.setChecked(not self._language.startswith("zh")) self.act_cal_panel.blockSignals(True) self.act_cal_panel.setChecked(self._show_cal) self.act_cal_panel.blockSignals(False) is_finger = self.sensor_picker.sensor_type() == "finger" self.act_finger_3d.setEnabled(is_finger) self.act_finger_3d.blockSignals(True) self.act_finger_3d.setChecked(bool(self._show_finger_3d) if is_finger else False) self.act_finger_3d.blockSignals(False) self._apply_arrow_rotation_for_sensor() def _apply_panel_visibility(self) -> None: is_finger = self.sensor_picker.sensor_type() == "finger" show_3d = is_finger and bool(self._show_finger_3d) self.mesh3d_panel.setVisible(show_3d) self.cal_panel.setVisible(self._show_cal) self.mesh3d_panel.setMinimumWidth(MIN_PANEL_WIDTHS["mesh3d"]) self.arrow_panel.setMinimumWidth(MIN_PANEL_WIDTHS["arrow"]) self.force_plots.setMinimumWidth(MIN_PANEL_WIDTHS["plots"]) self.cal_panel.setMinimumWidth(MIN_PANEL_WIDTHS["cal"]) self.act_finger_3d.setEnabled(is_finger) self.act_finger_3d.blockSignals(True) self.act_finger_3d.setChecked(show_3d) self.act_finger_3d.blockSignals(False) for i in range(self.content_splitter.count()): widget = self.content_splitter.widget(i) if widget is self.mesh3d_panel: self.content_splitter.setStretchFactor(i, 2 if show_3d else 0) elif widget is self.arrow_panel: self.content_splitter.setStretchFactor(i, 3) elif widget is self.force_plots: self.content_splitter.setStretchFactor(i, 4) elif widget is self.cal_panel: self.content_splitter.setStretchFactor(i, 3 if self._show_cal else 0) def _update_run_enabled(self) -> None: sensor_ok = self.sensor_picker.sensor_type() in RUNNABLE_TYPES source_ok = True reason = "" if not sensor_ok: reason = t("toolbar.run_disabled_sensor") elif self.toolbar.source_mode() == "video" and not self.toolbar.video_path(): source_ok = False reason = t("toolbar.run_disabled_video") self.toolbar.set_run_tooltip_reason(reason) self.toolbar.set_run_enabled(self._running or (sensor_ok and source_ok)) if not self._running: self.toolbar.refresh_labels() self._sync_shortcut_actions() def retranslate(self) -> None: self.setWindowTitle(t("app.title")) self.file_menu.setTitle(t("menu.file")) self.act_open_video.setText(t("menu.open_video")) self.act_reveal_config.setText(t("menu.reveal_config")) self.view_menu.setTitle(t("menu.view")) self.act_cal_panel.setText(t("menu.calibration_panel")) self.act_finger_3d.setText(t("menu.finger_3d")) self.act_show_fps.setText(t("menu.show_fps")) self.act_plot_settings.setText(t("menu.plot_settings")) self.act_rotate_arrows.setText(t("menu.rotate_arrows")) self.act_run.setText(t("menu.run")) self.act_preview.setText(t("menu.preview")) self.act_reset.setText(t("menu.reset")) self.theme_menu.setTitle(t("menu.theme")) self.act_theme_dark.setText(t("menu.theme_dark")) self.act_theme_light.setText(t("menu.theme_light")) self.lang_menu.setTitle(t("menu.language")) self.act_lang_zh.setText(t("menu.lang_zh")) self.act_lang_en.setText(t("menu.lang_en")) self.help_menu.setTitle(t("menu.help")) self.act_about.setText(t("menu.about")) self.toolbar.retranslate() self.sensor_picker.retranslate() self.mesh3d_panel.retranslate() self.arrow_panel.retranslate() self.force_plots.retranslate() self.cal_panel.retranslate() self.brand_bar.retranslate() self._refresh_status_chip() if not self._preview_open and not self._running: self.statusBar().showMessage(t("status.ready")) self._update_run_enabled() def _set_phase(self, phase: str) -> None: """Track the acquisition phase and reflect it in the status chip.""" self._session_phase = phase if phase in SESSION_PHASES else "idle" self._refresh_status_chip() def _refresh_status_chip(self) -> None: chip = getattr(self, "status_chip", None) if chip is None: return phase = self._session_phase if self._session_phase in SESSION_PHASES else "idle" token = f"status_{phase}" chip.setText(t(f"status.phase_{phase}")) chip.setStyleSheet( f"QLabel#statusChip {{ color: {color(self._theme, token)};" " background: transparent; font-weight: 600; padding: 0 10px 0 4px; }" ) def _set_theme(self, theme: str) -> None: self._theme = theme self.prefs["theme"] = theme clear_theme_cache() app = QApplication.instance() if app is not None: app.setStyleSheet(build_qss(theme)) self.force_plots.set_theme(theme) self.cal_panel.set_theme(theme) self.brand_bar.set_theme(theme) self._refresh_status_chip() self._sync_menu_checks() self._persist_prefs() def _set_language(self, language: str) -> None: self._language = language self.prefs["language"] = language set_locale(language) self.retranslate() self._sync_menu_checks() self._persist_prefs() def _show_about(self) -> None: AboutDialog(self).exec_() def _on_camera_id_changed(self, value: int) -> None: self.prefs["camera_id"] = str(int(value)) self._persist_prefs() def _on_source_mode_changed(self, mode: str) -> None: self.prefs["source_mode"] = mode self._persist_prefs() self._update_run_enabled() def _on_units_changed(self, units: str) -> None: self.prefs["units"] = units self.force_plots.set_fixed_newton(str(units) == "newton") self._persist_prefs() def _on_image_view_mode_changed(self, mode: str) -> None: self.prefs["image_view_mode"] = str(mode) self._last_rendered_view_mode = None self._persist_prefs() if self._running: if self._active_worker is not None: self._render_run_image(self._active_worker.latest()) def _on_plot_settings(self) -> None: dialog = PlotSettingsDialog( self, window_sec=float(self.prefs.get("plot_window_sec") or DEFAULT_PLOT_WINDOW_SEC), fn_max=float(self.prefs.get("plot_fn_max") or DEFAULT_PLOT_FN_MAX), shear_max=float(self.prefs.get("plot_shear_max") or DEFAULT_PLOT_SHEAR_MAX), units=self.toolbar.units(), ) if dialog.exec_() != PlotSettingsDialog.Accepted: return window_sec, fn_max, shear_max = dialog.values() self.prefs["plot_window_sec"] = window_sec self.prefs["plot_fn_max"] = fn_max self.prefs["plot_shear_max"] = shear_max self.force_plots.set_window_sec(window_sec) self.force_plots.set_axis_limits(fn_max, shear_max) self._last_plot_packet_key = None self._persist_prefs() def _apply_plot_prefs_to_panel(self) -> None: self.force_plots.set_window_sec( float(self.prefs.get("plot_window_sec") or DEFAULT_PLOT_WINDOW_SEC) ) self.force_plots.set_axis_limits( float(self.prefs.get("plot_fn_max") or DEFAULT_PLOT_FN_MAX), float(self.prefs.get("plot_shear_max") or DEFAULT_PLOT_SHEAR_MAX), ) self.force_plots.set_fixed_newton(self.toolbar.units() == "newton") def _on_cal_panel_toggled(self, visible: bool) -> None: self._show_cal = bool(visible) self.prefs["show_cal_panel"] = self._show_cal self._apply_panel_visibility() self._persist_prefs() def _on_calibration_exported(self, path: str) -> None: """Apply newly written force_calibration.json to the running Sensor.""" self._ensure_workers() for worker in (self._box_worker, self._finger_worker): if worker is not None and hasattr(worker, "set_pending_calibration"): worker.set_pending_calibration(path) def _on_finger_3d_toggled(self, visible: bool) -> None: if self.sensor_picker.sensor_type() != "finger": return self._show_finger_3d = bool(visible) self.prefs["show_finger_3d"] = self._show_finger_3d self._apply_panel_visibility() self._persist_prefs() def _on_show_fps_toggled(self, visible: bool) -> None: self._show_fps = bool(visible) self.prefs["show_fps"] = self._show_fps self.arrow_panel.set_fps_visible(self._show_fps) if not self._show_fps: self.arrow_panel.set_fps(None) self._persist_prefs() def _on_sensor_type_changed(self, sensor_type: str) -> None: self.prefs["sensor_type"] = str(sensor_type) self.prefs["config_name"] = self.sensor_picker.config_name() self.cal_panel.set_meta_context( sensor_type=str(sensor_type), config_ref=self._resolve_config_ref(), ) self._apply_panel_visibility() self._apply_arrow_rotation_for_sensor() self._update_run_enabled() self._persist_prefs() def _apply_arrow_rotation_for_sensor(self) -> None: enabled = self.sensor_picker.sensor_type() in ARROW_ROTATE_TYPES self.arrow_panel.set_rotation_enabled(enabled) self.act_rotate_arrows.setEnabled(enabled) def _on_rotate_arrows(self) -> None: if self.sensor_picker.sensor_type() not in ARROW_ROTATE_TYPES: return deg = self.arrow_panel.rotate_90_ccw() self.prefs["arrow_rotation_deg"] = deg self._persist_prefs() # ----------------------------------------------------------- shortcuts def _sync_shortcut_actions(self) -> None: if getattr(self, "act_run", None) is None: return self.act_run.setEnabled(self.toolbar.btn_run.isEnabled()) self.act_preview.setEnabled(self.toolbar.btn_preview.isEnabled()) self.act_reset.setEnabled(self.toolbar.btn_reset.isEnabled()) def _on_run_shortcut(self) -> None: if not self.toolbar.btn_run.isEnabled(): return self.toolbar.btn_run.setChecked(not self.toolbar.btn_run.isChecked()) def _on_preview_shortcut(self) -> None: if not self.toolbar.btn_preview.isEnabled(): return self.toolbar.btn_preview.setChecked(not self.toolbar.btn_preview.isChecked()) def _on_config_changed(self, config_name: str) -> None: self.prefs["config_name"] = str(config_name) self._persist_prefs() def _on_browse_video(self) -> None: if self._running or self._preview_open: return start = self.toolbar.video_path() or str(Path.home()) path, _ = QFileDialog.getOpenFileName( self, t("menu.open_video"), start, "Video (*.mp4 *.avi *.mov *.mkv);;All (*)", ) if not path: return self.toolbar.set_source_mode("video") self.toolbar.set_video_path(path) self.prefs["source_mode"] = "video" self.prefs["video_path"] = path self._persist_prefs() self._update_run_enabled() def _on_reveal_config(self) -> None: sensor_type = self.sensor_picker.sensor_type() path = sensor_json_path(sensor_type) if path is None: cfg = self.sensor_picker.config_name() candidate = Path(cfg) path = candidate if candidate.is_file() else None if path is None: self.statusBar().showMessage(t("status.config_missing", type=sensor_type)) return try: _reveal_in_explorer(path) except Exception as exc: self.statusBar().showMessage(t("status.reveal_failed", error=str(exc))) def _resolve_config_ref(self) -> str | None: sensor_type = self.sensor_picker.sensor_type() local = sensor_json_path(sensor_type) if local is not None: return str(local) cfg = self.sensor_picker.config_name() if cfg and Path(cfg).is_file(): return cfg if sensor_type in RUNNABLE_TYPES: return sensor_type return None def _on_search_cameras(self) -> None: if self._preview_open or self._running: return from orisys_app_shared.camera import list_cameras, pick_default_camera preferred = str(self.toolbar.camera_id()) self.statusBar().showMessage(t("toolbar.search_cameras") + "…") QApplication.processEvents() cameras = list_cameras() if not cameras: self.statusBar().showMessage(t("status.camera_none")) return names = ", ".join(cam.label() for cam in cameras) self.statusBar().showMessage( t("status.camera_found", count=len(cameras), names=names) ) selected = pick_default_camera(cameras, preferred) if selected is not None: self.toolbar.set_camera_id(int(selected)) self.prefs["camera_id"] = str(selected) self._persist_prefs() def _on_preview_toggled(self, checked: bool) -> None: if checked: self._start_preview() else: self._stop_preview(announce=True) def _start_preview(self) -> None: if self._running or self.toolbar.source_mode() != "camera": self.toolbar.set_preview_checked(False) return camera_id = self.toolbar.camera_id() try: self._get_preview().open(camera_id) except Exception as exc: msg = t("status.preview_failed", error=str(exc)) self.statusBar().showMessage(msg) self.toolbar.set_preview_checked(False) self.arrow_panel.clear_preview(msg) return self.toolbar.set_source_controls_enabled(False) self.toolbar.btn_preview.setEnabled(True) self._update_run_enabled() self._preview_timer.start() self.statusBar().showMessage(t("status.preview_started", camera_id=camera_id)) def _stop_preview(self, *, announce: bool) -> None: self._preview_timer.stop() if self._preview is not None: self._preview.close() if not self._running: self.toolbar.set_source_controls_enabled(True) self.arrow_panel.clear_preview(t("panels.arrow_empty")) self.toolbar.set_preview_checked(False) self._update_run_enabled() if announce: self.statusBar().showMessage(t("status.preview_stopped")) def _on_preview_tick(self) -> None: if self._preview is None: return frame = self._preview.read() if frame is None: self._stop_preview(announce=True) self.statusBar().showMessage(t("status.preview_failed", error="no frame")) return self.arrow_panel.show_frame(_bgr_to_qimage(frame)) def _on_run_toggled(self, checked: bool) -> None: if checked: self._start_run() else: self._stop_run(reason="user_stop") def _start_run(self) -> None: from orisys_app_shared.acquisition.sdk_worker import orisys_importable sensor_type = self.sensor_picker.sensor_type() if sensor_type not in RUNNABLE_TYPES: self.toolbar.set_run_checked(False) self.statusBar().showMessage(t("status.finger_run_unavailable")) return ok, detail = orisys_importable() if not ok: self.toolbar.set_run_checked(False) QMessageBox.warning(self, t("app.title"), detail) return config_ref = self._resolve_config_ref() if not config_ref: self.toolbar.set_run_checked(False) self.statusBar().showMessage( t("status.config_missing", type=sensor_type) ) return source = self.toolbar.active_source() if source is None: self.toolbar.set_run_checked(False) self.statusBar().showMessage(t("status.video_required")) return if self._preview_open: self._stop_preview(announce=False) self.force_plots.reset() self._last_plot_packet_key = None self._last_rendered_view_mode = None self._last_session_error = "" self._run_source_label = str(source) self._active_sensor_type = sensor_type self._ensure_workers() self._active_worker = ( self._finger_worker if sensor_type == "finger" else self._box_worker ) try: self._active_worker.start(source, config_ref=config_ref) except Exception as exc: report_error(exc, message="Run failed") self.toolbar.set_run_checked(False) self.statusBar().showMessage(t("status.run_failed", error=str(exc))) return self._running = True self._set_phase("connecting") self.cal_panel.set_meta_context(sensor_type=sensor_type, config_ref=config_ref) self.cal_panel.set_running(True) self.toolbar.set_source_controls_enabled(False) self.sensor_picker.set_picker_enabled(False) self.toolbar.set_reset_enabled(False) self._run_timer.start() self.statusBar().showMessage(t("status.connecting")) logger.info( "Connecting type=%s source=%s config=%s", sensor_type, source, config_ref, ) def _stop_run(self, *, reason: str = "user_stop") -> None: self._run_timer.stop() if self._active_worker is not None: self._active_worker.stop() self._running = False self._set_phase("error" if reason == "failed" else "idle") self.cal_panel.set_running(False) self.toolbar.set_run_checked(False) self.toolbar.set_reset_enabled(False) self.toolbar.set_source_controls_enabled(True) self.sensor_picker.set_picker_enabled(True) self._update_run_enabled() keep_frame = reason in ("ended", "disconnected") if not keep_frame: self.arrow_panel.clear_preview(t("panels.arrow_empty")) else: self.arrow_panel.set_fps(None) if reason == "ended": self.statusBar().showMessage(t("status.video_ended")) elif reason == "disconnected": self.statusBar().showMessage(t("status.camera_disconnected")) elif reason == "failed": pass # caller sets status elif reason == "user_stop": self.statusBar().showMessage(t("status.run_stopped")) else: self.statusBar().showMessage(t("status.run_stopped")) def _on_reset(self) -> None: if self._session_phase != "running" or self._active_worker is None: return try: self._active_worker.reset() self.force_plots.reset() self._last_plot_packet_key = None self.statusBar().showMessage(t("status.reset_ok")) except Exception as exc: self.statusBar().showMessage(t("status.run_failed", error=str(exc))) def _on_worker_error(self, message: str) -> None: self._last_session_error = message logger.error("Acquisition error: %s", message) def _on_worker_status(self, message: str) -> None: logger.info("%s", message) def _on_run_tick(self) -> None: if self._active_worker is None: return if self._last_session_error: err = self._last_session_error self._last_session_error = "" self._active_worker.consume_end_reason() self._stop_run(reason="failed") self.statusBar().showMessage(t("status.run_failed", error=err)) return end_reason = self._active_worker.consume_end_reason() if end_reason in ("ended", "disconnected"): self._stop_run(reason=end_reason) return if end_reason == "error": self._stop_run(reason="failed") self.statusBar().showMessage(t("status.run_failed", error="acquisition error")) return if not self._active_worker.is_running and self._running: leftover = self._active_worker.consume_end_reason() if leftover in ("ended", "disconnected"): self._stop_run(reason=leftover) elif leftover == "error": self._stop_run(reason="failed") self.statusBar().showMessage( t("status.run_failed", error="acquisition error") ) else: self._stop_run(reason="user_stop") return packet = self._active_worker.latest() if packet is None: return packet_key = (int(packet.session_id), float(packet.timestamp)) is_new = packet_key != self._last_plot_packet_key if self._session_phase == "connecting" and is_new: self._set_phase("running") self.toolbar.set_reset_enabled(True) self._sync_shortcut_actions() self.statusBar().showMessage( t("status.run_started", source=self._run_source_label) ) if is_new: self._last_plot_packet_key = packet_key display = select_display_forces( units_pref=self.toolbar.units(), force_calibrated=packet.force_calibrated, fn=packet.normal_force, fx=packet.shear_x, fy=packet.shear_y, fn_raw=packet.normal_force_raw, fx_raw=packet.shear_x_raw, fy_raw=packet.shear_y_raw, ) if display.using_raw_fallback: self.statusBar().showMessage(t("status.units_raw_fallback")) self.force_plots.push( packet.timestamp, display.fn, display.fx, display.fy, unit=display.unit, ) if self._show_cal: self.cal_panel.push_raw( packet.normal_force_raw, packet.shear_x_raw, packet.shear_y_raw, packet.timestamp, ) mode = self.arrow_panel.view_mode() if is_new or mode != self._last_rendered_view_mode: self._render_run_image(packet) if self._show_fps and is_new: self.arrow_panel.set_fps(packet.fps) def _render_run_image(self, packet) -> None: if packet is None: return from orisys_app_shared.acquisition.arrows import ( depth_map_to_bgr, draw_arrow_display, flow_colormap, hex_to_bgr, ) mode = self.arrow_panel.view_mode() self._last_rendered_view_mode = mode draw_style = "finger" if self._active_sensor_type == "finger" else "box" colormap = flow_colormap(self._theme) if mode == "depth": depth_bgr = depth_map_to_bgr(packet.depth) frame = depth_bgr if depth_bgr is not None else packet.processed_frame elif mode == "arrows_stitched": frame = draw_arrow_display( packet.processed_frame, packet.flow, background="captured", draw_style=draw_style, valid_mask=packet.valid_mask, colormap=colormap, ) else: canvas_bgr = hex_to_bgr(color(self._theme, "arrow_canvas", "#0B1830")) frame = draw_arrow_display( packet.processed_frame, packet.flow, background="pure_color", canvas_bgr=canvas_bgr, draw_style=draw_style, valid_mask=packet.valid_mask, colormap=colormap, ) self.arrow_panel.show_frame(_bgr_to_qimage(frame)) def _restore_geometry(self) -> None: geo = self.prefs.get("window_geometry") if isinstance(geo, str) and geo: try: from PyQt5.QtCore import QByteArray self.restoreGeometry(QByteArray.fromHex(geo.encode("ascii"))) except Exception: logger.debug("Could not restore window geometry", exc_info=True) def _persist_prefs(self) -> None: geo = self.saveGeometry().toHex().data().decode("ascii") self.prefs["window_geometry"] = geo self.prefs["splitter_sizes"] = [int(v) for v in self.content_splitter.sizes()] self.prefs["theme"] = self._theme self.prefs["language"] = self._language self.prefs["camera_id"] = str(self.toolbar.camera_id()) self.prefs["source_mode"] = self.toolbar.source_mode() self.prefs["video_path"] = self.toolbar.video_path() self.prefs["units"] = self.toolbar.units() self.prefs["image_view_mode"] = self.arrow_panel.view_mode() self.prefs["arrow_rotation_deg"] = self.arrow_panel.rotation_deg() self.prefs.setdefault("plot_window_sec", DEFAULT_PLOT_WINDOW_SEC) self.prefs.setdefault("plot_fn_max", DEFAULT_PLOT_FN_MAX) self.prefs.setdefault("plot_shear_max", DEFAULT_PLOT_SHEAR_MAX) self.prefs["show_cal_panel"] = bool(self._show_cal) self.prefs["show_finger_3d"] = bool(self._show_finger_3d) self.prefs["show_fps"] = bool(self._show_fps) self.prefs["sensor_type"] = self.sensor_picker.sensor_type() self.prefs["config_name"] = self.sensor_picker.config_name() save_preferences(self.prefs) def closeEvent(self, event) -> None: # noqa: N802 self._stop_preview(announce=False) self._stop_run(reason="user_stop") self._persist_prefs() logger.info("Customer shell closed; preferences saved") super().closeEvent(event)