diff --git a/ophyd_devices/sim/sim_monitor.py b/ophyd_devices/sim/sim_monitor.py index 0aa4a0f..c034c0f 100644 --- a/ophyd_devices/sim/sim_monitor.py +++ b/ophyd_devices/sim/sim_monitor.py @@ -3,8 +3,6 @@ from dataclasses import dataclass import numpy as np -from bec_lib import messages -from bec_lib.endpoints import MessageEndpoints from bec_lib.logger import bec_logger from ophyd import Component as Cpt from ophyd import Device, Kind, StatusBase @@ -13,6 +11,7 @@ from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase from ophyd_devices.sim.sim_data import SimulatedDataMonitor from ophyd_devices.sim.sim_signals import ReadOnlySignal, SetableSignal from ophyd_devices.utils import bec_utils +from ophyd_devices.utils.bec_signals import AsyncSignal, ProgressSignal logger = bec_logger.logger @@ -131,9 +130,10 @@ class SimMonitorAsyncControl(Device): readback = Cpt(ReadOnlySignal, value=BIT_DEPTH(0), kind=Kind.hinted, compute_readback=True) current_trigger = Cpt(SetableSignal, value=BIT_DEPTH(0), kind=Kind.config) async_update = Cpt(SetableSignal, value="extend", kind=Kind.config) + data = Cpt(AsyncSignal, ndim=1, max_size=1000, doc="Buffered asynchronous readings") + progress = Cpt(ProgressSignal, doc="Scan progress") SUB_READBACK = "readback" - SUB_PROGRESS = "progress" _default_sub = SUB_READBACK def __init__(self, name, *, sim_init: dict = None, parent=None, device_manager=None, **kwargs): @@ -141,7 +141,6 @@ class SimMonitorAsyncControl(Device): self.device_manager = device_manager else: self.device_manager = bec_utils.DMMock() - self.connector = self.device_manager.connector self.sim_init = sim_init self.sim = self.sim_cls(parent=self, **kwargs) self._registered_proxies = {} @@ -178,7 +177,6 @@ class SimMonitorAsync(PSIDeviceBase, SimMonitorAsyncControl): super().__init__( name=name, scan_info=scan_info, parent=parent, device_manager=device_manager, **kwargs ) - self._stream_ttl = 1800 self._random_send_interval = None self._counter = 0 self.prep_random_interval() @@ -213,26 +211,15 @@ class SimMonitorAsync(PSIDeviceBase, SimMonitorAsyncControl): return status def _send_data_to_bec(self) -> None: - """Sends bundled data to BEC""" + """Emit buffered readings through the asynchronous data signal.""" async_update = self.async_update.get() if async_update not in ["extend", "append"]: raise ValueError(f"Invalid async_update value for device {self.name}: {async_update}") - metadata = None - if async_update == "extend": - metadata = {"async_update": {"type": "add", "max_shape": [None]}} - elif async_update == "append": - metadata = {"async_update": {"type": "add", "max_shape": [None, None]}} - - msg = messages.DeviceMessage( - signals={self.readback.name: self.data_buffer}, metadata=metadata - ) - self.connector.xadd( - MessageEndpoints.device_async_readback( - scan_id=self.scan_info.msg.scan_id, device=self.name - ), - {"data": msg}, - expire=self._stream_ttl, + self.data.put( + self.data_buffer["value"].copy(), + timestamp=self.data_buffer["timestamp"][-1], + async_update={"type": "add", "max_shape": [None]}, ) self.clear_buffer() @@ -256,13 +243,7 @@ class SimMonitorAsync(PSIDeviceBase, SimMonitorAsyncControl): if not self.scan_info.msg: return max_value = self.scan_info.msg.num_points - # pylint: disable=protected-access - self._run_subs( - sub_type=self.SUB_PROGRESS, - value=value, - max_value=max_value, - done=bool(max_value == value), - ) + self.progress.put(value=value, max_value=max_value, done=bool(max_value == value)) def on_stop(self): """Stop the device.""" diff --git a/tests/test_simulation.py b/tests/test_simulation.py index e0b9ac8..817603c 100644 --- a/tests/test_simulation.py +++ b/tests/test_simulation.py @@ -251,6 +251,7 @@ def test_init_async_monitor(async_monitor): """Test the __init__ method of SimMonitorAsync.""" assert isinstance(async_monitor, SimMonitorAsync) assert isinstance(async_monitor, BECDeviceProtocol) + assert async_monitor.data.describe()[async_monitor.data.name]["signal_info"]["ndim"] == 1 def test_signal_delay_device_start_resets_trigger_and_reaches_max(signal_delay_device): @@ -785,28 +786,37 @@ def test_async_mon_on_trigger(async_monitor): assert mock_send.call_count == 1 -def test_async_mon_send_data_to_bec(async_monitor): - """Test the _send_data_to_bec method of SimMonitorAsync.""" +@pytest.mark.parametrize("mode", ["extend", "append"]) +def test_async_mon_send_data_to_bec(async_monitor, mode): + """Buffered readings are emitted through a BEC message signal.""" async_monitor.scan_info = get_mock_scan_info(device=async_monitor) - async_monitor.data_buffer.update({"value": [0, 5], "timestamp": [0, 0]}) - with mock.patch.object(async_monitor.connector, "xadd") as mock_xadd: - async_monitor._send_data_to_bec() - dev_msg = messages.DeviceMessage( - signals={async_monitor.readback.name: async_monitor.data_buffer}, + async_monitor.async_update.put(mode) + async_monitor.data_buffer.update({"value": [0, 5], "timestamp": [1, 2]}) + updates = [] + async_monitor.data.subscribe(lambda value, **kwargs: updates.append(value), run=False) + async_monitor._send_data_to_bec() + assert updates == [ + messages.DeviceMessage( + signals={async_monitor.data.name: {"value": [0, 5], "timestamp": 2}}, metadata={"async_update": {"type": "add", "max_shape": [None]}}, ) + ] + assert async_monitor.data_buffer == {"value": [], "timestamp": []} - call = [ - mock.call( - MessageEndpoints.device_async_readback( - scan_id=async_monitor.scan_info.msg.scan_id, device=async_monitor.name - ), - {"data": dev_msg}, - expire=async_monitor._stream_ttl, - ) - ] - assert mock_xadd.mock_calls == call - assert async_monitor.data_buffer["value"] == [] + +def test_async_monitor_progress_signal(async_monitor): + """Trigger progress is emitted through the progress signal.""" + async_monitor.scan_info = get_mock_scan_info(device=async_monitor) + updates = [] + async_monitor.progress.subscribe(lambda value, **kwargs: updates.append(value), run=False) + async_monitor._progress_update(1) + assert updates == [ + messages.ProgressMessage( + value=1, + max_value=async_monitor.scan_info.msg.num_points, + done=async_monitor.scan_info.msg.num_points == 1, + ) + ] def test_positioner_updated_timestamp(positioner):