ykcheungab
81063213f9
控制器与两个对话框共用 CalibrationService.read_identity，解析失败与缺 SN 分开上报； 本轮改动文件统一 LF，SN 必填回归已锁定。 定向 73 项、校准 UI 8 项与 lint 通过；未跑全量门禁。
452 lines
15 KiB
Python
452 lines
15 KiB
Python
"""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
|