"""Unit tests for the LaserDelay positioner (mocked EPICS).""" # pylint: disable=redefined-outer-name,protected-access from __future__ import annotations import functools 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 import messages from bec_lib.bec_yaml_loader import yaml_load from bec_lib.devicemanager import DeviceConfigError from bec_server.device_server.devices.config_update_handler import ConfigUpdateHandler 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 ConnectionTimeoutError, InternalSignalError from ophyd.utils import LimitError, ReadOnlyError from ophyd_devices.interfaces.base_classes.psi_device_base import DeviceStoppedError from ophyd_devices.interfaces.protocols.bec_protocols import ( BECDeviceProtocol, BECPositionerProtocol, ) from ophyd_devices.tests.utils import MockPV, patched_device from superxas_bec.devices.laser_delay import ( PHASE_VOLTAGE_LIMIT, LaserDelay, LaserDelayBusyError, LaserDelayHardwareError, LaserDelayNotCalibratedError, LaserDelayPersistenceError, delay_card_setting, fold_phase_delta, phase_from_voltages, phase_ramp_offsets, voltages_for_phase, wrap_phase, ) 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 @pytest.fixture(autouse=True) def fast_ramp(monkeypatch): """Tests ramp without sleeping; the beamline minimum is tested on its own.""" assert LaserDelay.MIN_STEP_SLEEP == 0.1 # the value the scientist tested monkeypatch.setattr(LaserDelay, "MIN_STEP_SLEEP", 0.0) 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_path"]["value"] == "script" 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", "get_calibration_info", "plan_move", } assert config["laser_delay_fine_rollover"]["value"] is True assert laser_delay.calibration_pv_prefix is None assert isinstance(laser_delay, BECPositionerProtocol) assert isinstance(laser_delay, BECDeviceProtocol) 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, 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:] # first move after calibrate ramps from the read-back 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", [ {"steps_per_rf_period": 35}, {"vmax": 0.0}, {"phase_path": "fast"}, {"steps_per_rf_period": 0}, {"steps_per_rf_period": 2.5}, {"rf_period": -2.0}, {"cavity_period": 0}, {"vmax": float("nan")}, {"vmax": 5.5}, {"fine_rollover": "yes"}, {"step_sleep": -0.1}, {"limits": [1.0]}, {"limits": [5.0, -5.0]}, {"limits": [-5.0, float("inf")]}, ], ) 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.steps_per_rf_period.set(18).wait(timeout=2) assert calibrated._constants().steps_per_rf_period == 36 with pytest.raises(ValueError): calibrated.phase_path.set("fastest").wait(timeout=2) calibrated.phase_path.set("shortest").wait(timeout=2) assert calibrated._constants().phase_path == "shortest" def test_runtime_limits_are_validated(calibrated): """BEC writes runtime limits (dev.laser_delay.limits = [...]) one signal at a time.""" with pytest.raises(ValueError): calibrated.low_limit_travel.set(float("nan")).wait(timeout=2) calibrated.low_limit_travel.set(5.0).wait(timeout=2) calibrated.high_limit_travel.set(-5.0).wait(timeout=2) with pytest.raises(LimitError, match="inverted"): calibrated.set(0.0) calibrated.high_limit_travel.set(10.0).wait(timeout=2) with pytest.raises(LimitError, match="outside"): calibrated.set(0.0) calibrated.set(7.0).wait(timeout=10) assert calibrated.read()["laser_delay"]["value"] == 7.0 def config_update_handler(device: LaserDelay, device_config: dict) -> ConfigUpdateHandler: """BEC's ConfigUpdateHandler with just enough device manager for runtime deviceConfig updates.""" entry = mock.MagicMock() entry.name, entry.enabled, entry.obj = device.name, True, device entry._config = {"deviceConfig": dict(device_config)} device_manager = mock.MagicMock() device_manager.devices = {device.name: entry} device_manager.update_config = functools.partial(DeviceManagerDS.update_config, device_manager) handler = ConfigUpdateHandler.__new__(ConfigUpdateHandler) handler.device_manager = device_manager return handler def test_runtime_config_update_errors_name_the_real_problem(): """A failed runtime config update rolls back cleanly and shows the real error.""" entry = yaml_load(str(CONFIG_DIR / "x10da_laser_delay.yaml"))["laser_delay"] with patched_device(LaserDelay, name="laser_delay", **entry["deviceConfig"]) as device: handler = config_update_handler(device, entry["deviceConfig"]) def update(config): msg = messages.DeviceConfigMessage( action="update", config={"laser_delay": {"deviceConfig": config}} ) handler._update_config(msg, threading.Event()) with pytest.raises(DeviceConfigError, match="phase_path must be one of"): update({"phase_path": "fastest"}) assert device.phase_path.get() == "script" with pytest.raises(DeviceConfigError, match="reloading the device config"): update({"phase_v1_pv": "X10DA-ES4:AO-03"}) assert device.phase_v1.pvname == "X10DA-ES4:AO-02" with pytest.raises(DeviceConfigError, match="finite"): update({"limits": [float("nan"), 1.0]}) update({"step_sleep": 0.05, "phase_v1_pv": "X10DA-ES4:AO-02", "limits": [-20, 20]}) assert device.step_sleep.get() == 0.05 assert device.limits == (-20.0, 20.0) 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 YAML deviceConfig key is a constructor argument.""" 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_sim.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) # NaN while the move runs: the delay is undefined until the ramp has finished assert values[0] == SCOPE and math.isnan(values[1]) and values[2:] == [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 refused = calibrated.set(0.0) # not raised: a failed status, the running move goes on assert refused.done and not refused.success with pytest.raises(LaserDelayBusyError): refused.wait(timeout=1) assert calibrated.moving and not status.done 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_move_requested_during_calibration_runs_after_it(calibrated): """A move arriving during calibrate() waits and then uses the new calibration.""" reading, release = threading.Event(), threading.Event() original_get = calibrated.delay_fine.get def slow_get(*args, **kwargs): reading.set() release.wait(timeout=5) return original_get(*args, **kwargs) set_hardware(calibrated, coarse=160, fine=12) # the hardware was moved by hand since log = record_puts(calibrated) result = {} with mock.patch.object(calibrated.delay_fine, "get", side_effect=slow_get): thread = threading.Thread(target=lambda: result.update(cal=calibrated.calibrate(2.5))) thread.start() assert reading.wait(timeout=5) status = calibrated.set(1.0) time.sleep(0.1) assert log == [], "the move wrote PVs while calibrate() was reading them" release.set() thread.join(timeout=5) status.wait(timeout=10) assert status.success cal = result["cal"] assert cal["offset_t"] == pytest.approx(-(2.0 * 160 + 0.01 * 12) - 2.5) assert (log[0][1], log[1][1]) == delay_card_setting(-cal["offset_t"] - 1.0, cal["laser_period"]) assert calibrated.read()["laser_delay"]["value"] == 1.0 def test_calibration_refused_while_a_stopped_move_finishes_its_step(calibrated): calibrated.step_sleep.put(0.05) blocker = threading.Event() original_put = calibrated.phase_v1._write_pv.put def blocking_put(value, *args, **kwargs): blocker.wait(timeout=5) # a slow CA put return original_put(value, *args, **kwargs) calibrated.phase_v1._write_pv.put = blocking_put status = calibrated.set(-4.9) assert wait_until(calibrated._hw_lock.locked) calibrated.stop() assert status.done and not status.success with pytest.raises(LaserDelayBusyError): calibrated.calibrate(SCOPE) with pytest.raises(LaserDelayBusyError): calibrated.set_calibration(2000.2, 1.0, 1.0, 1.0) blocker.set() assert wait_until(lambda: not calibrated.moving) calibrated.calibrate(SCOPE) 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): """After a stop the old ramp ends before the next move writes.""" 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_destroy_fails_the_move_status_even_if_the_task_hangs(): """destroy() releases waiters even when the move task is stuck.""" with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device: set_hardware(device) device.calibrate(SCOPE) blocker = threading.Event() original_put = device.phase_v1._write_pv.put def hung_put(value, *args, **kwargs): blocker.wait(timeout=5) return original_put(value, *args, **kwargs) device.phase_v1._write_pv.put = hung_put status = device.set(-4.9) assert wait_until(device._hw_lock.locked) with mock.patch.object(device.task_handler, "shutdown"): device.destroy() assert status.done and not status.success with pytest.raises(DeviceStoppedError, match="destroyed"): status.wait(timeout=0) blocker.set() assert wait_until(lambda: not device._hw_lock.locked()) def fail_on_put(signal, call: int, exc: Exception) -> None: """Make the ``call``-th put to ``signal`` raise ``exc`` (the IOC went away).""" pv = signal._write_pv original, count = pv.put, {"n": 0} def put(value, *args, **kwargs): count["n"] += 1 if count["n"] == call: raise exc return original(value, *args, **kwargs) pv.put = put def test_pv_lost_mid_ramp_fails_the_move_and_keeps_the_phase_bookkeeping(calibrated): """V2 fails mid-ramp: the move fails cleanly and the next move works.""" old_phase_t = calibrated.phase_t.get() plan = calibrated.plan_move(-4.9) log = record_puts(calibrated) error = ConnectionTimeoutError("X10DA-ES4:AO-01 could not connect within 1.0 s") fail_on_put(calibrated.phase_v2, 6, error) status = calibrated.set(-4.9) with pytest.raises(ConnectionTimeoutError): status.wait(timeout=5) assert wait_until(lambda: not calibrated.moving) assert calibrated.motor_is_moving.get() == 0 assert math.isnan(calibrated.readback.get()) # DELAY, FINE, 5 complete pairs, then V1 of the 6th step (V2 failed) assert [pv for pv, _ in log[:2]] == ["X10DA-ES2-TIM:SET1-DELAY", "X10DA-ES2-TIM:SET1-FINE"] assert len(log) == 2 + 2 * 5 + 1 offsets = phase_ramp_offsets(plan["phase_delta"], 2.0, 36) assert calibrated.phase_t.get() == pytest.approx(old_phase_t + offsets[4]) # the half-written pair leaves less than one ramp step of mismatch; phase_mismatch shows it assert abs(calibrated.read_phase_mismatch()) < 2.0 / 36 calibrated.set(-4.9).wait(timeout=10) assert calibrated.read()["laser_delay"]["value"] == -4.9 assert calibrated.phase_t.get() == plan["phase_target"] def test_delay_card_write_failure_leaves_the_phase_shifter_alone(calibrated): old_phase_t = calibrated.phase_t.get() log = record_puts(calibrated) fail_on_put(calibrated.delay_fine, 1, ConnectionTimeoutError("SET1-FINE not connected")) status = calibrated.set(-4.9) with pytest.raises(ConnectionTimeoutError): status.wait(timeout=5) assert [pv for pv, _ in log] == ["X10DA-ES2-TIM:SET1-DELAY"] assert calibrated.phase_t.get() == old_phase_t assert math.isnan(calibrated.readback.get()) assert calibrated.motor_is_moving.get() == 0 def test_phase_read_failure_refuses_the_move_before_any_write(calibrated): old_phase_t = calibrated.phase_t.get() log = record_puts(calibrated) with mock.patch.object( calibrated.phase_v1, "get", side_effect=ConnectionTimeoutError("AO-02 not connected") ): status = calibrated.set(-4.9) with pytest.raises(ConnectionTimeoutError): status.wait(timeout=5) assert log == [] assert calibrated.phase_t.get() == old_phase_t assert calibrated.readback.get() == pytest.approx(SCOPE) # nothing moved assert not calibrated.moving def test_move_refused_when_voltages_were_touched(calibrated): """A phase shifter turned outside BEC blocks moves until recalibration.""" calibrated.set(1.0).wait(timeout=10) 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) assert "away from the stored phase" in calibrated.plan_move(2.0)["refused"] log = record_puts(calibrated) status = calibrated.set(2.0) with pytest.raises(LaserDelayHardwareError, match="away from the stored phase"): status.wait(timeout=10) assert log == [] assert calibrated.readback.get() == 1.0 # nothing was written assert not calibrated.get_calibration_info()["verified"] with pytest.raises(LaserDelayNotCalibratedError, match="does not match the hardware"): calibrated.set(2.0) calibrated.calibrate(SCOPE) calibrated.set(2.0).wait(timeout=10) assert calibrated.readback.get() == 2.0 def test_mismatch_below_one_step_is_accepted(calibrated): """A V1 written without its V2 (IOC failure mid-pair) leaves less than one step: allowed.""" calibrated.set(1.0).wait(timeout=10) phase = calibrated.read_phase() v1, v2 = voltages_for_phase(phase + 0.04, 2.0, 1.0) calibrated.phase_v1._read_pv.mock_data = v1 calibrated.phase_v2._read_pv.mock_data = v2 calibrated.set(2.0).wait(timeout=10) assert calibrated.readback.get() == 2.0 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): """Module docstring: 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) # phase_path "shortest" @pytest.mark.parametrize( "delta, expected, wraps", [ (0.0, 0.0, 0), (6.005, 6.005, 0), # +cavity/2 is inside (-6.005, 6.005, -1), # -cavity/2 is not (-6.0, -6.0, 0), (6.006, 6.006 - 12.01, 1), (16.13, 16.13 - 12.01, 1), (11.51, 11.51 - 12.01, 1), (-24.03, -24.03 + 2 * 12.01, -2), (30.0, 30.0 - 2 * 12.01, 2), ], ) def test_fold_phase_delta(delta, expected, wraps): folded, k = fold_phase_delta(delta, 12.01) assert k == wraps and isinstance(k, int) assert folded == pytest.approx(expected, abs=1e-12) def test_fold_phase_delta_properties(): rng = np.random.default_rng(7) for delta in rng.uniform(-60, 60, 2000): folded, k = fold_phase_delta(float(delta), 12.01) assert -6.005 - 1e-9 < folded <= 6.005 + 1e-9 assert folded == pytest.approx(delta - k * 12.01, abs=1e-12) if -6.005 < delta <= 6.005: assert folded == delta and k == 0 # unchanged, bit for bit SHORTEST_TARGETS = [-5.0, 5.0, -5.0, 0.0, 0.5, 11.9, 12.2, -24.0, 37.25, -0.88, 400.0, -1500.3] def run_targets(device: LaserDelay, targets: list) -> list[dict]: """Plan and move to each target; return the plans with the final voltages and stored phase.""" results = [] for target in targets: plan = device.plan_move(target) device.set(target).wait(timeout=30) plan["v1"], plan["v2"] = device.phase_v1.get(), device.phase_v2.get() plan["stored"] = device.phase_t.get() plan["mismatch"] = device.read_phase_mismatch() plan["readback"] = device.readback.get() results.append(plan) return results @pytest.mark.parametrize("rf_period, cavity_period", [(2.0, 12.01), (2.0014, 12.0084)]) def test_shortest_path_ramps_at_most_half_a_cavity_period(rf_period, cavity_period): """Shortest path: script delay card, at most cavity/2 per move, documented wrap offset.""" kwargs = {"step_sleep": 0.0, "rf_period": rf_period, "cavity_period": cavity_period} runs = {} for path in ("script", "shortest"): with patched_device(LaserDelay, name="laser_delay", phase_path=path, **kwargs) as device: set_hardware(device) device.calibrate(SCOPE) runs[path] = run_targets(device, SHORTEST_TARGETS) max_steps = math.ceil((cavity_period / 2) / (rf_period / 36)) + 1 for script, short in zip(runs["script"], runs["shortest"]): assert (short["delay_coarse"], short["delay_fine"]) == ( script["delay_coarse"], script["delay_fine"], ) assert short["phase_path"] == "shortest" and script["cavity_wraps"] == 0 assert abs(short["phase_delta"]) <= cavity_period / 2 + 1e-9 assert short["phase_steps"] <= max_steps assert short["readback"] == short["target"] assert short["mismatch"] == pytest.approx(0, abs=1e-9) k = short["cavity_wraps"] assert short["stored"] == pytest.approx(script["stored"] - k * cavity_period, abs=1e-9) end_short = phase_from_voltages(short["v1"], short["v2"], rf_period) end_script = phase_from_voltages(script["v1"], script["v2"], rf_period) offset = wrap_phase(-k * cavity_period, rf_period) assert short["wrap_offset_ns"] == pytest.approx(offset, abs=1e-12) assert wrap_phase(end_short - end_script - offset, rf_period) == pytest.approx(0, abs=1e-9) if cavity_period == 12.0084: assert offset == pytest.approx(0, abs=1e-9) else: assert offset == pytest.approx(wrap_phase(-0.01 * k, 2.0), abs=1e-9) first = runs["shortest"][0] assert runs["script"][0]["phase_delta"] > cavity_period # script: long first move assert first["phase_delta"] == pytest.approx( runs["script"][0]["phase_delta"] - cavity_period, abs=1e-9 ) def test_shortest_path_is_the_script_path_when_no_fold_is_needed(): """Moves that stay within half a cavity period write exactly what the script path writes.""" targets = [1.5, 2.0, 3.0, 2.2, 7.0] with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device: set_hardware(device) device.calibrate(SCOPE) device.set(1.0).wait(timeout=10) state = device.get_calibration() hardware = { "coarse": device.delay_coarse.get(), "fine": device.delay_fine.get(), "v1": device.phase_v1.get(), "v2": device.phase_v2.get(), } log_script = record_puts(device) for target in targets: device.set(target).wait(timeout=10) with patched_device( LaserDelay, name="laser_delay", step_sleep=0.0, phase_path="shortest" ) as device: set_hardware(device, **hardware) device.set_calibration(**state, current_delay=1.0) log_short = record_puts(device) for target in targets: assert device.plan_move(target)["cavity_wraps"] == 0 device.set(target).wait(timeout=10) assert log_short == log_script def test_stop_during_a_shortest_path_move_keeps_the_bookkeeping(laser_delay): laser_delay.phase_path.put("shortest") laser_delay.calibrate(SCOPE) laser_delay.step_sleep.put(0.02) log = record_puts(laser_delay) plan = laser_delay.plan_move(-5.0) assert plan["cavity_wraps"] == 1 and 0 < plan["phase_delta"] < 6.005 status = laser_delay.set(-5.0) assert wait_until(lambda: len(log) >= 2 + 2 * 5) laser_delay.stop() with pytest.raises(DeviceStoppedError): status.wait(timeout=1) assert wait_until(lambda: not laser_delay.moving) assert laser_delay.read_phase_mismatch() == pytest.approx(0, abs=1e-9) laser_delay.step_sleep.put(0.0) laser_delay.set(-5.0).wait(timeout=10) end = phase_from_voltages(laser_delay.phase_v1.get(), laser_delay.phase_v2.get(), 2.0) expected = plan["phase_start"] + plan["phase_delta"] assert wrap_phase(end - expected, 2.0) == pytest.approx(0, abs=1e-9) assert laser_delay.phase_t.get() == pytest.approx(plan["phase_target"] - 12.01, abs=1e-9) @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() def test_sim_scope_follows_every_move_on_both_phase_paths(): """The simulated oscilloscope shows the commanded delay after every move.""" device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0) try: assert device.sim_scope.get() == pytest.approx(-0.88) device.calibrate(device.sim_scope.get()) rng = np.random.default_rng(7) for path in ("script", "shortest"): device.phase_path.set(path).wait(timeout=2) for target in rng.uniform(-40, 40, 25): device.set(float(target)).wait(timeout=30) assert device.sim_scope.get() == pytest.approx(target, abs=1e-6), path device.set(-5.0).wait(timeout=30) assert device.sim_response.get() == pytest.approx(0, abs=1e-6) device.set(5.0).wait(timeout=30) assert device.sim_response.get() == pytest.approx(1, abs=1e-6) finally: device.destroy() def test_sim_scope_shows_a_wrong_calibration(): device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0) try: device.calibrate(-0.5) # scope misread by 0.38 ns device.set(2.0).wait(timeout=30) assert device.sim_scope.get() == pytest.approx(2.0 - 0.38) finally: device.destroy() def test_sim_rejects_what_the_iocs_refuse(): device = SimLaserDelay(name="laser_delay", step_sleep=0.0) try: with pytest.raises(ValueError): device.sim_set_hardware(phase_v1=PHASE_VOLTAGE_LIMIT + 0.1) with pytest.raises(ValueError): device.sim_set_hardware(delay_fine=200) finally: device.destroy() # delay card fine value 200 def test_delay_card_fine_rollover(): # t1 mod 2 = 1.99985 -> the script writes FINE 200 assert delay_card_setting(301.99985, 2000.2) == (150, 200) assert delay_card_setting(301.99985, 2000.2, fine_rollover=True) == (151, 0) # the next coarse unit would be a full laser period: wrap to 0 assert delay_card_setting(1999.99985, 2000.0, fine_rollover=True) == (0, 0) rng = np.random.default_rng(3) for total in rng.uniform(-5000, 5000, 5000): script = delay_card_setting(total, 2000.2) coarse, fine = delay_card_setting(total, 2000.2, fine_rollover=True) assert 0 <= fine <= 199 assert 2 * coarse + 0.01 * fine == pytest.approx(2 * script[0] + 0.01 * script[1]) @pytest.mark.parametrize("rollover, expected", [(True, (151, 0)), (False, (150, 200))]) def test_move_writes_rolled_over_fine_value(calibrated, rollover, expected): calibrated.fine_rollover.set(rollover).wait(timeout=2) offset_t = calibrated.get_calibration()["offset_t"] target = -offset_t - 301.99985 # delay card total with FINE 200 plan = calibrated.plan_move(target) assert (plan["delay_coarse"], plan["delay_fine"]) == expected log = record_puts(calibrated) calibrated.set(target).wait(timeout=10) assert [value for _, value in log[:2]] == list(expected) # phase shifter voltage limit def test_vmax_is_capped_at_the_hard_limit(calibrated): calibrated.vmax.set(PHASE_VOLTAGE_LIMIT).wait(timeout=2) with pytest.raises(ValueError, match="hard limit"): calibrated.vmax.set(PHASE_VOLTAGE_LIMIT + 0.01).wait(timeout=2) assert calibrated.vmax.get() == PHASE_VOLTAGE_LIMIT # calibration kept across BEC restarts @pytest.fixture def caplog(): """Messages logged by the device module (bec_logger is loguru, not std logging).""" class Log: def __init__(self): self.patcher = mock.patch("superxas_bec.devices.laser_delay.logger") @property def text(self): return "\n".join(str(c.args[0]) for c in self.mock.method_calls if c.args) log = Log() log.mock = log.patcher.start() yield log 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 store(): """Calibration store standing in for the calibration PVs (same interface).""" return MemoryCalibrationStore("laser_delay", registry={}) 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, **(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(), "fine": device.delay_fine.get(), "v1": device.phase_v1.get(), "v2": device.phase_v2.get(), } def test_calibration_is_saved_after_calibrate_and_every_move(store): context, device = make_device(store) try: device.calibrate(SCOPE) 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 = 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 assert saved["constants"] == {"rf_period": 2.0, "cavity_period": 12.01, "vmax": 1.0} 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(store) assert info["verified"] is True finally: context.__exit__(None, None, None) 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) before, hardware = device.get_calibration(), hardware_of(device) plan_before = device.plan_move(-2.0) finally: context.__exit__(None, None, None) context, device = make_device(store, **hardware) try: assert math.isnan(device.get_calibration()["phase_t"]) device.on_connected() assert device.get_calibration() == before assert device.read()["laser_delay"]["value"] == 1.5 assert device.plan_move(-2.0) == plan_before finally: context.__exit__(None, None, None) 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) before, hardware = device.get_calibration(), hardware_of(device) finally: context.__exit__(None, None, None) hardware["v1"], hardware["v2"] = hardware["v2"], -hardware["v1"] # phase turned by hand context, device = make_device(store, **hardware) try: device.on_connected() assert device.get_calibration() == before assert math.isnan(device.read()["laser_delay"]["value"]) assert "NOT VERIFIED" in caplog.text info = device.get_calibration_info() assert ( not info["verified"] and "away from the stored phase" in info["unverified_reasons"][0] ) with pytest.raises(LaserDelayNotCalibratedError, match="does not match the hardware"): device.set(1.0) finally: context.__exit__(None, None, None) 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(store) try: device.calibrate(SCOPE) with mock.patch.object(device, "_write_phase", side_effect=SystemExit("crash")): status = device.set(3.0) with pytest.raises(Exception): status.wait(timeout=10) # simulate the crash: the end-of-move save never happened content = store.content content["move_in_progress"] = True store.content = content hardware = hardware_of(device) finally: context.__exit__(None, None, None) context, device = make_device(store, **hardware) try: device.on_connected() assert "stopped during a move" in device.get_calibration_info()["unverified_reasons"][0] with pytest.raises(LaserDelayNotCalibratedError): device.set(1.0) finally: context.__exit__(None, None, None) 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(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 store.content["move_in_progress"] is False finally: context.__exit__(None, None, None) 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(store, evr=1.0, **hardware) try: device.on_connected() assert "laser period changed" in device.get_calibration_info()["unverified_reasons"][0] finally: context.__exit__(None, None, None) 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(store) try: device.calibrate(SCOPE) device.set(1.5).wait(timeout=10) hardware = hardware_of(device) finally: context.__exit__(None, None, None) hardware["fine"] = (hardware["fine"] + 50) % 200 context, device = make_device(store, **hardware) try: device.on_connected() info = device.get_calibration_info() assert not info["verified"] and "delay card reads" in info["unverified_reasons"][0] assert math.isnan(device.read()["laser_delay"]["value"]) with pytest.raises(LaserDelayNotCalibratedError, match="does not match the hardware"): device.set(1.0) device.calibrate(SCOPE) device.set(1.0).wait(timeout=10) assert device.readback.get() == 1.0 finally: context.__exit__(None, None, None) 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(store) try: device.calibrate(SCOPE) device.set(5.0).wait(timeout=10) 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) assert (hardware["v1"], hardware["v2"]) == pytest.approx((at_5["v1"], at_5["v2"]), abs=1e-9) assert (hardware["coarse"], hardware["fine"]) != (at_5["coarse"], at_5["fine"]) finally: context.__exit__(None, None, None) 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 info = device.get_calibration_info() assert not info["verified"] and "delay card reads" in info["unverified_reasons"][0] with pytest.raises(LaserDelayNotCalibratedError): device.set(4.0) finally: context.__exit__(None, None, None) 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, device = make_device(store, {"cavity_period": 12.0084}, **hardware) try: device.on_connected() info = device.get_calibration_info() assert not info["verified"] and "cavity_period was 12.01" in info["unverified_reasons"][0] finally: context.__exit__(None, None, None) 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 = store.content content["calibrated_at"] = "2026-01-01T00:00:00+00:00" store.content = content context, device = make_device(store) try: device.on_connected() assert device.read()["laser_delay"]["value"] == pytest.approx(SCOPE) assert "days old" in caplog.text finally: context.__exit__(None, None, None) 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 with pytest.raises(LaserDelayNotCalibratedError): device.set(1.0) finally: context.__exit__(None, None, None) 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 def test_unwritable_calibration_store_refuses_the_move(caplog): """Without the move marker a later restore could trust a stale phase: fail safe.""" context, device = make_device(_FailingStore(fail_save=True)) try: with pytest.raises(LaserDelayPersistenceError, match="will be lost at the next restart"): device.calibrate(SCOPE) assert device.get_calibration_info()["verified"] # applied anyway log = record_puts(device) status = device.set(1.0) with pytest.raises(LaserDelayPersistenceError, match="move marker could not be saved"): status.wait(timeout=10) assert log == [] assert device.readback.get() == pytest.approx(SCOPE) # nothing moved assert device.motor_is_moving.get() == 0 and not device.moving assert "could not save" in caplog.text finally: context.__exit__(None, None, None) 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()) for target in (3.3, -7.1, 14.2): device.set(target).wait(timeout=30) finally: device.destroy() device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0) try: device.on_connected() assert device.read()["laser_delay"]["value"] == 14.2 assert device.sim_scope.get() == pytest.approx(14.2) device.set(-2.0).wait(timeout=30) assert device.sim_scope.get() == pytest.approx(-2.0) finally: device.destroy() # safety checks def test_hardware_signals_refuse_direct_writes(calibrated): """dev.laser_delay.phase_v1.set(...) from a client would bypass the ramp and the bookkeeping.""" log = record_puts(calibrated) for signal, value in ( (calibrated.phase_v1, 0.5), (calibrated.phase_v2, 7.5), (calibrated.delay_coarse, 10), (calibrated.delay_fine, 10), ): with pytest.raises(ReadOnlyError, match="written only by laser_delay"): signal.put(value) with pytest.raises(ReadOnlyError): signal.set(value).wait(timeout=2) assert log == [] def test_hardware_writes_are_range_checked_even_for_the_device(calibrated): with calibrated._hardware_writes(): with pytest.raises(ValueError, match="outside"): calibrated.phase_v1.put(PHASE_VOLTAGE_LIMIT + 0.01) with pytest.raises(ValueError, match="outside"): calibrated.delay_fine.put(201) def test_direct_write_from_another_thread_during_a_move_is_refused(laser_delay): """The permission is per thread: a client write while the move task writes is still refused.""" laser_delay.step_sleep.put(0.02) laser_delay.calibrate(SCOPE) status = laser_delay.set(0.5) # crosses t = 0: a long ramp assert wait_until(lambda: laser_delay.phase_v1._write_pv.mock_data != 0.6) with pytest.raises(ReadOnlyError): laser_delay.phase_v1.put(0.0) laser_delay.stop() with pytest.raises(DeviceStoppedError): status.wait(timeout=5) def test_beamline_minimum_step_sleep(monkeypatch): monkeypatch.setattr(LaserDelay, "MIN_STEP_SLEEP", 0.1) # the real value, see fast_ramp with pytest.raises(ValueError, match="at least 0.1"): with patched_device(LaserDelay, name="laser_delay", step_sleep=0.05): pass with patched_device(LaserDelay, name="laser_delay") as device: assert device.step_sleep.get() == 0.1 with pytest.raises(ValueError, match="at least 0.1"): device.step_sleep.set(0.05).wait(timeout=2) device.step_sleep.set(0.15).wait(timeout=2) assert SimLaserDelay.MIN_STEP_SLEEP == 0.0 def test_busy_refusal_does_not_stop_the_running_move(laser_delay): """A move while another runs returns a failed status and does not stop the ramp.""" laser_delay.step_sleep.put(0.02) laser_delay.calibrate(SCOPE) log = record_puts(laser_delay) first = laser_delay.set(0.5) # crosses t = 0: a long ramp assert wait_until(lambda: len(log) >= 2 + 2 * 3) refused = laser_delay.set(1.0) assert refused.done and not refused.success assert isinstance(refused.exception(), LaserDelayBusyError) assert not refused._call_stop_on_failure # pylint: disable=protected-access n_puts = len(log) time.sleep(0.2) assert len(log) > n_puts, "the running move was stopped by the refusal" assert not first.done and laser_delay.moving laser_delay.step_sleep.put(0.0) first.wait(timeout=30) assert laser_delay.readback.get() == 0.5 def test_move_refused_when_the_amplitude_is_wrong(calibrated): """ES4 IOC reboot (V1 = V2 = 0): the first ramp step would jump to vmax.""" calibrated.phase_v1._read_pv.mock_data = 0.0 calibrated.phase_v2._read_pv.mock_data = 0.0 log = record_puts(calibrated) status = calibrated.set(1.0) with pytest.raises(LaserDelayHardwareError, match="amplitude"): status.wait(timeout=5) assert log == [] with pytest.raises(LaserDelayHardwareError, match="amplitude"): calibrated.calibrate(SCOPE) def test_raising_vmax_at_runtime_does_not_jump_the_voltages(calibrated): calibrated.vmax.set(5.0).wait(timeout=2) log = record_puts(calibrated) status = calibrated.set(1.0) with pytest.raises(LaserDelayHardwareError, match="amplitude"): status.wait(timeout=5) assert log == [] def test_readback_is_nan_while_moving(laser_delay): laser_delay.step_sleep.put(0.02) laser_delay.calibrate(SCOPE) status = laser_delay.set(0.5) # crosses t = 0: about 200 steps of 0.02 s assert wait_until(lambda: math.isnan(laser_delay.read()["laser_delay"]["value"])) assert laser_delay.moving status.wait(timeout=30) assert laser_delay.read()["laser_delay"]["value"] == 0.5 def test_clamped_delay_card_fails_the_move_before_the_phase_shifter(calibrated): log = record_puts(calibrated) original = calibrated.delay_fine.get def clamped(*args, **kwargs): value = original(*args, **kwargs) return min(value, 10) if kwargs.get("use_monitor") is False else value with mock.patch.object(calibrated.delay_fine, "get", side_effect=clamped): status = calibrated.set(-0.5) # FINE above 10 with pytest.raises(LaserDelayHardwareError, match="delay card reads"): status.wait(timeout=5) assert [pv for pv, _ in log] == ["X10DA-ES2-TIM:SET1-DELAY", "X10DA-ES2-TIM:SET1-FINE"] assert math.isnan(calibrated.readback.get()) assert calibrated.get_calibration_info()["verified"] # phase untouched: next move is fine def test_clamped_phase_voltage_fails_the_move_and_refuses_the_next(calibrated): original = calibrated.phase_v2.get def clamped(*args, **kwargs): value = original(*args, **kwargs) return value + 0.2 if kwargs.get("use_monitor") is False else value with mock.patch.object(calibrated.phase_v2, "get", side_effect=clamped): status = calibrated.set(1.0) with pytest.raises(LaserDelayHardwareError, match="phase shifter reads"): status.wait(timeout=5) 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