From 531f8657c4f7e4142eb5bcd106851c1a1ae2cd2f Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Tue, 13 Oct 2026 13:39:40 +0200 Subject: [PATCH] feat(devices): keep the laser delay calibration in soft records on an IOC --- .../device_configs/laser_delay_sim.yaml | 3 +- .../device_configs/x10da_laser_delay.yaml | 4 +- superxas_bec/devices/laser_delay.py | 109 ++++++--- superxas_bec/devices/laser_delay_sim.py | 44 +++- tests/tests_devices/test_laser_delay.py | 228 ++++++++++++------ 5 files changed, 277 insertions(+), 111 deletions(-) diff --git a/superxas_bec/device_configs/laser_delay_sim.yaml b/superxas_bec/device_configs/laser_delay_sim.yaml index 6a89f6a..9d9bdec 100644 --- a/superxas_bec/device_configs/laser_delay_sim.yaml +++ b/superxas_bec/device_configs/laser_delay_sim.yaml @@ -4,7 +4,7 @@ # SimLaserDelay: LaserDelay with soft signals, for a local demo BEC. # bec.config.add_to_session("/superxas_bec/device_configs/laser_delay_sim.yaml") # dev.laser_delay.calibrate(dev.laser_delay.sim_scope.get()) -# step_sleep is shortened for demos (beamline: 0.1). +# step_sleep is shortened for demos (beamline: 0.1); the calibration is kept in memory. laser_delay: readoutPriority: baseline @@ -12,7 +12,6 @@ laser_delay: deviceClass: superxas_bec.devices.laser_delay_sim.SimLaserDelay deviceConfig: step_sleep: 0.01 - calibration_file: ~/.superxas_bec/laser_delay_sim_calibration.json deviceTags: - laser - timing diff --git a/superxas_bec/device_configs/x10da_laser_delay.yaml b/superxas_bec/device_configs/x10da_laser_delay.yaml index 8a243b9..c44386b 100644 --- a/superxas_bec/device_configs/x10da_laser_delay.yaml +++ b/superxas_bec/device_configs/x10da_laser_delay.yaml @@ -3,7 +3,7 @@ ################################### # Delay card + RF phase shifter 2 (ES4), Duetto laser. Docs: LaserDelay docstrings. # Calibrate after every laser restart: dev.laser_delay.calibrate(); the calibration -# is kept in calibration_file, see dev.laser_delay.get_calibration_info(). +# is kept in the X10DA-LAS-DELAY: soft records, see dev.laser_delay.get_calibration_info(). # onFailure buffer: an IOC outage does not abort unrelated scans. laser_delay: @@ -23,7 +23,7 @@ laser_delay: step_sleep: 0.1 phase_path: script # "shortest" is opt-in fine_rollover: true # FINE 200 -> next coarse unit (open question) - calibration_file: ~/.superxas_bec/laser_delay_calibration.json + calibration_pv_prefix: "X10DA-LAS-DELAY:" # autosaved soft records with the calibration deviceTags: - laser - timing diff --git a/superxas_bec/devices/laser_delay.py b/superxas_bec/devices/laser_delay.py index b12fdf9..996ee8f 100644 --- a/superxas_bec/devices/laser_delay.py +++ b/superxas_bec/devices/laser_delay.py @@ -16,8 +16,8 @@ A delay ``t`` (X-ray minus laser, as on the oscilloscope) is set as: Calibration (script ``CurrentSituation``, after every laser restart) from the scope reading ``s``: ``offset_t = -(2 DELAY + 0.01 FINE) - s``, ``offset_phase_t = -phase_t - (s mod cavity_period)``. -Stored in config signals and a JSON ``calibration_file`` (script: ``Timing.offset``), verified -against the hardware on connect. +Stored in config signals and in autosaved soft records on an IOC (``calibration_pv_prefix``; +script: ``Timing.offset``), verified against the hardware on connect. Safety: moves are refused if V1/V2 do not match the stored phase within one ramp step (``phase_mismatch``), if the amplitude is off, or if a restored calibration is unverified. @@ -31,19 +31,16 @@ from __future__ import annotations import contextlib import datetime import functools -import json import math -import os import threading import time from dataclasses import asdict, dataclass -from pathlib import Path from typing import TYPE_CHECKING, Callable import numpy as np from bec_lib.logger import bec_logger from ophyd import Component as Cpt -from ophyd import EpicsSignal, EpicsSignalRO +from ophyd import Device, EpicsSignal, EpicsSignalRO from ophyd import FormattedComponent as FCpt from ophyd import Kind, Signal from ophyd.positioner import PositionerBase @@ -397,37 +394,75 @@ def _phase_path(value): # calibration storage -class _FileCalibrationStore: - """Calibration kept in a JSON file on the device server host.""" +class LaserDelayCalibrationPVs(Device): + """Autosaved soft records on an IOC that keep the laser delay calibration. - def __init__(self, path: str): - self.path = Path(os.path.expanduser(path)) + ``ao``: the four calibration numbers, ``DELAY``, ``CAL-SCOPE`` (ns); ``stringout`` ``CAL-TIME``; + ``bo`` ``MOVING``. + """ + + laser_period = Cpt(EpicsSignal, "LASER-PERIOD", kind=Kind.config) + offset_t = Cpt(EpicsSignal, "OFFSET-T", kind=Kind.config) + offset_phase_t = Cpt(EpicsSignal, "OFFSET-PHASE-T", kind=Kind.config) + phase_t = Cpt(EpicsSignal, "PHASE-T", kind=Kind.config) + current_delay = Cpt(EpicsSignal, "DELAY", kind=Kind.config) + scope = Cpt(EpicsSignal, "CAL-SCOPE", kind=Kind.config) + calibrated_at = Cpt(EpicsSignal, "CAL-TIME", string=True, kind=Kind.config) + moving = Cpt(EpicsSignal, "MOVING", kind=Kind.config) + + +class _PVCalibrationStore: + """Calibration kept in soft records on an IOC (``LaserDelayCalibrationPVs``).""" + + CONNECT_TIMEOUT = 5.0 + + def __init__(self, prefix: str, name: str): + self.prefix = prefix + self.pvs = LaserDelayCalibrationPVs(prefix, name=f"{name}_calibration_pvs") def __str__(self) -> str: - return str(self.path) + return f"PVs {self.prefix}*" def load(self) -> dict | None: - """Saved content, or None if there is no file.""" - if not self.path.exists(): + """Content of the records, or None if they hold no calibration (laser period 0 or NaN).""" + self.pvs.wait_for_connection(timeout=self.CONNECT_TIMEOUT) + laser_period = float(self.pvs.laser_period.get()) + if not math.isfinite(laser_period) or laser_period <= 0: return None - return json.loads(self.path.read_text(encoding="utf-8")) + delay = float(self.pvs.current_delay.get()) + scope = float(self.pvs.scope.get()) + return { + "calibration": { + "laser_period": laser_period, + "offset_t": float(self.pvs.offset_t.get()), + "offset_phase_t": float(self.pvs.offset_phase_t.get()), + "phase_t": float(self.pvs.phase_t.get()), + }, + "current_delay": delay if math.isfinite(delay) else None, + "scope_xray_minus_laser": scope if math.isfinite(scope) else None, + "calibrated_at": str(self.pvs.calibrated_at.get()) or None, + "move_in_progress": bool(int(self.pvs.moving.get())), + } def save(self, content: dict) -> None: - """Write atomically (temp file, fsync, rename).""" - self.path.parent.mkdir(parents=True, exist_ok=True) - tmp = self.path.with_name(f".{self.path.name}.{os.getpid()}.{threading.get_ident()}.tmp") - try: - with open(tmp, "w", encoding="utf-8") as handle: - handle.write(json.dumps(content, indent=2)) - handle.flush() - os.fsync(handle.fileno()) - os.replace(tmp, self.path) - finally: - if tmp.exists(): - tmp.unlink() + """Write the records; the move marker first when set, last when cleared.""" + moving = 1 if content.get("move_in_progress") else 0 + if moving: + self.pvs.moving.put(moving, wait=True) + calibration = content["calibration"] + for key in ("laser_period", "offset_t", "offset_phase_t", "phase_t"): + getattr(self.pvs, key).put(float(calibration[key]), wait=True) + delay = content.get("current_delay") + self.pvs.current_delay.put(math.nan if delay is None else float(delay), wait=True) + scope = content.get("scope_xray_minus_laser") + self.pvs.scope.put(math.nan if scope is None else float(scope), wait=True) + self.pvs.calibrated_at.put(content.get("calibrated_at") or "", wait=True) + if not moving: + self.pvs.moving.put(moving, wait=True) def destroy(self) -> None: - """Nothing to release.""" + """Release the channel access connections.""" + self.pvs.destroy() # device @@ -461,7 +496,8 @@ class LaserDelay(PSIDeviceBase, PositionerBase): step_sleep (float): seconds between ramp steps, at least ``MIN_STEP_SLEEP``. phase_path (str): ``"script"`` (default) or ``"shortest"``. fine_rollover (bool): write ``FINE = 200`` as the next coarse unit (default True). - calibration_file (str | None): JSON file keeping the calibration across restarts. + calibration_pv_prefix (str | None): prefix of the soft records keeping the calibration + across restarts; None: not kept. limits (list[float] | None): soft limits ``[low, high]`` in ns. """ @@ -573,7 +609,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): evr_period_pv = _ConstructorOnly() phase_v1_pv = _ConstructorOnly() phase_v2_pv = _ConstructorOnly() - calibration_file = _ConstructorOnly() + calibration_pv_prefix = _ConstructorOnly() # pylint: disable=too-many-arguments,too-many-locals def __init__( @@ -593,7 +629,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): step_sleep: float = 0.1, phase_path: str = "script", fine_rollover: bool = True, - calibration_file: str | None = None, + calibration_pv_prefix: str | None = None, limits: list[float] | tuple[float, float] | None = None, scan_info: ScanInfo | None = None, device_manager: DeviceManagerBase | None = None, @@ -604,7 +640,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): self._evr_period_pv = evr_period_pv self._phase_v1_pv = phase_v1_pv self._phase_v2_pv = phase_v2_pv - self._calibration_file = calibration_file + self._calibration_pv_prefix = calibration_pv_prefix self._move_lock = threading.RLock() # guards _move_status and _move_stop self._hw_lock = threading.Lock() # held by the move task and by calibrate/set_calibration self._hw_permit = threading.local() # per-thread permission to write the hardware signals @@ -617,9 +653,10 @@ class LaserDelay(PSIDeviceBase, PositionerBase): ) self.readback.name = self.name self._position = math.nan - self._store: _FileCalibrationStore | None = None - if calibration_file: - self._store = _FileCalibrationStore(calibration_file) + # store: load() -> dict | None, save(dict), destroy(), str() + self._store = None + if calibration_pv_prefix: + self._store = _PVCalibrationStore(calibration_pv_prefix, self.name) self.rf_period.put(float(rf_period)) self.cavity_period.put(float(cavity_period)) self.vmax.put(float(vmax)) @@ -883,7 +920,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): return self._calibration().as_dict() def get_calibration_info(self) -> dict: - """Calibration with origin, age, verification state and storage file.""" + """Calibration with origin, age, verification state and storage.""" calibrated_at = self._calibration_meta.get("calibrated_at") age_days = None if calibrated_at: @@ -1093,7 +1130,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): # calibration storage def _extra_saved_state(self) -> dict: - """Extra content for the calibration file (simulation hook).""" + """Extra content for the calibration store (simulation hook).""" return {} def _save_calibration( diff --git a/superxas_bec/devices/laser_delay_sim.py b/superxas_bec/devices/laser_delay_sim.py index 34c416c..822956c 100644 --- a/superxas_bec/devices/laser_delay_sim.py +++ b/superxas_bec/devices/laser_delay_sim.py @@ -2,6 +2,7 @@ from __future__ import annotations +import copy import inspect import math import threading @@ -25,6 +26,45 @@ from superxas_bec.devices.laser_delay import ( logger = bec_logger.logger +# per device name, for the lifetime of the process (survives a config reload) +_SIM_CALIBRATIONS: dict[str, dict] = {} + + +class MemoryCalibrationStore: + """In-memory stand-in for the calibration PVs (simulation and tests), keyed by device name.""" + + def __init__(self, name: str, registry: dict[str, dict] | None = None): + self.name = name + self.registry = _SIM_CALIBRATIONS if registry is None else registry + + def __str__(self) -> str: + return f"memory ({self.name})" + + @property + def content(self) -> dict | None: + """The stored content (for tests); None if nothing is stored.""" + return self.registry.get(self.name) + + @content.setter + def content(self, value: dict | None) -> None: + if value is None: + self.registry.pop(self.name, None) + else: + self.registry[self.name] = copy.deepcopy(value) + + def load(self) -> dict | None: + """Stored content, or None.""" + content = self.registry.get(self.name) + return copy.deepcopy(content) if content is not None else None + + def save(self, content: dict) -> None: + """Replace the stored content.""" + self.registry[self.name] = copy.deepcopy(content) + + def destroy(self) -> None: + """Nothing to release; the content stays for the next device object.""" + + class _HardwareSoftSignal(_DeviceWritesOnly, CheckedSignal): """Soft stand-in for a hardware PV.""" @@ -119,6 +159,8 @@ class SimLaserDelay(LaserDelay): **kwargs, ): super().__init__(prefix=prefix, name=name, **kwargs) + if self._store is None: # no calibration PVs in the simulation: keep it in memory + self._store = MemoryCalibrationStore(self.name) self._sim_cavity_period = sim_cavity_period self._sim_time_zero = float(sim_time_zero) self._sim_rise = float(sim_rise) @@ -164,7 +206,7 @@ class SimLaserDelay(LaserDelay): Called after the device is connected and its signals are connected. Default values for signals should be set here. """ - # soft signals lose their state on restart; restore it before the calibration + # soft signals lose their state on a rebuild; restore it before the calibration try: content = self._read_saved_calibration() sim = (content or {}).get("sim") diff --git a/tests/tests_devices/test_laser_delay.py b/tests/tests_devices/test_laser_delay.py index 05933bd..584e5ae 100644 --- a/tests/tests_devices/test_laser_delay.py +++ b/tests/tests_devices/test_laser_delay.py @@ -5,7 +5,6 @@ from __future__ import annotations import functools -import json import math import threading import time @@ -44,7 +43,11 @@ from superxas_bec.devices.laser_delay import ( voltages_for_phase, wrap_phase, ) -from superxas_bec.devices.laser_delay_sim import SimLaserDelay +from superxas_bec.devices.laser_delay_sim import ( + _SIM_CALIBRATIONS, + MemoryCalibrationStore, + SimLaserDelay, +) CONFIG_DIR = Path(__file__).resolve().parents[2] / "superxas_bec" / "device_configs" SCOPE = 546.04 - 546.92 @@ -136,7 +139,7 @@ def test_default_pvs_constants_and_positioner_type(laser_delay): "plan_move", } assert config["laser_delay_fine_rollover"]["value"] is True - assert laser_delay.calibration_file is None + assert laser_delay.calibration_pv_prefix is None assert isinstance(laser_delay, BECPositionerProtocol) assert isinstance(laser_delay, BECDeviceProtocol) @@ -1003,21 +1006,47 @@ def caplog(): log.patcher.stop() +@pytest.fixture(autouse=True) +def clean_sim_calibrations(): + """Start each test with an empty simulation calibration registry.""" + _SIM_CALIBRATIONS.clear() + yield + _SIM_CALIBRATIONS.clear() + + @pytest.fixture -def calibration_file(tmp_path): - return tmp_path / "sub" / "laser_delay_calibration.json" +def store(): + """Calibration store standing in for the calibration PVs (same interface).""" + return MemoryCalibrationStore("laser_delay", registry={}) -def make_device(calibration_file, **hardware): - """A fresh device (as after a BEC restart) on the given mocked hardware.""" +def make_device(store, device_kwargs=None, **hardware): + """Fresh device (as after a restart) on mocked hardware, using ``store``.""" context = patched_device( - LaserDelay, name="laser_delay", step_sleep=0.0, calibration_file=str(calibration_file) + LaserDelay, name="laser_delay", step_sleep=0.0, **(device_kwargs or {}) ) device = context.__enter__() + device._store = store set_hardware(device, **hardware) return context, device +class _FailingStore(MemoryCalibrationStore): + def __init__(self, fail_load=False, fail_save=False): + super().__init__("laser_delay", registry={}) + self.fail_load, self.fail_save = fail_load, fail_save + + def load(self): + if self.fail_load: + raise ValueError("unreadable calibration") + return super().load() + + def save(self, content): + if self.fail_save: + raise ConnectionError("calibration PVs not connected") + super().save(content) + + def hardware_of(device): return { "coarse": device.delay_coarse.get(), @@ -1027,16 +1056,16 @@ def hardware_of(device): } -def test_calibration_is_saved_after_calibrate_and_every_move(calibration_file): - context, device = make_device(calibration_file) +def test_calibration_is_saved_after_calibrate_and_every_move(store): + context, device = make_device(store) try: device.calibrate(SCOPE) - saved = json.loads(calibration_file.read_text()) + saved = store.content assert saved["calibration"] == device.get_calibration() assert saved["current_delay"] == pytest.approx(SCOPE) assert saved["scope_xray_minus_laser"] == pytest.approx(SCOPE) device.set(1.5).wait(timeout=10) - saved = json.loads(calibration_file.read_text()) + saved = store.content assert saved["calibration"]["phase_t"] == device.get_calibration()["phase_t"] assert saved["current_delay"] == 1.5 assert saved["move_in_progress"] is False @@ -1044,14 +1073,14 @@ def test_calibration_is_saved_after_calibrate_and_every_move(calibration_file): info = device.get_calibration_info() assert info["calibrated_at"] == saved["calibrated_at"] assert info["age_days"] == pytest.approx(0, abs=0.01) - assert info["storage"] == str(calibration_file) + assert info["storage"] == str(store) assert info["verified"] is True finally: context.__exit__(None, None, None) -def test_calibration_is_restored_after_a_restart(calibration_file): - context, device = make_device(calibration_file) +def test_calibration_is_restored_after_a_restart(store): + context, device = make_device(store) try: device.calibrate(SCOPE) device.set(1.5).wait(timeout=10) @@ -1059,7 +1088,7 @@ def test_calibration_is_restored_after_a_restart(calibration_file): plan_before = device.plan_move(-2.0) finally: context.__exit__(None, None, None) - context, device = make_device(calibration_file, **hardware) + context, device = make_device(store, **hardware) try: assert math.isnan(device.get_calibration()["phase_t"]) device.on_connected() @@ -1070,8 +1099,8 @@ def test_calibration_is_restored_after_a_restart(calibration_file): context.__exit__(None, None, None) -def test_restore_with_turned_phase_shifter_refuses_moves(calibration_file, caplog): - context, device = make_device(calibration_file) +def test_restore_with_turned_phase_shifter_refuses_moves(store, caplog): + context, device = make_device(store) try: device.calibrate(SCOPE) device.set(1.5).wait(timeout=10) @@ -1079,7 +1108,7 @@ def test_restore_with_turned_phase_shifter_refuses_moves(calibration_file, caplo finally: context.__exit__(None, None, None) hardware["v1"], hardware["v2"] = hardware["v2"], -hardware["v1"] # phase turned by hand - context, device = make_device(calibration_file, **hardware) + context, device = make_device(store, **hardware) try: device.on_connected() assert device.get_calibration() == before @@ -1095,9 +1124,9 @@ def test_restore_with_turned_phase_shifter_refuses_moves(calibration_file, caplo context.__exit__(None, None, None) -def test_restore_after_a_crash_during_a_move_refuses_moves(calibration_file): +def test_restore_after_a_crash_during_a_move_refuses_moves(store): """The marker written before the first hardware write survives a crashed device server.""" - context, device = make_device(calibration_file) + context, device = make_device(store) try: device.calibrate(SCOPE) with mock.patch.object(device, "_write_phase", side_effect=SystemExit("crash")): @@ -1105,13 +1134,13 @@ def test_restore_after_a_crash_during_a_move_refuses_moves(calibration_file): with pytest.raises(Exception): status.wait(timeout=10) # simulate the crash: the end-of-move save never happened - content = json.loads(calibration_file.read_text()) + content = store.content content["move_in_progress"] = True - calibration_file.write_text(json.dumps(content)) + store.content = content hardware = hardware_of(device) finally: context.__exit__(None, None, None) - context, device = make_device(calibration_file, **hardware) + context, device = make_device(store, **hardware) try: device.on_connected() assert "stopped during a move" in device.get_calibration_info()["unverified_reasons"][0] @@ -1121,33 +1150,33 @@ def test_restore_after_a_crash_during_a_move_refuses_moves(calibration_file): context.__exit__(None, None, None) -def test_move_marker_is_set_before_the_first_write(calibration_file): - context, device = make_device(calibration_file) +def test_move_marker_is_set_before_the_first_write(store): + context, device = make_device(store) try: device.calibrate(SCOPE) seen = [] original = device.delay_coarse._write_pv.put def check_marker(value, *args, **kwargs): - seen.append(json.loads(calibration_file.read_text())["move_in_progress"]) + seen.append(store.content["move_in_progress"]) return original(value, *args, **kwargs) device.delay_coarse._write_pv.put = check_marker device.set(2.0).wait(timeout=10) assert seen == [True] - assert json.loads(calibration_file.read_text())["move_in_progress"] is False + assert store.content["move_in_progress"] is False finally: context.__exit__(None, None, None) -def test_restore_with_changed_laser_period_refuses_moves(calibration_file): - context, device = make_device(calibration_file) +def test_restore_with_changed_laser_period_refuses_moves(store): + context, device = make_device(store) try: device.calibrate(SCOPE) hardware = hardware_of(device) finally: context.__exit__(None, None, None) - context, device = make_device(calibration_file, evr=1.0, **hardware) + context, device = make_device(store, evr=1.0, **hardware) try: device.on_connected() assert "laser period changed" in device.get_calibration_info()["unverified_reasons"][0] @@ -1155,9 +1184,9 @@ def test_restore_with_changed_laser_period_refuses_moves(calibration_file): context.__exit__(None, None, None) -def test_restore_with_changed_delay_card_refuses_moves(calibration_file): +def test_restore_with_changed_delay_card_refuses_moves(store): """A delay card changed outside BEC leaves the restored calibration unverified.""" - context, device = make_device(calibration_file) + context, device = make_device(store) try: device.calibrate(SCOPE) device.set(1.5).wait(timeout=10) @@ -1165,7 +1194,7 @@ def test_restore_with_changed_delay_card_refuses_moves(calibration_file): finally: context.__exit__(None, None, None) hardware["fine"] = (hardware["fine"] + 50) % 200 - context, device = make_device(calibration_file, **hardware) + context, device = make_device(store, **hardware) try: device.on_connected() info = device.get_calibration_info() @@ -1180,13 +1209,13 @@ def test_restore_with_changed_delay_card_refuses_moves(calibration_file): context.__exit__(None, None, None) -def test_restore_after_an_external_move_by_whole_rf_periods_refuses_moves(calibration_file): +def test_restore_after_an_external_move_by_whole_rf_periods_refuses_moves(store): """A whole-RF-period change while BEC was down is caught by the delay card check.""" - context, device = make_device(calibration_file) + context, device = make_device(store) try: device.calibrate(SCOPE) device.set(5.0).wait(timeout=10) - saved = calibration_file.read_text() # BEC's last state: 5.0 ns + saved = store.load() # BEC's last state: 5.0 ns at_5 = hardware_of(device) device.set(3.0).wait(timeout=10) # what the script does meanwhile: one RF period hardware = hardware_of(device) @@ -1194,8 +1223,8 @@ def test_restore_after_an_external_move_by_whole_rf_periods_refuses_moves(calibr assert (hardware["coarse"], hardware["fine"]) != (at_5["coarse"], at_5["fine"]) finally: context.__exit__(None, None, None) - calibration_file.write_text(saved) # BEC never saw the script's move - context, device = make_device(calibration_file, **hardware) + store.content = saved # BEC never saw the script's move + context, device = make_device(store, **hardware) try: device.on_connected() assert device.read_phase_mismatch() == pytest.approx(0, abs=1e-9) # phase check is blind @@ -1207,22 +1236,14 @@ def test_restore_after_an_external_move_by_whole_rf_periods_refuses_moves(calibr context.__exit__(None, None, None) -def test_restore_with_changed_constants_refuses_moves(calibration_file): - context, device = make_device(calibration_file) +def test_restore_with_changed_constants_refuses_moves(store): + context, device = make_device(store) try: device.calibrate(SCOPE) hardware = hardware_of(device) finally: context.__exit__(None, None, None) - context = patched_device( - LaserDelay, - name="laser_delay", - step_sleep=0.0, - cavity_period=12.0084, - calibration_file=str(calibration_file), - ) - device = context.__enter__() - set_hardware(device, **hardware) + context, device = make_device(store, {"cavity_period": 12.0084}, **hardware) try: device.on_connected() info = device.get_calibration_info() @@ -1231,16 +1252,16 @@ def test_restore_with_changed_constants_refuses_moves(calibration_file): context.__exit__(None, None, None) -def test_old_calibration_is_restored_with_a_warning(calibration_file, caplog): - context, device = make_device(calibration_file) +def test_old_calibration_is_restored_with_a_warning(store, caplog): + context, device = make_device(store) try: device.calibrate(SCOPE) finally: context.__exit__(None, None, None) - content = json.loads(calibration_file.read_text()) + content = store.content content["calibrated_at"] = "2026-01-01T00:00:00+00:00" - calibration_file.write_text(json.dumps(content)) - context, device = make_device(calibration_file) + store.content = content + context, device = make_device(store) try: device.on_connected() assert device.read()["laser_delay"]["value"] == pytest.approx(SCOPE) @@ -1249,10 +1270,8 @@ def test_old_calibration_is_restored_with_a_warning(calibration_file, caplog): context.__exit__(None, None, None) -def test_broken_calibration_file_leaves_the_device_uncalibrated(calibration_file, caplog): - calibration_file.parent.mkdir(parents=True) - calibration_file.write_text("{not json") - context, device = make_device(calibration_file) +def test_unreadable_calibration_store_leaves_the_device_uncalibrated(caplog): + context, device = make_device(_FailingStore(fail_load=True)) try: device.on_connected() # must not raise: the device still connects assert "could not restore" in caplog.text @@ -1262,19 +1281,15 @@ def test_broken_calibration_file_leaves_the_device_uncalibrated(calibration_file context.__exit__(None, None, None) -def test_no_calibration_file_configured_writes_nothing(laser_delay, tmp_path, monkeypatch): - monkeypatch.chdir(tmp_path) +def test_no_calibration_store_configured(laser_delay): laser_delay.on_connected() laser_delay.calibrate(SCOPE) assert laser_delay.get_calibration_info()["storage"] is None - assert not list(tmp_path.iterdir()) -def test_unwritable_calibration_store_refuses_the_move(tmp_path, caplog): +def test_unwritable_calibration_store_refuses_the_move(caplog): """Without the move marker a later restore could trust a stale phase: fail safe.""" - blocker = tmp_path / "file" - blocker.write_text("") - context, device = make_device(blocker / "cal.json") # parent is a file + context, device = make_device(_FailingStore(fail_save=True)) try: with pytest.raises(LaserDelayPersistenceError, match="will be lost at the next restart"): device.calibrate(SCOPE) @@ -1291,9 +1306,9 @@ def test_unwritable_calibration_store_refuses_the_move(tmp_path, caplog): context.__exit__(None, None, None) -def test_sim_restart_restores_hardware_and_calibration(tmp_path): - path = tmp_path / "cal.json" - device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0, calibration_file=path) +def test_sim_reload_restores_hardware_and_calibration(): + """A rebuilt device with the same name gets the calibration back.""" + device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0) try: device.on_connected() device.calibrate(device.sim_scope.get()) @@ -1301,7 +1316,7 @@ def test_sim_restart_restores_hardware_and_calibration(tmp_path): device.set(target).wait(timeout=30) finally: device.destroy() - device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0, calibration_file=path) + device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0) try: device.on_connected() assert device.read()["laser_delay"]["value"] == 14.2 @@ -1448,3 +1463,76 @@ def test_clamped_phase_voltage_fails_the_move_and_refuses_the_next(calibrated): assert not calibrated.get_calibration_info()["verified"] with pytest.raises(LaserDelayNotCalibratedError): calibrated.set(2.0) + + +# calibration PVs + +PV_PREFIX = "X10DA-LAS-DELAY:" + + +def store_values(device) -> dict: + pvs = device._store.pvs + return {sig: getattr(pvs, sig)._read_pv.mock_data for sig in pvs.component_names} + + +def test_calibration_pvs_are_written_and_restored(): + with patched_device( + LaserDelay, name="laser_delay", step_sleep=0.0, calibration_pv_prefix=PV_PREFIX + ) as device: + set_hardware(device) + device.calibrate(SCOPE) + device.set(1.5).wait(timeout=10) + before, hardware, saved = ( + device.get_calibration(), + hardware_of(device), + store_values(device), + ) + assert device._store.pvs.laser_period.pvname == f"{PV_PREFIX}LASER-PERIOD" + assert saved["phase_t"] == before["phase_t"] + assert saved["current_delay"] == 1.5 + assert saved["moving"] == 0 + assert device.get_calibration_info()["storage"] == f"PVs {PV_PREFIX}*" + with patched_device( + LaserDelay, name="laser_delay", step_sleep=0.0, calibration_pv_prefix=PV_PREFIX + ) as device: + set_hardware(device, **hardware) + for sig, value in saved.items(): + getattr(device._store.pvs, sig)._read_pv.mock_data = value + device.on_connected() + assert device.get_calibration() == before + assert device.read()["laser_delay"]["value"] == 1.5 + assert device.get_calibration_info()["verified"] + + +def test_empty_calibration_pvs_leave_the_device_uncalibrated(): + with patched_device( + LaserDelay, name="laser_delay", step_sleep=0.0, calibration_pv_prefix=PV_PREFIX + ) as device: + set_hardware(device) + device._store.pvs.laser_period._read_pv.mock_data = 0.0 # fresh records, no autosave + device.on_connected() + with pytest.raises(LaserDelayNotCalibratedError, match="not calibrated"): + device.set(1.0) + + +def test_calibration_pv_marker_is_set_first_and_cleared_last(): + with patched_device( + LaserDelay, name="laser_delay", step_sleep=0.0, calibration_pv_prefix=PV_PREFIX + ) as device: + set_hardware(device) + device.calibrate(SCOPE) + order = [] + for sig in device._store.pvs.component_names: + pv = getattr(device._store.pvs, sig)._write_pv + original = pv.put + + def recorded(value, *args, _sig=sig, _original=original, **kwargs): + order.append((_sig, value)) + return _original(value, *args, **kwargs) + + pv.put = recorded + device.set(1.0).wait(timeout=10) + names = [sig for sig, _ in order] + assert order[0] == ("moving", 1) + assert order[-1] == ("moving", 0) + assert names.count("phase_t") == 2