bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-06-05 19:30:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: improve logging if fake scan msg is created

Browse Source
This commit is contained in:
appel_c 2025-02-25 11:34:45 +01:00
parent 42f6cb6ed3
commit 75f32800f8
3 changed files with 16 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ophyd_devices/interfaces/base_classes/psi_device_base.py
View File

@ -42,7 +42,7 @@ class PSIDeviceBase(Device):
self.task_handler = TaskHandler(parent=self) self.task_handler = TaskHandler(parent=self)
self.file_utils = FileHandler() self.file_utils = FileHandler()
if scan_info is None: if scan_info is None:
scan_info = get_mock_scan_info() scan_info = get_mock_scan_info(device=self)
self.scan_info = scan_info self.scan_info = scan_info
self.on_init() self.on_init()

18
ophyd_devices/tests/utils.py
View File

@ -1,12 +1,12 @@
""" Utilities to mock and test devices.""" """ Utilities to mock and test devices."""
from dataclasses import dataclass
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from unittest import mock from unittest import mock
from bec_lib.devicemanager import ScanInfo from bec_lib.devicemanager import ScanInfo
from bec_lib.logger import bec_logger from bec_lib.logger import bec_logger
from bec_lib.utils.import_utils import lazy_import_from from bec_lib.utils.import_utils import lazy_import_from
from ophyd import Device
if TYPE_CHECKING: if TYPE_CHECKING:
from bec_lib.messages import ScanStatusMessage from bec_lib.messages import ScanStatusMessage
@ -273,19 +273,27 @@ class MockPV:
return data["value"] if data is not None else None return data["value"] if data is not None else None
def get_mock_scan_info(): def get_mock_scan_info(device: Device | None) -> ScanInfo:
""" """
Get a mock scan info object. Get a mock scan info object.
""" """
return ScanInfo(msg=fake_scan_status_msg()) return ScanInfo(msg=fake_scan_status_msg(device=device))
def fake_scan_status_msg(): def fake_scan_status_msg(device: Device | None = None) -> ScanStatusMessage:
""" """
Create a fake scan status message. Create a fake scan status message.
Args:
device: The device creating the fake scan status message.
""" """
if device is None:
device = Device(name="mock_device")
logger.warning( logger.warning(
("Device is not connected to a Redis server. Fetching mocked ScanStatusMessage.") (
f"Device {device.name} is not connected to a Redis server. Fetching mocked ScanStatusMessage."
)
) )
return ScanStatusMessage( return ScanStatusMessage(
metadata={}, metadata={},

4
tests/test_simulation.py
View File

@ -697,7 +697,7 @@ def test_async_mon_on_trigger(async_monitor):
def test_async_mon_send_data_to_bec(async_monitor): def test_async_mon_send_data_to_bec(async_monitor):
"""Test the _send_data_to_bec method of SimMonitorAsync.""" """Test the _send_data_to_bec method of SimMonitorAsync."""
async_monitor.scan_info = get_mock_scan_info() async_monitor.scan_info = get_mock_scan_info(device=async_monitor)
async_monitor.data_buffer.update({"value": [0, 5], "timestamp": [0, 0]}) async_monitor.data_buffer.update({"value": [0, 5], "timestamp": [0, 0]})
with mock.patch.object(async_monitor.connector, "xadd") as mock_xadd: with mock.patch.object(async_monitor.connector, "xadd") as mock_xadd:
async_monitor._send_data_to_bec() async_monitor._send_data_to_bec()
@ -747,7 +747,7 @@ def test_waveform(waveform):
# Now also test the async readback # Now also test the async readback
mock_connector = waveform.connector = mock.MagicMock() mock_connector = waveform.connector = mock.MagicMock()
mock_run_subs = waveform._run_subs = mock.MagicMock() mock_run_subs = waveform._run_subs = mock.MagicMock()
waveform.scan_info = get_mock_scan_info() waveform.scan_info = get_mock_scan_info(device=waveform)
waveform.scan_info.msg.scan_id = "test" waveform.scan_info.msg.scan_id = "test"
status = waveform.trigger() status = waveform.trigger()
timer = 0 timer = 0