bec/debye_bec
0
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Added tests for advanced scans

Browse Source
This commit is contained in:
gac-x01da (Resp. Clark Adam Hugh)
2024-09-11 14:54:16 +02:00
committed by gac-x01da
parent 38bee8c5c7
commit 31dd582648
3 changed files with 421 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
debye_bec/devices/mo1_bragg.py
View File

@@ -541,6 +541,9 @@ class Mo1Bragg(Device, PositionerBase):
## ACS requires energy as input, outputs angle
## + reset bit signal, + calc done bit signal
## In the meantime, misuse s_scan_angle/energy PVs
## TODO Add fallback solution for automatic testing, otherwise test will fail
## because no monochromator will calculate the angle
## Unsure how to implement this
move_type = self.move_type.get()
if move_type == MoveType.ENERGY:
self.scan_settings.s_scan_energy_lo.put(e_kink)

122
tests/tests_devices/test_mo1_bragg.py
View File

@@ -149,6 +149,23 @@ def test_set_xas_settings(mock_bragg):
assert dev.scan_settings.s_scan_angle_hi.get() == 20
assert dev.scan_settings.s_scan_scantime.get() == 1
def test_set_advanced_xas_settings(mock_bragg):
dev = mock_bragg
dev.move_type.set(MoveType.ENERGY)
dev.set_advanced_xas_settings(low=10000, high=12000, scan_time=0.5)
p_prev = 0
for p in dev.scan_settings.a_scan_pos.get():
assert p > 0 and p < 40, "maximum positions of monochromator exceeded"
assert p >= p_prev, "values in a_scan_pos must be in ascending order"
p_prev = p
for v in dev.scan_settings.a_scan_vel.get():
assert v > -120 and v < 120, "maximum speed of monochromator exceeded"
t_tot
for t in dev.scan_settings.a_scan_time.get():
assert t >= 0, "time interval must be greater then 0"
t_tot = t_tot + t
assert t_tot <= 0.5, "total time must be smaller or equal to scan_time"
# TODO add test for MoveType.Angle (or delete MoveType.Angle)
def test_set_xrd_settings(mock_bragg):
dev = mock_bragg
@@ -201,6 +218,8 @@ def test_update_scan_parameters(mock_bragg):
"exp_time_high": 3,
"cycle_low": 1,
"cycle_high": 5,
"p_kink": 50,
"e_kink": 8000,
}
},
metadata={},
@@ -327,6 +346,50 @@ def test_unstage(mock_bragg):
queue="primary",
metadata={"RID": "test1234"},
),
ScanQueueMessage(
scan_type="xas_advanced_scan",
parameter={
"args": {},
"kwargs": {
"motor": "mo1_bragg",
"start": 8000,
"stop": 9000,
"scan_time": 1,
"scan_duration": 10,
"p_kink": 50,
"e_kink": 8500,
},
"num_points": 100,
},
queue="primary",
metadata={"RID": "test1234"},
),
ScanQueueMessage(
scan_type="xas_advanced_scan_with_xrd",
parameter={
"args": {},
"kwargs": {
"motor": "mo1_bragg",
"start": 8000,
"stop": 9000,
"scan_time": 1,
"scan_duration": 10,
"p_kink": 50,
"e_kink": 8500,
"xrd_enable_low": True,
"xrd_enable_high": False,
"num_trigger_low": 1,
"num_trigger_high": 7,
"exp_time_low": 1,
"exp_time_high": 3,
"cycle_low": 1,
"cycle_high": 5,
},
"num_points": 10,
},
queue="primary",
metadata={"RID": "test1234"},
),
],
)
def test_stage(mock_bragg, scan_worker_mock, msg):
@@ -365,7 +428,7 @@ def test_stage(mock_bragg, scan_worker_mock, msg):
):
scan_name = scan_status_msg.content["info"].get("scan_name", "")
# Chek the not implemented fly scan first, should raise Mo1BraggError
if scan_name not in ["xas_simple_scan", "xas_simple_scan_with_xrd"]:
if scan_name not in ["xas_simple_scan", "xas_simple_scan_with_xrd", "xas_advanced_scan", "xas_advanced_scan_with_xrd"]:
with pytest.raises(Mo1BraggError):
mock_bragg.stage()
assert mock_check_scan_msg.call_count == 1
@@ -373,6 +436,7 @@ def test_stage(mock_bragg, scan_worker_mock, msg):
else:
with (
mock.patch.object(mock_bragg, "set_xas_settings") as mock_xas_settings,
mock.patch.object(mock_bragg, "set_advanced_xas_settings") as mock_advanced_xas_settings,
mock.patch.object(mock_bragg, "set_xrd_settings") as mock_xrd_settings,
mock.patch.object(
mock_bragg, "set_scan_control_settings"
@@ -434,3 +498,59 @@ def test_stage(mock_bragg, scan_worker_mock, msg):
"scan_duration"
],
)
# Check xas_advanced_scan
elif scan_name == "xas_advanced_scan":
mock_bragg.stage()
assert mock_advanced_xas_settings.call_args == mock.call(
low=scan_status_msg.content["info"]["kwargs"]["start"],
high=scan_status_msg.content["info"]["kwargs"]["stop"],
scan_time=scan_status_msg.content["info"]["kwargs"]["scan_time"],
)
assert mock_xrd_settings.call_args == mock.call(
enable_low=False,
enable_high=False,
num_trigger_low=0,
num_trigger_high=0,
exp_time_low=0,
exp_time_high=0,
cycle_low=0,
cycle_high=0,
)
assert mock_set_scan_control_settings.call_args == mock.call(
mode=ScanControlMode.ADVANCED,
scan_duration=scan_status_msg.content["info"]["kwargs"][
"scan_duration"
],
)
# Check xas_advanced_scan_with_xrd
elif scan_name == "xas_advanced_scan_with_xrd":
mock_bragg.stage()
assert mock_advanced_xas_settings.call_args == mock.call(
low=scan_status_msg.content["info"]["kwargs"]["start"],
high=scan_status_msg.content["info"]["kwargs"]["stop"],
scan_time=scan_status_msg.content["info"]["kwargs"]["scan_time"],
)
assert mock_xrd_settings.call_args == mock.call(
enable_low=scan_status_msg.content["info"]["kwargs"]["xrd_enable_low"],
enable_high=scan_status_msg.content["info"]["kwargs"][
"xrd_enable_high"
],
num_trigger_low=scan_status_msg.content["info"]["kwargs"][
"num_trigger_low"
],
num_trigger_high=scan_status_msg.content["info"]["kwargs"][
"num_trigger_high"
],
exp_time_low=scan_status_msg.content["info"]["kwargs"]["exp_time_low"],
exp_time_high=scan_status_msg.content["info"]["kwargs"][
"exp_time_high"
],
cycle_low=scan_status_msg.content["info"]["kwargs"]["cycle_low"],
cycle_high=scan_status_msg.content["info"]["kwargs"]["cycle_high"],
)
assert mock_set_scan_control_settings.call_args == mock.call(
mode=ScanControlMode.ADVANCED,
scan_duration=scan_status_msg.content["info"]["kwargs"][
"scan_duration"
],
)

