1488 lines
62 KiB
Python
1488 lines
62 KiB
Python
"""Unit tests for the LaserDelay positioner (mocked EPICS)."""
|
|
|
|
# pylint: disable=redefined-outer-name,protected-access
|
|
|
|
from __future__ import annotations
|
|
|
|
import functools
|
|
import json
|
|
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,
|
|
SimLaserDelay,
|
|
delay_card_setting,
|
|
fold_phase_delta,
|
|
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
|
|
|
|
|
|
@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_file 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: the stored phase is the read-back, so 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",
|
|
[
|
|
{"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 deviceConfig update makes BEC re-apply the whole previous deviceConfig,
|
|
including the PV names. That rollback must succeed so the user sees the actual 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 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)
|
|
# 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):
|
|
"""BEC's device server runs instructions on a thread pool, so a set can arrive while calibrate()
|
|
reads the PVs. The move must not write while the calibration reads, and must then use 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):
|
|
"""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_destroy_fails_the_move_status_even_if_the_task_hangs():
|
|
"""destroy() (config reload, device server shutdown) must release whoever waits on the move,
|
|
also when the move task is stuck in a slow CA put and the thread kill does not reach it."""
|
|
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 stops answering in the middle of the ramp: the move fails with the EPICS error, the device
|
|
is idle again, the readback is NaN, the stored phase follows the last complete V1/V2 pair and
|
|
the next move (IOC back) 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):
|
|
"""Somebody turned the phase shifter outside BEC: the next move is refused before any write,
|
|
and every later move too, until a new calibration."""
|
|
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):
|
|
"""Same delay card as the script path, at most cavity/2 of phase per move (also the first move
|
|
after calibration), consistent bookkeeping, and the phase shifter ends where the script path
|
|
puts it, up to the documented wrap offset (0 when the cavity is a whole number of RF periods).
|
|
"""
|
|
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
|
|
def calibration_file(tmp_path):
|
|
return tmp_path / "sub" / "laser_delay_calibration.json"
|
|
|
|
|
|
def make_device(calibration_file, **hardware):
|
|
"""A fresh device (as after a BEC restart) on the given mocked hardware."""
|
|
context = patched_device(
|
|
LaserDelay, name="laser_delay", step_sleep=0.0, calibration_file=str(calibration_file)
|
|
)
|
|
device = context.__enter__()
|
|
set_hardware(device, **hardware)
|
|
return context, device
|
|
|
|
|
|
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(calibration_file):
|
|
context, device = make_device(calibration_file)
|
|
try:
|
|
device.calibrate(SCOPE)
|
|
saved = json.loads(calibration_file.read_text())
|
|
assert saved["calibration"] == device.get_calibration()
|
|
assert saved["current_delay"] == pytest.approx(SCOPE)
|
|
assert saved["scope_xray_minus_laser"] == pytest.approx(SCOPE)
|
|
device.set(1.5).wait(timeout=10)
|
|
saved = json.loads(calibration_file.read_text())
|
|
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(calibration_file)
|
|
assert info["verified"] is True
|
|
finally:
|
|
context.__exit__(None, None, None)
|
|
|
|
|
|
def test_calibration_is_restored_after_a_restart(calibration_file):
|
|
context, device = make_device(calibration_file)
|
|
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(calibration_file, **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(calibration_file, caplog):
|
|
context, device = make_device(calibration_file)
|
|
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(calibration_file, **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(calibration_file):
|
|
"""The marker written before the first hardware write survives a crashed device server."""
|
|
context, device = make_device(calibration_file)
|
|
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 = json.loads(calibration_file.read_text())
|
|
content["move_in_progress"] = True
|
|
calibration_file.write_text(json.dumps(content))
|
|
hardware = hardware_of(device)
|
|
finally:
|
|
context.__exit__(None, None, None)
|
|
context, device = make_device(calibration_file, **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(calibration_file):
|
|
context, device = make_device(calibration_file)
|
|
try:
|
|
device.calibrate(SCOPE)
|
|
seen = []
|
|
original = device.delay_coarse._write_pv.put
|
|
|
|
def check_marker(value, *args, **kwargs):
|
|
seen.append(json.loads(calibration_file.read_text())["move_in_progress"])
|
|
return original(value, *args, **kwargs)
|
|
|
|
device.delay_coarse._write_pv.put = check_marker
|
|
device.set(2.0).wait(timeout=10)
|
|
assert seen == [True]
|
|
assert json.loads(calibration_file.read_text())["move_in_progress"] is False
|
|
finally:
|
|
context.__exit__(None, None, None)
|
|
|
|
|
|
def test_restore_with_changed_laser_period_refuses_moves(calibration_file):
|
|
context, device = make_device(calibration_file)
|
|
try:
|
|
device.calibrate(SCOPE)
|
|
hardware = hardware_of(device)
|
|
finally:
|
|
context.__exit__(None, None, None)
|
|
context, device = make_device(calibration_file, 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(calibration_file):
|
|
"""A delay card that no longer shows the saved delay means the delay was changed outside BEC
|
|
(script, caput). The phase part of such a change can be whole RF periods, invisible to the phase
|
|
check, so the calibration is not verified until the next calibrate."""
|
|
context, device = make_device(calibration_file)
|
|
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(calibration_file, **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(calibration_file):
|
|
"""While BEC was down the script moved the delay by 2.0 ns (one RF period): the voltages are
|
|
exactly where the saved phase expects them, only the delay card tells. Without the card check the
|
|
next move would be 2.0 ns off while the readback claims the target."""
|
|
context, device = make_device(calibration_file)
|
|
try:
|
|
device.calibrate(SCOPE)
|
|
device.set(5.0).wait(timeout=10)
|
|
saved = calibration_file.read_text() # 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)
|
|
calibration_file.write_text(saved) # BEC never saw the script's move
|
|
context, device = make_device(calibration_file, **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(calibration_file):
|
|
context, device = make_device(calibration_file)
|
|
try:
|
|
device.calibrate(SCOPE)
|
|
hardware = hardware_of(device)
|
|
finally:
|
|
context.__exit__(None, None, None)
|
|
context = patched_device(
|
|
LaserDelay,
|
|
name="laser_delay",
|
|
step_sleep=0.0,
|
|
cavity_period=12.0084,
|
|
calibration_file=str(calibration_file),
|
|
)
|
|
device = context.__enter__()
|
|
set_hardware(device, **hardware)
|
|
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(calibration_file, caplog):
|
|
context, device = make_device(calibration_file)
|
|
try:
|
|
device.calibrate(SCOPE)
|
|
finally:
|
|
context.__exit__(None, None, None)
|
|
content = json.loads(calibration_file.read_text())
|
|
content["calibrated_at"] = "2026-01-01T00:00:00+00:00"
|
|
calibration_file.write_text(json.dumps(content))
|
|
context, device = make_device(calibration_file)
|
|
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_broken_calibration_file_leaves_the_device_uncalibrated(calibration_file, caplog):
|
|
calibration_file.parent.mkdir(parents=True)
|
|
calibration_file.write_text("{not json")
|
|
context, device = make_device(calibration_file)
|
|
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_file_configured_writes_nothing(laser_delay, tmp_path, monkeypatch):
|
|
monkeypatch.chdir(tmp_path)
|
|
laser_delay.on_connected()
|
|
laser_delay.calibrate(SCOPE)
|
|
assert laser_delay.get_calibration_info()["storage"] is None
|
|
assert not list(tmp_path.iterdir())
|
|
|
|
|
|
def test_unwritable_calibration_store_refuses_the_move(tmp_path, caplog):
|
|
"""Without the move marker a later restore could trust a stale phase: fail safe."""
|
|
blocker = tmp_path / "file"
|
|
blocker.write_text("")
|
|
context, device = make_device(blocker / "cal.json") # parent is a file
|
|
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_restart_restores_hardware_and_calibration(tmp_path):
|
|
path = tmp_path / "cal.json"
|
|
device = SimLaserDelay(name="laser_delay", step_sleep=0.0, sim_noise=0, calibration_file=path)
|
|
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, calibration_file=path)
|
|
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 (deployment audit 2026-10-08) #
|
|
# --------------------------------------------------------------------------------------------- #
|
|
|
|
|
|
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):
|
|
"""BEC v4's mv ends without waiting; the next queued scan's set arrives while the ramp runs. The
|
|
refusal must not end that ramp: the device server turns an exception from set() into a status
|
|
whose failure calls stop(), so the refusal is an already failed status instead."""
|
|
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)
|