fix(monitor async): move from SUB_ to BECMessageSignal

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."""

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):