diff --git a/debye_bec/devices/mo1_bragg.py b/debye_bec/devices/mo1_bragg.py index 2ae766e..d98812f 100644 --- a/debye_bec/devices/mo1_bragg.py +++ b/debye_bec/devices/mo1_bragg.py @@ -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) diff --git a/tests/tests_devices/test_mo1_bragg.py b/tests/tests_devices/test_mo1_bragg.py index afc7920..25d5f74 100644 --- a/tests/tests_devices/test_mo1_bragg.py +++ b/tests/tests_devices/test_mo1_bragg.py @@ -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" + ], + ) \ No newline at end of file diff --git a/tests/tests_scans/test_mono_bragg_scans.py b/tests/tests_scans/test_mono_bragg_scans.py index b40b2e0..3ed6464 100644 --- a/tests/tests_scans/test_mono_bragg_scans.py +++ b/tests/tests_scans/test_mono_bragg_scans.py @@ -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={}, + ), + ] \ No newline at end of file