60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
from bec_server.scan_server.tests.scan_fixtures import (
|
|
MockCustomDevice,
|
|
device_manager,
|
|
nth_done_status_mock,
|
|
readout_priority,
|
|
v4_scan_assembler,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def otf_scan_assembler(device_manager, readout_priority, v4_scan_assembler):
|
|
"""Use BEC's assembler with the beamline-specific signals and RPC responses."""
|
|
device_signals = {
|
|
"mono": ["sync_with_undulator"],
|
|
"mirror_theta": ["base_velocity", "velocity"],
|
|
"grating_beta": ["base_velocity", "velocity"],
|
|
"signals": ["store_updates"],
|
|
}
|
|
for name, signals in device_signals.items():
|
|
device = MockCustomDevice(
|
|
name,
|
|
device_info={
|
|
"signals": {
|
|
signal: {
|
|
"obj_name": f"{name}_{signal}",
|
|
"kind_str": "config",
|
|
"describe": {"precision": 3},
|
|
}
|
|
for signal in signals
|
|
}
|
|
},
|
|
limits=(250, 2000) if name == "mono" else (-10, 10),
|
|
)
|
|
for signal in signals:
|
|
getattr(device, signal).put = mock.MagicMock()
|
|
device_manager.add_device(device)
|
|
|
|
mono = device_manager.devices["mono"]
|
|
mono.set = mock.MagicMock()
|
|
mono.get_max_speeds = mock.MagicMock(return_value=(0.02, 0.03))
|
|
mono.set_progress_range = mock.MagicMock()
|
|
readout_priority.monitored_devices = ["mono", "signals"]
|
|
|
|
def assemble(e1=500.0, e2=800.0, time=2.0):
|
|
scan = v4_scan_assembler("otf_scan", e1=e1, e2=e2, time=time)
|
|
# The shared fixture's device dictionary needs aliases for dev.<name> access.
|
|
for name in device_signals:
|
|
setattr(scan.dev, name, scan.dev[name])
|
|
return scan
|
|
|
|
return assemble
|
|
|
|
|
|
@pytest.fixture
|
|
def otf_scan(otf_scan_assembler):
|
|
return otf_scan_assembler()
|