fix: adjust device config and pseudos #29

Merged
wakonig_k merged 1 commits from fix/device_config_pseudo into main 2026-10-16 09:38:11 +02:00
7 changed files with 360 additions and 78 deletions

No files matched your search

+275
View File
@@ -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
+24 -27
View File
@@ -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
+34 -35
View File
@@ -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
+2 -1
View File
@@ -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 *
-9
View File
@@ -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
+16 -6
View File
@@ -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)
@@ -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