298
tests/tests_scans/test_mono_bragg_scans.py
View File

@@ -4,7 +4,7 @@ from unittest import mock
from bec_lib.messages import DeviceInstructionMessage
from bec_server.device_server.tests.utils import DMMock
from debye_bec.scans import XASSimpleScan, XASSimpleScanWithXRD
from debye_bec.scans import XASSimpleScan, XASSimpleScanWithXRD, XASAdvancedScan, XASAdvancedScanWithXRD
def test_xas_simple_scan():
@@ -315,3 +315,299 @@ def test_xas_simple_scan_with_xrd():
parameter={},
),
]
def test_xas_advanced_scan():
# create a fake device manager that we can use to add devices
device_manager = DMMock()
device_manager.add_device("mo1_bragg")
request = XASAdvancedScan(
start=8000, stop=9000, scan_time=1, scan_duration=10, p_kink=50, e_kink=8500, device_manager=device_manager
)
request.metadata["RID"] = "my_test_request_id"
with (
mock.patch.object(request.stubs, "get_req_status", side_effect=[False, True]),
mock.patch.object(request.stubs, "_get_from_rpc", return_value=True),
):
reference_commands = list(request.run())
for cmd in reference_commands:
if not cmd:
continue
if "RID" in cmd.metadata:
cmd.metadata["RID"] = "my_test_request_id"
if "rpc_id" in cmd.parameter:
cmd.parameter["rpc_id"] = "my_test_rpc_id"
assert reference_commands == [
None,
None,
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 0, "RID": "my_test_request_id"},
device=None,
action="scan_report_instruction",
parameter={"device_progress": ["mo1_bragg"]},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 1, "RID": "my_test_request_id"},
device=None,
action="open_scan",
parameter={
"scan_motors": [],
"readout_priority": {
"monitored": [],
"baseline": [],
"on_request": [],
"async": [],
},
"num_points": None,
"positions": [8000, 9000],
"scan_name": "xas_advanced_scan",
"scan_type": "fly",
},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 2, "RID": "my_test_request_id"},
device=None,
action="stage",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "baseline", "DIID": 3, "RID": "my_test_request_id"},
device=None,
action="baseline_reading",
parameter={},
),
DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"DIID": 4,
"RID": "my_test_request_id",
"response": True,
},
device="mo1_bragg",
action="rpc",
parameter={
"device": "mo1_bragg",
"func": "move_type.set",
"rpc_id": "my_test_rpc_id",
"args": ("energy",),
"kwargs": {},
},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 5, "RID": "my_test_request_id"},
device="mo1_bragg",
action="kickoff",
parameter={"configure": {}, "wait_group": "kickoff"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 6, "RID": "my_test_request_id"},
device="mo1_bragg",
action="complete",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 7, "RID": "my_test_request_id"},
device=None,
action="read",
parameter={"group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 8, "RID": "my_test_request_id"},
device=None,
action="wait",
parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 9, "RID": "my_test_request_id"},
device=None,
action="read",
parameter={"group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 10, "RID": "my_test_request_id"},
device=None,
action="wait",
parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 11, "RID": "my_test_request_id"},
device=None,
action="wait",
parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 12, "RID": "my_test_request_id"},
device=None,
action="complete",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 13, "RID": "my_test_request_id"},
device=None,
action="unstage",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 14, "RID": "my_test_request_id"},
device=None,
action="close_scan",
parameter={},
),
]
def test_xas_advanced_scan_with_xrd():
# create a fake device manager that we can use to add devices
device_manager = DMMock()
device_manager.add_device("mo1_bragg")
request = XASAdvancedScanWithXRD(
start=8000,
stop=9000,
scan_time=1,
scan_duration=10,
p_kink=50,
e_kink=8500,
xrd_enable_low=True,
num_trigger_low=1,
exp_time_low=1,
cycle_low=1,
xrd_enable_high=True,
num_trigger_high=2,
exp_time_high=3,
cycle_high=4,
device_manager=device_manager,
)
request.metadata["RID"] = "my_test_request_id"
with (
mock.patch.object(request.stubs, "get_req_status", side_effect=[False, True]),
mock.patch.object(request.stubs, "_get_from_rpc", return_value=True),
):
reference_commands = list(request.run())
for cmd in reference_commands:
if not cmd:
continue
if "RID" in cmd.metadata:
cmd.metadata["RID"] = "my_test_request_id"
if "rpc_id" in cmd.parameter:
cmd.parameter["rpc_id"] = "my_test_rpc_id"
assert reference_commands == [
None,
None,
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 0, "RID": "my_test_request_id"},
device=None,
action="scan_report_instruction",
parameter={"device_progress": ["mo1_bragg"]},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 1, "RID": "my_test_request_id"},
device=None,
action="open_scan",
parameter={
"scan_motors": [],
"readout_priority": {
"monitored": [],
"baseline": [],
"on_request": [],
"async": [],
},
"num_points": None,
"positions": [8000, 9000],
"scan_name": "xas_advanced_scan_with_xrd",
"scan_type": "fly",
},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 2, "RID": "my_test_request_id"},
device=None,
action="stage",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "baseline", "DIID": 3, "RID": "my_test_request_id"},
device=None,
action="baseline_reading",
parameter={},
),
DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"DIID": 4,
"RID": "my_test_request_id",
"response": True,
},
device="mo1_bragg",
action="rpc",
parameter={
"device": "mo1_bragg",
"func": "move_type.set",
"rpc_id": "my_test_rpc_id",
"args": ("energy",),
"kwargs": {},
},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 5, "RID": "my_test_request_id"},
device="mo1_bragg",
action="kickoff",
parameter={"configure": {}, "wait_group": "kickoff"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 6, "RID": "my_test_request_id"},
device="mo1_bragg",
action="complete",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 7, "RID": "my_test_request_id"},
device=None,
action="read",
parameter={"group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 8, "RID": "my_test_request_id"},
device=None,
action="wait",
parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 9, "RID": "my_test_request_id"},
device=None,
action="read",
parameter={"group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 10, "RID": "my_test_request_id"},
device=None,
action="wait",
parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 11, "RID": "my_test_request_id"},
device=None,
action="wait",
parameter={"type": "read", "group": "primary", "wait_group": "readout_primary"},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 12, "RID": "my_test_request_id"},
device=None,
action="complete",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 13, "RID": "my_test_request_id"},
device=None,
action="unstage",
parameter={},
),
DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "DIID": 14, "RID": "my_test_request_id"},
device=None,
action="close_scan",
parameter={},
),
]