From 299d01d889ca5a46e1f6cc750beac83e294e71b1 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Sat, 3 Oct 2026 17:01:29 +0200 Subject: [PATCH] test(devices): add LaserDelay unit tests for calibration, stop and config --- tests/tests_devices/test_laser_delay.py | 530 ++++++++++++++++++++++++ 1 file changed, 530 insertions(+) create mode 100644 tests/tests_devices/test_laser_delay.py diff --git a/tests/tests_devices/test_laser_delay.py b/tests/tests_devices/test_laser_delay.py new file mode 100644 index 0000000..d0f3411 --- /dev/null +++ b/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()