428 lines
17 KiB
Python
428 lines
17 KiB
Python
"""Golden tests: the device writes exactly what the reference ``DelayLaser.py`` writes."""
|
|
|
|
# pylint: disable=redefined-outer-name,protected-access,no-member
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import sys
|
|
import types
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from ophyd_devices.tests.utils import patched_device
|
|
|
|
from superxas_bec.devices.laser_delay import (
|
|
LaserDelay,
|
|
LaserDelayHardwareError,
|
|
delay_card_setting,
|
|
phase_from_voltages,
|
|
phase_ramp_offsets,
|
|
voltages_for_phase,
|
|
wrap_phase,
|
|
)
|
|
|
|
REFERENCE_SCRIPT = Path(__file__).parent / "reference" / "DelayLaser.py.txt"
|
|
|
|
DELAY = "X10DA-ES2-TIM:SET1-DELAY"
|
|
FINE = "X10DA-ES2-TIM:SET1-FINE"
|
|
EVR = "X10DA-ES2-TIM:EVR-FPS2"
|
|
V1 = "X10DA-ES4:AO-02"
|
|
V2 = "X10DA-ES4:AO-01"
|
|
|
|
FLOAT_TOL = 1e-12
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fast_ramp(monkeypatch):
|
|
"""Tests ramp without sleeping; the beamline minimum is tested on its own."""
|
|
monkeypatch.setattr(LaserDelay, "MIN_STEP_SLEEP", 0.0)
|
|
|
|
|
|
# start states
|
|
STATE_A = {EVR: 2.0002, DELAY: 150, FINE: 37, V1: 0.6, V2: -0.8}
|
|
STATE_B = {EVR: 1.0, DELAY: 10, FINE: 150, V1: -0.3, V2: 0.95}
|
|
|
|
|
|
@dataclass
|
|
class Scenario:
|
|
"""Calibrate with ``scope``, then run ``steps`` (delays, or dicts of PVs changed by hand)."""
|
|
|
|
initial: dict
|
|
scope: float
|
|
steps: list = field(default_factory=list)
|
|
|
|
|
|
SCENARIOS = {
|
|
"scope_cal_and_basic_moves": Scenario(STATE_A, 546.04 - 546.92, [-4.9, 0.0, 0.5, 5.3]),
|
|
"cross_cavity_period": Scenario(STATE_A, 546.04 - 546.92, [11.9, 12.2, 11.9]),
|
|
"large_negative_delays": Scenario(STATE_A, 546.04 - 546.92, [-1500.3, -250.37, -24.02]),
|
|
"laser_period_wrap": Scenario(STATE_A, 546.04 - 546.92, [400.0, -1800.0, 299.49, 299.5]),
|
|
"large_phase_moves_both_directions": Scenario(
|
|
STATE_A, 546.04 - 546.92, [0.1, 6.0, 0.1, 11.95, 0.05]
|
|
),
|
|
"other_state_positive_scope": Scenario(STATE_B, 37.25, [37.25, 30.0, -3.33, 1000.0]),
|
|
"other_state_wrap_short_period": Scenario(STATE_B, -14.58 + 15.46, [-990.0, 2.5, 999.99]),
|
|
}
|
|
|
|
|
|
# reference script with fake EPICS
|
|
|
|
|
|
class FakeIOC:
|
|
"""PV values shared by all fake channels; records puts in order."""
|
|
|
|
def __init__(self, initial: dict):
|
|
self.values = dict(initial)
|
|
self.puts: list[tuple[str, object]] = []
|
|
|
|
|
|
def load_reference_script(ioc: FakeIOC, monkeypatch, tmp_path, sleeps: list) -> types.ModuleType:
|
|
"""Execute the reference script as a fresh module talking to ``ioc``."""
|
|
|
|
class CaChannel: # pylint: disable=invalid-name
|
|
"""Fake of CaChannel.CaChannel: searchw/getw/putw."""
|
|
|
|
def __init__(self):
|
|
self.pvname = None
|
|
|
|
def searchw(self, pvname):
|
|
self.pvname = pvname
|
|
|
|
def getw(self):
|
|
return ioc.values[self.pvname]
|
|
|
|
def putw(self, value):
|
|
ioc.puts.append((self.pvname, value))
|
|
ioc.values[self.pvname] = value
|
|
|
|
ca_module = types.ModuleType("CaChannel")
|
|
ca_module.CaChannel = CaChannel
|
|
ca_module.ca = types.SimpleNamespace()
|
|
epics_module = types.ModuleType("epicsPV")
|
|
epics_module.epicsPV = object
|
|
monkeypatch.setitem(sys.modules, "CaChannel", ca_module)
|
|
monkeypatch.setitem(sys.modules, "epicsPV", epics_module)
|
|
monkeypatch.chdir(tmp_path) # Timing.offset is written to the working directory
|
|
|
|
module = types.ModuleType("delay_laser_reference")
|
|
module.__file__ = str(REFERENCE_SCRIPT)
|
|
source = REFERENCE_SCRIPT.read_text(encoding="utf-8")
|
|
code = compile(source, str(REFERENCE_SCRIPT), "exec")
|
|
exec(code, module.__dict__) # pylint: disable=exec-used
|
|
module.time = types.SimpleNamespace(sleep=sleeps.append)
|
|
module.ConnectEpics()
|
|
return module
|
|
|
|
|
|
def run_reference(scenario: Scenario, monkeypatch, tmp_path) -> dict:
|
|
"""Run a scenario through the script. Returns calibration, puts and state per move."""
|
|
ioc = FakeIOC(scenario.initial)
|
|
sleeps: list = []
|
|
script = load_reference_script(ioc, monkeypatch, tmp_path, sleeps)
|
|
script.CurrentSituation(scenario.scope)
|
|
result = {"calibration": list(np.loadtxt("Timing.offset")), "moves": []}
|
|
for step in scenario.steps:
|
|
if isinstance(step, dict):
|
|
ioc.values.update(step)
|
|
continue
|
|
ioc.puts.clear()
|
|
sleeps.clear()
|
|
script.SetDelay(step)
|
|
result["moves"].append(
|
|
{
|
|
"target": step,
|
|
"puts": list(ioc.puts),
|
|
"sleeps": len(sleeps),
|
|
"state": list(np.loadtxt("Timing.offset")),
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
# device with recorded puts
|
|
|
|
|
|
def record_puts(device: LaserDelay) -> list:
|
|
"""Record every put to the four written PVs of ``device``, in order."""
|
|
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_pv_values(device: LaserDelay, values: dict) -> None:
|
|
"""Set mocked PV values without recording them as puts."""
|
|
signals = {
|
|
DELAY: device.delay_coarse,
|
|
FINE: device.delay_fine,
|
|
EVR: device.evr_period,
|
|
V1: device.phase_v1,
|
|
V2: device.phase_v2,
|
|
}
|
|
for pvname, value in values.items():
|
|
signals[pvname]._read_pv.mock_data = value
|
|
|
|
|
|
def run_device(scenario: Scenario, **device_kwargs) -> dict:
|
|
"""Run a scenario through the device (default configuration unless ``device_kwargs``)."""
|
|
device_kwargs.setdefault("fine_rollover", False) # the script writes FINE 200
|
|
with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0, **device_kwargs) as device:
|
|
set_pv_values(device, scenario.initial)
|
|
log = record_puts(device)
|
|
calibration = device.calibrate(scenario.scope)
|
|
result = {
|
|
"calibration": [
|
|
calibration[key]
|
|
for key in ("laser_period", "offset_t", "offset_phase_t", "phase_t")
|
|
],
|
|
"moves": [],
|
|
}
|
|
for step in scenario.steps:
|
|
if isinstance(step, dict):
|
|
set_pv_values(device, step)
|
|
continue
|
|
log.clear()
|
|
status = device.set(step)
|
|
status.wait(timeout=30)
|
|
assert status.success
|
|
state = device.get_calibration()
|
|
result["moves"].append(
|
|
{
|
|
"target": step,
|
|
"puts": list(log),
|
|
"state": [
|
|
state[key]
|
|
for key in ("laser_period", "offset_t", "offset_phase_t", "phase_t")
|
|
],
|
|
"readback": device.read()["laser_delay"]["value"],
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
# comparison helpers
|
|
|
|
|
|
def assert_same_value(expected, actual, context: str) -> None:
|
|
"""Integers must match exactly (and be integers); floats within FLOAT_TOL."""
|
|
if isinstance(expected, (int, np.integer)) and not isinstance(expected, bool):
|
|
assert isinstance(actual, (int, np.integer)), f"{context}: {actual!r} is not an integer"
|
|
assert int(actual) == int(expected), f"{context}: {actual} != {expected}"
|
|
return
|
|
assert math.isclose(
|
|
float(actual), float(expected), rel_tol=FLOAT_TOL, abs_tol=FLOAT_TOL
|
|
), f"{context}: {actual!r} != {expected!r} (diff {float(actual) - float(expected):.3e})"
|
|
|
|
|
|
def assert_same_puts(expected: list, actual: list, context: str) -> None:
|
|
"""Same PVs in the same order, same values."""
|
|
assert [pv for pv, _ in actual] == [
|
|
pv for pv, _ in expected
|
|
], f"{context}: PV sequence differs (expected {len(expected)} puts, got {len(actual)})"
|
|
for index, ((pv, expected_value), (_, actual_value)) in enumerate(zip(expected, actual)):
|
|
assert_same_value(expected_value, actual_value, f"{context}, put #{index} to {pv}")
|
|
|
|
|
|
# tests
|
|
|
|
|
|
@pytest.mark.parametrize("scenario_id", list(SCENARIOS))
|
|
def test_device_matches_reference_script(scenario_id, monkeypatch, tmp_path):
|
|
"""Calibration and every move write the same PV sequence as the script."""
|
|
scenario = SCENARIOS[scenario_id]
|
|
expected = run_reference(scenario, monkeypatch, tmp_path)
|
|
actual = run_device(scenario)
|
|
|
|
for key, exp, act in zip(
|
|
("laser_period", "offset_t", "offset_phase_t", "phase_t"),
|
|
expected["calibration"],
|
|
actual["calibration"],
|
|
):
|
|
assert_same_value(exp, act, f"{scenario_id}: calibration {key}")
|
|
|
|
assert len(actual["moves"]) == len(expected["moves"])
|
|
for exp_move, act_move in zip(expected["moves"], actual["moves"]):
|
|
context = f"{scenario_id}: move to {exp_move['target']}"
|
|
assert_same_puts(exp_move["puts"], act_move["puts"], context)
|
|
# delay card first, then pairs of V1/V2; one script sleep per pair
|
|
pvs = [pv for pv, _ in exp_move["puts"]]
|
|
assert pvs[:2] == [DELAY, FINE]
|
|
assert pvs[2::2] == [V1] * exp_move["sleeps"] and pvs[3::2] == [V2] * exp_move["sleeps"]
|
|
for key, exp, act in zip(
|
|
("laser_period", "offset_t", "offset_phase_t", "phase_t"),
|
|
exp_move["state"],
|
|
act_move["state"],
|
|
):
|
|
assert_same_value(exp, act, f"{context}: stored {key}")
|
|
assert act_move["readback"] == exp_move["target"]
|
|
|
|
|
|
@pytest.mark.parametrize("scenario_id", list(SCENARIOS))
|
|
def test_shortest_phase_path_against_reference_script(scenario_id, monkeypatch, tmp_path):
|
|
"""Opt-in phase_path="shortest": script delay card, folded ramps, documented wrap offset."""
|
|
scenario = SCENARIOS[scenario_id]
|
|
expected = run_reference(scenario, monkeypatch, tmp_path)
|
|
actual = run_device(scenario, phase_path="shortest")
|
|
for key, exp, act in zip(
|
|
("laser_period", "offset_t", "offset_phase_t", "phase_t"),
|
|
expected["calibration"],
|
|
actual["calibration"],
|
|
):
|
|
assert_same_value(exp, act, f"{scenario_id}: calibration {key}")
|
|
|
|
max_pairs = math.ceil(6.005 / (2.0 / 36)) + 1
|
|
touched = False
|
|
voltages = {"exp": (scenario.initial[V1], scenario.initial[V2])}
|
|
voltages["act"] = voltages["exp"]
|
|
stored = {"exp": expected["calibration"][3], "act": actual["calibration"][3]}
|
|
moves = iter(zip(expected["moves"], actual["moves"], strict=True))
|
|
for step in scenario.steps:
|
|
if isinstance(step, dict): # both programs see the same hand-made change
|
|
voltages = {
|
|
key: (step.get(V1, v[0]), step.get(V2, v[1])) for key, v in voltages.items()
|
|
}
|
|
touched = True
|
|
continue
|
|
exp_move, act_move = next(moves)
|
|
context = f"{scenario_id}: move to {exp_move['target']}"
|
|
assert_same_puts(exp_move["puts"][:2], act_move["puts"][:2], f"{context}: delay card")
|
|
assert 1 <= (len(act_move["puts"]) - 2) // 2 <= max_pairs, context
|
|
for index in range(3): # laser_period, offset_t, offset_phase_t
|
|
assert_same_value(exp_move["state"][index], act_move["state"][index], context)
|
|
assert act_move["readback"] == exp_move["target"]
|
|
|
|
wraps = (exp_move["state"][3] - act_move["state"][3]) / 12.01
|
|
assert wraps == pytest.approx(round(wraps), abs=1e-9), context
|
|
turned = act_move["state"][3] - stored["act"]
|
|
assert abs(turned) <= 6.005 + 1e-9, context
|
|
start = {key: phase_from_voltages(*v, 2.0) for key, v in voltages.items()}
|
|
voltages = {
|
|
"exp": (exp_move["puts"][-2][1], exp_move["puts"][-1][1]),
|
|
"act": (act_move["puts"][-2][1], act_move["puts"][-1][1]),
|
|
}
|
|
end = {key: phase_from_voltages(*v, 2.0) for key, v in voltages.items()}
|
|
assert wrap_phase(end["act"] - start["act"] - turned, 2.0) == pytest.approx(
|
|
0, abs=1e-9
|
|
), context
|
|
if not touched:
|
|
assert wrap_phase(end["act"] - end["exp"], 2.0) == pytest.approx(
|
|
wrap_phase(-0.01 * round(wraps), 2.0), abs=1e-9
|
|
), context
|
|
stored = {"exp": exp_move["state"][3], "act": act_move["state"][3]}
|
|
|
|
|
|
def test_scenario_table_covers_the_edge_cases(monkeypatch, tmp_path):
|
|
"""The scenarios really cover laser-period wraps, cavity crossings and long ramps."""
|
|
totals, phase_deltas, crossings = [], [], 0
|
|
for scenario in SCENARIOS.values():
|
|
expected = run_reference(scenario, monkeypatch, tmp_path)
|
|
laser_period, offset_t, _, phase_t = expected["calibration"]
|
|
targets = [move["target"] for move in expected["moves"]]
|
|
for previous, current in zip(targets, targets[1:]):
|
|
if math.floor(previous / 12.01) != math.floor(current / 12.01):
|
|
crossings += 1
|
|
for move in expected["moves"]:
|
|
totals.append((-offset_t - move["target"], laser_period))
|
|
phase_deltas.append(move["state"][3] - phase_t)
|
|
phase_t = move["state"][3]
|
|
assert any(total < 0 for total, _ in totals), "no case wraps below zero"
|
|
assert any(total >= period for total, period in totals), "no case wraps above the laser period"
|
|
assert crossings > 0, "no move crosses a cavity period"
|
|
assert min(phase_deltas) < -10 and max(phase_deltas) > 10, "no long ramps in both directions"
|
|
|
|
|
|
def test_zero_phase_move_script_fails_device_writes_final_step(monkeypatch, tmp_path):
|
|
"""Same delay twice: the script crashes in arange, the device writes only the final step."""
|
|
scenario = Scenario(STATE_A, 546.04 - 546.92, [0.5])
|
|
ioc = FakeIOC(scenario.initial)
|
|
script = load_reference_script(ioc, monkeypatch, tmp_path, [])
|
|
script.CurrentSituation(scenario.scope)
|
|
script.SetDelay(0.5)
|
|
ioc.puts.clear()
|
|
with pytest.raises((ValueError, ZeroDivisionError)):
|
|
script.SetDelay(0.5)
|
|
script_card_puts = list(ioc.puts)
|
|
assert [pv for pv, _ in script_card_puts] == [DELAY, FINE]
|
|
|
|
with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device:
|
|
set_pv_values(device, scenario.initial)
|
|
device.calibrate(scenario.scope)
|
|
device.set(0.5).wait(timeout=30)
|
|
log = record_puts(device)
|
|
v1_before, v2_before = device.phase_v1.get(), device.phase_v2.get()
|
|
status = device.set(0.5)
|
|
status.wait(timeout=5)
|
|
assert status.success
|
|
assert_same_puts(script_card_puts, log[:2], "repeat move: delay card")
|
|
assert [pv for pv, _ in log[2:]] == [V1, V2]
|
|
assert_same_value(v1_before, log[2][1], "repeat move: V1")
|
|
assert_same_value(v2_before, log[3][1], "repeat move: V2")
|
|
|
|
|
|
def test_delay_card_conversion_matches_script_for_many_values(monkeypatch, tmp_path):
|
|
"""int() truncations of MoveDelayCard, checked on many delays incl. near-integer boundaries."""
|
|
ioc = FakeIOC(STATE_A)
|
|
script = load_reference_script(ioc, monkeypatch, tmp_path, [])
|
|
rng = np.random.default_rng(20261013)
|
|
totals = list(rng.uniform(-5000.0, 5000.0, 3000))
|
|
# values whose fine part sits on a 0.01 ns boundary, where truncation is fragile
|
|
totals += [round(k * 0.01, 2) for k in range(0, 2001)]
|
|
totals += [k * 0.01 for k in range(-400, 400)]
|
|
for laser_period in (2000.2, 1000.0, 1000 * 2.0002, 12.01 * 83.0):
|
|
script.LaserPeriod = laser_period
|
|
for total in totals:
|
|
ioc.puts.clear()
|
|
script.MoveDelayCard(total)
|
|
(_, coarse), (_, fine) = ioc.puts # pylint: disable=unbalanced-tuple-unpacking
|
|
assert delay_card_setting(total, laser_period) == (coarse, fine), (total, laser_period)
|
|
|
|
|
|
def test_phase_ramp_matches_script_for_many_moves(monkeypatch, tmp_path):
|
|
"""Ramp values (numpy.arange steps plus final step) of MovePhaseShifter for random moves."""
|
|
ioc = FakeIOC(STATE_A)
|
|
script = load_reference_script(ioc, monkeypatch, tmp_path, [])
|
|
rng = np.random.default_rng(42)
|
|
for _ in range(200):
|
|
angle = rng.uniform(-np.pi, np.pi)
|
|
ioc.values[V1], ioc.values[V2] = float(np.cos(angle)), float(np.sin(angle))
|
|
start = script.GetPhaseT()
|
|
relative = float(rng.uniform(-24.0, 24.0))
|
|
ioc.puts.clear()
|
|
script.MovePhaseShifter(relative)
|
|
expected = [value for _, value in ioc.puts]
|
|
actual = []
|
|
for offset in list(phase_ramp_offsets(relative, 2.0, 36)) + [relative]:
|
|
actual.extend(voltages_for_phase(start + offset, 2.0, 1.0))
|
|
assert len(actual) == len(expected)
|
|
for exp, act in zip(expected, actual):
|
|
assert_same_value(exp, act, f"phase ramp {relative}")
|
|
|
|
|
|
def test_device_refuses_where_the_script_follows_touched_voltages(monkeypatch, tmp_path):
|
|
"""Voltages turned by hand: the device refuses the move and writes nothing."""
|
|
scenario = Scenario(STATE_A, 546.04 - 546.92, [1.0, {V1: 0.0, V2: 1.0}, 2.0])
|
|
expected = run_reference(scenario, monkeypatch, tmp_path)
|
|
assert len(expected["moves"]) == 2 # the script moves twice
|
|
with patched_device(
|
|
LaserDelay, name="laser_delay", step_sleep=0.0, fine_rollover=False
|
|
) as device:
|
|
set_pv_values(device, scenario.initial)
|
|
device.calibrate(scenario.scope)
|
|
device.set(1.0).wait(timeout=30)
|
|
set_pv_values(device, {V1: 0.0, V2: 1.0})
|
|
log = record_puts(device)
|
|
status = device.set(2.0)
|
|
with pytest.raises(LaserDelayHardwareError):
|
|
status.wait(timeout=30)
|
|
assert log == []
|