feat(devices): keep the laser delay calibration in soft records on an IOC

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>/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

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(<scope value>); 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

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(

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")

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