diff --git a/tests/tests_devices/test_undulator.py b/tests/tests_devices/test_undulator.py new file mode 100644 index 0000000..3595e8d --- /dev/null +++ b/tests/tests_devices/test_undulator.py @@ -0,0 +1,275 @@ +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest +import yaml +from ophyd import EpicsSignalRO, PVPositioner +from ophyd.utils.errors import StatusTimeoutError +from ophyd_devices.devices.undulator import UndulatorBase +from ophyd_devices.tests.utils import patched_device + +from xtreme_bec.devices.undulator import ( + UndulatorAlpha, + UndulatorGap, + UndulatorHarmonic, + UndulatorMode, + UndulatorModeEnum, +) + + +@pytest.fixture +def mode(): + with patched_device(UndulatorMode, name="id_mode") as device: + device.select_control._read_pv.mock_data = 1 + device.cmu_online._read_pv.mock_data = 4 + device.done._read_pv.mock_data = 1 + yield device + device.destroy() + + +def finish_move(device, status): + device.done._read_pv.mock_data = 0 + assert not status.done + device.done._read_pv.mock_data = 1 + status.wait(timeout=1) + assert status.success + + +def test_mode_uses_undulator_signals_and_standard_set(mode): + assert isinstance(mode, UndulatorBase) + assert mode.setpoint.pvname == "X07MA-UIND:MODE" + assert mode.value_string.pvname == "X07MA-UIND:MODE" + assert isinstance(mode.value_string, EpicsSignalRO) + assert mode.done.pvname == "X07MA-UIND:DONE" + assert mode.stop_signal.pvname == "X07MA-UIND:STOP" + assert mode.select_control.pvname == "X07MA-UIND:SCTRL" + assert mode.hints == {"fields": ["id_mode"]} + assert not hasattr(mode, "set_mode") + + status = mode.set("Linear H") + + assert mode.setpoint.get() == 1 + assert mode.read()["id_mode"]["value"] == 1 + assert mode.position == 1 + assert not status.done + finish_move(mode, status) + + +@pytest.mark.parametrize("selected", list(UndulatorModeEnum)) +@pytest.mark.parametrize("input_type", ["label", "enum", "index"]) +def test_set_accepts_every_mode(mode, selected, input_type): + index = list(UndulatorModeEnum).index(selected) + target = {"label": selected.value, "enum": selected, "index": index}[input_type] + + status = mode.set(target) + + assert mode.setpoint.get() == index + assert not status.done + finish_move(mode, status) + + +@pytest.mark.parametrize("busy", [False, True]) +def test_same_mode_still_writes_and_waits_for_done(mode, busy): + if busy: + mode.done._read_pv.mock_data = 0 + with patch.object(mode.setpoint, "put", wraps=mode.setpoint.put) as put: + status = mode.set("Off") + + put.assert_called_once_with(0, wait=True) + assert not status.done + finish_move(mode, status) + + +def test_move_normalizes_labels_and_preserves_callback_and_timeout(mode): + moved_cb = Mock() + status = mode.move("Circular+", wait=False, timeout=1, moved_cb=moved_cb) + + assert mode.setpoint.get() == 4 + assert status.timeout == 1 + moved_cb.assert_not_called() + finish_move(mode, status) + moved_cb.assert_called_once_with(status, obj=mode) + + +@pytest.mark.parametrize("method", ["set", "move"]) +@pytest.mark.parametrize( + ["target", "error_type", "message"], + [ + (-1, ValueError, "Invalid mode index"), + (-8, ValueError, "Invalid mode index"), + (8, ValueError, "Invalid mode index"), + ("unknown", ValueError, "Invalid mode string"), + (1.0, TypeError, "Mode must be"), + (None, TypeError, "Mode must be"), + (True, TypeError, "Mode must be"), + (False, TypeError, "Mode must be"), + ], +) +def test_invalid_modes_do_not_write(mode, method, target, error_type, message): + with patch.object(mode.setpoint, "put") as put: + with pytest.raises(error_type, match=message): + getattr(mode, method)(target, wait=False) + put.assert_not_called() + + +@pytest.mark.parametrize("target", ["Off", "Circular+", 4]) +@pytest.mark.parametrize("method", ["set", "move"]) +def test_operator_control_blocks_moves_before_status_creation(mode, method, target): + mode.select_control._read_pv.mock_data = 0 + + with patch.object(PVPositioner, "move") as parent_move: + with pytest.raises(PermissionError, match="operator controlled"): + getattr(mode, method)(target, wait=False) + + parent_move.assert_not_called() + assert mode.setpoint.get() == 0 + + +@pytest.mark.parametrize("force", [False, True]) +def test_direct_setpoint_writes_obey_control_validation(mode, force): + mode.select_control._read_pv.mock_data = 0 + + with pytest.raises(PermissionError, match="operator controlled"): + mode.setpoint.put(4, force=force) + + assert mode.setpoint.get() == 0 + + +def test_stop_writes_stop_and_waits_for_done(mode): + status = mode.set("Circular-") + mode.done._read_pv.mock_data = 0 + + mode.stop() + + assert mode.stop_signal.get() == 1 + assert not status.done + finish_move(mode, status) + + +def test_operator_control_skips_stop_write(mode): + mode.select_control._read_pv.mock_data = 0 + + mode.stop() + mode.stop() + + assert mode.stop_signal.get() == 0 + + +def test_move_times_out_without_done_transition(mode): + status = mode.set("Linear H", timeout=0.01) + + with pytest.raises(StatusTimeoutError): + status.wait(timeout=1) + + assert status.done + assert not status.success + + +@pytest.mark.parametrize( + ["device_name", "device_class", "suffix"], + [ + ("id_mode", UndulatorMode, "MODE"), + ("id_harmonic", UndulatorHarmonic, "HARMONIC"), + ("id_gap", UndulatorGap, "GAP-SP"), + ], +) +def test_frontend_undulator_configuration(device_name, device_class, suffix): + config_path = Path(__file__).parents[2] / "xtreme_bec/device_configs/frontend.yaml" + device_config = yaml.safe_load(config_path.read_text())[device_name] + assert device_config["deviceClass"] == f"xtreme_bec.devices.undulator.{device_class.__name__}" + config = device_config["deviceConfig"] + + assert config == {"prefix": "X07MA-UIND:"} + with patched_device(device_class, name=device_name, **config) as device: + assert device.setpoint.pvname == f"X07MA-UIND:{suffix}" + assert device.done.pvname == "X07MA-UIND:DONE" + assert device.cmu_online.pvname == "X07MA-OP-CMU:ONLINE" + device.destroy() + + +@pytest.fixture(params=[UndulatorGap, UndulatorHarmonic, UndulatorAlpha, UndulatorMode]) +def undulator(request): + with patched_device(request.param, name="undulator", prefix="TEST:UIND:") as device: + device.select_control._read_pv.mock_data = 1 + device.cmu_online._read_pv.mock_data = 4 + device.done._read_pv.mock_data = 1 + yield device + device.destroy() + + +def test_cmu_state_uses_absolute_read_only_pv(undulator): + assert undulator.cmu_online.pvname == "X07MA-OP-CMU:ONLINE" + assert isinstance(undulator.cmu_online, EpicsSignalRO) + assert "undulator_cmu_online" not in undulator.read() + + +@pytest.mark.parametrize("state", [0, 1, 2, 3, 5, 6]) +@pytest.mark.parametrize("target", [0, 1]) +@pytest.mark.parametrize("method", ["move", "set"]) +def test_cmu_must_be_online_before_moves(undulator, state, target, method): + undulator.cmu_online._read_pv.mock_data = state + + with patch.object(PVPositioner, "move") as parent_move: + with pytest.raises(PermissionError, match="CMU must be On-line"): + getattr(undulator, method)(target, wait=False) + + parent_move.assert_not_called() + assert undulator.setpoint.get() == 0 + + +@pytest.mark.parametrize("state", [0, 1, 2, 3, 5, 6]) +@pytest.mark.parametrize("force", [False, True]) +def test_cmu_validation_applies_to_direct_setpoint_writes(undulator, state, force): + undulator.cmu_online._read_pv.mock_data = state + + with pytest.raises(PermissionError, match="CMU must be On-line"): + undulator.setpoint.put(1, force=force) + + assert undulator.setpoint.get() == 0 + + +@pytest.mark.parametrize("state", [0, 1, 2, 3, 5, 6]) +def test_cmu_validation_skips_stop_writes(undulator, state): + undulator.cmu_online._read_pv.mock_data = state + + assert undulator.stop_signal.put(1) is None + undulator.stop() + undulator.stop() + + assert undulator.stop_signal.get() == 0 + + +def test_online_cmu_allows_moves_and_done_completion(undulator): + status = undulator.set(1) + + assert undulator.setpoint.get() == 1 + assert not status.done + if undulator.readback is not None: + undulator.readback._read_pv.mock_data = 1 + finish_move(undulator, status) + assert undulator.position == 1 + + +def test_online_cmu_preserves_pending_stop_behavior(undulator): + status = undulator.set(1) + undulator.done._read_pv.mock_data = 0 + + undulator.stop() + + assert undulator.stop_signal.get() == 1 + assert not status.done + finish_move(undulator, status) + + +def test_online_cmu_still_requires_beamline_control(undulator): + undulator.select_control._read_pv.mock_data = 0 + + with patch.object(PVPositioner, "move") as parent_move: + with pytest.raises(PermissionError, match="operator controlled"): + undulator.set(1) + parent_move.assert_not_called() + with pytest.raises(PermissionError, match="operator controlled"): + undulator.setpoint.put(1) + undulator.stop() + assert undulator.setpoint.get() == 0 + assert undulator.stop_signal.get() == 0 diff --git a/xtreme_bec/device_configs/frontend.yaml b/xtreme_bec/device_configs/frontend.yaml index 6bb48fe..c26beb7 100644 --- a/xtreme_bec/device_configs/frontend.yaml +++ b/xtreme_bec/device_configs/frontend.yaml @@ -21,7 +21,6 @@ # Default values may be omitted for brevity. ########################################################################################################################## - # An example device configuration with all fields: # device1: @@ -30,7 +29,7 @@ # deviceConfig: # delay: 1 # update_frequency: 10 -# deviceTags: +# deviceTags: # - frontend # - motor # enabled: true @@ -48,15 +47,13 @@ ########################################################################################################################## - energy: description: Energy in eV - deviceClass: ophyd.EpicsSignal + deviceClass: xtreme_bec.devices.Energy deviceConfig: - auto_monitor: true - read_pv: X07MA-PHS-E:GO.A + prefix: X07MA- deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline energy_mode: @@ -66,7 +63,7 @@ energy_mode: auto_monitor: true read_pv: X07MA-PHS-E:OPT deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline fe_slxc: @@ -75,7 +72,7 @@ fe_slxc: deviceConfig: prefix: X07MA-FE-SL1:CENTERX deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -85,7 +82,7 @@ fe_slxi: deviceConfig: prefix: X07MA-FE-SL1:TRXR deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -95,7 +92,7 @@ fe_slxo: deviceConfig: prefix: X07MA-FE-SL1:TRXW deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -105,7 +102,7 @@ fe_slxs: deviceConfig: prefix: X07MA-FE-SL1:SIZEX deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -115,7 +112,7 @@ fe_slyb: deviceConfig: prefix: X07MA-FE-SL1:TRYB deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -125,7 +122,7 @@ fe_slyc: deviceConfig: prefix: X07MA-FE-SL1:CENTERY deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -135,7 +132,7 @@ fe_slys: deviceConfig: prefix: X07MA-FE-SL1:SIZEY deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -145,7 +142,7 @@ fe_slyt: deviceConfig: prefix: X07MA-FE-SL1:TRYT deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -153,9 +150,9 @@ id_alpha: description: Angle of linear polarization relative to the horizontal plane deviceClass: xtreme_bec.devices.undulator.UndulatorAlpha deviceConfig: - prefix: 'X07MA-UIND:' + prefix: "X07MA-UIND:" deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline id_energy: @@ -166,34 +163,34 @@ id_energy: read_pv: X07MA-UIND:ENERGY-RBV write_pv: X07MA-UIND:ENERGY-SP deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline id_gap: description: Motor to control the IDGap of X07MA deviceClass: xtreme_bec.devices.undulator.UndulatorGap deviceConfig: - prefix: 'X07MA-UIND:' + prefix: "X07MA-UIND:" deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline id_harmonic: description: Undulator harmonic deviceClass: xtreme_bec.devices.undulator.UndulatorHarmonic deviceConfig: - prefix: 'X07MA-UIND:' + prefix: "X07MA-UIND:" deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline id_mode: description: UIND mode of X07MA deviceClass: xtreme_bec.devices.undulator.UndulatorMode deviceConfig: - prefix: 'X07MA-UIND:' + prefix: "X07MA-UIND:" deviceTags: - - frontend + - frontend enabled: true readoutPriority: baseline xbpm1x: @@ -202,7 +199,7 @@ xbpm1x: deviceConfig: prefix: X07MA-FE-XBPM1:TRX deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline @@ -212,7 +209,7 @@ xbpm1y: deviceConfig: prefix: X07MA-FE-XBPM1:TRY deviceTags: - - frontend + - frontend enabled: true onFailure: raise readoutPriority: baseline diff --git a/xtreme_bec/device_configs/optics.yaml b/xtreme_bec/device_configs/optics.yaml index b74edba..e36bf6d 100644 --- a/xtreme_bec/device_configs/optics.yaml +++ b/xtreme_bec/device_configs/optics.yaml @@ -21,7 +21,6 @@ # Default values may be omitted for brevity. ########################################################################################################################## - # An example device configuration with all fields: # device1: @@ -30,7 +29,7 @@ # deviceConfig: # delay: 1 # update_frequency: 10 -# deviceTags: +# deviceTags: # - frontend # - motor # enabled: true @@ -48,85 +47,85 @@ ########################################################################################################################## - cff: description: PGM CFF deviceClass: xtreme_bec.devices.Cff deviceConfig: prefix: X07MA- deviceTags: - - optics + - optics enabled: true readoutPriority: baseline cmu_rx: description: CMU Rx pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oRx user_setpoint: :tRx.A prefix: X07MA-OP-CMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline cmu_ry: description: CMU Ry pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oRy user_setpoint: :tRy.A prefix: X07MA-OP-CMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline cmu_rz: description: CMU Rz pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oRz user_setpoint: :tRz.A prefix: X07MA-OP-CMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline cmu_y: description: CMU y pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oy user_setpoint: :ty.A prefix: X07MA-OP-CMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline cmu_z: description: CMU z pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oz user_setpoint: :tz.A prefix: X07MA-OP-CMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline exit_slit: description: Exit slit - deviceClass: ophyd_devices.EpicsSignal + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: - auto_monitor: true - read_pv: X07MA-OP-SL1SV1:TR_ISAP - write_pv: X07MA-OP-SL1SV1:TR_AP + override_suffixes: + user_readback: :TR_ISAP + user_setpoint: :TR_AP + prefix: X07MA-OP-SL1SV1 deviceTags: - - optics + - optics enabled: true readoutPriority: baseline exit_slit_long: @@ -134,71 +133,71 @@ exit_slit_long: deviceClass: ophyd_devices.EpicsUserMotorVME deviceConfig: limits: - - 180.0 - - 500.0 + - 180.0 + - 500.0 prefix: X07MA-OP-SL1:TRX deviceTags: - - optics + - optics enabled: true readoutPriority: baseline fmu_rx: description: FMU Rx pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oRx user_setpoint: :tRx.A prefix: X07MA-OP-FMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline fmu_ry: description: FMU Ry pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oRy user_setpoint: :tRy.A prefix: X07MA-OP-FMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline fmu_rz: description: FMU Rz pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oRz user_setpoint: :tRz.A prefix: X07MA-OP-FMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline fmu_y: description: FMU y pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oy user_setpoint: :ty.A prefix: X07MA-OP-FMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline fmu_z: description: FMU z pseudo motor - deviceClass: xtreme_bec.devices.mirror_pseudo.MirrorPseudo + deviceClass: xtreme_bec.devices.PseudoPutComplete deviceConfig: override_suffixes: user_readback: :oz user_setpoint: :tz.A prefix: X07MA-OP-FMU deviceTags: - - optics + - optics enabled: true readoutPriority: baseline grating_beta: @@ -207,7 +206,7 @@ grating_beta: deviceConfig: prefix: X07MA:m2 deviceTags: - - optics + - optics enabled: true readoutPriority: baseline mirror_theta: @@ -216,7 +215,7 @@ mirror_theta: deviceConfig: prefix: X07MA:m1 deviceTags: - - optics + - optics enabled: true readoutPriority: baseline mono: @@ -225,6 +224,6 @@ mono: deviceConfig: prefix: X07MA- deviceTags: - - optics + - optics enabled: true readoutPriority: baseline diff --git a/xtreme_bec/devices/__init__.py b/xtreme_bec/devices/__init__.py index ac5b666..d7301e8 100644 --- a/xtreme_bec/devices/__init__.py +++ b/xtreme_bec/devices/__init__.py @@ -1,4 +1,5 @@ from .cff import Cff -from .pgm import PGMMonochromator +from .pgm import Energy, PGMMonochromator +from .pseudo_put_complete import PseudoPutComplete from .undulator import UndulatorAlpha, UndulatorGap, UndulatorHarmonic, UndulatorMode from .x07ma_devices import * diff --git a/xtreme_bec/devices/mirror_pseudo.py b/xtreme_bec/devices/mirror_pseudo.py deleted file mode 100644 index fd5b2e3..0000000 --- a/xtreme_bec/devices/mirror_pseudo.py +++ /dev/null @@ -1,9 +0,0 @@ -from ophyd import Component as Cpt -from ophyd import EpicsSignal -from ophyd_devices.devices.simple_positioner import PSISimplePositionerBase - - -class MirrorPseudo(PSISimplePositionerBase): - user_readback = Cpt(EpicsSignal, "R") - user_setpoint = Cpt(EpicsSignal, "S") - use_put_complete = True diff --git a/xtreme_bec/devices/pgm.py b/xtreme_bec/devices/pgm.py index c0e057a..d19583a 100644 --- a/xtreme_bec/devices/pgm.py +++ b/xtreme_bec/devices/pgm.py @@ -11,21 +11,31 @@ from ophyd_devices import DeviceStatus, ProgressSignal, PSIDeviceBase, StatusBas logger = bec_logger.logger -class PGMMonochromatorBase(PVPositioner): +class Energy(PVPositioner): + """ + Energy positioner for the PGM monochromator. + """ + + setpoint = Cpt(EpicsSignal, "PHS-E:GO.A", kind=Kind.normal, auto_monitor=True) + readback = Cpt(EpicsSignalRO, "PGM:CERBK", kind=Kind.hinted, auto_monitor=True) + done = Cpt(EpicsSignalRO, "PHS:alldone", kind=Kind.omitted, auto_monitor=True) + stop_signal = Cpt(EpicsSignal, "PGM:stop", kind=Kind.omitted) + + def __init__(self, prefix: str = "", **kwargs): + super().__init__(prefix, **kwargs) + self.readback.name = self.name + + +class PGMMonochromatorBase(Energy): """ PGM monochromator """ - setpoint = Cpt(EpicsSignal, "PHS-E:GO.A", auto_monitor=True) - readback = Cpt(EpicsSignalRO, "PGM:CERBK", kind=Kind.hinted, auto_monitor=True) - done = Cpt(EpicsSignalRO, "PHS:alldone", kind=Kind.omitted, auto_monitor=True) mono_energy_readback = Cpt(EpicsSignalRO, "PGM:rbkenergy", kind=Kind.omitted, auto_monitor=True) theta = Cpt(EpicsSignalRO, "PGM:theta", kind=Kind.normal, auto_monitor=True) beta = Cpt(EpicsSignalRO, "PGM:beta", kind=Kind.normal, auto_monitor=True) - stop_signal = Cpt(EpicsSignal, "PGM:stop", kind=Kind.omitted) - cff = Cpt(EpicsSignal, "PGM:rbkcff", write_pv="PGM:cff.A", kind=Kind.config) with_undulator = Cpt(EpicsSignal, "PHS-E:OPT", kind=Kind.config) energy = Cpt(EpicsSignal, "PGM:energy", kind=Kind.omitted, auto_monitor=True) diff --git a/xtreme_bec/devices/pseudo_put_complete.py b/xtreme_bec/devices/pseudo_put_complete.py new file mode 100644 index 0000000..fb57850 --- /dev/null +++ b/xtreme_bec/devices/pseudo_put_complete.py @@ -0,0 +1,9 @@ +from ophyd import Component as Cpt +from ophyd import EpicsSignal, Kind +from ophyd_devices.devices.simple_positioner import PSISimplePositionerBase + + +class PseudoPutComplete(PSISimplePositionerBase): + user_readback = Cpt(EpicsSignal, "R", kind=Kind.hinted, auto_monitor=True) + user_setpoint = Cpt(EpicsSignal, "S", kind=Kind.normal, auto_monitor=True) + use_put_complete = True