让直播标定文件从传感器对话框进入打开链路。

对话框可选配置默认、原始力或浏览文件,并按配置条目写入 prefs;点运行时把 cal_path(Finger 含 device_id)交给 worker,采集中则热更新。对话框与打开链路单测通过。
This commit is contained in:
ykcheungab
2026-10-08 16:37:26 +08:00
parent 19ae1f4968
commit 3d2fb0c4eb
9 changed files with 1006 additions and 54 deletions
@@ -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
@@ -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:
+186
View File
@@ -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)