From 3d2fb0c4ebba439a5d636ff7fcbeeb9e4f87efa6 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Thu, 8 Oct 2026 16:37:26 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=A9=E7=9B=B4=E6=92=AD=E6=A0=87=E5=AE=9A?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E4=BB=8E=E4=BC=A0=E6=84=9F=E5=99=A8=E5=AF=B9?= =?UTF-8?q?=E8=AF=9D=E6=A1=86=E8=BF=9B=E5=85=A5=E6=89=93=E5=BC=80=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对话框可选配置默认、原始力或浏览文件，并按配置条目写入 prefs；点运行时把 cal_path（Finger 含 device_id）交给 worker，采集中则热更新。对话框与打开链路单测通过。 --- apps/studio/config/locales/en_US.json | 8 + apps/studio/config/locales/zh_CN.json | 8 + apps/studio/src/orisys_studio/app_window.py | 163 ++++++++- .../src/orisys_studio/ui/sensor_dialog.py | 333 +++++++++++++++--- .../tests/test_live_calibration_dialog.py | 321 +++++++++++++++++ .../tests/test_sensor_catalog_picker.py | 2 +- .../acquisition/finger_worker.py | 17 + .../acquisition/sdk_worker.py | 22 +- shared/tests/test_live_open_calibration.py | 186 ++++++++++ 9 files changed, 1006 insertions(+), 54 deletions(-) create mode 100644 apps/studio/tests/test_live_calibration_dialog.py create mode 100644 shared/tests/test_live_open_calibration.py diff --git a/apps/studio/config/locales/en_US.json b/apps/studio/config/locales/en_US.json index ee169ae..0de07e6 100644 --- a/apps/studio/config/locales/en_US.json +++ b/apps/studio/config/locales/en_US.json @@ -193,6 +193,14 @@ "catalog_state_unknown": "verdict not recognised", "catalog_calibration_selected": "Calibration selected: {name}", "catalog_calibration_none": "No calibration selected — raw force", + "live_calibration_section": "Live calibration", + "live_calibration_raw_tip": "Run will open with raw force. Apply while running only remembers the choice; restart to drop loaded coefficients.", + "live_calibration_default_tip": "Run will load the profile default {name}.", + "live_calibration_file_tip": "Run will load {name}; Apply while running hot-updates the device.", + "live_calibration_apply_tip": "Remember this choice; hot-update the device if a session is running", + "live_calibration_remembered": "Calibration choice remembered; it loads on the next run", + "live_calibration_applied": "Requested hot-update of calibration: {name}", + "live_calibration_invalid": "That calibration choice is not valid: {reason}", "catalog_duplicate": "Also provided by {refs}", "catalog_shape_descriptor": "descriptor bundle", "catalog_shape_compatibility": "compatibility bundle, no descriptor", diff --git a/apps/studio/config/locales/zh_CN.json b/apps/studio/config/locales/zh_CN.json index 180aa16..ef3b650 100644 --- a/apps/studio/config/locales/zh_CN.json +++ b/apps/studio/config/locales/zh_CN.json @@ -193,6 +193,14 @@ "catalog_state_unknown": "无法识别的校验结果", "catalog_calibration_selected": "已选择标定：{name}", "catalog_calibration_none": "未选择标定——使用原始力", + "live_calibration_section": "直播用标定", + "live_calibration_raw_tip": "点运行时按原始力打开。采集中点「应用到设备」只会记住选择，需重新运行才去掉已加载的系数。", + "live_calibration_default_tip": "点运行时加载配置默认标定 {name}。", + "live_calibration_file_tip": "点运行时加载 {name}；采集中可点「应用到设备」热更新。", + "live_calibration_apply_tip": "记住当前选择；若正在采集则热更新到设备", + "live_calibration_remembered": "已记住标定选择；下次运行时加载", + "live_calibration_applied": "已请求热更新标定：{name}", + "live_calibration_invalid": "当前标定选择无效：{reason}", "catalog_duplicate": "同时由 {refs} 提供", "catalog_shape_descriptor": "描述符配置包", "catalog_shape_compatibility": "兼容配置包（无描述符）", diff --git a/apps/studio/src/orisys_studio/app_window.py b/apps/studio/src/orisys_studio/app_window.py index eadb18c..5ba280b 100644 --- a/apps/studio/src/orisys_studio/app_window.py +++ b/apps/studio/src/orisys_studio/app_window.py @@ -1831,11 +1831,156 @@ class StudioMainWindow(QMainWindow): """Note the coefficients that were written. Nothing is applied here. Exporting and applying are separate actions: writing a file must never - change what a running device is using. + change what a running device is using. The new file becomes the dialog's + selected live calibration so Apply / the next Run can use it. """ + from orisys_studio.ui.sensor_dialog import ROLE_FILE + from orisys_studio.ui.video_analysis_dialog import calibration_identity + self._last_calibration_export = str(path) self.recorder.log_event("calibration_exported", str(path)) - self.statusBar().showMessage(t("cal.export_only", path=Path(path).name)) + export = Path(path) + _sensor_type, sn = calibration_identity(export) + self.set_live_calibration_choice( + role=ROLE_FILE, + path=export, + device_id=sn or None, + ) + if self.sensor_dialog is not None: + self.sensor_dialog.select_calibration_file(export) + self.statusBar().showMessage(t("cal.export_only", path=export.name)) + + # ------------------------------------------------- live calibration choice + def _live_calibration_prefs_key(self) -> str | None: + entry = self.sensor_picker.selected_entry() + return None if entry is None else entry.ref + + def live_calibration_choice(self) -> tuple[str, Path | None, str]: + """``(role, file_path_or_None, device_id)`` remembered for the selected bundle. + + ``file_path`` is only set for the browsed-file role; the profile default is + resolved when the chain opens. A missing file falls back to raw force. + """ + from orisys_studio.ui.sensor_dialog import ROLE_DEFAULT, ROLE_FILE, ROLE_NONE + + key = self._live_calibration_prefs_key() + store = self.prefs.get("live_calibration") + if not key or not isinstance(store, dict): + return ROLE_NONE, None, "" + entry = store.get(key) + if not isinstance(entry, dict): + return ROLE_NONE, None, "" + role = str(entry.get("role") or ROLE_NONE) + if role not in (ROLE_DEFAULT, ROLE_FILE, ROLE_NONE): + role = ROLE_NONE + path: Path | None = None + if role == ROLE_FILE: + raw = str(entry.get("path") or "").strip() + candidate = Path(raw) if raw else None + if candidate is None or not candidate.is_file(): + return ROLE_NONE, None, "" + path = candidate + return role, path, str(entry.get("device_id") or "") + + def set_live_calibration_choice( + self, + *, + role: str, + path: Path | str | None = None, + device_id: str | None = None, + ) -> None: + """Remember the live calibration choice for the currently selected bundle.""" + from orisys_studio.ui.sensor_dialog import ROLE_DEFAULT, ROLE_FILE, ROLE_NONE + + key = self._live_calibration_prefs_key() + if not key: + return + store = self.prefs.get("live_calibration") + if not isinstance(store, dict): + store = {} + self.prefs["live_calibration"] = store + wanted = str(role or ROLE_NONE) + if wanted not in (ROLE_DEFAULT, ROLE_FILE, ROLE_NONE): + wanted = ROLE_NONE + payload: dict[str, str] = {"role": wanted} + if wanted == ROLE_FILE and path is not None and str(path).strip(): + payload["path"] = str(path) + if device_id and str(device_id).strip(): + payload["device_id"] = str(device_id).strip() + store[key] = payload + save_preferences(self.prefs) + + def _default_live_calibration_path(self) -> Path | None: + entry = self.sensor_picker.selected_entry() + if entry is None or entry.profile is None: + return None + declared = getattr(entry.profile, "default_calibration", None) + if declared is not None and Path(declared).is_file(): + return Path(declared) + return None + + def _resolved_live_calibration( + self, + ) -> tuple[Path | None, str | None, str]: + """``(cal_path, device_id, problem)`` for the next open or a hot apply. + + ``problem`` is non-empty when a Finger file choice fails the serial check. + """ + from orisys_studio.ui.sensor_dialog import ROLE_DEFAULT, ROLE_FILE, ROLE_NONE + from orisys_studio.ui.video_analysis_dialog import calibration_identity + + role, path, device_id = self.live_calibration_choice() + if role == ROLE_NONE: + return None, None, "" + if role == ROLE_DEFAULT: + return self._default_live_calibration_path(), None, "" + if role != ROLE_FILE or path is None: + return None, None, "" + entry = self.sensor_picker.selected_entry() + is_finger = ( + entry is not None + and str(entry.product_family or "").strip().lower() == "finger" + ) + if not is_finger: + return path, None, "" + serial = str(device_id or "").strip() + if not serial: + return path, None, t("video_analysis.device_required") + sensor_type, sn = calibration_identity(path) + if sensor_type and sensor_type != "finger": + return path, None, t("video_analysis.device_family", family=sensor_type) + if not sn: + return path, None, t("video_analysis.device_missing_sn") + if sn != serial: + return path, None, t("video_analysis.device_mismatch", sn=sn) + return path, serial, "" + + def apply_live_calibration_choice(self) -> None: + """Remember the dialog's choice; hot-update the device when a session is open.""" + if self.sensor_dialog is not None: + problem = self.sensor_dialog.choice_problem() + if problem: + self.statusBar().showMessage( + t("sensor.live_calibration_invalid", reason=problem) + ) + return + cal_path, _device_id, problem = self._resolved_live_calibration() + if problem: + self.statusBar().showMessage( + t("sensor.live_calibration_invalid", reason=problem) + ) + return + if cal_path is None or not self._running: + self.statusBar().showMessage(t("sensor.live_calibration_remembered")) + return + worker = self._active_worker + if worker is None or not hasattr(worker, "set_pending_calibration"): + self.statusBar().showMessage(t("cal.apply_no_target")) + return + worker.set_pending_calibration(cal_path) + self.statusBar().showMessage( + t("sensor.live_calibration_applied", name=cal_path.name) + ) # ------------------------------------------------- calibration (forwards) # "What does this coefficient file claim, and does it match the bench?" lives in @@ -2280,6 +2425,13 @@ class StudioMainWindow(QMainWindow): self.toolbar.set_run_checked(False) self.statusBar().showMessage(t("status.video_required")) return + cal_path, device_id, cal_problem = self._resolved_live_calibration() + if cal_problem: + self.toolbar.set_run_checked(False) + self.statusBar().showMessage( + t("sensor.live_calibration_invalid", reason=cal_problem) + ) + return if self._preview_open: self._stop_preview(announce=False) @@ -2339,7 +2491,12 @@ class StudioMainWindow(QMainWindow): # The acquisition thread applies it and records what the device reports. self._active_worker.set_filter_settings(self._filter_settings()) try: - self._active_worker.start(source, config_ref=config_ref) + self._active_worker.start( + source, + config_ref=config_ref, + cal_path=cal_path, + device_id=device_id, + ) except Exception as exc: report_error(exc, message="Run failed") self.toolbar.set_run_checked(False) diff --git a/apps/studio/src/orisys_studio/ui/sensor_dialog.py b/apps/studio/src/orisys_studio/ui/sensor_dialog.py index c208ffa..c5c3608 100644 --- a/apps/studio/src/orisys_studio/ui/sensor_dialog.py +++ b/apps/studio/src/orisys_studio/ui/sensor_dialog.py @@ -5,11 +5,12 @@ product and config the window is configured for is worth reading at a glance, but the pickers themselves — and the calibration that belongs to them — need more room than a 26 px rail can give. -The calibration half is read-only on purpose: it states which calibration the -*device* confirmed, which axes it covers, and what the last exported -coefficients file says it belongs to. Exporting and applying stay single-entry -actions (they are the calibration panel's own methods), so there is exactly one -implementation of each. +The calibration half lets the operator choose the force curve for the **next** +live run (and hot-apply it while a session is already open): the profile's +default file when the bundle declares one, raw force, or a browsed file — the +same three answers the bare-video analysis prompt asks for. Exporting still +writes a coefficients file without changing a running device until Apply is +used. """ from __future__ import annotations @@ -26,10 +27,14 @@ from orisys_app_shared.sensors import ( ) from PySide6.QtCore import Qt from PySide6.QtWidgets import ( + QComboBox, QDialog, + QFileDialog, + QFormLayout, QFrame, QHBoxLayout, QLabel, + QLineEdit, QPushButton, QScrollArea, QSizePolicy, @@ -38,6 +43,7 @@ from PySide6.QtWidgets import ( ) from orisys_studio.ui.fit_text import ShrinkableLabel +from orisys_studio.ui.video_analysis_dialog import calibration_identity #: How the catalog's shape identifiers read here. One literal key per kind, so the locale scan #: sees them and a new kind is a visible fallback rather than a printed identifier. @@ -57,15 +63,22 @@ _STATE_KEYS: dict[str, str] = { STATE_UNREADABLE: "sensor.catalog_state_invalid", } +ROLE_DEFAULT = "default" +ROLE_NONE = "none" +ROLE_FILE = "file" + class SensorDialog(QDialog): - """Sensor type / config picker plus the calibration facts that follow it.""" + """Sensor type / config picker plus the live calibration choice that follows it.""" def __init__(self, *, window, picker, theme: str = "dark") -> None: super().__init__(window) self._window = window self._theme = str(theme) self.picker = picker + self._custom_calibration: Path | None = None + self._device_source: Path | None = None + self._syncing = False self.setObjectName("sensorDialog") self.setModal(False) @@ -116,6 +129,37 @@ class SensorDialog(QDialog): ) root.addWidget(self.profile_scroll) + root.addWidget(self._section_rule()) + self.choice_title = QLabel() + self.choice_title.setObjectName("panelTitle") + root.addWidget(self.choice_title) + + choice_form = QFormLayout() + choice_form.setLabelAlignment(Qt.AlignmentFlag.AlignRight) + self.calibration_label = QLabel() + self.calibration_label.setObjectName("fieldLabel") + self.calibration_combo = QComboBox() + self.calibration_combo.setObjectName("liveCalibrationCombo") + calibration_row = QHBoxLayout() + calibration_row.addWidget(self.calibration_combo, 1) + self.btn_browse = QPushButton() + self.btn_browse.setProperty("compact", "true") + self.btn_browse.clicked.connect(self._on_browse) + calibration_row.addWidget(self.btn_browse) + choice_form.addRow(self.calibration_label, calibration_row) + + self.device_label = QLabel() + self.device_label.setObjectName("fieldLabel") + self.device_edit = QLineEdit() + self.device_edit.setObjectName("liveDeviceEdit") + choice_form.addRow(self.device_label, self.device_edit) + self._choice_form = choice_form + root.addLayout(choice_form) + self.choice_note = ShrinkableLabel() + self.choice_note.setObjectName("hintText") + self.choice_note.setWordWrap(True) + root.addWidget(self.choice_note) + root.addWidget(self._section_rule()) self.calib_title = QLabel() self.calib_title.setObjectName("panelTitle") @@ -169,6 +213,8 @@ class SensorDialog(QDialog): # follows them while it is open. self.picker.sensor_type_changed.connect(lambda _value: self.refresh()) self.picker.config_changed.connect(lambda _value: self.refresh()) + self.calibration_combo.currentIndexChanged.connect(self._on_choice_changed) + self.device_edit.textChanged.connect(self._on_device_edited) self.retranslate() self.refresh() @@ -183,17 +229,41 @@ class SensorDialog(QDialog): # ------------------------------------------------------------------ actions def _on_export(self) -> None: - self._window.cal_panel.export_coefficients() + panel = getattr(self._window, "cal_panel", None) + if panel is not None and hasattr(panel, "export_coefficients"): + panel.export_coefficients() + export = str(getattr(self._window, "_last_calibration_export", "") or "") + if export: + self.select_calibration_file(Path(export)) self.refresh() def _on_apply(self) -> None: - self._window.cal_panel.apply_coefficients() + self._persist_choice() + apply = getattr(self._window, "apply_live_calibration_choice", None) + if callable(apply): + apply() self.refresh() + def _on_browse(self) -> None: + chosen, _filter = QFileDialog.getOpenFileName( + self, + t("video_analysis.calibration_browse"), + str(Path.cwd()), + t("video_analysis.calibration_filter"), + ) + if chosen: + self.select_calibration_file(Path(chosen)) + + def select_calibration_file(self, path: Path) -> None: + """Put a browsed (or just-exported) file in the combo and persist it.""" + self._refresh_calibrations(custom=Path(path)) + self._persist_choice() + # ------------------------------------------------------------------ content def refresh(self) -> None: """Re-read the shell's calibration state. Called on open and on change.""" window = self._window + self._load_choice_into_controls() version = str(getattr(window, "_active_calibration_id", "") or "") self.version_label.setText( f"{t('sensor.calib_version')}: {version or t('sensor.calib_none')}" @@ -213,40 +283,209 @@ class SensorDialog(QDialog): ) self.btn_reveal.setEnabled(bool(config_ref)) self._refresh_profile() + self._sync_device_row() + self._refresh_choice_note() export = str(getattr(window, "_last_calibration_export", "") or "") if not export: self.export_label.setText(t("sensor.export_none")) self.identity_label.setText("") - self.btn_apply.setEnabled(False) - return - path = Path(export) - self.export_label.setText( - t( - "sensor.export_file", - name=path.name, - size=f"{path.stat().st_size / 1024:.1f}" if path.is_file() else "?", - time=( - datetime.fromtimestamp(path.stat().st_mtime).strftime("%Y-%m-%d %H:%M") - if path.is_file() - else "?" - ), + else: + path = Path(export) + self.export_label.setText( + t( + "sensor.export_file", + name=path.name, + size=f"{path.stat().st_size / 1024:.1f}" if path.is_file() else "?", + time=( + datetime.fromtimestamp(path.stat().st_mtime).strftime("%Y-%m-%d %H:%M") + if path.is_file() + else "?" + ), + ) ) - ) - identity = {} - read_identity = getattr(window, "calibration_identity", None) - if callable(read_identity): - identity = dict(read_identity(export) or {}) - self.identity_label.setText( - t( - "sensor.export_identity", - sensor_type=identity.get("sensor_type") or "?", - sn=identity.get("sn") or "?", - config=Path(str(identity.get("config_ref") or "?")).name, + identity = {} + read_identity = getattr(window, "calibration_identity", None) + if callable(read_identity): + identity = dict(read_identity(export) or {}) + self.identity_label.setText( + t( + "sensor.export_identity", + sensor_type=identity.get("sensor_type") or "?", + sn=identity.get("sn") or "?", + config=Path(str(identity.get("config_ref") or "?")).name, + ) ) - ) self.btn_apply.setEnabled(True) + def _entry(self): + read_entry = getattr(self.picker, "selected_entry", None) + return read_entry() if callable(read_entry) else None + + def _default_calibration_path(self) -> Path | None: + entry = self._entry() + if entry is None or entry.profile is None: + return None + declared = getattr(entry.profile, "default_calibration", None) + if declared is not None and Path(declared).is_file(): + return Path(declared) + return None + + def _load_choice_into_controls(self) -> None: + read = getattr(self._window, "live_calibration_choice", None) + role, path, device_id = ROLE_NONE, None, "" + if callable(read): + role, path, device_id = read() + custom = Path(path) if role == ROLE_FILE and path else None + self._refresh_calibrations(custom=custom) + self._syncing = True + try: + if role == ROLE_FILE and custom is not None: + # already selected by _refresh_calibrations + pass + elif role == ROLE_DEFAULT and self._default_calibration_path() is not None: + index = self.calibration_combo.findData(ROLE_DEFAULT) + if index >= 0: + self.calibration_combo.setCurrentIndex(index) + else: + index = self.calibration_combo.findData(ROLE_NONE) + if index >= 0: + self.calibration_combo.setCurrentIndex(index) + if device_id: + self.device_edit.setText(str(device_id)) + finally: + self._syncing = False + + def _refresh_calibrations(self, *, custom: Path | None) -> None: + self.calibration_combo.blockSignals(True) + self.calibration_combo.clear() + default = self._default_calibration_path() + if default is not None: + self.calibration_combo.addItem( + t("video_analysis.calibration_default", name=default.name), + ROLE_DEFAULT, + ) + self.calibration_combo.addItem(t("video_analysis.calibration_none"), ROLE_NONE) + if custom is not None: + self._custom_calibration = Path(custom) + self.calibration_combo.addItem( + t("video_analysis.calibration_file", name=Path(custom).name), ROLE_FILE + ) + self.calibration_combo.setCurrentIndex(self.calibration_combo.count() - 1) + elif self.calibration_combo.count(): + # Prefer the profile default when one exists; otherwise raw force. + wanted = ROLE_DEFAULT if default is not None else ROLE_NONE + index = self.calibration_combo.findData(wanted) + self.calibration_combo.setCurrentIndex(max(index, 0)) + self.calibration_combo.blockSignals(False) + self._sync_device_row() + + def current_role(self) -> str: + return str(self.calibration_combo.currentData() or ROLE_NONE) + + def current_calibration_path(self) -> Path | None: + role = self.current_role() + if role == ROLE_FILE: + return self._custom_calibration + if role == ROLE_NONE: + return None + return self._default_calibration_path() + + def current_device_id(self) -> str | None: + if not self._is_finger() or self.current_role() != ROLE_FILE: + return None + serial = self.device_edit.text().strip() + return serial or None + + def _is_finger(self) -> bool: + entry = self._entry() + if entry is None: + return False + return str(entry.product_family or "").strip().lower() == "finger" + + def _persist_choice(self) -> None: + save = getattr(self._window, "set_live_calibration_choice", None) + if not callable(save): + return + save( + role=self.current_role(), + path=self.current_calibration_path(), + device_id=self.current_device_id(), + ) + + def _on_choice_changed(self) -> None: + if self._syncing: + return + self._sync_device_row() + self._refresh_choice_note() + self._refresh_profile() + self._persist_choice() + + def _on_device_edited(self, _text: str) -> None: + if self._syncing: + return + self._refresh_choice_note() + self._persist_choice() + + def _sync_device_row(self) -> None: + path = self.current_calibration_path() if self.current_role() == ROLE_FILE else None + visible = self._is_finger() and path is not None + self.device_label.setVisible(visible) + self.device_edit.setVisible(visible) + form = getattr(self, "_choice_form", None) + if form is not None and hasattr(form, "setRowVisible"): + form.setRowVisible(self.device_label, visible) + if not visible: + self._device_source = None + return + if path != self._device_source: + self._device_source = path + _sensor_type, sn = calibration_identity(path) + if sn: + self._syncing = True + try: + self.device_edit.setText(sn) + finally: + self._syncing = False + + def choice_problem(self) -> str: + """Why the current live choice must not be opened or applied; empty when usable.""" + return self._device_problem() + + def _device_problem(self) -> str: + path = self.current_calibration_path() if self.current_role() == ROLE_FILE else None + if not self._is_finger() or path is None: + return "" + serial = self.device_edit.text().strip() + if not serial: + return t("video_analysis.device_required") + sensor_type, sn = calibration_identity(path) + if sensor_type and sensor_type != "finger": + return t("video_analysis.device_family", family=sensor_type) + if not sn: + return t("video_analysis.device_missing_sn") + if sn != serial: + return t("video_analysis.device_mismatch", sn=sn) + return "" + + def _refresh_choice_note(self) -> None: + problem = self._device_problem() + if problem: + self.choice_note.setText(problem) + return + role = self.current_role() + path = self.current_calibration_path() + if role == ROLE_NONE or path is None: + self.choice_note.setText(t("sensor.live_calibration_raw_tip")) + elif role == ROLE_DEFAULT: + self.choice_note.setText( + t("sensor.live_calibration_default_tip", name=path.name) + ) + else: + self.choice_note.setText( + t("sensor.live_calibration_file_tip", name=path.name) + ) + def _refresh_profile(self) -> None: """State the selected catalog entry, and the reason when it cannot be started. @@ -256,12 +495,19 @@ class SensorDialog(QDialog): sentence. It is read from the picker rather than re-derived here, so the window and the dialog can never say two different things about one bundle. """ - read_entry = getattr(self.picker, "selected_entry", None) - entry = read_entry() if callable(read_entry) else None + entry = self._entry() if entry is None: self.profile_label.setText("") self.profile_label.setToolTip("") return + path = self.current_calibration_path() + role = self.current_role() + if role == ROLE_NONE or path is None: + calibration_line = t("sensor.catalog_calibration_none") + else: + calibration_line = t( + "sensor.catalog_calibration_selected", name=path.name + ) lines = [ t("sensor.catalog_name", name=entry.name, folder=entry.folder), t( @@ -276,11 +522,7 @@ class SensorDialog(QDialog): state=entry.state, note=t(_STATE_KEYS.get(entry.state, "sensor.catalog_state_unknown")), ), - ( - t("sensor.catalog_calibration_selected", name=entry.selected_calibration) - if entry.selected_calibration - else t("sensor.catalog_calibration_none") - ), + calibration_line, ] if entry.duplicates: lines.append(t("sensor.catalog_duplicate", refs=", ".join(entry.duplicates))) @@ -300,13 +542,18 @@ class SensorDialog(QDialog): def retranslate(self) -> None: self.setWindowTitle(t("sensor.dialog_title")) + self.choice_title.setText(t("sensor.live_calibration_section")) + self.calibration_label.setText(t("video_analysis.calibration")) + self.device_label.setText(t("video_analysis.device")) + self.device_edit.setPlaceholderText(t("video_analysis.device_tip")) + self.btn_browse.setText(t("video_analysis.calibration_browse")) self.calib_title.setText(t("sensor.calib_section")) self.export_title.setText(t("sensor.export_section")) self.btn_reveal.setText(t("menu.reveal_config")) self.btn_export.setText(t("cal.export")) self.btn_export.setToolTip(t("cal.export_tip")) self.btn_apply.setText(t("cal.apply")) - self.btn_apply.setToolTip(t("cal.apply_tip")) + self.btn_apply.setToolTip(t("sensor.live_calibration_apply_tip")) self.btn_close.setText(t("dialog.cancel")) self.picker.retranslate() self.refresh() @@ -315,7 +562,5 @@ class SensorDialog(QDialog): self._theme = str(theme) def showEvent(self, event) -> None: # noqa: N802 - Qt naming - # Every open re-reads the shell: the device may have confirmed a - # calibration since the dialog was last shown. - self.refresh() super().showEvent(event) + self.refresh() diff --git a/apps/studio/tests/test_live_calibration_dialog.py b/apps/studio/tests/test_live_calibration_dialog.py new file mode 100644 index 0000000..853a1f1 --- /dev/null +++ b/apps/studio/tests/test_live_calibration_dialog.py @@ -0,0 +1,321 @@ +"""Live calibration choice in the sensor dialog: pick, remember, open, hot-apply.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from orisys_app_shared.i18n import t +from orisys_app_shared.sensors import scan_sensor_catalog +from orisys_studio.ui import sensor_picker as picker_module +from orisys_studio.ui.sensor_dialog import ROLE_DEFAULT, ROLE_FILE, ROLE_NONE, SensorDialog +from orisys_studio.ui.sensor_picker import SensorPickerWidget +from PySide6.QtWidgets import QWidget + +CONFIG = { + "name": "fixture", + "camera": { + "resolution_raw": [640, 480], + "fps": 30, + "resolution_stitch": [100, 100], + }, + "tracker": {"algorithm": "dis"}, + "stitcher": {"n": [1, 1]}, +} + + +def _write(path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + + +def _descriptor(profile_id: str, **overrides: object) -> dict[str, object]: + data: dict[str, object] = { + "schema_version": 1, + "profile_id": profile_id, + "display_name": "Mini lab", + "product_family": "mini", + "processing_backend": "single_camera", + "sensor_json": "sensor.json", + } + data.update(overrides) + return data + + +class _Host(QWidget): + """Window stand-in with the live-calibration prefs API the dialog expects.""" + + def __init__(self) -> None: + super().__init__() + self.prefs: dict = {} + self.applied = 0 + self._choices: dict[str, tuple[str, Path | None, str]] = {} + self.picker: SensorPickerWidget | None = None + self._dialog: SensorDialog | None = None + + def _on_reveal_config(self) -> None: + return + + def _prefs_key(self) -> str | None: + if self.picker is None: + return None + entry = self.picker.selected_entry() + return None if entry is None else entry.ref + + def live_calibration_choice(self) -> tuple[str, Path | None, str]: + key = self._prefs_key() + if not key: + return ROLE_NONE, None, "" + return self._choices.get(key, (ROLE_NONE, None, "")) + + def set_live_calibration_choice( + self, + *, + role: str, + path: Path | str | None = None, + device_id: str | None = None, + ) -> None: + key = self._prefs_key() + if not key: + return + file_path = Path(path) if role == ROLE_FILE and path else None + self._choices[key] = (role, file_path, str(device_id or "")) + store = self.prefs.setdefault("live_calibration", {}) + payload: dict[str, str] = {"role": role} + if file_path is not None: + payload["path"] = str(file_path) + if device_id: + payload["device_id"] = str(device_id) + store[key] = payload + + def apply_live_calibration_choice(self) -> None: + if self._dialog is not None and self._dialog.choice_problem(): + return + self.applied += 1 + + +def _make_picker(qt_app, tmp_path, monkeypatch, build): + root = tmp_path / "user" + build(root) + catalog = scan_sensor_catalog( + shipped_root=tmp_path / "no-shipped-root", user_root=root + ) + monkeypatch.setattr(picker_module, "scan_sensor_catalog", lambda: catalog) + return SensorPickerWidget() + + +def _dialog_for(picker) -> tuple[_Host, SensorDialog]: + host = _Host() + host.picker = picker + dialog = SensorDialog(window=host, picker=picker) + host._dialog = dialog + return host, dialog + + +def test_without_a_profile_default_the_combo_starts_on_raw_force( + qt_app, tmp_path, monkeypatch +): + def build(root: Path) -> None: + _write(root / "mini-lab" / "sensor.json", CONFIG) + _write(root / "mini-lab" / "manifest.json", _descriptor("orisys.mini.lab-r1")) + + picker = _make_picker(qt_app, tmp_path, monkeypatch, build) + picker.set_sensor_type("mini") + _host, dialog = _dialog_for(picker) + try: + assert dialog.calibration_combo.findData(ROLE_DEFAULT) < 0 + assert dialog.current_role() == ROLE_NONE + assert dialog.current_calibration_path() is None + assert t("sensor.catalog_calibration_none") in dialog.profile_label.text() + finally: + dialog.deleteLater() + picker.deleteLater() + + +def test_a_box_profile_default_appears_as_a_combo_row(qt_app): + """The shipped box bundle declares ``force_calibration.json`` as its default.""" + picker = SensorPickerWidget() + picker.set_sensor_type("box") + entry = picker.selected_entry() + assert entry is not None + assert entry.profile is not None + assert entry.profile.default_calibration is not None, ( + "premise: shipped box must expose a profile default calibration" + ) + _host, dialog = _dialog_for(picker) + try: + assert dialog.calibration_combo.findData(ROLE_DEFAULT) >= 0 + assert dialog.calibration_combo.findData(ROLE_NONE) >= 0 + assert dialog.current_role() == ROLE_NONE + dialog.calibration_combo.setCurrentIndex( + dialog.calibration_combo.findData(ROLE_DEFAULT) + ) + path = dialog.current_calibration_path() + assert path is not None and path.name == "force_calibration.json" + assert t( + "sensor.catalog_calibration_selected", name="force_calibration.json" + ) in dialog.profile_label.text() + finally: + dialog.deleteLater() + picker.deleteLater() + + +def test_browsing_a_file_shows_its_name_and_persists_per_bundle( + qt_app, tmp_path, monkeypatch +): + def build(root: Path) -> None: + for name, family, profile in ( + ("mini-a", "mini", "orisys.mini.a-r1"), + ("finger-a", "finger", "orisys.finger.a-r1"), + ): + folder = root / name + _write(folder / "sensor.json", CONFIG) + extras = { + "display_name": name, + "product_family": family, + "processing_backend": ( + "uv_unroll" if family == "finger" else "single_camera" + ), + } + if family == "finger": + extras["computational_assets"] = {} + _write(folder / "manifest.json", _descriptor(profile, **extras)) + + picker = _make_picker(qt_app, tmp_path, monkeypatch, build) + host, dialog = _dialog_for(picker) + cal = tmp_path / "custom.json" + cal.write_text( + json.dumps({"meta": {"sensor_type": "mini", "sn": "X"}}), encoding="utf-8" + ) + try: + picker.set_sensor_type("mini") + dialog.select_calibration_file(cal) + assert dialog.current_role() == ROLE_FILE + assert dialog.calibration_combo.currentText() == t( + "video_analysis.calibration_file", name="custom.json" + ) + mini_key = picker.selected_entry().ref + assert host.prefs["live_calibration"][mini_key]["path"] == str(cal) + + picker.set_sensor_type("finger") + dialog.refresh() + assert dialog.current_role() == ROLE_NONE + finger_key = picker.selected_entry().ref + assert finger_key != mini_key + assert ROLE_FILE != host.prefs.get("live_calibration", {}).get( + finger_key, {} + ).get("role") + finally: + dialog.deleteLater() + picker.deleteLater() + + +def test_a_finger_file_with_a_mismatched_sn_is_not_a_valid_choice( + qt_app, tmp_path, monkeypatch +): + def build(root: Path) -> None: + folder = root / "finger-lab" + _write(folder / "sensor.json", CONFIG) + _write( + folder / "manifest.json", + _descriptor( + "orisys.finger.lab-r1", + display_name="Finger lab", + product_family="finger", + processing_backend="uv_unroll", + computational_assets={}, + ), + ) + + picker = _make_picker(qt_app, tmp_path, monkeypatch, build) + picker.set_sensor_type("finger") + host, dialog = _dialog_for(picker) + cal = tmp_path / "DD02-14.json" + cal.write_text( + json.dumps({"meta": {"sensor_type": "finger", "sn": "DD02-14"}}), + encoding="utf-8", + ) + try: + dialog.select_calibration_file(cal) + # Offscreen dialogs are not shown; ``isHidden`` is the visibility flag. + assert not dialog.device_edit.isHidden() + assert dialog.device_edit.text() == "DD02-14" + assert dialog.choice_problem() == "" + dialog.device_edit.setText("OTHER") + assert dialog.choice_problem() == t( + "video_analysis.device_mismatch", sn="DD02-14" + ) + dialog.btn_apply.click() + assert host.applied == 0 + finally: + dialog.deleteLater() + picker.deleteLater() + + +def test_window_resolves_finger_file_choice_for_open(make_window, tmp_path): + window = make_window() + cal = tmp_path / "DD02-14.json" + cal.write_text( + json.dumps({"meta": {"sensor_type": "finger", "sn": "DD02-14"}}), + encoding="utf-8", + ) + finger = next( + ( + entry + for entry in scan_sensor_catalog().entries + if str(entry.product_family) == "finger" + and entry.start_blocked_reason is None + ), + None, + ) + if finger is None: + pytest.skip("no startable Finger profile in the catalog") + window.sensor_picker.set_sensor_type("finger") + window.sensor_picker.set_config_name(finger.key) + window.set_live_calibration_choice( + role=ROLE_FILE, path=cal, device_id="DD02-14" + ) + cal_path, device_id, problem = window._resolved_live_calibration() + assert problem == "" + assert cal_path == cal + assert device_id == "DD02-14" + + window.set_live_calibration_choice( + role=ROLE_FILE, path=cal, device_id="OTHER" + ) + _path, _device, problem = window._resolved_live_calibration() + assert problem == t("video_analysis.device_mismatch", sn="DD02-14") + + +def test_apply_while_running_queues_set_pending_calibration(make_window, tmp_path): + window = make_window() + cal = tmp_path / "force_calibration.json" + cal.write_text('{"schema_version": 2}', encoding="utf-8") + window.set_live_calibration_choice(role=ROLE_FILE, path=cal) + pending: list[str] = [] + + class _Worker: + def add_sink(self, _sink) -> None: + return + + def set_pending_calibration(self, path) -> None: + pending.append(str(path)) + + worker = _Worker() + # Both slots filled so ``ensure_workers`` does not replace the shim. + window.session.set_worker("box", worker) + window.session.set_worker("finger", worker) + window.session.select_worker("box") + window.session.set_running(True) + assert window._running + assert window._active_worker is worker + + window.apply_live_calibration_choice() + assert pending == [str(cal)] + + pending.clear() + window.set_live_calibration_choice(role=ROLE_NONE) + window.apply_live_calibration_choice() + assert pending == [] + assert window.statusBar().currentMessage() == t("sensor.live_calibration_remembered") diff --git a/apps/studio/tests/test_sensor_catalog_picker.py b/apps/studio/tests/test_sensor_catalog_picker.py index bfdf6ec..2da4416 100644 --- a/apps/studio/tests/test_sensor_catalog_picker.py +++ b/apps/studio/tests/test_sensor_catalog_picker.py @@ -302,7 +302,7 @@ def test_the_details_sit_directly_below_the_selector(mixed_picker): dialog = _details_dialog(mixed_picker) root = dialog.layout() assert root.itemAt(0).layout() is dialog.picker_row, "the selector is not the first row" - assert root.itemAt(1).widget() is dialog.profile_label, ( + assert root.itemAt(1).widget() is dialog.profile_scroll, ( "the details are not directly under the selector" ) diff --git a/shared/src/orisys_app_shared/acquisition/finger_worker.py b/shared/src/orisys_app_shared/acquisition/finger_worker.py index 72720ba..f19949b 100644 --- a/shared/src/orisys_app_shared/acquisition/finger_worker.py +++ b/shared/src/orisys_app_shared/acquisition/finger_worker.py @@ -235,6 +235,9 @@ class FingerAcquisitionWorker(RecordSinkMixin): self._end_reason: str | None = None self._phase: str = "idle" self._pending_calibration: str | None = None + #: Path and device serial for the **next** ``start()`` (live open kwargs). + self._session_cal_path: str | None = None + self._session_device_id: str | None = None #: A zeroing requested from another thread, applied by this one. self._pending_reset: bool = False self._mono_t0: float = 0.0 @@ -305,6 +308,8 @@ class FingerAcquisitionWorker(RecordSinkMixin): video_source: str | int, *, config_ref: str | Path, + cal_path: str | Path | None = None, + device_id: str | None = None, ) -> FixedCameraMode: # Non-blocking: a previous session that is stuck opening must never hold # the caller (the GUI thread) hostage. @@ -317,6 +322,12 @@ class FingerAcquisitionWorker(RecordSinkMixin): self._mode = mode self._video_source = video_source self._export_dir = resolve_finger_export_dir(config_ref) + self._session_cal_path = ( + None if cal_path is None or not str(cal_path).strip() else str(cal_path) + ) + self._session_device_id = ( + None if device_id is None or not str(device_id).strip() else str(device_id) + ) with self._lock: self._latest = None self._end_reason = None @@ -564,6 +575,12 @@ class FingerAcquisitionWorker(RecordSinkMixin): } if self._export_dir is not None: open_kw["export_dir"] = str(self._export_dir) + if self._session_cal_path is not None: + # Same open path as the offline analysis pass: the operator's file, + # and the serial they confirmed when the file needs one. + open_kw["cal_path"] = self._session_cal_path + if self._session_device_id: + open_kw["device_id"] = self._session_device_id runtime = FingerRuntime.open(**open_kw) with self._lock: self._runtimes[session] = runtime diff --git a/shared/src/orisys_app_shared/acquisition/sdk_worker.py b/shared/src/orisys_app_shared/acquisition/sdk_worker.py index 1e54556..5042e43 100644 --- a/shared/src/orisys_app_shared/acquisition/sdk_worker.py +++ b/shared/src/orisys_app_shared/acquisition/sdk_worker.py @@ -730,6 +730,8 @@ class SdkAcquisitionWorker(RecordSinkMixin): self._end_reason: str | None = None self._phase: str = "idle" self._pending_calibration: str | None = None + #: Path for the **next** ``start()`` (overrides the bundle default when set). + self._session_cal_path: str | None = None #: A zeroing requested from another thread, applied by this one. self._pending_reset: bool = False #: Monotonic origin for ``SdkPacket.mono_t``, reset per session. @@ -820,9 +822,12 @@ class SdkAcquisitionWorker(RecordSinkMixin): video_source: str | int, *, config_ref: str | Path, + cal_path: str | Path | None = None, + device_id: str | None = None, ) -> FixedCameraMode: # Non-blocking: a previous session that is stuck opening the device must # never hold the caller (the GUI thread) hostage. + del device_id # box/mini open path has no Finger-style serial check self.stop() previous = self._thread ok, detail = orisys_importable() @@ -831,6 +836,9 @@ class SdkAcquisitionWorker(RecordSinkMixin): mode = load_fixed_mode(config_ref) self._mode = mode self._video_source = video_source + self._session_cal_path = ( + None if cal_path is None or not str(cal_path).strip() else str(cal_path) + ) with self._lock: self._latest = None self._end_reason = None @@ -1087,12 +1095,14 @@ class SdkAcquisitionWorker(RecordSinkMixin): is_file = _is_file_source(source) try: self._status(f"Opening Orisys Sensor ({mode.label()})…") - sensor = orisys.Sensor( - source, - config_name=mode.config_ref, - verbose=False, - backend="auto", - ) + open_kw: dict = { + "config_name": mode.config_ref, + "verbose": False, + "backend": "auto", + } + if self._session_cal_path is not None: + open_kw["cal_path"] = self._session_cal_path + sensor = orisys.Sensor(source, **open_kw) with self._lock: self._sensors[session] = sensor if self._stop.is_set() or session != self._session_id: diff --git a/shared/tests/test_live_open_calibration.py b/shared/tests/test_live_open_calibration.py new file mode 100644 index 0000000..1b14aae --- /dev/null +++ b/shared/tests/test_live_open_calibration.py @@ -0,0 +1,186 @@ +"""Live workers pass the operator's cal_path into FingerRuntime / Sensor.open.""" + +from __future__ import annotations + +import sys +import types +from pathlib import Path + +APP_ROOT = Path(__file__).resolve().parents[2] +BOX_CONFIG = APP_ROOT / "apps" / "studio" / "config" / "sensors" / "box" / "sensor.json" + + +class _TrackingSensor: + opened: dict = {} + + def __init__(self, source, **kwargs): + type(self).opened = {"source": source, **kwargs} + self.connected = True + + def configure(self, **kwargs): + return () + + def get_img(self): + return None + + def disconnect(self): + self.connected = False + + +class _TrackingRuntime: + opened: dict = {} + tactile = types.SimpleNamespace(raw_sensor=None) + capture = None + + @classmethod + def open(cls, **kwargs): + cls.opened = dict(kwargs) + return cls() + + def warmup(self, *, color=False): + return None + + def close(self): + return None + + +def _install_fake_orisys(*, sensor_cls=None, runtime_cls=None): + saved = {name: sys.modules.get(name) for name in ("orisys", "orisys.finger")} + module = types.ModuleType("orisys") + module.__version__ = "fake" + module.Sensor = sensor_cls or _TrackingSensor + finger = types.ModuleType("orisys.finger") + finger.FingerRuntime = runtime_cls or _TrackingRuntime + module.finger = finger + sys.modules["orisys"] = module + sys.modules["orisys.finger"] = finger + return saved + + +def _restore(saved) -> None: + for name, module in saved.items(): + if module is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = module + + +def _mode(config: Path, label: str = "fixture"): + return types.SimpleNamespace( + config_ref=str(config), + label=lambda: label, + fps=30.0, + ) + + +def test_sdk_worker_open_passes_cal_path(monkeypatch, tmp_path): + from orisys_app_shared.acquisition import sdk_worker as mod + + saved = _install_fake_orisys(sensor_cls=_TrackingSensor) + _TrackingSensor.opened = {} + config = BOX_CONFIG if BOX_CONFIG.is_file() else tmp_path / "sensor.json" + if not config.is_file(): + config.write_text("{}", encoding="utf-8") + cal = tmp_path / "force_calibration.json" + cal.write_text("{}", encoding="utf-8") + mode = _mode(config, "box-fixture") + try: + monkeypatch.setattr(mod, "orisys_importable", lambda: (True, "")) + monkeypatch.setattr(mod, "loaded_calibration_version", lambda _s: "") + monkeypatch.setattr(mod, "loaded_path", lambda _s: None) + monkeypatch.setattr(mod, "load_fixed_mode", lambda _ref: mode) + worker = mod.SdkAcquisitionWorker() + # Set the session fields the open path reads — do not start a thread. + worker._mode = mode + worker._video_source = 0 + worker._session_id = 1 + worker._session_cal_path = str(cal) + worker._stop.clear() + opened = worker._open_session(1, None) + assert opened is not None + assert _TrackingSensor.opened.get("cal_path") == str(cal) + # ``start`` itself stores the path for that open. + worker2 = mod.SdkAcquisitionWorker() + monkeypatch.setattr(worker2, "stop", lambda: None) + monkeypatch.setattr(worker2, "_run", lambda *_a, **_k: None) + worker2.start(0, config_ref=config, cal_path=cal) + assert worker2._session_cal_path == str(cal) + finally: + _restore(saved) + + +def test_finger_worker_open_passes_cal_path_and_device_id(monkeypatch, tmp_path): + from orisys_app_shared.acquisition import finger_worker as mod + + saved = _install_fake_orisys(runtime_cls=_TrackingRuntime) + _TrackingRuntime.opened = {} + config = tmp_path / "sensor.json" + config.write_text("{}", encoding="utf-8") + cal = tmp_path / "DD02-14.json" + cal.write_text("{}", encoding="utf-8") + mode = _mode(config, "finger-fixture") + try: + monkeypatch.setattr(mod, "orisys_importable", lambda: (True, "")) + monkeypatch.setattr(mod, "load_fixed_mode", lambda _ref: mode) + monkeypatch.setattr(mod, "resolve_finger_export_dir", lambda _ref: None) + monkeypatch.setattr(mod, "loaded_calibration_version", lambda _s: "") + monkeypatch.setattr(mod, "loaded_path", lambda _s: None) + import orisys.finger as finger_mod + + monkeypatch.setattr(finger_mod, "FingerRuntime", _TrackingRuntime) + worker = mod.FingerAcquisitionWorker() + worker._mode = mode + worker._video_source = 0 + worker._session_id = 1 + worker._session_cal_path = str(cal) + worker._session_device_id = "DD02-14" + worker._export_dir = None + worker._stop.clear() + opened = worker._open_runtime(1, None) + assert opened is not None + assert _TrackingRuntime.opened.get("cal_path") == str(cal) + assert _TrackingRuntime.opened.get("device_id") == "DD02-14" + + worker2 = mod.FingerAcquisitionWorker() + monkeypatch.setattr(worker2, "stop", lambda: None) + monkeypatch.setattr(worker2, "_run", lambda *_a, **_k: None) + worker2.start( + 0, config_ref=config, cal_path=cal, device_id="DD02-14" + ) + assert worker2._session_cal_path == str(cal) + assert worker2._session_device_id == "DD02-14" + finally: + _restore(saved) + + +def test_workers_omit_cal_path_when_none_is_chosen(monkeypatch, tmp_path): + from orisys_app_shared.acquisition import finger_worker as mod + + saved = _install_fake_orisys(runtime_cls=_TrackingRuntime) + _TrackingRuntime.opened = {} + config = tmp_path / "sensor.json" + config.write_text("{}", encoding="utf-8") + mode = _mode(config, "finger-fixture") + try: + monkeypatch.setattr(mod, "orisys_importable", lambda: (True, "")) + monkeypatch.setattr(mod, "load_fixed_mode", lambda _ref: mode) + monkeypatch.setattr(mod, "resolve_finger_export_dir", lambda _ref: None) + monkeypatch.setattr(mod, "loaded_calibration_version", lambda _s: "") + monkeypatch.setattr(mod, "loaded_path", lambda _s: None) + import orisys.finger as finger_mod + + monkeypatch.setattr(finger_mod, "FingerRuntime", _TrackingRuntime) + worker = mod.FingerAcquisitionWorker() + worker._mode = mode + worker._video_source = 0 + worker._session_id = 1 + worker._session_cal_path = None + worker._session_device_id = None + worker._export_dir = None + worker._stop.clear() + opened = worker._open_runtime(1, None) + assert opened is not None + assert "cal_path" not in _TrackingRuntime.opened + assert "device_id" not in _TrackingRuntime.opened + finally: + _restore(saved)