EXtra/tests/test_components_scantool.py at 704af9c958ea6c129b168a2223d62dffa5ab5501 · European-XFEL/EXtra · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import logging
from unittest.mock import MagicMock

from extra.components import Scantool

import pytest
import numpy as np

def test_scantool(mock_scantool_run, mock_spb_aux_run, caplog):
    mock_name = "CRISPY/KARABACON"
    mock_source = MagicMock()

    # Values taken from p3334, run 167
    mock_run_values = {
        "deviceEnv.acquisitionTime.value": 1.0,
        "deviceEnv.activeMotors.value": np.array([b'polarizer_theta', b'Theta_ana', b'', b'']),
        "scanEnv.scanType.value": "dscan",
        "scanEnv.steps.value": np.array([40, 1, 1, 1]),
        "scanEnv.startPoints.value": np.array([-0.0032, -0.002, 1, 1]),
        "scanEnv.stopPoints.value": np.array([0.0032, 0.002, 1, 1]),
        "actualConfiguration.value": "--- Motors: ['MID_AUXT1_UPP/MOTOR/R1:default', 'MID_EXP_UPP/MOTOR/T7:default']--- Data Sources: ['MID-EXP-UPP/CTRL/KEITHLEY-1:value', 'MID-EXP-UPP/CTRL/KEITHLEY-2:value']--- Triggers: ['MID_RR_SYS/MDL/TRIGGER']"
    }

    mock_run = MagicMock()
    mock_run.__getitem__ = lambda self, _: mock_source
    mock_run.get_run_values.return_value = mock_run_values

    # Test a run without a scantool
    mock_run.control_sources = { "foo" }
    with pytest.raises(RuntimeError) as e:
        Scantool(mock_run)
    assert "please pass an explicit source name" in str(e)

    # Test a run with multiple scantools
    mock_run.control_sources = { mock_name, f"{mock_name}2" }
    with pytest.raises(RuntimeError) as e:
        Scantool(mock_run)
    assert "multiple possible scantools" in str(e)

    mock_run.control_sources = { mock_name }

    # Test a run where the scantool is not active
    mock_source["isMoving"].ndarray.return_value = np.zeros(10, dtype=np.uint8)
    scantool = Scantool(mock_run)
    assert not scantool.active

    # Now with a run where it is active
    mock_source["isMoving"].ndarray.return_value = np.ones(10, dtype=np.uint8)
    scantool = Scantool(mock_run)
    assert scantool.active
    assert scantool.motors == ["polarizer_theta", "Theta_ana"]
    assert scantool.motor_devices is not None

    # And with a run with an unsupported actualConfiguration format
    mock_run_values["actualConfiguration.value"] = "--- Motors:['MID_AUXT1_UPP/MOTOR/R1:default', 'MID_EXP_UPP/MOTOR/T7:default']"
    with pytest.warns():
        scantool = Scantool(mock_run)

    # Test compatibility with Karabacon 3.0.7-2.16.2, which is an older version
    # still used at FXE. In this version the activeMotors and acquisitionTime
    # are stored as first-level properties rather than being under deviceEnv.
    mock_run_values["acquisitionTime.value"] = mock_run_values["deviceEnv.acquisitionTime.value"]
    mock_run_values["activeMotors.value"] = mock_run_values["deviceEnv.activeMotors.value"]
    del mock_run_values["deviceEnv.acquisitionTime.value"]
    del mock_run_values["deviceEnv.activeMotors.value"]

    # This shouldn't throw an exception
    scantool = Scantool(mock_run)

    # Karabacon 3.0.10 renamed the acquisition time again to
    # deviceEnv.acquisitionTimes and data type changed from Double to VectorDouble.
    # Allows defining acq time per step.
    mock_run_values["deviceEnv.acquisitionMode.value"] = "Continuous"
    mock_run_values["deviceEnv.acquisitionTimes.value"] = np.ones(1000, dtype=np.uint8)
    mock_run_values["deviceEnv.acquisitionTimes.value"][0] = 20
    del mock_run_values["acquisitionTime.value"]

    scantool = Scantool(mock_run)
    assert scantool.acquisition_time == 20

    # Test creating a Scantool from a union of two runs with only control data,
    # see mockdata/karabacon.py for the settings it was created with.
    with caplog.at_level(logging.WARNING):
        scantool = Scantool(mock_scantool_run.union(mock_spb_aux_run))
        assert len(caplog.records) == 1

    assert scantool.scan_type == "dscan"
    assert scantool.motors == ["ppl_odl"]
    assert scantool.acquisition_time == 30
    assert scantool.start_positions == { "ppl_odl": -15 }
    assert scantool.stop_positions == { "ppl_odl": 25 }

    # Smoke tests
    scantool.info()
    scantool.format()
    repr(scantool)