302 lines
11 KiB
Python
302 lines
11 KiB
Python
# pylint: skip-file
|
|
import threading
|
|
from typing import Generator
|
|
from unittest import mock
|
|
|
|
import ophyd
|
|
import pytest
|
|
from bec_server.scan_server.scan_worker import ScanWorker
|
|
from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo
|
|
from ophyd.status import WaitTimeoutError
|
|
from ophyd_devices.interfaces.base_classes.psi_device_base import DeviceStoppedError
|
|
from ophyd_devices.tests.utils import MockPV
|
|
|
|
# from bec_server.device_server.tests.utils import DMMock
|
|
from debye_bec.devices.nidaq.nidaq import Nidaq, NidaqError
|
|
|
|
# TODO move this function to ophyd_devices, it is duplicated in csaxs_bec and needed for other pluging repositories
|
|
from debye_bec.devices.test_utils.utils import patch_dual_pvs
|
|
from debye_bec.devices.utils.utils import fetch_scan_info
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def scan_info_mock():
|
|
"""Fixture for the ScanInfo object."""
|
|
return ScanServerScanInfo(scan_name="xas_simple_scan", scan_id="test")
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def scan_worker_mock(scan_server_mock):
|
|
"""Scan worker fixture, utility to generate scan_info for a given scan name."""
|
|
scan_server_mock.device_manager.connector = mock.MagicMock()
|
|
scan_worker = ScanWorker(parent=scan_server_mock)
|
|
yield scan_worker
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_nidaq() -> Generator[Nidaq, None, None]:
|
|
"""Fixture for the Nidaq device."""
|
|
name = "nidaq"
|
|
prefix = "nidaq:prefix_test:"
|
|
with mock.patch.object(ophyd, "cl") as mock_cl:
|
|
mock_cl.get_pv = MockPV
|
|
mock_cl.thread_class = threading.Thread
|
|
dev = Nidaq(name=name, prefix=prefix)
|
|
patch_dual_pvs(dev)
|
|
yield dev
|
|
|
|
|
|
def test_init(mock_nidaq):
|
|
"""Test the initialization of the Nidaq device."""
|
|
dev = mock_nidaq
|
|
assert dev.name == "nidaq"
|
|
assert dev.prefix == "nidaq:prefix_test:"
|
|
assert dev.valid_scan_names == [
|
|
"xas_simple_scan",
|
|
"xas_simple_scan_with_xrd",
|
|
"xas_advanced_scan",
|
|
"xas_advanced_scan_with_xrd",
|
|
"nidaq_continuous_scan",
|
|
]
|
|
|
|
|
|
def test_check_if_scan_name_is_valid(mock_nidaq, scan_info_mock):
|
|
"""Test the check_if_scan_name_is_valid method."""
|
|
dev = mock_nidaq
|
|
scan_info_mock.scan_name = "xas_simple_scan"
|
|
dev.scan_info.msg.info.update(scan_info_mock.model_dump())
|
|
scan_parameters = fetch_scan_info(dev.scan_info)
|
|
assert dev._check_if_scan_name_is_valid(scan_parameters)
|
|
scan_info_mock.scan_name = "invalid_scan_name"
|
|
dev.scan_info.msg.info.update(scan_info_mock.model_dump())
|
|
scan_parameters = fetch_scan_info(dev.scan_info)
|
|
assert not dev._check_if_scan_name_is_valid(scan_parameters)
|
|
|
|
|
|
def test_set_config(mock_nidaq):
|
|
dev = mock_nidaq
|
|
# TODO #21 Add test logic for set_config, issue created #
|
|
|
|
|
|
def test_on_connected(mock_nidaq):
|
|
"""Test the on_connected method of the Nidaq device."""
|
|
dev = mock_nidaq
|
|
dev.power.put(0)
|
|
dev.heartbeat._read_pv.mock_data = 0
|
|
# First scenario, raise timeout error
|
|
|
|
# This will raise a WaitTimeoutError error as we currently do not support callbacks in the MockPV
|
|
dev.timeout_wait_for_signal = 0.1
|
|
# To check that it raised, we check that dev.power PV is set to 1
|
|
# Set state PV to 0, 1 is expected value
|
|
dev.state._read_pv.mock_data = 0
|
|
with pytest.raises(WaitTimeoutError):
|
|
dev.on_connected()
|
|
assert dev.power.get() == 1
|
|
# TODO, once the MOCKPv supports callbacks, we can test the rest of the logic issue #22
|
|
|
|
|
|
# def test_on_stage(mock_nidaq):
|
|
# dev = mock_nidaq
|
|
# #TODO Add once MockPV supports callbacks #22
|
|
|
|
|
|
def test_on_kickoff(mock_nidaq):
|
|
"""Test the on_kickoff method of the Nidaq device."""
|
|
dev = mock_nidaq
|
|
dev.kickoff_call.put(0)
|
|
dev.kickoff()
|
|
assert dev.kickoff_call.get() == 1
|
|
|
|
|
|
def test_on_unstage(mock_nidaq):
|
|
"""Test the on_unstage method of the Nidaq device."""
|
|
dev = mock_nidaq
|
|
dev.state._read_pv.mock_data = 0 # Set state to 0, 1 is Standby
|
|
dev._timeout_wait_for_pv = 0.1 # Set a short timeout for testing
|
|
dev.enable_compression._read_pv.mock_data = 0 # Compression enabled
|
|
with pytest.raises(WaitTimeoutError):
|
|
dev.on_unstage()
|
|
dev.state._read_pv.mock_data = 1
|
|
# FIXME #22 add callback mechanism to MockPV to test the rest of the logic
|
|
# dev.on_unstage()
|
|
# assert dev.enable_compression.get() == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
["scan_name", "raise_error", "nidaq_state"],
|
|
[
|
|
("line_scan", False, 0),
|
|
("xas_simple_scan", False, 3),
|
|
("xas_simple_scan", True, 0),
|
|
("nidaq_continuous_scan", False, 0),
|
|
],
|
|
)
|
|
def test_on_pre_scan(mock_nidaq, scan_name, raise_error, nidaq_state, scan_info_mock):
|
|
"""Test the on_pre_scan method of the Nidaq device."""
|
|
dev = mock_nidaq
|
|
dev.state.put(nidaq_state)
|
|
scan_info_mock.scan_name = scan_name
|
|
dev.scan_info.msg.info.update(scan_info_mock.model_dump())
|
|
dev.scan_parameters = fetch_scan_info(dev.scan_info)
|
|
dev._timeout_wait_for_pv = 0.1 # Set a short timeout for testing
|
|
if not raise_error:
|
|
dev.pre_scan()
|
|
else:
|
|
with pytest.raises(WaitTimeoutError):
|
|
dev.pre_scan()
|
|
|
|
|
|
def test_on_complete(mock_nidaq, scan_info_mock):
|
|
"""Test the on_complete method of the Nidaq device."""
|
|
dev = mock_nidaq
|
|
scan_info_mock.scan_name = "nidaq_continuous_scan"
|
|
dev.scan_info.msg.info.update(scan_info_mock.model_dump())
|
|
dev.scan_parameters = fetch_scan_info(dev.scan_info)
|
|
# Check for nidaq_continuous_scan
|
|
dev.state.put(0) # Set state to DISABLED
|
|
status = dev.complete()
|
|
assert status.done is False
|
|
dev.state.put(1)
|
|
# Should resolve now
|
|
status.wait(timeout=5) # Wait for the status to complete
|
|
assert status.done is True
|
|
|
|
# Check for XAS simple scan
|
|
scan_info_mock.scan_name = "xas_simple_scan"
|
|
dev.scan_info.msg.info.update(scan_info_mock.model_dump())
|
|
dev.scan_parameters = fetch_scan_info(dev.scan_info)
|
|
dev.state.put(0) # Set state to ACQUIRE
|
|
dev.stop_call.put(0)
|
|
dev._timeout_wait_for_pv = 5
|
|
status = dev.on_complete()
|
|
assert status.done is False
|
|
assert dev.stop_call.get() == 1 # Should have called stop
|
|
dev.state.put(1) # Set state to STANDBY
|
|
# Should resolve now
|
|
status.wait(timeout=5) # Wait for the status to complete
|
|
assert status.done is True
|
|
|
|
# Test that it resolves if device is stopped
|
|
dev.state.put(0) # Set state to DISABLED
|
|
dev.stop()
|
|
status.wait(timeout=5)
|
|
assert status.done is True
|
|
|
|
|
|
########## DynamicSignal statistic groups ##########
|
|
|
|
from ophyd import Kind
|
|
from ophyd_devices.utils.bec_signals import DynamicSignal
|
|
|
|
|
|
def test_statistic_groups_and_omitted_epics(mock_nidaq):
|
|
"""Groups own the unified namespace; EPICS scalars are silent data sources.
|
|
|
|
DynamicSignal is deliberate (per the NIDAQ maintainers, the active channel
|
|
set cannot be predicted upfront): channel names are validated, batch
|
|
completeness is not.
|
|
"""
|
|
dev = mock_nidaq
|
|
assert isinstance(dev.mean, DynamicSignal)
|
|
assert dev.mean.strict_signal_validation is False
|
|
assert dev.mean.name == "nidaq_mean"
|
|
assert dev.max.name == "nidaq_max"
|
|
assert dev.std_dev.name == "nidaq_std_dev"
|
|
assert dev.stream.name == "nidaq_stream"
|
|
# EPICS scalars demoted: no longer in the readout namespace
|
|
assert dev.ai0.kind == Kind.omitted
|
|
assert dev.enc_epics.kind == Kind.omitted
|
|
assert "nidaq_ai0" not in dev.read()
|
|
|
|
|
|
########## software-mode forwarding into the statistic groups ##########
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from ophyd import Staged
|
|
|
|
|
|
def _enable_channels(dev, ai_mask=0b1, ci_mask=0b1000, di_mask=0b110):
|
|
"""Enable ai0, ci3 and di1/di2 via the channel bitmasks."""
|
|
dev.ai_chans._read_pv.mock_data = ai_mask
|
|
dev.ci_chans._read_pv.mock_data = ci_mask
|
|
dev.di_chans._read_pv.mock_data = di_mask
|
|
|
|
|
|
def test_enabled_sw_channels_follow_bitmasks(mock_nidaq):
|
|
dev = mock_nidaq
|
|
_enable_channels(dev, ai_mask=0b101, ci_mask=0b10, di_mask=0b110)
|
|
enabled = dev._enabled_sw_channels()
|
|
assert enabled["mean"] == ["ai0", "ai2", "ci1"]
|
|
assert enabled["max"] == ["di1", "di2"]
|
|
|
|
|
|
def test_read_forwards_enabled_epics_scalars_when_gates_open(mock_nidaq, scan_info_mock):
|
|
"""Software-triggered scans: the monitored readout forwards the ENABLED
|
|
EPICS values into the SAME groups the hardware-triggered stream feeds."""
|
|
dev = mock_nidaq
|
|
_enable_channels(dev) # ai0, ci3, di1, di2
|
|
dev.ai0._read_pv.mock_data = 5.0
|
|
dev.ci3._read_pv.mock_data = 42
|
|
dev.di1._read_pv.mock_data = 0
|
|
dev.di2._read_pv.mock_data = 1
|
|
dev.ai1._read_pv.mock_data = 99.0 # disabled channel, must not be forwarded
|
|
|
|
scan_info_mock.scan_name = "xas_simple_scan"
|
|
dev.scan_parameters = scan_info_mock
|
|
with mock.patch.object(dev.scan_info, "msg", SimpleNamespace(status="open")):
|
|
# not staged -> no forwarding
|
|
dev.read()
|
|
assert dev.mean.get() is None
|
|
|
|
dev._staged = Staged.yes
|
|
# backend-driven scan -> no forwarding (the hardware stream owns the groups)
|
|
dev.read()
|
|
assert dev.mean.get() is None
|
|
|
|
# software-triggered scan -> enabled channels forwarded under the unified names
|
|
scan_info_mock.scan_name = "line_scan"
|
|
dev.read()
|
|
out = dev.mean.get()
|
|
assert out.signals["nidaq_mean_ai0"]["value"] == 5.0
|
|
assert out.signals["nidaq_mean_ci3"]["value"] == 42
|
|
assert "nidaq_mean_ai1" not in out.signals # disabled -> filtered
|
|
assert out.metadata["acquisition_group"] == "monitored"
|
|
assert dev.max.get().signals["nidaq_max_di1"]["value"] == 0
|
|
assert dev.max.get().signals["nidaq_max_di2"]["value"] == 1
|
|
# hardware-only groups untouched in software mode
|
|
assert dev.std_dev.get() is None
|
|
assert dev.stream.get() is None
|
|
|
|
# scan not open -> no forwarding
|
|
dev.ai0._read_pv.mock_data = 7.0
|
|
with mock.patch.object(dev.scan_info, "msg", SimpleNamespace(status="closed")):
|
|
dev.read()
|
|
assert dev.mean.get().signals["nidaq_mean_ai0"]["value"] == 5.0
|
|
dev._staged = Staged.no
|
|
|
|
|
|
def test_guarded_put_reports_unknown_names_once(mock_nidaq, scan_info_mock):
|
|
"""An undeclared channel name is dropped and reported once per group per scan."""
|
|
dev = mock_nidaq
|
|
dev._guarded_group_put("mean", {"bogus": {"value": [1.0], "timestamp": 1.0}})
|
|
assert dev.mean.get() is None
|
|
assert "mean" in dev._group_put_errors
|
|
# staging for the next scan resets the reporting
|
|
scan_info_mock.scan_name = "line_scan"
|
|
dev.scan_info.msg.info.update(scan_info_mock.model_dump())
|
|
dev.on_stage() # returns early: not a backend-driven scan
|
|
assert dev._group_put_errors == set()
|
|
|
|
|
|
def test_group_put_notifies_subscribers(mock_nidaq):
|
|
"""The device server publishes to the per-signal device_async_signal endpoint
|
|
from BECMessageSignal subscriptions; a put must notify subscribers."""
|
|
dev = mock_nidaq
|
|
received = []
|
|
dev.mean.subscribe(lambda *a, **kw: received.append(kw.get("value")), run=False)
|
|
dev._guarded_group_put("mean", {"ai0": {"value": 5.0, "timestamp": 1.0}})
|
|
assert len(received) == 1
|
|
assert received[0].signals["nidaq_mean_ai0"]["value"] == 5.0
|