153 lines
5.2 KiB
Python
153 lines
5.2 KiB
Python
# pylint: skip-file
|
|
from unittest import mock
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from bec_server.scan_server.tests.scan_fixtures import *
|
|
from bec_server.scan_server.tests.scan_hook_tests import *
|
|
|
|
XAS_SIMPLE_SCAN_DEFAULT_HOOK_TESTS = [
|
|
("prepare_scan", [assert_prepare_scan_reads_baseline_devices]),
|
|
("open_scan", [assert_scan_open_called]),
|
|
("stage", [assert_stage_all_devices_called]),
|
|
("pre_scan", [assert_pre_scan_called]),
|
|
("unstage", [assert_unstage_all_devices_called]),
|
|
("close_scan", [assert_close_scan_waits_for_baseline_and_closes]),
|
|
]
|
|
|
|
|
|
def _assemble_xas_simple_scan(v4_scan_assembler, **overrides):
|
|
params = {
|
|
"start": 8000.0,
|
|
"stop": 9000.0,
|
|
"scan_time": 1.0,
|
|
"scan_duration": 10.0,
|
|
"motor": "mo1_bragg",
|
|
"daq": "nidaq",
|
|
"monitored_readout_cycle": 1.0,
|
|
}
|
|
params.update(overrides)
|
|
return v4_scan_assembler("xas_simple_scan", **params)
|
|
|
|
|
|
@pytest.mark.parametrize(("hook_name", "hook_tests"), XAS_SIMPLE_SCAN_DEFAULT_HOOK_TESTS)
|
|
def test_xas_simple_scan_v4_default_hooks(
|
|
v4_scan_assembler, nth_done_status_mock, hook_name, hook_tests
|
|
):
|
|
scan = _assemble_xas_simple_scan(v4_scan_assembler)
|
|
|
|
run_scan_tests(scan, [(hook_name, hook_tests)], nth_done_status_mock=nth_done_status_mock)
|
|
|
|
|
|
def test_xas_simple_scan_v4_prepare_scan_updates_metadata(v4_scan_assembler):
|
|
scan = _assemble_xas_simple_scan(v4_scan_assembler)
|
|
scan.actions.add_scan_report_instruction_device_progress = mock.MagicMock()
|
|
baseline_status = mock.MagicMock()
|
|
scan.actions.read_baseline_devices = mock.MagicMock(return_value=baseline_status)
|
|
|
|
scan.prepare_scan()
|
|
|
|
scan.actions._build_scan_status_message("open")
|
|
|
|
np.testing.assert_array_equal(scan.scan_info.positions, np.array([8000.0, 9000.0]))
|
|
assert scan.scan_info.additional_scan_parameters["scan_time"] == 1.0
|
|
assert scan.scan_info.additional_scan_parameters["scan_duration"] == 10.0
|
|
assert scan.scan_info.readout_priority_modification["async"] == ["nidaq"]
|
|
scan.actions.add_scan_report_instruction_device_progress.assert_called_once_with(scan.motor)
|
|
scan.actions.read_baseline_devices.assert_called_once_with(wait=False)
|
|
assert scan._baseline_readout_status is baseline_status
|
|
|
|
|
|
def test_xas_simple_scan_v4_scan_core_reads_until_complete(v4_scan_assembler, nth_done_status_mock):
|
|
scan = _assemble_xas_simple_scan(v4_scan_assembler)
|
|
completion_status = nth_done_status_mock(resolve_after=3)
|
|
scan.actions.kickoff = mock.MagicMock()
|
|
scan.actions.complete = mock.MagicMock(return_value=completion_status)
|
|
scan.actions.read_monitored_devices = mock.MagicMock()
|
|
|
|
with mock.patch("debye_bec.scans.xas_simple_scan.time.sleep"):
|
|
scan.scan_core()
|
|
|
|
scan.actions.kickoff.assert_called_once_with(scan.motor)
|
|
scan.actions.complete.assert_called_once_with(scan.motor, wait=False)
|
|
assert scan.actions.read_monitored_devices.call_count == 2
|
|
|
|
|
|
def test_xas_simple_scan_v4_post_scan_completes_all_devices(v4_scan_assembler):
|
|
scan = _assemble_xas_simple_scan(v4_scan_assembler)
|
|
scan.actions.complete_all_devices = mock.MagicMock()
|
|
|
|
scan.post_scan()
|
|
|
|
scan.actions.complete_all_devices.assert_called_once_with()
|
|
|
|
|
|
def test_xas_simple_scan_with_xrd_v4_updates_xrd_metadata(v4_scan_assembler):
|
|
scan = v4_scan_assembler(
|
|
"xas_simple_scan_with_xrd",
|
|
start=8000.0,
|
|
stop=9000.0,
|
|
scan_time=1.0,
|
|
scan_duration=10.0,
|
|
break_enable_low=True,
|
|
break_time_low=1.0,
|
|
cycle_low=2,
|
|
break_enable_high=False,
|
|
break_time_high=3.0,
|
|
cycle_high=4,
|
|
exp_time=0.5,
|
|
n_of_trigger=6,
|
|
motor="mo1_bragg",
|
|
daq="nidaq",
|
|
)
|
|
|
|
assert scan.scan_name == "xas_simple_scan_with_xrd"
|
|
assert scan.scan_info.additional_scan_parameters["break_enable_low"] is True
|
|
assert scan.scan_info.additional_scan_parameters["cycle_high"] == 4
|
|
assert scan.scan_info.additional_scan_parameters["n_of_trigger"] == 6
|
|
|
|
|
|
def test_xas_advanced_scan_v4_updates_spline_metadata(v4_scan_assembler):
|
|
scan = v4_scan_assembler(
|
|
"xas_advanced_scan",
|
|
start=8000.0,
|
|
stop=9000.0,
|
|
scan_time=1.0,
|
|
scan_duration=10.0,
|
|
p_kink=50.0,
|
|
e_kink=8500.0,
|
|
motor="mo1_bragg",
|
|
daq="nidaq",
|
|
)
|
|
|
|
assert scan.scan_name == "xas_advanced_scan"
|
|
assert scan.scan_info.additional_scan_parameters["p_kink"] == 50.0
|
|
assert scan.scan_info.additional_scan_parameters["e_kink"] == 8500.0
|
|
|
|
|
|
def test_xas_advanced_scan_with_xrd_v4_updates_all_metadata(v4_scan_assembler):
|
|
scan = v4_scan_assembler(
|
|
"xas_advanced_scan_with_xrd",
|
|
start=8000.0,
|
|
stop=9000.0,
|
|
scan_time=1.0,
|
|
scan_duration=10.0,
|
|
p_kink=55.0,
|
|
e_kink=8450.0,
|
|
break_enable_low=True,
|
|
break_time_low=1.5,
|
|
cycle_low=2,
|
|
break_enable_high=True,
|
|
break_time_high=2.5,
|
|
cycle_high=3,
|
|
exp_time=0.25,
|
|
n_of_trigger=8,
|
|
motor="mo1_bragg",
|
|
daq="nidaq",
|
|
)
|
|
|
|
assert scan.scan_name == "xas_advanced_scan_with_xrd"
|
|
assert scan.scan_info.additional_scan_parameters["p_kink"] == 55.0
|
|
assert scan.scan_info.additional_scan_parameters["break_enable_high"] is True
|
|
assert scan.scan_info.exp_time == 0.25
|