test(devices): add LaserDelay unit tests for calibration, stop and config

tests/tests_devices/test_laser_delay.py

@@ -0,0 +1,530 @@
+"""Unit tests for the LaserDelay positioner (mocked EPICS)."""
+
+# pylint: disable=redefined-outer-name,protected-access
+
+from __future__ import annotations
+
+import math
+import threading
+import time
+from pathlib import Path
+from unittest import mock
+
+import numpy as np
+import ophyd
+import pytest
+from bec_lib.bec_yaml_loader import yaml_load
+from bec_server.device_server.devices.device_serializer import get_device_base_class
+from bec_server.device_server.devices.devicemanager import DeviceManagerDS
+from ophyd.signal import InternalSignalError
+from ophyd.utils import LimitError
+from ophyd_devices.interfaces.base_classes.psi_device_base import DeviceStoppedError
+from ophyd_devices.tests.utils import MockPV, patched_device
+
+from superxas_bec.devices.laser_delay import (
+    LaserDelay,
+    LaserDelayBusyError,
+    LaserDelayNotCalibratedError,
+    SimLaserDelay,
+    phase_from_voltages,
+    phase_ramp_offsets,
+    voltages_for_phase,
+    wrap_phase,
+)
+
+CONFIG_DIR = Path(__file__).resolve().parents[2] / "superxas_bec" / "device_configs"
+SCOPE = 546.04 - 546.92
+
+
+def record_puts(device: LaserDelay) -> list:
+    """Record every put to the four written PVs, in order (thread safe enough for list.append)."""
+    log: list[tuple[str, object]] = []
+    for signal in (device.delay_coarse, device.delay_fine, device.phase_v1, device.phase_v2):
+        pv = signal._write_pv
+        original = pv.put
+
+        def recorded(value, *args, _pv=pv, _original=original, **kwargs):
+            log.append((_pv.pvname, value))
+            return _original(value, *args, **kwargs)
+
+        pv.put = recorded
+    return log
+
+
+def set_hardware(device: LaserDelay, evr=2.0002, coarse=150, fine=37, v1=0.6, v2=-0.8) -> None:
+    """Set the mocked PV values."""
+    device.evr_period._read_pv.mock_data = evr
+    device.delay_coarse._read_pv.mock_data = coarse
+    device.delay_fine._read_pv.mock_data = fine
+    device.phase_v1._read_pv.mock_data = v1
+    device.phase_v2._read_pv.mock_data = v2
+
+
+@pytest.fixture
+def laser_delay():
+    """Uncalibrated device with mocked PVs and no sleep between ramp steps."""
+    with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device:
+        set_hardware(device)
+        yield device
+        device.stop()
+
+
+@pytest.fixture
+def calibrated(laser_delay):
+    """Calibrated device."""
+    laser_delay.calibrate(SCOPE)
+    return laser_delay
+
+
+def wait_until(condition, timeout=5.0):
+    """Poll ``condition`` until it is true."""
+    start = time.monotonic()
+    while time.monotonic() - start < timeout:
+        if condition():
+            return True
+        time.sleep(0.005)
+    return False
+
+
+# --------------------------------------------------------------------------------------------- #
+#  Construction and configuration                                                               #
+# --------------------------------------------------------------------------------------------- #
+
+
+def test_default_pvs_constants_and_positioner_type(laser_delay):
+    assert laser_delay.delay_coarse.pvname == "X10DA-ES2-TIM:SET1-DELAY"
+    assert laser_delay.delay_fine.pvname == "X10DA-ES2-TIM:SET1-FINE"
+    assert laser_delay.evr_period.pvname == "X10DA-ES2-TIM:EVR-FPS2"
+    assert laser_delay.phase_v1.pvname == "X10DA-ES4:AO-02"
+    assert laser_delay.phase_v2.pvname == "X10DA-ES4:AO-01"
+    config = laser_delay.read_configuration()
+    assert config["laser_delay_rf_period"]["value"] == 2.0
+    assert config["laser_delay_cavity_period"]["value"] == 12.01
+    assert config["laser_delay_vmax"]["value"] == 1.0
+    assert config["laser_delay_steps_per_rf_period"]["value"] == 36
+    assert config["laser_delay_step_sleep"]["value"] == 0.0
+    assert config["laser_delay_phase_reference"]["value"] == "stored"
+    assert get_device_base_class(laser_delay) == "positioner"
+    assert laser_delay.hints == {"fields": ["laser_delay"]}
+    assert laser_delay.egu == "ns"
+    assert laser_delay.limits == (0.0, 0.0)
+    assert math.isnan(laser_delay.read()["laser_delay"]["value"])
+    assert set(LaserDelay.USER_ACCESS) == {
+        "calibrate",
+        "set_calibration",
+        "get_calibration",
+        "plan_move",
+    }
+
+
+def test_config_overrides_pvs_and_constants():
+    with patched_device(
+        LaserDelay,
+        name="ld",
+        delay_coarse_pv="X99:DELAY",
+        delay_fine_pv="X99:FINE",
+        evr_period_pv="X99:EVR",
+        phase_v1_pv="X99:V1",
+        phase_v2_pv="X99:V2",
+        rf_period=2.0014,
+        cavity_period=12.0084,
+        vmax=0.8,
+        steps_per_rf_period=72,
+        step_sleep=0.0,
+        phase_reference="readback",
+        limits=[-20, 20],
+    ) as device:
+        assert [s.pvname for s in (device.delay_coarse, device.delay_fine, device.evr_period)] == [
+            "X99:DELAY",
+            "X99:FINE",
+            "X99:EVR",
+        ]
+        assert (device.phase_v1.pvname, device.phase_v2.pvname) == ("X99:V1", "X99:V2")
+        assert device.limits == (-20.0, 20.0)
+        set_hardware(device, v1=0.8, v2=0.0)
+        log = record_puts(device)
+        device.calibrate(1.5)
+        plan = device.plan_move(3.0)
+        device.set(3.0).wait(timeout=10)
+        phase_puts = log[2:]
+        # readback reference: the ramp goes from the read-back phase straight to the target
+        assert plan["phase_delta"] == pytest.approx(plan["phase_target"] - plan["phase_start"])
+        expected_steps = len(phase_ramp_offsets(plan["phase_delta"], 2.0014, 72)) + 1
+        assert len(phase_puts) == 2 * expected_steps == 2 * plan["phase_steps"]
+        for (_, v1), (_, v2) in zip(phase_puts[::2], phase_puts[1::2]):
+            assert math.hypot(v1, v2) == pytest.approx(0.8)
+        # step size is rf/72
+        phases = [
+            phase_from_voltages(v1, v2, 2.0014) for (_, v1), (_, v2) in zip(log[2::2], log[3::2])
+        ]
+        steps = np.abs(np.diff(np.unwrap(np.asarray(phases) * 2 * np.pi / 2.0014)))
+        assert np.max(steps[:-1]) == pytest.approx(2 * np.pi / 72)
+        with pytest.raises(LimitError):
+            device.set(25.0)
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"phase_reference": "both"},
+        {"steps_per_rf_period": 0},
+        {"steps_per_rf_period": 2.5},
+        {"rf_period": -2.0},
+        {"cavity_period": 0},
+        {"vmax": float("nan")},
+        {"step_sleep": -0.1},
+        {"limits": [1.0]},
+    ],
+)
+def test_invalid_config_is_rejected(kwargs):
+    with pytest.raises(ValueError):
+        with patched_device(LaserDelay, name="laser_delay", **kwargs):
+            pass
+
+
+def test_runtime_constant_update_is_validated(calibrated):
+    """deviceConfig updates at runtime (BEC calls .set on the signal) are validated."""
+    calibrated.step_sleep.set(0.2).wait(timeout=2)
+    assert calibrated._constants().step_sleep == 0.2
+    with pytest.raises(ValueError):
+        calibrated.step_sleep.set(-1).wait(timeout=2)
+    with pytest.raises(ValueError):
+        calibrated.phase_reference.set("sometimes").wait(timeout=2)
+    assert calibrated._constants().phase_reference == "stored"
+
+
+def test_calibration_signals_are_internal(calibrated):
+    """Calibration can only change through calibrate()/set_calibration()/moves."""
+    with pytest.raises(InternalSignalError):
+        calibrated.offset_t.put(1.0)
+    with pytest.raises(InternalSignalError):
+        calibrated.readback.put(1.0)
+
+
+def test_yaml_entries_map_onto_constructor_arguments():
+    """Every deviceConfig key of both YAML files is a constructor argument (no leftovers), and the
+    standard config pulls laser_delay in through the experimental hutch include."""
+    standard = yaml_load(str(CONFIG_DIR / "x10da_standard_config.yaml"))
+    hutch = yaml_load(str(CONFIG_DIR / "x10da_experimental_hutch.yaml"))
+    sim = yaml_load(str(CONFIG_DIR / "laser_delay_sim.yaml"))
+    assert standard["laser_delay"] == hutch["laser_delay"]
+    entry = standard["laser_delay"]
+    assert entry["deviceClass"] == "superxas_bec.devices.laser_delay.LaserDelay"
+    assert entry["onFailure"] == "buffer"
+    assert sim["laser_delay"]["deviceClass"] == "superxas_bec.devices.laser_delay.SimLaserDelay"
+    for config in (entry, sim["laser_delay"]):
+        with mock.patch.object(ophyd, "cl") as mock_cl:
+            mock_cl.get_pv = MockPV
+            mock_cl.thread_class = threading.Thread
+            obj, leftover = DeviceManagerDS.construct_device_obj(
+                {"name": "laser_delay", **config}, device_manager=None
+            )
+        assert leftover == {}
+        assert isinstance(obj, LaserDelay)
+        obj.destroy()
+
+
+# --------------------------------------------------------------------------------------------- #
+#  Calibration                                                                                  #
+# --------------------------------------------------------------------------------------------- #
+
+
+def test_refuses_to_move_before_calibration(laser_delay):
+    log = record_puts(laser_delay)
+    with pytest.raises(LaserDelayNotCalibratedError, match="calibrate"):
+        laser_delay.set(1.0)
+    with pytest.raises(LaserDelayNotCalibratedError):
+        laser_delay.plan_move(1.0)
+    assert log == []
+    assert not laser_delay.moving
+
+
+def test_calibrate_stores_config_signals_and_sets_readback(laser_delay):
+    result = laser_delay.calibrate(SCOPE)
+    delay_card_t = 2.0 * 150 + 0.01 * 37
+    phase_t = 2.0 * math.atan2(-0.8, 0.6) / (2 * math.pi)
+    assert result["laser_period"] == pytest.approx(2000.2)
+    assert result["offset_t"] == pytest.approx(-delay_card_t - SCOPE)
+    assert result["phase_t"] == pytest.approx(phase_t)
+    assert result["offset_phase_t"] == pytest.approx(-phase_t - SCOPE % 12.01)
+    config = laser_delay.read_configuration()
+    for key, value in result.items():
+        assert config[f"laser_delay_{key}"]["value"] == value
+    assert laser_delay.get_calibration() == result
+    reading = laser_delay.read()
+    assert reading["laser_delay"]["value"] == SCOPE
+    assert reading["laser_delay_phase_mismatch"]["value"] == pytest.approx(0.0, abs=1e-12)
+    assert laser_delay.position == SCOPE
+
+
+def test_calibrate_rejects_bad_input(laser_delay):
+    with pytest.raises(ValueError):
+        laser_delay.calibrate(float("nan"))
+    laser_delay.evr_period._read_pv.mock_data = 0.0
+    with pytest.raises(ValueError, match="laser period"):
+        laser_delay.calibrate(SCOPE)
+    assert math.isnan(laser_delay.get_calibration()["offset_t"])
+
+
+def test_calibration_publishes_readback_subscription(laser_delay):
+    """BEC listens to the 'readback' event; calibrate() and moves must fire it."""
+    values = []
+    laser_delay.subscribe(lambda value, **_: values.append(value), event_type="readback", run=False)
+    laser_delay.calibrate(SCOPE)
+    laser_delay.set(1.0).wait(timeout=10)
+    assert values == [SCOPE, 1.0]
+
+
+def test_set_calibration_restores_previous_state(laser_delay):
+    laser_delay.calibrate(SCOPE)
+    laser_delay.set(2.5).wait(timeout=10)
+    saved = laser_delay.get_calibration()
+    v1, v2 = laser_delay.phase_v1.get(), laser_delay.phase_v2.get()
+
+    # BEC restart: a fresh device, hardware untouched
+    with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as fresh:
+        set_hardware(
+            fresh,
+            coarse=laser_delay.delay_coarse.get(),
+            fine=laser_delay.delay_fine.get(),
+            v1=v1,
+            v2=v2,
+        )
+        with pytest.raises(LaserDelayNotCalibratedError):
+            fresh.set(1.0)
+        # Timing.offset order: laser_period, offset_t, offset_phase_t, phase_t
+        fresh.set_calibration(*saved.values(), current_delay=2.5)
+        assert fresh.get_calibration() == saved
+        assert fresh.read()["laser_delay"]["value"] == 2.5
+        log_fresh = record_puts(fresh)
+        fresh.set(-1.0).wait(timeout=10)
+    log = record_puts(laser_delay)
+    laser_delay.set(-1.0).wait(timeout=10)
+    assert log_fresh == log
+
+
+@pytest.mark.parametrize(
+    "values", [(0.0, 1.0, 1.0, 1.0), (-5.0, 1.0, 1.0, 1.0), (2000.0, float("nan"), 1.0, 1.0)]
+)
+def test_set_calibration_validates(laser_delay, values):
+    with pytest.raises(ValueError):
+        laser_delay.set_calibration(*values)
+
+
+# --------------------------------------------------------------------------------------------- #
+#  Moves                                                                                        #
+# --------------------------------------------------------------------------------------------- #
+
+
+def test_move_sets_delay_card_and_ramps_phase(calibrated):
+    log = record_puts(calibrated)
+    plan = calibrated.plan_move(-4.9)
+    status = calibrated.set(-4.9)
+    status.wait(timeout=10)
+    assert status.success
+    assert log[0] == ("X10DA-ES2-TIM:SET1-DELAY", plan["delay_coarse"])
+    assert log[1] == ("X10DA-ES2-TIM:SET1-FINE", plan["delay_fine"])
+    assert len(log) == 2 + 2 * plan["phase_steps"]
+    v1, v2 = voltages_for_phase(plan["phase_start"] + plan["phase_delta"], 2.0, 1.0)
+    assert log[-2][1] == v1 and log[-1][1] == v2
+    assert calibrated.phase_t.get() == plan["phase_target"]
+    reading = calibrated.read()
+    assert reading["laser_delay"]["value"] == -4.9
+    assert reading["laser_delay_delay_coarse"]["value"] == plan["delay_coarse"]
+    assert reading["laser_delay_phase_mismatch"]["value"] == pytest.approx(0.0, abs=1e-9)
+    assert calibrated.motor_is_moving.get() == 0
+    assert not calibrated.moving
+
+
+def test_move_does_not_block_and_reports_moving(calibrated):
+    calibrated.step_sleep.put(0.01)
+    moving = []
+    calibrated.motor_is_moving.subscribe(lambda value, **_: moving.append(value), run=False)
+    start = time.monotonic()
+    status = calibrated.set(-4.9)  # 290 steps, ~3 s
+    assert time.monotonic() - start < 0.5
+    assert not status.done
+    assert calibrated.moving
+    assert calibrated.motor_is_moving.get() == 1
+    with pytest.raises(LaserDelayBusyError):
+        calibrated.set(0.0)
+    with pytest.raises(LaserDelayBusyError):
+        calibrated.calibrate(SCOPE)
+    calibrated.stop()
+    with pytest.raises(DeviceStoppedError):
+        status.wait(timeout=2)
+    assert wait_until(lambda: calibrated.motor_is_moving.get() == 0)
+    assert moving[:1] == [1] and moving[-1] == 0
+
+
+def test_stop_mid_ramp_keeps_last_step_and_fails_status(calibrated):
+    calibrated.step_sleep.put(0.05)
+    log = record_puts(calibrated)
+    old_phase_t = calibrated.phase_t.get()
+    plan = calibrated.plan_move(-4.9)
+    status = calibrated.set(-4.9)
+    assert wait_until(lambda: len(log) >= 2 + 2 * 5)
+    start = time.monotonic()
+    calibrated.stop()
+    with pytest.raises(DeviceStoppedError):
+        status.wait(timeout=1)
+    assert time.monotonic() - start < 0.5
+    assert wait_until(lambda: not calibrated._hw_lock.locked())
+    n_puts = len(log)
+    time.sleep(0.2)
+    assert len(log) == n_puts, "puts continued after stop"
+    assert n_puts % 2 == 0, "a step was left half written"
+    steps_written = (n_puts - 2) // 2
+    assert 5 <= steps_written < plan["phase_steps"]
+    # voltages stay at the last written step
+    assert calibrated.phase_v1.get() == log[-2][1]
+    assert calibrated.phase_v2.get() == log[-1][1]
+    # bookkeeping: stored phase advanced by what was actually ramped
+    last_offset = phase_ramp_offsets(plan["phase_delta"], 2.0, 36)[steps_written - 1]
+    assert calibrated.phase_t.get() == pytest.approx(old_phase_t + last_offset)
+    assert calibrated.read()["laser_delay_phase_mismatch"]["value"] == pytest.approx(0, abs=1e-9)
+    assert math.isnan(calibrated.read()["laser_delay"]["value"])
+    assert calibrated.motor_is_moving.get() == 0
+
+    # the next move works and ends where an uninterrupted move would have ended
+    calibrated.step_sleep.put(0.0)
+    calibrated.set(-4.9).wait(timeout=10)
+    assert calibrated.phase_t.get() == plan["phase_target"]
+    v1, v2 = voltages_for_phase(plan["phase_target"], 2.0, 1.0)
+    assert calibrated.phase_v1.get() == pytest.approx(v1, abs=1e-12)
+    assert calibrated.phase_v2.get() == pytest.approx(v2, abs=1e-12)
+    assert calibrated.read()["laser_delay"]["value"] == -4.9
+
+
+def test_new_move_right_after_stop_does_not_revive_the_old_ramp(calibrated):
+    """The scan server may send the next move as soon as the stopped status fails. The old ramp must
+    not continue, and the new move must start only after the old task has let go of the PVs."""
+    calibrated.step_sleep.put(0.02)
+    log = record_puts(calibrated)
+    old = calibrated.set(-4.9)
+    assert wait_until(lambda: len(log) >= 2 + 2 * 5)
+    calibrated.stop()
+    with pytest.raises(DeviceStoppedError):
+        old.wait(timeout=1)
+    calibrated.step_sleep.put(0.0)
+    new = calibrated.set(0.0)
+    new.wait(timeout=10)
+    assert new.success
+    delay_puts = [i for i, (pv, _) in enumerate(log) if pv == "X10DA-ES2-TIM:SET1-DELAY"]
+    assert len(delay_puts) == 2
+    assert delay_puts[1] % 2 == 0  # the old move ended on a complete V1/V2 pair
+    expected_phase = (-0.0) % 12.01 - calibrated.offset_phase_t.get()
+    assert calibrated.phase_t.get() == expected_phase
+    assert calibrated.read()["laser_delay_phase_mismatch"]["value"] == pytest.approx(0, abs=1e-9)
+    assert calibrated.read()["laser_delay"]["value"] == 0.0
+
+
+def test_stop_is_idempotent_without_move(calibrated):
+    calibrated.stop()
+    calibrated.stop()
+    calibrated.set(0.5).wait(timeout=10)
+    assert calibrated.read()["laser_delay"]["value"] == 0.5
+
+
+def test_status_timeout_also_stops_the_ramp(calibrated):
+    calibrated.step_sleep.put(0.05)
+    log = record_puts(calibrated)
+    status = calibrated.move(-4.9, timeout=0.3)
+    with pytest.raises(Exception):
+        status.wait(timeout=2)
+    assert wait_until(lambda: not calibrated._hw_lock.locked())
+    n_puts = len(log)
+    time.sleep(0.2)
+    assert len(log) == n_puts < 2 + 2 * 290
+
+
+def test_destroy_ends_a_running_move():
+    with patched_device(LaserDelay, name="laser_delay", step_sleep=0.05) as device:
+        set_hardware(device)
+        device.calibrate(SCOPE)
+        status = device.set(-4.9)
+        device.destroy()
+        assert wait_until(lambda: status.done)
+        assert not status.success
+
+
+def test_stored_vs_readback_reference_when_voltages_were_touched(calibrated):
+    calibrated.set(1.0).wait(timeout=10)
+    # somebody turns the phase shifter by hand
+    calibrated.phase_v1._read_pv.mock_data = 0.0
+    calibrated.phase_v2._read_pv.mock_data = 1.0
+    assert calibrated.read()["laser_delay_phase_mismatch"]["value"] != pytest.approx(0, abs=1e-3)
+    plan_stored = calibrated.plan_move(2.0)
+    calibrated.set(2.0).wait(timeout=10)
+    end_stored = calibrated.read_phase()
+    # stored: relative move from the stored phase, starting at the read-back -> offset persists
+    assert wrap_phase(end_stored - (0.5 + plan_stored["phase_delta"]), 2.0) == pytest.approx(
+        0, abs=1e-9
+    )
+
+    calibrated.phase_v1._read_pv.mock_data = 0.0
+    calibrated.phase_v2._read_pv.mock_data = 1.0
+    calibrated.phase_reference.put("readback")
+    plan_readback = calibrated.plan_move(3.0)
+    calibrated.set(3.0).wait(timeout=10)
+    # readback: lands on the target phase whatever the voltages were
+    assert plan_readback["phase_start"] == pytest.approx(0.5)
+    assert wrap_phase(
+        calibrated.read_phase() - plan_readback["phase_target"], 2.0
+    ) == pytest.approx(0, abs=1e-9)
+    assert calibrated.phase_t.get() == plan_readback["phase_target"]
+
+
+def test_plan_move_writes_nothing(calibrated):
+    log = record_puts(calibrated)
+    plan = calibrated.plan_move(5.3)
+    assert log == []
+    assert plan["estimated_duration_s"] == 0.0  # step_sleep 0 in the fixture
+    assert plan["phase_steps"] == len(phase_ramp_offsets(plan["phase_delta"], 2.0, 36)) + 1
+
+
+def test_first_move_after_calibration_can_ramp_two_cavity_periods(laser_delay):
+    """Documented in LASER_DELAY.md: dP = (-t mod cav) + (s mod cav) right after calibration."""
+    laser_delay.calibrate(SCOPE)
+    plan = laser_delay.plan_move(-4.9)
+    assert plan["phase_delta"] == pytest.approx(4.9 + SCOPE % 12.01)
+    assert plan["phase_delta"] > 12.01
+    laser_delay.set(-4.9).wait(timeout=10)
+    # later moves stay within one cavity period
+    for target in (0.0, 5.3, -4.9, 11.9, 12.2):
+        assert abs(laser_delay.plan_move(target)["phase_delta"]) < 12.01
+        laser_delay.set(target).wait(timeout=10)
+
+
+@pytest.mark.parametrize("value", [float("nan"), float("inf"), "abc"])
+def test_rejects_non_finite_targets(calibrated, value):
+    with pytest.raises(ValueError):
+        calibrated.set(value)
+
+
+# --------------------------------------------------------------------------------------------- #
+#  Simulation                                                                                   #
+# --------------------------------------------------------------------------------------------- #
+
+
+def test_sim_device_moves_without_epics():
+    device = SimLaserDelay(name="laser_delay", step_sleep=0.0)
+    try:
+        device.wait_for_connection()
+        assert not any(hasattr(sig.item, "pvname") for sig in device.walk_signals())
+        result = device.calibrate(-0.88)
+        assert result["laser_period"] == pytest.approx(2000.2)
+        assert result["offset_t"] == pytest.approx(-300.37 + 0.88)
+        device.set(0.5).wait(timeout=30)
+        reading = device.read()
+        assert reading["laser_delay"]["value"] == 0.5
+        assert reading["laser_delay_phase_mismatch"]["value"] == pytest.approx(0, abs=1e-9)
+        coarse, fine = (
+            reading["laser_delay_delay_coarse"]["value"],
+            reading["laser_delay_delay_fine"]["value"],
+        )
+        assert 2 * coarse + 0.01 * fine == pytest.approx(
+            (-result["offset_t"] - 0.5) % 2000.2, abs=0.011
+        )
+    finally:
+        device.destroy()