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

统一实时标定身份解析并规范行尾

Browse Source 
控制器与两个对话框共用 CalibrationService.read_identity，解析失败与缺 SN 分开上报；
本轮改动文件统一 LF，SN 必填回归已锁定。
定向 73 项、校准 UI 8 项与 lint 通过；未跑全量门禁。
This commit is contained in:
ykcheungab
2026-10-19 18:42:13 +08:00
parent 70b4341fd0
commit 81063213f9
13 changed files with 1095 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/studio/config/locales/en_US.json
+5
View File
@@ -291,6 +291,10 @@
"mode_replay_tip": "The charts read the open recording; capture and recording continue",
"mode_recomputed": "Recomputed from this video (browsing)",
"mode_recomputed_tip": "Curves come from this analysis, not forces.csv.",
"field_experiment": "Experimental model field_gain/1: weighted unloaded baseline {baseline} N ({frames} frames), mean extrapolation {extrapolated}%",
"field_experiment_tip": "This is an experimental estimate, not a validated local pressure or total force. Model fingerprint {fingerprint}; strategy {strategy}; unloaded reference {reference}; HHD grid {grid}, downsample {downsample}",
"field_unknown": "Unknown experiment source: this result carries no field diagnostics (probably an older cached result)",
"field_unknown_tip": "Older caches do not record the weighted baseline or per-frame extrapolation; re-run the analysis for full diagnostics",
"mode_no_trace": "No force trace — {reason}",
"mode_no_trace_tip": "There is nothing to plot, and the reason is a fact about the input rather than a loading state",
"trace_not_analysed": "this video has not been analysed yet",
@@ -358,6 +362,7 @@
"device_missing_sn": "This calibration file has no SN, so it cannot be matched to a sensor",
"device_mismatch": "The calibration file's SN is {sn}, which is not the device id entered",
"device_family": "This calibration belongs to {family}, not Finger",
"device_unreadable": "This calibration file cannot be read, so the sensor it belongs to cannot be confirmed ({error})",
"reference": "Reference frame",
"reference_time": "{seconds} s",
"reference_missing": "The recording has no frame {frame}.",
apps/studio/config/locales/zh_CN.json
+5
View File
@@ -291,6 +291,10 @@
"mode_replay_tip": "曲线读的是打开的录制；回放期间采集与记录继续",
"mode_recomputed": "从这段视频重算（浏览）",
"mode_recomputed_tip": "曲线来自这次分析，不是 forces.csv。",
"field_experiment": "实验模型 field_gain/1：加权空载基线 {baseline} N（{frames} 帧），平均外推 {extrapolated}%",
"field_experiment_tip": "这是一个实验估计，不是已验证的局部压力或总力。模型指纹 {fingerprint}；策略 {strategy}；无载参考 {reference}；HHD 网格 {grid}，downsample {downsample}",
"field_unknown": "实验来源未知：这个结果没有 field 诊断（可能是旧版本缓存）",
"field_unknown_tip": "旧缓存不记录加权基线与逐帧外推比例；重跑分析可得到完整诊断",
"mode_no_trace": "没有力曲线 — {reason}",
"mode_no_trace_tip": "没有可画的东西，而原因是输入本身的事实，不是一种加载状态",
"trace_not_analysed": "这段视频还没有分析过",
@@ -358,6 +362,7 @@
"device_missing_sn": "这份标定文件没有 SN，无法确认它属于哪台传感器",
"device_mismatch": "标定文件的 SN 是 {sn}，与填写的设备编号不一致",
"device_family": "这份标定属于 {family}，不能用于 Finger",
"device_unreadable": "这份标定文件无法读取，不能确认它属于哪台传感器（{error}）",
"reference": "参考帧",
"reference_time": "{seconds} 秒",
"reference_missing": "录制中没有第 {frame} 帧。",
apps/studio/src/orisys_studio/app_window.py
+66 -193
View File
@@ -46,6 +46,7 @@ from orisys_studio.acquire_presenter import AcquirePresenter
from orisys_studio.calibration_service import CalibrationService
from orisys_studio.display_prefs import DisplayPrefsController
from orisys_studio.frame_view import FrameView
from orisys_studio.live_calibration import LiveCalibrationController
from orisys_studio.recording import RecordingCoordinator, record_config_from_prefs
from orisys_studio.replay import ReplayEngine
from orisys_studio.replay_controller import ReplayController
@@ -211,8 +212,6 @@ class StudioMainWindow(QMainWindow):
#: frame that was already in flight cannot repaint the panels they paused on.
self._display_paused = False
self._active_calibration_id = ""
#: Last coefficients file written by the calibration panel (not applied).
self._last_calibration_export = ""
#: Built once the panels exist (see below); until then the status rail
#: treats calibration as unknown.
self.acquire_presenter: AcquirePresenter | None = None
@@ -234,6 +233,24 @@ class StudioMainWindow(QMainWindow):
#: Built on first use so the pickers are parented where they belong.
self.sensor_dialog = None
self.status_presenter = StatusPresenter(self.toolbar)
# The live calibration choice's rules (the stored per-bundle choice, the
# profile default, the Finger serial check) live in ``live_calibration``;
# the window keeps the forwarders the sensor dialog, the Run entry, the
# bare-video prompt and the UI checks name.
self.live_calibration = LiveCalibrationController(
prefs=self.prefs,
selected_entry=self.sensor_picker.selected_entry,
sensor_type=self.sensor_picker.sensor_type,
config_ref=self._resolve_config_ref,
status=self._live_calibration_status,
log_event=self.recorder.log_event,
running=lambda: self._running,
active_worker=lambda: self._active_worker,
worker_for=self._worker_for,
confirm=self._confirm_live_calibration,
dialog_problem=self._live_calibration_dialog_problem,
select_in_dialog=self._select_calibration_in_dialog,
)
# Bind the shared mask and chart history before controllers can deliver frames.
self.frame_view = FrameView(self.arrow_panel, self.mesh3d_panel)
self.mesh3d_panel.set_domain_latch(self.frame_view.domain_mask)
@@ -1520,60 +1537,50 @@ class StudioMainWindow(QMainWindow):
)
def _on_calibration_exported(self, path: str) -> None:
"""Note the coefficients that were written. Nothing is applied here.
"""The panel wrote coefficients: remember the file, apply nothing.
Exporting and applying are separate actions: writing a file must never
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.
Exporting and applying are separate actions; the rule about which file the
next open/apply uses lives in ``live_calibration.py``.
"""
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))
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))
self.live_calibration.note_exported(path)
# ------------------------------------------------- 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
# The rules live in ``live_calibration.py``; these names are the narrow
# interface the sensor dialog, the Run entry, the bare-video prompt and the UI
# checks already use. Each is a read or a delegation, never a second rule.
@property
def _last_calibration_export(self) -> str:
"""Last coefficients file written by the calibration panel (not applied)."""
return self.live_calibration.last_export
def _live_calibration_status(self, text: str) -> None:
self.statusBar().showMessage(str(text or ""))
def _live_calibration_dialog_problem(self) -> str | None:
"""The sensor dialog's current problem, or ``None`` when it is not open."""
dialog = self.sensor_dialog
return None if dialog is None else dialog.choice_problem()
def _select_calibration_in_dialog(self, path: Path) -> None:
if self.sensor_dialog is not None:
self.sensor_dialog.select_calibration_file(path)
def _confirm_live_calibration(self, title: str, text: str) -> bool:
"""The operator's answer to one live-calibration confirmation."""
answer = QMessageBox.question(
self,
title,
text,
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
return answer == QMessageBox.StandardButton.Yes
def live_calibration_for(
self, ref: str
) -> tuple[str, Path | None, str] | None:
"""Saved live choice for one catalog ref, or ``None`` when nothing was saved.
``None`` is not raw force: a caller that has its own default (the analysis
prompt) keeps that default until the sensor dialog has stored a choice.
"""
from orisys_studio.ui.sensor_dialog import ROLE_DEFAULT, ROLE_FILE, ROLE_NONE
store = self.prefs.get("live_calibration")
if not ref or not isinstance(store, dict):
return None
entry = store.get(ref)
if not isinstance(entry, dict):
return 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 "")
"""Saved live choice for one catalog ref, or ``None`` when nothing was saved."""
return self.live_calibration.for_ref(ref)
def live_calibration_choice(self) -> tuple[str, Path | None, str]:
"""``(role, file_path_or_None, device_id)`` remembered for the selected bundle.
@@ -1581,13 +1588,7 @@ class StudioMainWindow(QMainWindow):
``file_path`` is only set for the browsed-file role; the profile default is
resolved when the chain opens. No saved choice is raw force for a live open.
"""
from orisys_studio.ui.sensor_dialog import ROLE_NONE
key = self._live_calibration_prefs_key()
found = None if key is None else self.live_calibration_for(key)
if found is None:
return ROLE_NONE, None, ""
return found
return self.live_calibration.choice()
def set_live_calibration_choice(
self,
@@ -1597,34 +1598,7 @@ class StudioMainWindow(QMainWindow):
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
self.live_calibration.save_choice(role=role, path=path, device_id=device_id)
def _resolved_live_calibration(
self,
@@ -1633,70 +1607,17 @@ class StudioMainWindow(QMainWindow):
``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, ""
return self.live_calibration.resolve()
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
if device_id and hasattr(worker, "request_calibration_zero"):
worker.set_pending_calibration(cal_path, device_id=device_id)
else:
worker.set_pending_calibration(cal_path)
self.statusBar().showMessage(
t("sensor.live_calibration_applied", name=cal_path.name)
)
self.live_calibration.apply_choice()
# ------------------------------------------------- calibration (forwards)
# "What does this coefficient file claim, and does it match the bench?" lives in
# ``CalibrationService`` (with the rest of the calibration file work, and without
# Qt). The dialog and the worker hand-over stay here: they are interaction, not a
# rule about files.
# Qt). The dialog and the UI checker read these two names; the live calibration
# controller calls the service directly.
def calibration_identity(self, path: str) -> dict:
return CalibrationService.identity_of(path)
@@ -1708,61 +1629,13 @@ class StudioMainWindow(QMainWindow):
)
def _on_calibration_apply(self, path: str) -> None:
"""Push freshly exported coefficients into the device — on confirmation.
"""Confirm and request freshly exported coefficients; the device confirms.
The user is shown the exact target device first, and only the matching
worker is updated. A calibration that names a *different* device or
configuration is called out explicitly and defaults to "no".
Identity and matching come from ``CalibrationService``, the confirmation is
the window's, and the actual hand-over is a request to the worker: the
recorded version still comes from the acquisition thread's confirmation.
"""
path = str(path or self._last_calibration_export)
if not path:
self.statusBar().showMessage(t("cal.apply_no_export"))
return
sensor_type = self.sensor_picker.sensor_type()
config_ref = self._resolve_config_ref() or ""
reasons = self.calibration_target_mismatch(
path, sensor_type=sensor_type, config_ref=config_ref
)
if reasons:
answer = QMessageBox.question(
self,
t("cal.apply_mismatch_title"),
t(
"cal.apply_mismatch",
device=self.calibration_identity(path).get("sensor_type") or "?",
detail="; ".join(reasons),
sensor_type=sensor_type,
config=config_ref,
),
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
if answer != QMessageBox.StandardButton.Yes:
self.recorder.log_event(
"calibration_apply_blocked", "; ".join(reasons)
)
self.statusBar().showMessage(t("cal.apply_mismatch_skipped"))
return
answer = QMessageBox.question(
self,
t("cal.apply_title"),
t("cal.apply_confirm", sensor_type=sensor_type, config=config_ref),
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
if answer != QMessageBox.StandardButton.Yes:
self.statusBar().showMessage(t("cal.apply_skipped"))
return
worker = self._worker_for(sensor_type)
if worker is None or not hasattr(worker, "set_pending_calibration"):
self.statusBar().showMessage(t("cal.apply_no_target"))
return
# Only a request. The acquisition thread confirms once the coefficients
# are actually loaded; the event and the recorded version come from there.
worker.set_pending_calibration(path)
self.statusBar().showMessage(
t("cal.apply_pending", sensor_type=sensor_type)
)
self.live_calibration.request_apply(path)
def _on_calibration_applied(
self, path: str, version: str, frame_id: int, mono_t: float, session: int
apps/studio/src/orisys_studio/calibration_service.py
+21 -8
View File
@@ -134,27 +134,40 @@ class CalibrationService:
# match what is on the bench?" — which is the question the *apply* path has to settle
# before coefficients measured on another sensor silently change every newton value.
@staticmethod
def identity_of(path: str | Path) -> dict:
"""What device a coefficient file says it belongs to.
def read_identity(path: str | Path) -> tuple[dict, str]:
"""What device a coefficient file says it belongs to, and why not.
Read with the SDK's own parser so the check sees the same document the device
would load. A file that cannot be read yields no identity, and the caller then
applies it without a mismatch claim.
One strict parse answers both questions: ``error`` is the SDK's own one-line
reason when the document cannot be loaded, and ``""`` when it can. A document
that parses without an SN yields an identity with empty fields and no error —
"cannot be read" and "has no SN" must not be reported as the same thing.
"""
try:
import orisys.calibration as sdk_calibration
document = sdk_calibration.load(str(path), strict=True)
except Exception: # noqa: BLE001 - reported by the apply path itself
except Exception as exc: # noqa: BLE001 - reported to the caller
logger.exception("Could not read calibration identity from %s", path)
return {}
reason = " ".join(str(exc).split()) or type(exc).__name__
return {}, reason
identity = dict(getattr(document, "identity", {}) or {})
pipeline = dict(getattr(document, "pipeline", {}) or {})
return {
"sensor_type": str(identity.get("sensor_type") or ""),
"sn": str(identity.get("sn") or ""),
"config_ref": str(pipeline.get("config_ref") or ""),
}
}, ""
@staticmethod
def identity_of(path: str | Path) -> dict:
"""What device a coefficient file says it belongs to.
Read with the SDK's own parser so the check sees the same document the device
would load. A file that cannot be read yields no identity, and the caller then
applies it without a mismatch claim. Callers that show *why* it could not be
read use :meth:`read_identity` instead.
"""
return CalibrationService.read_identity(path)[0]
@staticmethod
def same_path(first: str, second: str) -> bool:
apps/studio/src/orisys_studio/live_calibration.py
+299
View File
@@ -0,0 +1,299 @@
"""The live calibration choice: which coefficients the next live run opens with.
The window used to answer this: the per-bundle choice stored in the preferences,
the file a bundle's profile declares as its default, and whether a browsed Finger
file names the device the operator typed. Those are rules about a stored choice
and a file, not about the window, so they live here and are tested without one.
The preference payload is the same ``live_calibration`` mapping the sensor dialog
has always written (one entry per catalog ref), saved through the same
``save_preferences`` entry: extracting the rules must not create a second copy of
the state.
The window keeps forwarders for the callers that name them — the sensor dialog's
narrow choice interface, the Run entry, the bare-video prompt and the UI checks.
They read and delegate to this controller; the controller is given the facts it
reads and the actions it needs, not the window itself. File identity and matching
reuse :class:`CalibrationService`, so a live serial check and the Apply path agree
on whose file this is.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Callable
from orisys_app_shared.i18n import t
from orisys_app_shared.prefs import save_preferences
from orisys_studio.calibration_service import CalibrationService
#: The answers the live calibration choice can store, in the vocabulary the
#: sensor dialog's combo uses as item data. This is the preference payload's
#: schema; the dialog keeps the same strings as its own UI constants.
ROLE_DEFAULT = "default"
ROLE_FILE = "file"
ROLE_NONE = "none"
_ROLES = (ROLE_DEFAULT, ROLE_FILE, ROLE_NONE)
class LiveCalibrationController:
"""The stored live calibration choice and the file rules around it."""
def __init__(
self,
*,
prefs: dict,
selected_entry: Callable[[], Any],
sensor_type: Callable[[], str],
config_ref: Callable[[], str | None],
status: Callable[[str], None],
log_event: Callable[[str, str], None],
running: Callable[[], bool],
active_worker: Callable[[], Any],
worker_for: Callable[[str], Any],
confirm: Callable[[str, str], bool],
dialog_problem: Callable[[], str | None] | None = None,
select_in_dialog: Callable[[Path], None] | None = None,
save: Callable[[dict], None] = save_preferences,
) -> None:
#: The window's own preference mapping: the same object is read and saved,
#: so this controller can never hold a stale copy of the state.
self.prefs = prefs
#: The catalog entry currently selected in the rail's picker (or ``None``).
self._selected_entry = selected_entry
#: What the run would open with, for the apply confirmation and the worker
#: hand-over. These are reads, not state of this controller's.
self._sensor_type = sensor_type
self._config_ref = config_ref
self._status = status
self._log_event = log_event
self._running = running
self._active_worker = active_worker
self._worker_for = worker_for
#: Ask the operator ``(title, text) -> accepted``; the window owns the dialog.
self._confirm = confirm
#: ``None`` when no sensor dialog is open; otherwise its current problem.
self._dialog_problem = dialog_problem
#: How the sensor dialog is told about a just-exported file (may be absent).
self._select_in_dialog = select_in_dialog
self._save = save
#: Last coefficients file written by the calibration panel (not applied).
self.last_export = ""
# ------------------------------------------------------------------ reading
def for_ref(self, ref: str) -> tuple[str, Path | None, str] | None:
"""The saved choice for one catalog ref, or ``None`` when none was saved.
``None`` is not raw force: a caller with its own default (the analysis
prompt) keeps that default until the sensor dialog stores a choice.
"""
store = self.prefs.get("live_calibration")
if not ref or not isinstance(store, dict):
return None
entry = store.get(ref)
if not isinstance(entry, dict):
return None
role = str(entry.get("role") or ROLE_NONE)
if role not in _ROLES:
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():
# A remembered file that no longer exists is "no choice", not a
# failure to open: the operator can still pick another one.
return ROLE_NONE, None, ""
path = candidate
return role, path, str(entry.get("device_id") or "")
def choice(self) -> tuple[str, Path | None, str]:
"""``(role, file_path_or_None, device_id)`` for the selected bundle.
``file_path`` is only set for the browsed-file role; the profile default is
resolved when the chain opens. No saved choice is raw force for a live open.
"""
entry = self._selected_entry()
ref = "" if entry is None else str(getattr(entry, "ref", "") or "")
found = None if not ref else self.for_ref(ref)
if found is None:
return ROLE_NONE, None, ""
return found
def default_path(self) -> Path | None:
"""The calibration file the selected bundle's profile declares, if it exists."""
entry = self._selected_entry()
if entry is None or getattr(entry, "profile", None) 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
# ----------------------------------------------------------------- writing
def save_choice(
self,
*,
role: str,
path: Path | str | None = None,
device_id: str | None = None,
) -> None:
"""Remember the live calibration choice for the selected bundle."""
entry = self._selected_entry()
key = "" if entry is None else str(getattr(entry, "ref", "") or "")
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 _ROLES:
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
self._save(self.prefs)
# ---------------------------------------------------------------- resolve
def resolve(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 or
cannot be parsed. Identity and matching come from
:class:`CalibrationService`, the same rules the Apply path uses: one answer
to "whose file is this", not two parsers. A parse failure is reported as
itself, with the parser's reason; it is never confused with a missing SN.
"""
role, path, device_id = self.choice()
if role == ROLE_NONE:
return None, None, ""
if role == ROLE_DEFAULT:
return self.default_path(), None, ""
if role != ROLE_FILE or path is None:
return None, None, ""
entry = self._selected_entry()
is_finger = (
entry is not None
and str(getattr(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")
identity, error = CalibrationService.read_identity(path)
if error:
# "cannot be read" and "parsed without an SN" are different answers;
# the service returns the parser's own one-line reason for the former.
return path, None, t("video_analysis.device_unreadable", error=error)
sensor_type = str(identity.get("sensor_type") or "")
sn = str(identity.get("sn") or "")
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, ""
# ------------------------------------------------------------- exporting
def note_exported(self, path: str | Path) -> Path:
"""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. The new file becomes the dialog's
selected live calibration so Apply / the next Run can use it.
"""
export = Path(path)
self.last_export = str(export)
self._log_event("calibration_exported", str(export))
identity = CalibrationService.identity_of(export)
self.save_choice(
role=ROLE_FILE,
path=export,
device_id=str(identity.get("sn") or "") or None,
)
if self._select_in_dialog is not None:
self._select_in_dialog(export)
self._status(t("cal.export_only", path=export.name))
return export
# --------------------------------------------------------------- applying
def apply_choice(self) -> None:
"""Remember the dialog's choice; hot-update the device when a session is open."""
problem = None if self._dialog_problem is None else self._dialog_problem()
if problem:
self._status(t("sensor.live_calibration_invalid", reason=problem))
return
cal_path, device_id, problem = self.resolve()
if problem:
self._status(t("sensor.live_calibration_invalid", reason=problem))
return
if cal_path is None or not self._running():
self._status(t("sensor.live_calibration_remembered"))
return
worker = self._active_worker()
if worker is None or not hasattr(worker, "set_pending_calibration"):
self._status(t("cal.apply_no_target"))
return
if device_id and hasattr(worker, "request_calibration_zero"):
worker.set_pending_calibration(cal_path, device_id=device_id)
else:
worker.set_pending_calibration(cal_path)
self._status(t("sensor.live_calibration_applied", name=cal_path.name))
def request_apply(self, path: str | Path | None = None) -> None:
"""Push freshly exported coefficients into the device — on confirmation.
The operator is shown the exact target device first, and only the matching
worker is updated. A calibration that names a *different* device or
configuration is called out explicitly and defaults to "no". The device is
asked, not changed here: the acquisition thread confirms what it loaded and
the record channel notes the version from that confirmation.
"""
path = str(path or self.last_export)
if not path:
self._status(t("cal.apply_no_export"))
return
sensor_type = self._sensor_type()
config_ref = self._config_ref() or ""
reasons = CalibrationService.target_mismatch(
path, sensor_type=sensor_type, config_ref=config_ref
)
if reasons:
recorded = CalibrationService.identity_of(path).get("sensor_type") or "?"
accepted = self._confirm(
t("cal.apply_mismatch_title"),
t(
"cal.apply_mismatch",
device=recorded,
detail="; ".join(reasons),
sensor_type=sensor_type,
config=config_ref,
),
)
if not accepted:
self._log_event("calibration_apply_blocked", "; ".join(reasons))
self._status(t("cal.apply_mismatch_skipped"))
return
accepted = self._confirm(
t("cal.apply_title"),
t("cal.apply_confirm", sensor_type=sensor_type, config=config_ref),
)
if not accepted:
self._status(t("cal.apply_skipped"))
return
worker = self._worker_for(sensor_type)
if worker is None or not hasattr(worker, "set_pending_calibration"):
self._status(t("cal.apply_no_target"))
return
# Only a request. The acquisition thread confirms once the coefficients
# are actually loaded; the event and the recorded version come from there.
worker.set_pending_calibration(path)
self._status(t("cal.apply_pending", sensor_type=sensor_type))
apps/studio/src/orisys_studio/ui/sensor_dialog.py
+8 -3
View File
@@ -42,8 +42,8 @@ from PySide6.QtWidgets import (
QWidget,
)
from orisys_studio.calibration_service import CalibrationService
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.
@@ -445,7 +445,8 @@ class SensorDialog(QDialog):
return
if path != self._device_source:
self._device_source = path
_sensor_type, sn = calibration_identity(path)
identity, _error = CalibrationService.read_identity(path)
sn = str(identity.get("sn") or "")
if sn:
self._syncing = True
try:
@@ -464,7 +465,11 @@ class SensorDialog(QDialog):
serial = self.device_edit.text().strip()
if not serial:
return t("video_analysis.device_required")
sensor_type, sn = calibration_identity(path)
identity, error = CalibrationService.read_identity(path)
if error:
return t("video_analysis.device_unreadable", error=error)
sensor_type = str(identity.get("sensor_type") or "")
sn = str(identity.get("sn") or "")
if sensor_type and sensor_type != "finger":
return t("video_analysis.device_family", family=sensor_type)
if not sn:
apps/studio/src/orisys_studio/ui/video_analysis_dialog.py
+8 -21
View File
@@ -49,6 +49,7 @@ from PySide6.QtWidgets import (
QVBoxLayout,
)
from orisys_studio.calibration_service import CalibrationService
from orisys_studio.ui.fit_text import ShrinkableLabel
#: The calibration combo's two fixed rows. The first is a *file* the profile declares; the
@@ -58,25 +59,6 @@ ROLE_NONE = "none"
ROLE_FILE = "file"
def calibration_identity(path: Path) -> tuple[str, str]:
"""``(sensor_type, sn)`` from a force-calibration file, or empty strings.
The SDK reads the same two places (``identity``, then ``meta``) when it checks
that a chosen file belongs to the device the operator named.
"""
try:
data = json.loads(Path(path).read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return "", ""
if not isinstance(data, dict):
return "", ""
meta = data.get("meta") if isinstance(data.get("meta"), dict) else {}
identity = data.get("identity") if isinstance(data.get("identity"), dict) else {}
sensor_type = str(identity.get("sensor_type") or meta.get("sensor_type") or "")
sn = str(identity.get("sn") or meta.get("sn") or "")
return sensor_type, sn
class VideoAnalysisDialog(QDialog):
"""Ask how one bare video should be analysed, and hand back the three answers."""
@@ -454,7 +436,8 @@ class VideoAnalysisDialog(QDialog):
return
if path != self._device_source:
self._device_source = path
_sensor_type, sn = calibration_identity(path)
identity, _error = CalibrationService.read_identity(path)
sn = str(identity.get("sn") or "")
if sn:
self.device_edit.setText(sn)
@@ -472,7 +455,11 @@ class VideoAnalysisDialog(QDialog):
serial = self.device_edit.text().strip()
if not serial:
return t("video_analysis.device_required")
sensor_type, sn = calibration_identity(path)
identity, error = CalibrationService.read_identity(path)
if error:
return t("video_analysis.device_unreadable", error=error)
sensor_type = str(identity.get("sensor_type") or "")
sn = str(identity.get("sn") or "")
if sensor_type and sensor_type != "finger":
return t("video_analysis.device_family", family=sensor_type)
if not sn:
apps/studio/tests/test_bare_video_analysis.py
+26 -5
View File
@@ -742,6 +742,31 @@ def test_analysis_setup_reopens_the_prompt_after_the_first_pass(
assert window.replay_ctl.available_result() is not None
def _coefficient_document(directory: Path, *, sensor_type: str, sn: str) -> Path:
"""A v2 coefficient document the SDK's own parser accepts (a real export shape).
The analysis prompt now reads a Finger file's identity through
``CalibrationService`` — the strict parser — so a ``meta`` block in arbitrary
JSON is not an identity.
"""
from orisys_app_shared.force.calibration_io import (
CalSample,
CalSessionStore,
build_export_model,
)
store = CalSessionStore()
store.aggregate("fn", CalSample(weight_n=0.0, raw_mean=0.0, n=10))
store.aggregate("fn", CalSample(weight_n=5.0, raw_mean=500.0, n=10))
model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
document = model.to_dict()
document["identity"] = {"sensor_type": sensor_type, "sn": sn}
document["pipeline"] = {"config_ref": "D:/lab/sensor.json"}
path = directory / f"{sensor_type}-{sn}.json"
path.write_text(json.dumps(document), encoding="utf-8")
return path
def test_a_finger_calibration_file_must_name_the_same_device(qt_app, tmp_path):
"""A browsed Finger file is accepted only when the device id matches its SN."""
import cv2
@@ -763,11 +788,7 @@ def test_a_finger_calibration_file_must_name_the_same_device(qt_app, tmp_path):
None,
)
assert finger is not None, "the catalog has no startable Finger profile"
path = tmp_path / "DD02-14.json"
path.write_text(
json.dumps({"meta": {"sensor_type": "finger", "sn": "DD02-14"}}),
encoding="utf-8",
)
path = _coefficient_document(tmp_path, sensor_type="finger", sn="DD02-14")
dialog = VideoAnalysisDialog(video=clip, facts=facts, catalog=catalog)
try:
assert dialog.select_profile(finger.ref)
apps/studio/tests/test_calibration_service.py
+49
View File
@@ -79,3 +79,52 @@ def test_the_same_configuration_named_differently_is_not_a_mismatch(tmp_path):
# ...but it does have to be a path: a non-path value simply differs.
assert CalibrationService.same_path("D:/lab/a/sensor.json", "not a path\x00") is False
# --------------------------------------------------- one parse, both answers
def test_read_identity_returns_the_identity_and_no_error(tmp_path):
"""The strict parse answers with the identity *and* says it succeeded."""
path = _calibration_file(tmp_path, sensor_type="box", config_ref="D:/lab/a/sensor.json")
identity, error = CalibrationService.read_identity(path)
assert error == ""
assert identity == {
"sensor_type": "box",
"sn": "SN-A",
"config_ref": "D:/lab/a/sensor.json",
}
# The old name keeps its behaviour and is a thin wrapper over the one parse.
assert CalibrationService.identity_of(path) == identity
def test_read_identity_keeps_the_parsers_reason_when_the_file_cannot_be_read(tmp_path):
broken = tmp_path / "broken.json"
broken.write_text(
json.dumps({"schema_version": 2, "unit": "N", "channels": {"fn": "bogus"}}),
encoding="utf-8",
)
identity, error = CalibrationService.read_identity(broken)
assert identity == {}
assert error
assert "\n" not in error, "the reason must stay one line for the status bar"
assert CalibrationService.identity_of(broken) == {}
def test_read_identity_distinguishes_a_parsed_document_without_an_sn(tmp_path):
"""A readable document that simply names no device is not a parse failure."""
path = _calibration_file(tmp_path, sensor_type="box", config_ref="D:/lab/a/sensor.json")
document = json.loads(path.read_text(encoding="utf-8"))
document.pop("identity")
document.pop("pipeline")
path.write_text(json.dumps(document), encoding="utf-8")
identity, error = CalibrationService.read_identity(path)
assert error == ""
assert identity["sn"] == ""
assert identity["sensor_type"] == ""
apps/studio/tests/test_live_calibration.py
+451
View File
@@ -0,0 +1,451 @@
"""The live calibration choice's rules, tested without a window.
These are the facts the sensor dialog, the Run entry and the bare-video prompt
read: what was remembered for one bundle, what the profile declares as its
default, and whether a Finger file names the device the operator typed. The
window now only forwards to this controller, so it is exercised with a dict and
three lambdas.
"""
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
from orisys_app_shared.force.calibration_io import (
CalSample,
CalSessionStore,
build_export_model,
)
from orisys_app_shared.i18n import t
from orisys_studio.live_calibration import (
ROLE_DEFAULT,
ROLE_FILE,
ROLE_NONE,
LiveCalibrationController,
)
def _coefficients(directory: Path, *, sensor_type: str, sn: str) -> Path:
"""A coefficient document the SDK's own loader accepts (a real export shape)."""
store = CalSessionStore()
store.aggregate("fn", CalSample(weight_n=0.0, raw_mean=0.0, n=10))
store.aggregate("fn", CalSample(weight_n=5.0, raw_mean=500.0, n=10))
model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
document = model.to_dict()
document["identity"] = {"sensor_type": sensor_type, "sn": sn}
document["pipeline"] = {"config_ref": "D:/lab/sensor.json"}
path = directory / f"{sensor_type}_calibration.json"
path.write_text(json.dumps(document), encoding="utf-8")
return path
def _entry(ref: str, *, family: str = "box", default=None):
profile = SimpleNamespace(default_calibration=default)
return SimpleNamespace(ref=ref, product_family=family, profile=profile)
class _Worker:
"""The narrow acquisition-worker surface these paths use."""
def __init__(self) -> None:
self.pending: list[tuple[str, str | None]] = []
def set_pending_calibration(self, path, *, device_id=None) -> None:
self.pending.append((str(path), device_id))
class _ZeroCapableWorker(_Worker):
"""A worker that can also ask the device to zero before loading coefficients."""
def request_calibration_zero(self) -> None:
return
class _Host:
"""The callbacks the controller is injected with, and nothing else."""
def __init__(self, entry=None) -> None:
self.entry = entry
self.prefs: dict = {}
self.saves = 0
self.sensor_type = "box"
self.config_ref = "D:/lab/sensor.json"
self.running = False
self.worker = None
self.workers: dict = {}
self.statuses: list[str] = []
self.events: list[tuple[str, str]] = []
self.answers: list[bool] = []
self.confirmations: list[tuple[str, str]] = []
self.dialog_problem: str | None = None
self.selected: list[Path] = []
def controller(self) -> LiveCalibrationController:
return LiveCalibrationController(
prefs=self.prefs,
selected_entry=lambda: self.entry,
sensor_type=lambda: self.sensor_type,
config_ref=lambda: self.config_ref,
status=self.statuses.append,
log_event=lambda name, detail: self.events.append((name, detail)),
running=lambda: self.running,
active_worker=lambda: self.worker,
worker_for=lambda sensor_type: self.workers.get(
sensor_type, self.worker
),
confirm=self._confirm,
dialog_problem=lambda: self.dialog_problem,
select_in_dialog=self.selected.append,
save=self._save,
)
def _save(self, prefs: dict) -> None:
self.saves += 1
assert prefs is self.prefs
def _confirm(self, title: str, text: str) -> bool:
self.confirmations.append((title, text))
return self.answers.pop(0) if self.answers else False
def test_no_saved_choice_is_not_the_same_as_raw_force():
host = _Host(_entry("shipped/box"))
controller = host.controller()
assert controller.choice() == (ROLE_NONE, None, "")
assert controller.for_ref("shipped/box") is None
assert controller.resolve() == (None, None, "")
def test_a_file_choice_is_remembered_per_bundle(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal)
assert controller.for_ref("shipped/box") == (ROLE_FILE, cal, "")
assert controller.for_ref("shipped/finger") is None
assert host.prefs["live_calibration"]["shipped/box"]["path"] == str(cal)
assert host.saves == 1
def test_a_remembered_file_that_vanished_reads_as_no_choice(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal, device_id="SN-A")
cal.unlink()
assert controller.for_ref("shipped/box") == (ROLE_NONE, None, "")
assert controller.choice() == (ROLE_NONE, None, "")
def test_a_profile_default_is_resolved_when_it_is_asked_for(tmp_path):
default = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
host = _Host(_entry("shipped/box", default=default))
controller = host.controller()
controller.save_choice(role=ROLE_DEFAULT)
assert controller.choice() == (ROLE_DEFAULT, None, "")
assert controller.default_path() == default
assert controller.resolve() == (default, None, "")
def test_a_saved_choice_never_saves_one_key_for_another_bundle():
host = _Host(_entry("shipped/box"))
controller = host.controller()
controller.save_choice(role=ROLE_FILE, path="D:/lab/a.json", device_id=" SN-A ")
payload = host.prefs["live_calibration"]["shipped/box"]
assert payload == {"role": ROLE_FILE, "path": "D:/lab/a.json", "device_id": "SN-A"}
assert host.saves == 1
def test_an_unknown_role_is_remembered_as_raw_force():
host = _Host(_entry("shipped/box"))
controller = host.controller()
controller.save_choice(role="exported")
assert host.prefs["live_calibration"]["shipped/box"] == {"role": ROLE_NONE}
def test_without_a_selected_bundle_nothing_is_saved():
host = _Host(None)
controller = host.controller()
controller.save_choice(role=ROLE_FILE, path="D:/lab/a.json")
assert "live_calibration" not in host.prefs
assert host.saves == 0
def test_a_finger_file_must_name_the_typed_serial(tmp_path):
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="finger", sn="DD02-14")
controller.save_choice(role=ROLE_FILE, path=cal, device_id="DD02-14")
assert controller.resolve() == (cal, "DD02-14", "")
controller.save_choice(role=ROLE_FILE, path=cal, device_id="OTHER")
_path, _device, problem = controller.resolve()
assert problem == t("video_analysis.device_mismatch", sn="DD02-14")
def test_a_finger_file_without_a_typed_serial_is_refused(tmp_path):
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="finger", sn="DD02-14")
controller.save_choice(role=ROLE_FILE, path=cal)
_path, _device, problem = controller.resolve()
assert problem == t("video_analysis.device_required")
def test_a_box_file_chosen_for_a_finger_is_refused(tmp_path):
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal, device_id="SN-A")
_path, _device, problem = controller.resolve()
assert problem == t("video_analysis.device_family", family="box")
def test_an_unreadable_document_is_not_reported_as_a_missing_sn(tmp_path):
"""A parse failure is its own answer, not "this file has no SN".
The fixture carries the right SN in a ``meta`` block but is not a coefficient
document the SDK accepts (no usable channel curves). A missing coordinate-map
attachment fails strict parsing the same way; either reason must reach the
operator instead of being reported as "no SN".
"""
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
path = tmp_path / "not-a-coefficient-document.json"
path.write_text(
json.dumps({"meta": {"sensor_type": "finger", "sn": "DD02-14"}}),
encoding="utf-8",
)
controller.save_choice(role=ROLE_FILE, path=path, device_id="DD02-14")
_path, _device, problem = controller.resolve()
assert problem != t("video_analysis.device_missing_sn")
template = t("video_analysis.device_unreadable", error="\x00")
prefix, _, suffix = template.partition("\x00")
assert problem.startswith(prefix) and problem.endswith(suffix), problem
detail = problem[len(prefix) : len(problem) - len(suffix)]
assert detail.strip(), "the parser's reason was dropped"
def test_a_loadable_document_without_a_serial_is_reported_as_a_missing_sn(tmp_path):
"""Parsing succeeds and the document names no device: that is a missing SN."""
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="finger", sn="")
controller.save_choice(role=ROLE_FILE, path=cal, device_id="DD02-14")
_path, _device, problem = controller.resolve()
assert problem == t("video_analysis.device_missing_sn")
def test_the_serial_check_only_applies_to_finger(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal)
assert controller.resolve() == (cal, None, "")
# ------------------------------------------------------ exporting and applying
def test_exporting_selects_the_file_without_applying_it(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
worker = _Worker()
host.worker = worker
returned = controller.note_exported(cal)
assert returned == cal
assert controller.last_export == str(cal)
assert host.prefs["live_calibration"]["shipped/box"] == {
"role": ROLE_FILE,
"path": str(cal),
"device_id": "SN-A",
}
assert host.selected == [cal]
assert host.statuses == [t("cal.export_only", path=cal.name)]
assert host.events == [("calibration_exported", str(cal))]
# Exporting is not applying: the device was never asked to load anything.
assert worker.pending == []
def test_apply_choice_when_idle_only_remembers(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal)
worker = _Worker()
host.worker = worker
controller.apply_choice()
assert host.statuses == [t("sensor.live_calibration_remembered")]
assert worker.pending == []
assert host.confirmations == []
def test_apply_choice_hot_updates_a_running_box_session(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal)
worker = _Worker()
host.worker = worker
host.running = True
controller.apply_choice()
assert worker.pending == [(str(cal), None)]
assert host.statuses == [
t("sensor.live_calibration_applied", name=cal.name)
]
def test_apply_choice_passes_the_serial_to_a_finger_worker(tmp_path):
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="finger", sn="DD02-14")
controller.save_choice(role=ROLE_FILE, path=cal, device_id="DD02-14")
worker = _ZeroCapableWorker()
host.worker = worker
host.running = True
controller.apply_choice()
assert worker.pending == [(str(cal), "DD02-14")]
def test_apply_choice_reports_the_open_dialogs_problem(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal)
worker = _Worker()
host.worker = worker
host.running = True
host.dialog_problem = t("video_analysis.device_required")
controller.apply_choice()
assert host.statuses == [
t("sensor.live_calibration_invalid", reason=t("video_analysis.device_required"))
]
assert worker.pending == []
def test_apply_choice_without_a_target_reports_it(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.save_choice(role=ROLE_FILE, path=cal)
host.running = True
host.worker = None
controller.apply_choice()
assert host.statuses == [t("cal.apply_no_target")]
def test_apply_requires_a_written_file():
host = _Host(_entry("shipped/box"))
controller = host.controller()
controller.request_apply()
assert host.statuses == [t("cal.apply_no_export")]
assert host.confirmations == []
def test_a_mismatched_file_needs_an_explicit_yes(tmp_path):
host = _Host(_entry("shipped/finger", family="finger"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
worker = _Worker()
host.workers = {"finger": worker}
host.sensor_type = "finger"
host.config_ref = "D:/lab/B/sensor.json"
host.answers = [False]
controller.request_apply(cal)
assert host.confirmations and host.confirmations[0][0] == t(
"cal.apply_mismatch_title"
)
assert len(host.events) == 1
assert host.events[0][0] == "calibration_apply_blocked"
assert "sensor_type" in host.events[0][1]
assert host.statuses == [t("cal.apply_mismatch_skipped")]
assert worker.pending == []
def test_a_matching_file_is_confirmed_once_then_requested(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
worker = _Worker()
host.workers = {"box": worker}
host.sensor_type = "box"
host.config_ref = "D:/lab/sensor.json"
host.answers = [True]
controller.request_apply(cal)
assert len(host.confirmations) == 1
assert host.confirmations[0][0] == t("cal.apply_title")
assert worker.pending == [(str(cal), None)]
assert host.statuses == [t("cal.apply_pending", sensor_type="box")]
# A request is not a confirmation: no applied event and no version here.
assert host.events == []
def test_apply_defaults_to_the_last_exported_file(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
controller.note_exported(cal)
worker = _Worker()
host.workers = {"box": worker}
host.answers = [True]
controller.request_apply()
assert worker.pending == [(str(cal), None)]
def test_apply_without_a_target_worker_reports_it(tmp_path):
host = _Host(_entry("shipped/box"))
controller = host.controller()
cal = _coefficients(tmp_path, sensor_type="box", sn="SN-A")
host.answers = [True]
host.worker = None
controller.request_apply(cal)
assert host.statuses == [t("cal.apply_no_target")]
assert host.worker is None
apps/studio/tests/test_live_calibration_dialog.py
+105 -7
View File
@@ -45,6 +45,42 @@ def _descriptor(profile_id: str, **overrides: object) -> dict[str, object]:
return data
def _coefficients(directory: Path, *, sensor_type: str, sn: str) -> Path:
"""A coefficient document the SDK's own loader accepts (a real export shape).
The live open/apply path reads a file's identity through ``CalibrationService``
— the same parser the device would use — so a Finger serial check needs a
document that really is a calibration file, not a ``meta`` block in any JSON.
"""
from orisys_app_shared.force.calibration_io import (
CalSample,
CalSessionStore,
build_export_model,
)
store = CalSessionStore()
store.aggregate("fn", CalSample(weight_n=0.0, raw_mean=0.0, n=10))
store.aggregate("fn", CalSample(weight_n=5.0, raw_mean=500.0, n=10))
model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
document = model.to_dict()
document["identity"] = {"sensor_type": sensor_type, "sn": sn}
document["pipeline"] = {"config_ref": "D:/lab/sensor.json"}
path = directory / f"{sensor_type}-{sn}.json"
path.write_text(json.dumps(document), encoding="utf-8")
return path
def _assert_unreadable(problem: str) -> str:
"""The problem is ``device_unreadable`` with the parser's reason, not "no SN"."""
template = t("video_analysis.device_unreadable", error="\x00")
prefix, _, suffix = template.partition("\x00")
assert problem != t("video_analysis.device_missing_sn")
assert problem.startswith(prefix) and problem.endswith(suffix), problem
detail = problem[len(prefix) : len(problem) - len(suffix)]
assert detail.strip(), "the parser's reason was dropped"
return detail
class _Host(QWidget):
"""Window stand-in with the live-calibration prefs API the dialog expects."""
@@ -233,11 +269,7 @@ def test_a_finger_file_with_a_mismatched_sn_is_not_a_valid_choice(
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",
)
cal = _coefficients(tmp_path, sensor_type="finger", sn="DD02-14")
try:
dialog.select_calibration_file(cal)
# Offscreen dialogs are not shown; ``isHidden`` is the visibility flag.
@@ -255,13 +287,79 @@ def test_a_finger_file_with_a_mismatched_sn_is_not_a_valid_choice(
picker.deleteLater()
def test_window_resolves_finger_file_choice_for_open(make_window, tmp_path):
window = make_window()
def _finger_dialog(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)
return picker, host, dialog
def test_a_finger_file_the_sdk_rejects_is_reported_as_unreadable(
qt_app, tmp_path, monkeypatch
):
"""A ``meta`` SN is not an identity: the dialog precheck uses the strict parse."""
picker, _host, dialog = _finger_dialog(qt_app, tmp_path, monkeypatch)
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)
dialog.device_edit.setText("DD02-14")
_assert_unreadable(dialog.choice_problem())
finally:
dialog.deleteLater()
picker.deleteLater()
def test_a_finger_file_that_is_not_json_is_reported_as_unreadable(
qt_app, tmp_path, monkeypatch
):
picker, _host, dialog = _finger_dialog(qt_app, tmp_path, monkeypatch)
cal = tmp_path / "not-json.json"
cal.write_text("not a calibration document\n", encoding="utf-8")
try:
dialog.select_calibration_file(cal)
dialog.device_edit.setText("DD02-14")
_assert_unreadable(dialog.choice_problem())
finally:
dialog.deleteLater()
picker.deleteLater()
def test_a_finger_file_without_an_sn_is_reported_as_missing_sn(
qt_app, tmp_path, monkeypatch
):
"""Readable but nameless stays its own answer, not a parse failure."""
picker, _host, dialog = _finger_dialog(qt_app, tmp_path, monkeypatch)
cal = _coefficients(tmp_path, sensor_type="finger", sn="")
try:
dialog.select_calibration_file(cal)
dialog.device_edit.setText("DD02-14")
assert dialog.choice_problem() == t("video_analysis.device_missing_sn")
finally:
dialog.deleteLater()
picker.deleteLater()
def test_window_resolves_finger_file_choice_for_open(make_window, tmp_path):
window = make_window()
cal = _coefficients(tmp_path, sensor_type="finger", sn="DD02-14")
finger = next(
(
entry