orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix stop button disabled during connection

Browse Source
This commit is contained in:
ykcheungab
2026-09-12 20:27:34 +08:00
parent 904c27125f
commit 85d700bc06
15 changed files with 750 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -33,7 +33,7 @@ python D:\repo\orisys-app\apps\studio\run.py
`orisys_app_shared.force` imports `orisys.calibration` at module scope, so the
studio app cannot start against an SDK build that predates that module
(the published 0.5.1 wheel does not include it). When working from a repo
(the published 0.5.2 wheel does not include it). When working from a repo
checkout, install the SDK from source instead of from an index:
```powershell
apps/studio/config/locales/en_US.json
+10
View File
@@ -26,6 +26,7 @@
"plot_settings": "Plot settings…",
"rotate_arrows": "Rotate arrows 90°",
"run": "Run / Stop",
"stop": "Cancel acquisition",
"preview": "Preview / Stop preview",
"reset": "Reset tracker",
"help": "Help",
@@ -39,6 +40,10 @@
"plot_fn_max_tip": "Normal force axis max (Newton mode)",
"plot_shear_max": "Shear max",
"plot_shear_max_tip": "Shear force axis ±max for FX/FY (Newton mode)",
"plot_fn_raw_max": "FN raw max",
"plot_fn_raw_max_tip": "Normal-force axis max in raw (uncalibrated) mode. 0 = auto range on the data; the default is sized so idle jitter is invisible in the plot",
"plot_shear_raw_max": "Shear raw max",
"plot_shear_raw_max_tip": "Shear-force axis ±max for FX/FY in raw (uncalibrated) mode. 0 = auto range",
"reset_defaults": "Reset to defaults",
"ok": "OK",
"cancel": "Cancel"
@@ -57,6 +62,8 @@
"stop": "Stop",
"run_label": "▶ Run",
"stop_label": "■ Stop",
"cancel_label": "■ Cancel",
"cancel_tip": "Abort the connection (always available while connecting, shortcut Esc)",
"run_unavailable": "Not available yet",
"run_disabled_video": "Select a video file first",
"run_disabled_sensor": "Run is not available for this sensor type",
@@ -134,8 +141,11 @@
"preview_stopped": "Preview stopped",
"preview_failed": "Preview failed: {error}",
"connecting": "Connecting…",
"connect_slow": "This is taking a while — click “Cancel” or press Esc to abort",
"run_started": "Running ({source})",
"run_stopped": "Stopped",
"run_stopping": "Stopping… (releasing the device)",
"run_stopping_cancelled": "Connection cancelled (releasing the device)",
"run_failed": "Run failed: {error}",
"video_ended": "Video ended",
"camera_disconnected": "Camera disconnected",
apps/studio/config/locales/zh_CN.json
+10
View File
@@ -26,6 +26,7 @@
"plot_settings": "绘图设置…",
"rotate_arrows": "旋转箭头图 90°",
"run": "运行 / 停止",
"stop": "取消采集",
"preview": "预览 / 停止预览",
"reset": "重置追踪器",
"help": "帮助",
@@ -39,6 +40,10 @@
"plot_fn_max_tip": "法向力纵轴上限（牛顿模式）",
"plot_shear_max": "剪切上限",
"plot_shear_max_tip": "剪切力 FX/FY 纵轴 ±上限（牛顿模式）",
"plot_fn_raw_max": "FN 原始上限",
"plot_fn_raw_max_tip": "未标定（原始）模式下的法向力纵轴上限。0 = 按数据自动量程；默认值按实测峰值设定，使静止时的抖动在图中不可见",
"plot_shear_raw_max": "原始剪切上限",
"plot_shear_raw_max_tip": "未标定（原始）模式下的剪切力 FX/FY 纵轴 ±上限。0 = 自动量程",
"reset_defaults": "恢复默认",
"ok": "确定",
"cancel": "取消"
@@ -57,6 +62,8 @@
"stop": "停止",
"run_label": "▶ 运行",
"stop_label": "■ 停止",
"cancel_label": "■ 取消连接",
"cancel_tip": "中断连接（连接卡住时随时可用，快捷键 Esc）",
"run_unavailable": "当前不可运行",
"run_disabled_video": "请先选择一个视频文件",
"run_disabled_sensor": "此传感器类型暂不支持运行",
@@ -135,8 +142,11 @@
"preview_stopped": "预览已停止",
"preview_failed": "预览失败: {error}",
"connecting": "正在连接…",
"connect_slow": "连接耗时较长，可点击「取消连接」或按 Esc 中断",
"run_started": "运行中（{source}）",
"run_stopped": "已停止",
"run_stopping": "正在停止…（正在释放设备）",
"run_stopping_cancelled": "已取消连接（正在释放设备）",
"run_failed": "启动失败: {error}",
"video_ended": "视频已结束",
"camera_disconnected": "摄像机已断开",
apps/studio/config/preferences.default.json
+2
View File
@@ -15,6 +15,8 @@
"plot_window_sec": 4.0,
"plot_fn_max": 20.0,
"plot_shear_max": 20.0,
"plot_fn_raw_max": 1000000.0,
"plot_shear_raw_max": 500000.0,
"image_view_mode": "arrows_pure",
"arrow_rotation_deg": 90,
"splitter_sizes": [340, 480, 400]
apps/studio/src/orisys_studio/app_window.py
+68 -7
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import logging
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
@@ -42,7 +43,9 @@ from orisys_studio.ui.force_plots import ForcePlotsPanel
from orisys_studio.ui.mesh3d_panel import Mesh3dPanel
from orisys_studio.ui.plot_settings_dialog import (
DEFAULT_PLOT_FN_MAX,
DEFAULT_PLOT_FN_RAW_MAX,
DEFAULT_PLOT_SHEAR_MAX,
DEFAULT_PLOT_SHEAR_RAW_MAX,
DEFAULT_PLOT_WINDOW_SEC,
PlotSettingsDialog,
)
@@ -61,6 +64,8 @@ MIN_PANEL_WIDTHS = {
"cal": 300,
}
SESSION_PHASES = ("idle", "connecting", "running", "error")
#: After this long in ``connecting``, tell the user that Stop is the way out.
CONNECT_SLOW_HINT_S = 8.0
def _bgr_to_qimage(frame: np.ndarray) -> QImage:
@@ -93,6 +98,8 @@ class StudioMainWindow(QMainWindow):
self._show_fps = bool(self.prefs.get("show_fps", False))
self._session_phase = "idle"
self._running = False
self._connect_started_at = 0.0
self._connect_slow_notified = False
self._preview = None
self._preview_timer = QTimer(self)
self._preview_timer.setInterval(33)
@@ -309,6 +316,11 @@ class StudioMainWindow(QMainWindow):
self.act_run.triggered.connect(self._on_run_shortcut)
self.view_menu.addAction(self.act_run)
self.act_stop = QAction(self)
self.act_stop.setShortcut("Esc")
self.act_stop.triggered.connect(self._on_stop_shortcut)
self.view_menu.addAction(self.act_stop)
self.act_preview = QAction(self)
self.act_preview.setShortcut("F6")
self.act_preview.triggered.connect(self._on_preview_shortcut)
@@ -421,6 +433,7 @@ class StudioMainWindow(QMainWindow):
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_stop.setText(t("menu.stop"))
self.act_preview.setText(t("menu.preview"))
self.act_reset.setText(t("menu.reset"))
self.theme_menu.setTitle(t("menu.theme"))
@@ -446,6 +459,7 @@ class StudioMainWindow(QMainWindow):
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.toolbar.set_run_phase(self._session_phase)
self._refresh_status_chip()
def _refresh_status_chip(self) -> None:
@@ -513,16 +527,24 @@ class StudioMainWindow(QMainWindow):
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),
fn_raw_max=float(
self.prefs.get("plot_fn_raw_max") or DEFAULT_PLOT_FN_RAW_MAX
),
shear_raw_max=float(
self.prefs.get("plot_shear_raw_max") or DEFAULT_PLOT_SHEAR_RAW_MAX
),
units=self.toolbar.units(),
)
if dialog.exec_() != PlotSettingsDialog.Accepted:
return
window_sec, fn_max, shear_max = dialog.values()
window_sec, fn_max, shear_max, fn_raw_max, shear_raw_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.prefs["plot_fn_raw_max"] = fn_raw_max
self.prefs["plot_shear_raw_max"] = shear_raw_max
self.force_plots.set_window_sec(window_sec)
self.force_plots.set_axis_limits(fn_max, shear_max)
self.force_plots.set_axis_limits(fn_max, shear_max, fn_raw_max, shear_raw_max)
self._last_plot_packet_key = None
self._persist_prefs()
@@ -533,6 +555,8 @@ class StudioMainWindow(QMainWindow):
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),
float(self.prefs.get("plot_fn_raw_max") or DEFAULT_PLOT_FN_RAW_MAX),
float(self.prefs.get("plot_shear_raw_max") or DEFAULT_PLOT_SHEAR_RAW_MAX),
)
self.force_plots.set_fixed_newton(self.toolbar.units() == "newton")
@@ -590,10 +614,19 @@ class StudioMainWindow(QMainWindow):
self._persist_prefs()
# ----------------------------------------------------------- shortcuts
@property
def _session_active(self) -> bool:
"""True until the worker has fully settled, not just until Stop was hit."""
if self._running:
return True
worker = self._active_worker
return worker is not None and worker.is_running
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_stop.setEnabled(self._session_active)
self.act_preview.setEnabled(self.toolbar.btn_preview.isEnabled())
self.act_reset.setEnabled(self.toolbar.btn_reset.isEnabled())
@@ -602,6 +635,12 @@ class StudioMainWindow(QMainWindow):
return
self.toolbar.btn_run.setChecked(not self.toolbar.btn_run.isChecked())
def _on_stop_shortcut(self) -> None:
"""Esc must always be able to cancel a session, even a stuck connect."""
if not self._session_active:
return
self._stop_run(reason="user_stop")
def _on_preview_shortcut(self) -> None:
if not self.toolbar.btn_preview.isEnabled():
return
@@ -781,12 +820,16 @@ class StudioMainWindow(QMainWindow):
self._running = True
self._set_phase("connecting")
self._connect_started_at = time.perf_counter()
self._connect_slow_notified = False
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()
# Keep the stop affordance consistent with the phase we just entered.
self._update_run_enabled()
self.statusBar().showMessage(t("status.connecting"))
logger.info(
"Connecting type=%s source=%s config=%s",
@@ -795,10 +838,13 @@ class StudioMainWindow(QMainWindow):
config_ref,
)
def _stop_run(self, *, reason: str = "user_stop") -> None:
def _stop_run(self, *, reason: str = "user_stop", wait: bool = False) -> None:
self._run_timer.stop()
if self._active_worker is not None:
self._active_worker.stop()
# Never wait here: a device open that hangs must not freeze the GUI.
# ``wait=True`` is reserved for shutdown paths.
self._active_worker.stop(wait=wait)
was_connecting = self._session_phase == "connecting"
self._running = False
self._set_phase("error" if reason == "failed" else "idle")
self.cal_panel.set_running(False)
@@ -812,14 +858,18 @@ class StudioMainWindow(QMainWindow):
self.arrow_panel.clear_preview(t("panels.arrow_empty"))
else:
self.arrow_panel.set_fps(None)
winding_down = self._session_active
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"))
elif winding_down and was_connecting:
# The worker thread is still releasing the device in the background.
self.statusBar().showMessage(t("status.run_stopping_cancelled"))
elif winding_down:
self.statusBar().showMessage(t("status.run_stopping"))
else:
self.statusBar().showMessage(t("status.run_stopped"))
@@ -874,6 +924,13 @@ class StudioMainWindow(QMainWindow):
self._stop_run(reason="user_stop")
return
if not self._connect_slow_notified and self._session_phase == "connecting":
# The worker may be stuck opening a device; make sure the user knows
# the (non-blocking) Stop button is the way out.
if time.perf_counter() - self._connect_started_at >= CONNECT_SLOW_HINT_S:
self._connect_slow_notified = True
self.statusBar().showMessage(t("status.connect_slow"))
packet = self._active_worker.latest()
if packet is None:
return
@@ -882,6 +939,7 @@ class StudioMainWindow(QMainWindow):
if self._session_phase == "connecting" and is_new:
self._set_phase("running")
self._connect_slow_notified = True
self.toolbar.set_reset_enabled(True)
self._sync_shortcut_actions()
self.statusBar().showMessage(
@@ -987,6 +1045,8 @@ class StudioMainWindow(QMainWindow):
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.setdefault("plot_fn_raw_max", DEFAULT_PLOT_FN_RAW_MAX)
self.prefs.setdefault("plot_shear_raw_max", DEFAULT_PLOT_SHEAR_RAW_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)
@@ -996,7 +1056,8 @@ class StudioMainWindow(QMainWindow):
def closeEvent(self, event) -> None: # noqa: N802
self._stop_preview(announce=False)
self._stop_run(reason="user_stop")
# Blocking: on the way out we must not leave a device handle behind.
self._stop_run(reason="user_stop", wait=True)
self._persist_prefs()
logger.info("Studio shell closed; preferences saved")
super().closeEvent(event)
apps/studio/src/orisys_studio/ui/force_plots.py
+38 -7
View File
@@ -63,6 +63,8 @@ class ForcePlotsPanel(QFrame):
self._theme = theme
self._fn_max = 20.0
self._shear_max = 20.0
self._fn_raw_max = 1000000.0
self._shear_raw_max = 500000.0
self._fixed_newton = True
self._unit = "raw"
self._last_values: tuple[float, float, float] = (0.0, 0.0, 0.0)
@@ -197,9 +199,19 @@ class ForcePlotsPanel(QFrame):
self._clear_curves()
self._apply_axis_mode()
def set_axis_limits(self, fn_max: float, shear_max: float) -> None:
def set_axis_limits(
self,
fn_max: float,
shear_max: float,
fn_raw_max: float | None = None,
shear_raw_max: float | None = None,
) -> None:
self._fn_max = max(0.1, float(fn_max))
self._shear_max = max(0.1, float(shear_max))
if fn_raw_max is not None:
self._fn_raw_max = max(0.0, float(fn_raw_max))
if shear_raw_max is not None:
self._shear_raw_max = max(0.0, float(shear_raw_max))
self._apply_axis_mode()
def set_fixed_newton(self, enabled: bool) -> None:
@@ -247,17 +259,36 @@ class ForcePlotsPanel(QFrame):
self.plots[i].setYRange(*state.y_ranges[i], padding=0.0)
def _uses_fixed_y(self) -> bool:
return bool(self._fixed_newton) and self._unit == "N"
"""Fixed Y whenever we have limits for the active unit.
Raw mode used to auto-range, which collapsed the axis onto the idle
jitter and then rescaled on every contact. A fixed raw range keeps the
baseline visually flat: the raw maxima are sized so the idle jitter is
well under one pixel.
"""
if self._unit == "N":
return bool(self._fixed_newton)
return self._fn_raw_max > 0.0 and self._shear_raw_max > 0.0
def _fixed_y_range(self, channel_index: int) -> tuple[float, float]:
if self._unit == "N":
fn_max, shear_max = self._fn_max, self._shear_max
else:
fn_max, shear_max = self._fn_raw_max, self._shear_raw_max
if channel_index == 0:
return 0.0, self._fn_max
return -self._shear_max, self._shear_max
return 0.0, fn_max
return -shear_max, shear_max
def _preview_fixed_y(self) -> bool:
"""Whether to paint fixed axes before any sample has arrived."""
if self.history.sample_count > 0:
return self._uses_fixed_y()
if self._unit == "N":
return bool(self._fixed_newton)
return True
def _apply_axis_mode(self) -> None:
# Preview fixed newton axes when newton mode is selected, even before samples.
preview_fixed = bool(self._fixed_newton)
use_fixed = self._uses_fixed_y() or (preview_fixed and self.history.sample_count == 0)
use_fixed = self._preview_fixed_y()
for i, plot in enumerate(self.plots):
if use_fixed:
ymin, ymax = self._fixed_y_range(i)
apps/studio/src/orisys_studio/ui/plot_settings_dialog.py
+44 -6
View File
@@ -18,6 +18,10 @@ from orisys_app_shared.i18n import t
DEFAULT_PLOT_WINDOW_SEC = 4.0
DEFAULT_PLOT_FN_MAX = 20.0
DEFAULT_PLOT_SHEAR_MAX = 20.0
# Raw (uncalibrated) HHD units: sized from measured peaks so the idle jitter
# stays sub-pixel while a full contact still fits (see prefs store).
DEFAULT_PLOT_FN_RAW_MAX = 1000000.0
DEFAULT_PLOT_SHEAR_RAW_MAX = 500000.0
class PlotSettingsDialog(QDialog):
@@ -28,11 +32,13 @@ class PlotSettingsDialog(QDialog):
window_sec: float = DEFAULT_PLOT_WINDOW_SEC,
fn_max: float = DEFAULT_PLOT_FN_MAX,
shear_max: float = DEFAULT_PLOT_SHEAR_MAX,
fn_raw_max: float = DEFAULT_PLOT_FN_RAW_MAX,
shear_raw_max: float = DEFAULT_PLOT_SHEAR_RAW_MAX,
units: str = "newton",
) -> None:
super().__init__(parent)
self.setModal(True)
self.resize(360, 200)
self.resize(380, 260)
root = QVBoxLayout(self)
form = QFormLayout()
@@ -61,16 +67,40 @@ class PlotSettingsDialog(QDialog):
self.shear_max_spin.setSuffix(" N")
self.shear_max_spin.setValue(float(shear_max))
# Raw limits: 0 means "auto range on the data".
self.fn_raw_max_label = QLabel()
self.fn_raw_max_spin = QDoubleSpinBox()
self.fn_raw_max_spin.setRange(0.0, 1e9)
self.fn_raw_max_spin.setSingleStep(50000.0)
self.fn_raw_max_spin.setDecimals(0)
self.fn_raw_max_spin.setValue(float(fn_raw_max))
self.shear_raw_max_label = QLabel()
self.shear_raw_max_spin = QDoubleSpinBox()
self.shear_raw_max_spin.setRange(0.0, 1e9)
self.shear_raw_max_spin.setSingleStep(50000.0)
self.shear_raw_max_spin.setDecimals(0)
self.shear_raw_max_spin.setValue(float(shear_raw_max))
form.addRow(self.window_label, self.window_spin)
form.addRow(self.fn_max_label, self.fn_max_spin)
form.addRow(self.shear_max_label, self.shear_max_spin)
form.addRow(self.fn_raw_max_label, self.fn_raw_max_spin)
form.addRow(self.shear_raw_max_label, self.shear_raw_max_spin)
root.addLayout(form)
newton = str(units or "newton").lower().startswith("n")
self.fn_max_spin.setEnabled(newton)
self.shear_max_spin.setEnabled(newton)
self.fn_max_label.setEnabled(newton)
self.shear_max_label.setEnabled(newton)
raw_widgets = (
self.fn_raw_max_spin,
self.shear_raw_max_spin,
self.fn_raw_max_label,
self.shear_raw_max_label,
)
for widget in (self.fn_max_spin, self.shear_max_spin, self.fn_max_label,
self.shear_max_label):
widget.setEnabled(newton)
for widget in raw_widgets:
widget.setEnabled(not newton)
buttons_row = QHBoxLayout()
self.btn_reset = QPushButton()
@@ -85,17 +115,21 @@ class PlotSettingsDialog(QDialog):
self.retranslate()
def values(self) -> tuple[float, float, float]:
def values(self) -> tuple[float, float, float, float, float]:
return (
float(self.window_spin.value()),
float(self.fn_max_spin.value()),
float(self.shear_max_spin.value()),
float(self.fn_raw_max_spin.value()),
float(self.shear_raw_max_spin.value()),
)
def _reset_defaults(self) -> None:
self.window_spin.setValue(DEFAULT_PLOT_WINDOW_SEC)
self.fn_max_spin.setValue(DEFAULT_PLOT_FN_MAX)
self.shear_max_spin.setValue(DEFAULT_PLOT_SHEAR_MAX)
self.fn_raw_max_spin.setValue(DEFAULT_PLOT_FN_RAW_MAX)
self.shear_raw_max_spin.setValue(DEFAULT_PLOT_SHEAR_RAW_MAX)
def retranslate(self) -> None:
self.setWindowTitle(t("dialog.plot_settings_title"))
@@ -105,6 +139,10 @@ class PlotSettingsDialog(QDialog):
self.fn_max_spin.setToolTip(t("dialog.plot_fn_max_tip"))
self.shear_max_label.setText(t("dialog.plot_shear_max"))
self.shear_max_spin.setToolTip(t("dialog.plot_shear_max_tip"))
self.fn_raw_max_label.setText(t("dialog.plot_fn_raw_max"))
self.fn_raw_max_spin.setToolTip(t("dialog.plot_fn_raw_max_tip"))
self.shear_raw_max_label.setText(t("dialog.plot_shear_raw_max"))
self.shear_raw_max_spin.setToolTip(t("dialog.plot_shear_raw_max_tip"))
self.btn_reset.setText(t("dialog.reset_defaults"))
ok_btn = self.button_box.button(QDialogButtonBox.Ok)
cancel_btn = self.button_box.button(QDialogButtonBox.Cancel)
apps/studio/src/orisys_studio/ui/toolbar.py
+16 -2
View File
@@ -59,6 +59,7 @@ class ToolbarWidget(QWidget):
super().__init__(parent)
self._run_tooltip_reason = ""
self._run_available = True
self._run_phase = "idle"
outer = QVBoxLayout(self)
outer.setContentsMargins(0, 0, 0, 0)
outer.setSpacing(0)
@@ -244,6 +245,11 @@ class ToolbarWidget(QWidget):
self._run_tooltip_reason = str(reason or "")
self._apply_run_tooltip()
def set_run_phase(self, phase: str) -> None:
"""Track the session phase so Stop can advertise itself while connecting."""
self._run_phase = str(phase or "idle")
self.refresh_labels()
def active_source(self) -> str | int | None:
if self.source_mode() == "video":
path = self.video_path()
@@ -307,7 +313,12 @@ class ToolbarWidget(QWidget):
def _refresh_run_label(self) -> None:
if self.btn_run.isChecked():
self.btn_run.setText(t("toolbar.stop_label"))
# While connecting, cancel is the action the user actually needs —
# a device open can hang, and this is the way out of it.
if self._run_phase == "connecting":
self.btn_run.setText(t("toolbar.cancel_label"))
else:
self.btn_run.setText(t("toolbar.stop_label"))
else:
self.btn_run.setText(t("toolbar.run_label"))
self.btn_run.setObjectName(
@@ -319,7 +330,10 @@ class ToolbarWidget(QWidget):
def _apply_run_tooltip(self) -> None:
if self.btn_run.isChecked():
self.btn_run.setToolTip(t("toolbar.stop"))
if self._run_phase == "connecting":
self.btn_run.setToolTip(t("toolbar.cancel_tip"))
else:
self.btn_run.setToolTip(t("toolbar.stop"))
elif not self._run_available and self._run_tooltip_reason:
self.btn_run.setToolTip(self._run_tooltip_reason)
elif not self._run_available:
apps/studio/tests/test_ui_smoke.py
+46
View File
@@ -201,6 +201,52 @@ def test_toolbar_run_button_is_labelled(make_window):
assert window.toolbar.btn_run.text() != label
def test_stop_cancels_a_hung_connect(make_window):
"""Stop must stay live and take effect instantly while connecting."""
import time
import ui_check
from orisys_app_shared.i18n import t
window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
stuck = ui_check._StuckWorker()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window.toolbar.btn_run.setChecked(True)
assert window._session_phase == "connecting"
assert window.toolbar.btn_run.isEnabled(), "Stop is disabled while connecting"
assert window.toolbar.btn_run.text() == t("toolbar.cancel_label")
assert window.act_stop.isEnabled()
started = time.perf_counter()
window.toolbar.btn_run.setChecked(False)
elapsed = time.perf_counter() - started
assert elapsed < 0.5, f"Stop blocked the GUI for {elapsed:.2f}s"
assert not window._running
assert stuck.thread_still_owned, "Stop abandoned the still-running worker thread"
def test_escape_cancels_a_hung_connect(make_window):
import time
import ui_check
window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
stuck = ui_check._StuckWorker()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window.toolbar.btn_run.setChecked(True)
assert window.act_stop.shortcut().toString() == "Esc"
started = time.perf_counter()
window._on_stop_shortcut()
assert time.perf_counter() - started < 0.5
assert not window._running
def test_theme_switch_updates_panels(make_window):
window = _show(make_window(), (1400, 880))
window._set_theme("light")
apps/studio/tools/ui_check.py
+119
View File
@@ -15,6 +15,8 @@ from __future__ import annotations
import os
import sys
import threading
import time
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[1]
@@ -45,6 +47,55 @@ def section(name: str) -> None:
print(f"\n== {name} ==")
class _StuckWorker:
"""A worker whose device open never returns.
Mirrors the real worker's non-blocking ``stop()`` contract so the window can
be exercised against the failure mode the Stop button exists to survive.
"""
def __init__(self) -> None:
self._thread: threading.Thread | None = None
self._end_reason: str | None = None
self.thread_still_owned = False
self.stop_blocked_for = 0.0
def start(self, _source, *, config_ref=None) -> None:
self._thread = threading.Thread(target=self._hang, daemon=True)
self._thread.start()
def _hang(self) -> None:
time.sleep(600.0)
@property
def is_running(self) -> bool:
return self._thread is not None and self._thread.is_alive()
@property
def phase(self) -> str:
return "connecting"
def latest(self):
return None
def consume_end_reason(self):
reason, self._end_reason = self._end_reason, None
return reason
def reset(self) -> None:
return None
def stop(self, *, wait: bool = False) -> None:
self._end_reason = self._end_reason or "stopped"
started = time.perf_counter()
if wait and self._thread is not None and self._thread.is_alive():
self._thread.join(timeout=5.0)
if self._thread is None or not self._thread.is_alive():
self._thread = None
self.thread_still_owned = self._thread is not None
self.stop_blocked_for = time.perf_counter() - started
# --------------------------------------------------------------------------
# Palette rules
#
@@ -440,6 +491,74 @@ def main() -> int:
window.close()
app.processEvents()
section("stopping a hung connect")
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
window.resize(1440, 880)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
app.processEvents()
check(not window.act_stop.isEnabled(), "Stop shortcut is live before a session")
window.toolbar.btn_run.setChecked(True)
app.processEvents()
check(window._session_phase == "connecting", "run did not enter the connecting phase")
check(
window.toolbar.btn_run.isEnabled(),
"Stop is disabled while connecting — a hung connect could not be cancelled",
)
check(window.toolbar.btn_run.isChecked(), "run button did not latch into the stop state")
check(
window.toolbar.btn_run.text() == t("toolbar.cancel_label"),
f"connecting label is not the cancel label: {window.toolbar.btn_run.text()!r}",
)
check(
window.toolbar.btn_run.toolTip() == t("toolbar.cancel_tip"),
"connecting tooltip does not explain how to cancel",
)
check(window.act_stop.isEnabled(), "Stop shortcut is disabled while connecting")
started = time.perf_counter()
window.toolbar.btn_run.setChecked(False)
elapsed = time.perf_counter() - started
check(elapsed < 0.5, f"Stop blocked the GUI for {elapsed:.2f}s on a hung connect")
check(not window._running, "session stayed active after Stop")
check(window._session_phase == "idle", "phase did not return to idle after Stop")
check(
stuck.thread_still_owned,
"Stop abandoned the still-running worker thread",
)
# The slow-connect hint must point at the escape hatch.
window.toolbar.btn_run.setChecked(True)
app.processEvents()
window._connect_started_at = time.perf_counter() - 3600.0
window._on_run_tick()
check(
window.statusBar().currentMessage() == t("status.connect_slow"),
f"no slow-connect hint: {window.statusBar().currentMessage()!r}",
)
escaped = time.perf_counter()
window._on_stop_shortcut()
check(
time.perf_counter() - escaped < 0.5,
"Esc did not cancel the hung connect immediately",
)
check(not window._running, "Esc left the session running")
window.close()
app.processEvents()
print(f"\nscenes: {scene_count} checks: {CHECKS} failures: {len(FAILURES)}")
for failure in FAILURES:
print(f" FAIL: {failure}")
apps/studio/tools/ui_snapshot.py
+20 -1
View File
@@ -21,7 +21,7 @@ from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = APP_ROOT.parents[1]
for _path in (APP_ROOT / "src", REPO_ROOT / "shared" / "src"):
for _path in (APP_ROOT / "src", REPO_ROOT / "shared" / "src", APP_ROOT / "tools"):
if _path.is_dir() and str(_path) not in sys.path:
sys.path.insert(0, str(_path))
@@ -96,6 +96,12 @@ def main() -> int:
parser.add_argument("--size", default="1440x880", help="WxH window size")
parser.add_argument("--view", default="arrows_pure")
parser.add_argument("--demo", action="store_true", help="inject demo plot/table data")
parser.add_argument(
"--scene",
choices=("idle", "connecting"),
default="idle",
help="connecting renders the Stop/cancel state of a hung device open",
)
parser.add_argument("--out", required=True, help="output PNG path")
args = parser.parse_args()
@@ -135,6 +141,19 @@ def main() -> int:
for _ in range(4):
app.processEvents()
if args.scene == "connecting":
# A device open that never returns: the state the Stop button must
# survive. ``ui_check._StuckWorker`` drives the same window state.
from ui_check import _StuckWorker
stuck = _StuckWorker()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window.toolbar.btn_run.setChecked(True)
for _ in range(4):
app.processEvents()
out = Path(args.out)
out.parent.mkdir(parents=True, exist_ok=True)
ok = window.grab().save(str(out))
shared/src/orisys_app_shared/acquisition/finger_worker.py
+67 -18
View File
@@ -15,6 +15,8 @@ from orisys_app_shared.acquisition.constants import (
)
from orisys_app_shared.acquisition.sdk_worker import (
CAMERA_DISCONNECT_TIMEOUT_S,
HANDOVER_TIMEOUT_S,
STOP_JOIN_TIMEOUT_S,
VIDEO_EOF_TIMEOUT_S,
FixedCameraMode,
SdkPacket,
@@ -67,7 +69,9 @@ class FingerAcquisitionWorker:
self._lock = threading.Lock()
self._latest: Optional[SdkPacket] = None
self._session_id = 0
self._runtime = None
# Keyed by session so a session that is still winding down can only
# ever close its own runtime, never the runtime of a newer session.
self._runtimes: dict[int, object] = {}
self._mode: Optional[FixedCameraMode] = None
self._video_source: Optional[str | int] = None
self._export_dir: Optional[Path] = None
@@ -100,7 +104,10 @@ class FingerAcquisitionWorker:
*,
config_ref: str | Path,
) -> FixedCameraMode:
# Non-blocking: a previous session that is stuck opening must never hold
# the caller (the GUI thread) hostage.
self.stop()
previous = self._thread
ok, detail = orisys_importable()
if not ok:
raise RuntimeError(f"orisys SDK is not importable: {detail}")
@@ -119,26 +126,41 @@ class FingerAcquisitionWorker:
target=self._run,
name=f"FingerAcquisition-{session}",
daemon=True,
args=(session,),
args=(session, previous),
)
self._thread.start()
return mode
def stop(self) -> None:
def stop(self, *, wait: bool = False) -> None:
"""Ask the acquisition thread to stop.
The default is deliberately **non-blocking**: an open that never returns
must not freeze the caller, and the worker thread owns its own teardown.
Use ``wait=True`` only on shutdown paths that must not leave the runtime
behind.
"""
self._set_end_reason("stopped")
self._stop.set()
thread = self._thread
if thread is not None and thread.is_alive():
thread.join(timeout=8.0)
if wait and thread is not None and thread.is_alive():
thread.join(timeout=STOP_JOIN_TIMEOUT_S)
if thread is None or not thread.is_alive():
self._settle(owner=thread)
# Otherwise the thread settles itself on the way out.
def _settle(self, *, owner: Optional[threading.Thread]) -> None:
"""Release worker state once no thread owns the runtime any more."""
if owner is not None and self._thread is not owner:
return # a newer session owns the worker now
self._thread = None
self._close_runtime()
self._close_all_runtimes()
with self._lock:
self._latest = None
self._phase = "idle"
def reset(self) -> None:
with self._lock:
runtime = self._runtime
runtime = self._runtimes.get(self._session_id)
if runtime is None:
raise RuntimeError("Finger runtime is not running")
runtime.tactile.reset()
@@ -166,9 +188,10 @@ class FingerAcquisitionWorker:
with self._lock:
self._phase = phase
def _close_runtime(self) -> None:
runtime = self._runtime
self._runtime = None
def _close_runtime(self, session: int) -> None:
"""Close the runtime opened by ``session`` (no-op if it owns none)."""
with self._lock:
runtime = self._runtimes.pop(session, None)
if runtime is None:
return
try:
@@ -176,13 +199,38 @@ class FingerAcquisitionWorker:
except Exception:
pass
def _close_all_runtimes(self) -> None:
with self._lock:
sessions = list(self._runtimes)
for session in sessions:
self._close_runtime(session)
def _status(self, message: str) -> None:
if self._on_status is not None:
self._on_status(message)
def _run(self, session: int) -> None:
def _run(self, session: int, previous: Optional[threading.Thread] = None) -> None:
try:
self._run_session(session, previous)
finally:
# The finishing thread owns the teardown — this is what lets
# ``stop()`` return immediately without leaking the runtime.
self._close_runtime(session)
if self._thread is threading.current_thread():
self._thread = None
with self._lock:
self._latest = None
self._phase = "idle"
def _run_session(self, session: int, previous: Optional[threading.Thread]) -> None:
from orisys.finger import FingerRuntime
if previous is not None and previous.is_alive():
# Wait here rather than on the GUI thread, so the runtime is never
# opened twice at the same time.
self._status("Waiting for the previous session to finish…")
previous.join(timeout=HANDOVER_TIMEOUT_S)
mode = self._mode
source = self._video_source
if mode is None or source is None:
@@ -197,15 +245,17 @@ class FingerAcquisitionWorker:
}
if self._export_dir is not None:
open_kw["export_dir"] = str(self._export_dir)
self._runtime = FingerRuntime.open(**open_kw)
runtime = FingerRuntime.open(**open_kw)
with self._lock:
self._runtimes[session] = runtime
if self._stop.is_set() or session != self._session_id:
self._close_runtime()
self._close_runtime(session)
return
self._runtime.warmup(color=False)
runtime.warmup(color=False)
if self._stop.is_set() or session != self._session_id:
self._close_runtime()
self._close_runtime(session)
return
raw_sensor = getattr(self._runtime.tactile, "raw_sensor", None)
raw_sensor = getattr(runtime.tactile, "raw_sensor", None)
if raw_sensor is not None and hasattr(raw_sensor, "configure"):
try:
raw_sensor.configure(
@@ -226,7 +276,6 @@ class FingerAcquisitionWorker:
saw_frame = False
last_frame_mono = 0.0
none_streak_start: Optional[float] = None
runtime = self._runtime
while not self._stop.is_set() and session == self._session_id and runtime is not None:
try:
@@ -305,7 +354,7 @@ class FingerAcquisitionWorker:
self._on_error(str(exc))
break
self._close_runtime()
self._close_runtime(session)
self._set_phase("idle")
if self._stop.is_set():
self._set_end_reason("stopped")
shared/src/orisys_app_shared/acquisition/sdk_worker.py
+78 -23
View File
@@ -19,6 +19,11 @@ from orisys_app_shared.acquisition.constants import (
VIDEO_EOF_TIMEOUT_S = 0.5
CAMERA_DISCONNECT_TIMEOUT_S = 1.5
#: How long a *blocking* ``stop(wait=True)`` may hold the caller.
STOP_JOIN_TIMEOUT_S = 5.0
#: How long a new session waits (in its own thread, never on the GUI) for a
#: previous session that is still winding down before opening the device.
HANDOVER_TIMEOUT_S = 60.0
@dataclass(frozen=True)
@@ -223,7 +228,9 @@ class SdkAcquisitionWorker:
self._lock = threading.Lock()
self._latest: Optional[SdkPacket] = None
self._session_id = 0
self._sensor = None
# Keyed by session so a session that is still winding down can only
# ever close its own device, never the device of a newer session.
self._sensors: dict[int, object] = {}
self._mode: Optional[FixedCameraMode] = None
self._video_source: Optional[str | int] = None
self._end_reason: Optional[str] = None
@@ -255,7 +262,10 @@ class SdkAcquisitionWorker:
*,
config_ref: str | Path,
) -> FixedCameraMode:
# Non-blocking: a previous session that is stuck opening the device must
# never hold the caller (the GUI thread) hostage.
self.stop()
previous = self._thread
ok, detail = orisys_importable()
if not ok:
raise RuntimeError(f"orisys SDK is not importable: {detail}")
@@ -269,30 +279,47 @@ class SdkAcquisitionWorker:
self._session_id += 1
session = self._session_id
self._stop.clear()
self._thread = threading.Thread(
thread = threading.Thread(
target=self._run,
name=f"SdkAcquisition-{session}",
daemon=True,
args=(session,),
args=(session, previous),
)
self._thread.start()
self._thread = thread
thread.start()
return mode
def stop(self) -> None:
def stop(self, *, wait: bool = False) -> None:
"""Ask the acquisition thread to stop.
The default is deliberately **non-blocking**: a device open that never
returns must not freeze the caller, and the worker thread owns its own
teardown. Use ``wait=True`` only on shutdown paths that must not leave a
device handle behind.
"""
self._set_end_reason("stopped")
self._stop.set()
thread = self._thread
if thread is not None and thread.is_alive():
thread.join(timeout=5.0)
if wait and thread is not None and thread.is_alive():
thread.join(timeout=STOP_JOIN_TIMEOUT_S)
if thread is None or not thread.is_alive():
self._settle(owner=thread)
# Otherwise the thread settles itself on the way out. Leaving
# ``self._thread`` set is what stops the next ``start`` from racing it.
def _settle(self, *, owner: Optional[threading.Thread]) -> None:
"""Release worker state once no thread owns the device any more."""
if owner is not None and self._thread is not owner:
return # a newer session owns the worker now
self._thread = None
self._close_sensor()
self._close_all_sensors()
with self._lock:
self._latest = None
self._phase = "idle"
def reset(self) -> None:
with self._lock:
sensor = self._sensor
sensor = self._sensors.get(self._session_id)
if sensor is None:
raise RuntimeError("Sensor is not running")
sensor.reset()
@@ -320,9 +347,10 @@ class SdkAcquisitionWorker:
with self._lock:
self._phase = phase
def _close_sensor(self) -> None:
sensor = self._sensor
self._sensor = None
def _close_sensor(self, session: int) -> None:
"""Disconnect the device opened by ``session`` (no-op if it owns none)."""
with self._lock:
sensor = self._sensors.pop(session, None)
if sensor is None:
return
try:
@@ -330,13 +358,38 @@ class SdkAcquisitionWorker:
except Exception:
pass
def _close_all_sensors(self) -> None:
with self._lock:
sessions = list(self._sensors)
for session in sessions:
self._close_sensor(session)
def _status(self, message: str) -> None:
if self._on_status is not None:
self._on_status(message)
def _run(self, session: int) -> None:
def _run(self, session: int, previous: Optional[threading.Thread] = None) -> None:
try:
self._run_session(session, previous)
finally:
# The finishing thread owns the teardown — this is what lets
# ``stop()`` return immediately without leaking the device.
self._close_sensor(session)
if self._thread is threading.current_thread():
self._thread = None
with self._lock:
self._latest = None
self._phase = "idle"
def _run_session(self, session: int, previous: Optional[threading.Thread]) -> None:
import orisys
if previous is not None and previous.is_alive():
# Wait here rather than on the GUI thread, so the device is never
# opened twice at the same time.
self._status("Waiting for the previous session to finish…")
previous.join(timeout=HANDOVER_TIMEOUT_S)
mode = self._mode
source = self._video_source
if mode is None or source is None:
@@ -344,16 +397,18 @@ class SdkAcquisitionWorker:
is_file = _is_file_source(source)
try:
self._status(f"Opening Orisys Sensor ({mode.label()})…")
self._sensor = orisys.Sensor(
sensor = orisys.Sensor(
source,
config_name=mode.config_ref,
verbose=False,
backend="auto",
)
with self._lock:
self._sensors[session] = sensor
if self._stop.is_set() or session != self._session_id:
self._close_sensor()
self._close_sensor(session)
return
self._sensor.configure(
sensor.configure(
lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ,
lowpass_sample_rate_hz=FORCE_SAMPLE_RATE_HZ,
)
@@ -386,13 +441,13 @@ class SdkAcquisitionWorker:
break
t0 = time.perf_counter()
processed = self._sensor.get_img()
processed = sensor.get_img()
if processed is None:
if saw_frame and is_file and none_streak_start is None:
none_streak_start = time.perf_counter()
time.sleep(0.005)
continue
raw = getattr(self._sensor, "frame", None)
raw = getattr(sensor, "frame", None)
if raw is None:
if saw_frame and is_file and none_streak_start is None:
none_streak_start = time.perf_counter()
@@ -401,9 +456,9 @@ class SdkAcquisitionWorker:
none_streak_start = None
try:
self._consume_pending_calibration(self._sensor)
self._sensor.compute_deformation(decompose=True, check_motion=False)
self._sensor.compute_contact()
self._consume_pending_calibration(sensor)
sensor.compute_deformation(decompose=True, check_motion=False)
sensor.compute_contact()
except Exception:
pass
t1 = time.perf_counter()
@@ -411,7 +466,7 @@ class SdkAcquisitionWorker:
cutoff = t1 - 1.0
frame_times = [t for t in frame_times if t >= cutoff]
fps = float(len(frame_times)) if frame_times else 0.0
forces = read_forces_from_sensor(self._sensor, config_ref=mode.config_ref)
forces = read_forces_from_sensor(sensor, config_ref=mode.config_ref)
packet = SdkPacket(
raw_frame=_ensure_bgr(raw),
processed_frame=_ensure_bgr(processed),
@@ -441,7 +496,7 @@ class SdkAcquisitionWorker:
self._on_error(str(exc))
break
self._close_sensor()
self._close_sensor(session)
self._set_phase("idle")
if self._stop.is_set():
self._set_end_reason("stopped")
shared/src/orisys_app_shared/prefs/store.py
+5
View File
@@ -24,6 +24,11 @@ _DEFAULTS: dict[str, Any] = {
"plot_window_sec": 4.0,
"plot_fn_max": 20.0,
"plot_shear_max": 20.0,
# Raw (uncalibrated) HHD units. Sized so the idle jitter is sub-pixel on
# both raw-mode products while never clipping: measured peaks are ~447k
# (mini, 365x365) and ~41k (finger, 280x300); idle sigma is ~80 (finger).
"plot_fn_raw_max": 1000000.0,
"plot_shear_raw_max": 500000.0,
"image_view_mode": "arrows_pure",
"arrow_rotation_deg": 90,
"show_fps": False,
shared/tests/test_worker_cancellation.py
+226
View File
@@ -0,0 +1,226 @@
"""Cancellation semantics for the acquisition workers.
The GUI must never block on a device open that hangs: ``stop()`` returns
immediately, the worker thread owns its own teardown, and a new session waits
for the previous one *in its own thread* instead of opening the device twice.
These tests drive the real worker loop with a fake ``orisys`` module, so they
need no camera and no sensor bundle beyond a ``sensor.json``.
"""
from __future__ import annotations
import sys
import threading
import time
import types
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[2]
BOX_CONFIG = APP_ROOT / "apps" / "studio" / "config" / "sensors" / "box" / "sensor.json"
STOP_BUDGET_S = 0.5
class FakeSensor:
"""Blocks in the constructor until ``release`` is set."""
def __init__(self, release: threading.Event, tracker: "OpenTracker", **kwargs):
tracker.opened(self)
if not release.wait(timeout=10.0):
raise TimeoutError("fake sensor was never released")
self.configured: dict = {}
self.connected = True
def configure(self, **kwargs):
self.configured = kwargs
def get_img(self):
return None
def disconnect(self):
self.connected = False
class OpenTracker:
"""Records how many devices are open at the same time."""
def __init__(self) -> None:
self.lock = threading.Lock()
self.open_count = 0
self.peak_open = 0
self.instances: list[FakeSensor] = []
def opened(self, sensor: FakeSensor) -> None:
with self.lock:
self.instances.append(sensor)
self.open_count += 1
self.peak_open = max(self.peak_open, self.open_count)
def closed(self, sensor: FakeSensor) -> None:
with self.lock:
if sensor.connected:
sensor.connected = False
self.open_count -= 1
def _install_fake_orisys(tracker: OpenTracker, release: threading.Event) -> object:
"""Replace ``orisys`` (and ``orisys.finger``) with blocking fakes."""
saved = {name: sys.modules.get(name) for name in ("orisys", "orisys.finger")}
class _Sensor(FakeSensor):
def __init__(self, *args, **kwargs):
super().__init__(release, tracker, **kwargs)
def disconnect(self):
tracker.closed(self)
super().disconnect()
module = types.ModuleType("orisys")
module.__version__ = "fake"
module.Sensor = _Sensor # type: ignore[attr-defined]
class _Runtime:
def __init__(self, *args, **kwargs):
self._door = _Sensor()
self.tactile = types.SimpleNamespace(raw_sensor=None)
@classmethod
def open(cls, **kwargs):
return cls(**kwargs)
def warmup(self, **_kwargs):
return None
def close(self):
self._door.disconnect()
finger_module = types.ModuleType("orisys.finger")
finger_module.FingerRuntime = _Runtime # type: ignore[attr-defined]
module.finger = finger_module # type: ignore[attr-defined]
sys.modules["orisys"] = module
sys.modules["orisys.finger"] = finger_module
return saved
def _restore(saved) -> None:
for name, original in saved.items():
if original is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = original
def _new_worker(kind: str):
from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker
from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker
if kind == "finger":
return FingerAcquisitionWorker()
return SdkAcquisitionWorker()
def _wait_until(predicate, timeout=5.0, interval=0.01) -> bool:
deadline = time.perf_counter() + timeout
while time.perf_counter() < deadline:
if predicate():
return True
time.sleep(interval)
return predicate()
def _stop_is_immediate_and_keeps_ownership(kind: str) -> None:
tracker = OpenTracker()
release = threading.Event()
saved = _install_fake_orisys(tracker, release)
try:
worker = _new_worker(kind)
worker.start("0", config_ref=BOX_CONFIG)
assert _wait_until(lambda: tracker.peak_open >= 1), "session never opened the device"
started = time.perf_counter()
worker.stop()
elapsed = time.perf_counter() - started
assert elapsed < STOP_BUDGET_S, f"stop() blocked the caller for {elapsed:.2f}s"
assert worker._thread is not None, "stop() abandoned a still-running thread"
assert worker.consume_end_reason() == "stopped"
release.set()
assert _wait_until(lambda: worker._thread is None), "thread never released ownership"
assert _wait_until(lambda: tracker.open_count == 0), "device was not closed"
finally:
release.set()
_restore(saved)
def test_sdk_stop_is_immediate_and_keeps_ownership():
_stop_is_immediate_and_keeps_ownership("sdk")
def test_finger_stop_is_immediate_and_keeps_ownership():
_stop_is_immediate_and_keeps_ownership("finger")
def test_second_session_waits_instead_of_double_opening():
tracker = OpenTracker()
release = threading.Event()
saved = _install_fake_orisys(tracker, release)
try:
worker = _new_worker("sdk")
worker.start("0", config_ref=BOX_CONFIG)
assert _wait_until(lambda: tracker.peak_open == 1)
worker.stop()
worker.start("0", config_ref=BOX_CONFIG) # user hits Run again while stuck
time.sleep(0.3)
assert tracker.peak_open == 1, "two sessions opened the same device at once"
release.set()
assert _wait_until(lambda: len(tracker.instances) == 2, timeout=5.0), (
"the second session never opened after the stale one released the device"
)
assert tracker.peak_open == 1, "the device was open twice concurrently"
assert tracker.open_count <= 1, "stale session kept its device open"
worker.stop(wait=True)
assert _wait_until(lambda: tracker.open_count == 0), "device leaked"
finally:
release.set()
_restore(saved)
def test_stop_wait_is_bounded_and_closes_the_device():
tracker = OpenTracker()
release = threading.Event()
saved = _install_fake_orisys(tracker, release)
try:
worker = _new_worker("sdk")
worker.start("0", config_ref=BOX_CONFIG)
assert _wait_until(lambda: tracker.peak_open == 1)
release.set()
started = time.perf_counter()
worker.stop(wait=True)
elapsed = time.perf_counter() - started
assert elapsed < 5.0, f"stop(wait=True) took {elapsed:.2f}s"
assert _wait_until(lambda: tracker.open_count == 0), "device was not closed"
assert not worker.is_running
finally:
release.set()
_restore(saved)
if __name__ == "__main__": # pragma: no cover - manual runner
failures = 0
for name, function in sorted(globals().items()):
if name.startswith("test_") and callable(function):
try:
function()
print(f" ok {name}")
except AssertionError as exc:
failures += 1
print(f" FAIL {name}: {exc}")
except Exception as exc: # noqa: BLE001
failures += 1
print(f" ERR {name}: {type(exc).__name__}: {exc}")
print(f"failures: {failures}")
raise SystemExit(1 if failures else 0)