refactor(widgets): port multi waveform and motor map to device streams

2026-09-07 00:42:37 +02:00 · 2026-08-14 12:00:27 +02:00
parent 83e5b5f2b8
commit 192d0d74de
5 changed files with 500 additions and 105 deletions
@@ -32,9 +32,10 @@ class QtDataSubscription(QObject):
        self,
        client: BECClient,
        sources: list[SourceKey],
-        scan: str = "live",
+        scan: str | None = "live",
        parent: QObject | None = None,
        min_emit_interval: float = 0.1,
+        max_points: int | None = None,
    ):
        """
        Subscribe to data for the given sources.
@@ -43,11 +44,16 @@ class QtDataSubscription(QObject):
            client (BECClient): The widget's BEC client.
            sources (list[SourceKey]): (device, entry) pairs forming one
                correlation group.
-            scan (str): ``"live"`` to follow the active scan, or a concrete
-                (possibly finished) scan id.
+            scan (str | None): ``"live"`` to follow the active scan, a
+                concrete (possibly finished) scan id, or ``None`` for
+                scan-less device streams (readback, ``"monitor_1d"``,
+                preview signals).
            parent (QObject | None): Qt parent; closing follows the parent's
                destruction.
            min_emit_interval (float): Backend emission coalescing interval.
+            max_points (int | None): Per-source retention cap; oldest points
+                are dropped beyond it. Recommended for endless device-stream
+                subscriptions (``scan=None``).

        Raises:
            ValueError: If a concrete scan id cannot be served.
@@ -63,7 +69,11 @@ class QtDataSubscription(QObject):
        self._raw.connect(self._filter, Qt.QueuedConnection)
        self._api = DataAPI(client)
        self._subscription = self._api.subscribe(
-            sources=sources, scan=scan, callback=self._deliver, min_emit_interval=min_emit_interval
+            sources=sources,
+            scan=scan,
+            callback=self._deliver,
+            min_emit_interval=min_emit_interval,
+            max_points=max_points,
        )
        self.destroyed.connect(lambda: self.close())

@@ -13,6 +13,7 @@ from qtpy.QtWidgets import QHBoxLayout, QMainWindow, QWidget
 from bec_widgets.utils.bec_connector import ConnectionConfig
 from bec_widgets.utils.colors import Colors, apply_theme
 from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
+from bec_widgets.utils.qt_data_subscription import QtDataSubscription
 from bec_widgets.utils.settings_dialog import SettingsDialog
 from bec_widgets.utils.toolbars.toolbar import MaterialIconAction
 from bec_widgets.widgets.plots.motor_map.settings.motor_map_settings import MotorMapSettings
@@ -146,6 +147,11 @@ class MotorMap(PlotBase):
        self.coord_label = None
        self.motor_map_settings = None

+        # Data delivery through the DataAPI device streams (readback).
+        self._data_bridge: QtDataSubscription | None = None
+        self._connected_limit_endpoints: list = []
+        self._last_ordinals: dict[str, int | None] = {"x": None, "y": None}
+
        # Connect slots
        self.proxy_update_plot = pg.SignalProxy(
            self.update_signal, rateLimit=25, slot=self._update_plot
@@ -565,62 +571,123 @@ class MotorMap(PlotBase):
        # Update the crosshair
        self._set_motor_indicator_position(current_x, current_y)

-    @SafeSlot(dict, dict)
-    def on_device_readback(self, msg: dict, metadata: dict) -> None:
+    @SafeSlot(object)
+    def _on_data_update(self, update) -> None:
        """
-        Update the motor map plot with the new motor position.
+        Update the motor map trail from one columnar DataAPI update.
+
+        Each motor readback is a standalone device-stream source, so every
+        update carries exactly one motor's series. Ordinals are arrival
+        counters; positions newer than the last consumed ordinal are appended
+        with the other axis forward-filled from its last known position —
+        the same trail the legacy per-message handler produced.

        Args:
-            msg(dict): Message from the device readback.
-            metadata(dict): Metadata of the message.
+            update (SubscriptionUpdate): Full-state snapshot of one motor's
+                readback stream (standalone group).
        """
        device_x = self.config.device_x.device
        device_y = self.config.device_y.device

        if device_x is None or device_y is None:
            return
+        if not self._buffer["x"] or not self._buffer["y"]:
+            # Not seeded yet (map() seeds via dev[name].read(cached=True)).
+            return

-        if device_x in msg["signals"]:
-            x = msg["signals"][device_x]["value"]
-            self._buffer["x"].append(x)
-            self._buffer["y"].append(self._buffer["y"][-1])
+        source_x = update.get(device_x, device_x)
+        if source_x is not None:
+            appended = self._append_new_positions(source_x, axis="x")
+        else:
+            source_y = update.get(device_y, device_y)
+            if source_y is None:
+                return
+            appended = self._append_new_positions(source_y, axis="y")

-        elif device_y in msg["signals"]:
-            y = msg["signals"][device_y]["value"]
-            self._buffer["y"].append(y)
-            self._buffer["x"].append(self._buffer["x"][-1])
+        if appended:
+            self.update_signal.emit()

-        self.update_signal.emit()
+    def _append_new_positions(self, source, axis: str) -> bool:
+        """
+        Append the not-yet-consumed positions of one motor to the trail buffer.
+
+        Args:
+            source (SourceData): Columnar snapshot of the motor readback.
+            axis (str): "x" or "y" — the buffer axis fed by this source.
+
+        Returns:
+            bool: True if at least one position was appended.
+        """
+        other = "y" if axis == "x" else "x"
+        last_ordinal = self._last_ordinals[axis]
+        appended = False
+        for ordinal, value in zip(source.ordinals, source.values):
+            if last_ordinal is not None and ordinal <= last_ordinal:
+                continue
+            self._buffer[axis].append(value)
+            self._buffer[other].append(self._buffer[other][-1])
+            last_ordinal = ordinal
+            appended = True
+        self._last_ordinals[axis] = last_ordinal
+        return appended

    def _connect_motor_to_slots(self):
-        """Connect motors to slots."""
+        """Connect the motors to the DataAPI stream and the limits dispatcher."""
        self._disconnect_current_motors()

-        endpoints_readback = [
-            MessageEndpoints.device_readback(self.config.device_x.device),
-            MessageEndpoints.device_readback(self.config.device_y.device),
-        ]
+        # device_limits is config metadata (not a data series) and stays on
+        # the dispatcher.
        endpoints_limits = [
            MessageEndpoints.device_limits(self.config.device_x.device),
            MessageEndpoints.device_limits(self.config.device_y.device),
        ]
-
-        self.bec_dispatcher.connect_slot(self.on_device_readback, endpoints_readback)
        self.bec_dispatcher.connect_slot(self.on_device_limits, endpoints_limits)
+        self._connected_limit_endpoints = endpoints_limits
+
+        self._setup_data_api_subscription()

    def _disconnect_current_motors(self):
-        """Disconnect the current motors from the slots."""
-        if self.config.device_x.device is not None and self.config.device_y.device is not None:
-            endpoints_readback = [
-                MessageEndpoints.device_readback(self.config.device_x.device),
-                MessageEndpoints.device_readback(self.config.device_y.device),
-            ]
-            endpoints_limits = [
-                MessageEndpoints.device_limits(self.config.device_x.device),
-                MessageEndpoints.device_limits(self.config.device_y.device),
-            ]
-            self.bec_dispatcher.disconnect_slot(self.on_device_readback, endpoints_readback)
-            self.bec_dispatcher.disconnect_slot(self.on_device_limits, endpoints_limits)
+        """Disconnect the current motors from the data stream and the limits slot.
+
+        Disconnects the endpoints that were actually connected: at this point
+        the config may already carry the new motor names.
+        """
+        self._cleanup_data_api_subscription()
+        if self._connected_limit_endpoints:
+            self.bec_dispatcher.disconnect_slot(
+                self.on_device_limits, self._connected_limit_endpoints
+            )
+            self._connected_limit_endpoints = []
+
+    def _setup_data_api_subscription(self):
+        """(Re)create the scan-less DataAPI subscription for the configured motors."""
+        self._cleanup_data_api_subscription()
+        device_x = self.config.device_x.device
+        device_y = self.config.device_y.device
+        if not device_x or not device_y:
+            return
+        try:
+            self._data_bridge = QtDataSubscription(
+                self.client,
+                sources=[(device_x, device_x), (device_y, device_y)],
+                scan=None,
+                parent=self,
+                min_emit_interval=0.1,
+                max_points=self.config.max_points,
+            )
+            self._data_bridge.updated.connect(self._on_data_update)
+        except Exception as exc:
+            logger.warning(f"Failed to configure motor map data subscription: {exc}")
+            self._cleanup_data_api_subscription()
+
+    def _cleanup_data_api_subscription(self):
+        self._last_ordinals = {"x": None, "y": None}
+        if self._data_bridge is None:
+            return
+        try:
+            self._data_bridge.close()
+        finally:
+            self._data_bridge = None

    ################################################################################
    # Utility Methods
@@ -853,6 +920,19 @@ class MotorMap(PlotBase):
        data = {"x": self._buffer["x"], "y": self._buffer["y"]}
        return data

+    ################################################################################
+    # Cleanup
+    ################################################################################
+
+    def cleanup(self):
+        """
+        Cleanup the widget: close the DataAPI bridge. The limits dispatcher
+        slot is released by BECWidget.cleanup via disconnect_owner (the legacy
+        readback slot relied on the same owner cleanup).
+        """
+        self._cleanup_data_api_subscription()
+        super().cleanup()
+

 class DemoApp(QMainWindow):  # pragma: no cover
    def __init__(self):
@@ -4,7 +4,6 @@ from collections import deque
 from typing import TYPE_CHECKING, cast

 import pyqtgraph as pg
-from bec_lib.endpoints import MessageEndpoints
 from bec_lib.logger import bec_logger
 from pydantic import Field, ValidationError, field_validator
 from qtpy.QtCore import Signal
@@ -13,6 +12,7 @@ from qtpy.QtWidgets import QWidget
 from bec_widgets.utils.bec_connector import ConnectionConfig
 from bec_widgets.utils.colors import Colors
 from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
+from bec_widgets.utils.qt_data_subscription import QtDataSubscription
 from bec_widgets.utils.side_panel import SidePanel
 from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
 from bec_widgets.widgets.plots.multi_waveform.settings.control_panel import (
@@ -106,6 +106,10 @@ class MultiWaveform(PlotBase):
        self.visible_curves = []
        self.number_of_visible_curves = 0

+        # Data delivery through the DataAPI device stream (monitor_1d).
+        self._data_bridge: QtDataSubscription | None = None
+        self._last_ordinal: int | None = None
+
        self._init_multiwaveform_toolbar()

    ################################################################################
@@ -304,9 +308,8 @@ class MultiWaveform(PlotBase):
            color_palette (str|None): The color palette to use for the plot.
        """
        self.entry_validator.validate_monitor(monitor)
-        self._disconnect_monitor()
        self.config.monitor = monitor
-        self._connect_monitor()
+        self._setup_data_api_subscription()
        if color_palette is not None:
            self.color_palette = color_palette
        self._sync_monitor_selection_toolbar()
@@ -348,18 +351,27 @@ class MultiWaveform(PlotBase):
    ################################################################################
    # BEC Update Methods
    ################################################################################
-    @SafeSlot(dict, dict)
-    def on_monitor_1d_update(self, msg: dict, metadata: dict):
+    @SafeSlot(object)
+    def _on_data_update(self, update) -> None:
        """
-        Update the plot widget with the monitor data.
+        Render one columnar DataAPI update of the monitor_1d device stream.
+
+        Each value of the source is one 1-D trace (newest last); ordinals are
+        arrival counters, so only traces newer than the last rendered ordinal
+        are appended to the curve deque.

        Args:
-            msg(dict): The message data.
-            metadata(dict): The metadata of the message.
+            update (SubscriptionUpdate): Full-state snapshot of the monitor
+                stream (standalone group).
        """
-        data = msg.get("data", None)
-        current_scan_id = metadata.get("scan_id", None)
+        monitor = self.config.monitor
+        if not monitor:
+            return
+        source = update.get(monitor, "monitor_1d")
+        if source is None or not source.values:
+            return

+        current_scan_id = source.metadata.get("scan_id", None)
        if current_scan_id != self.scan_id:
            self.scan_id = current_scan_id
            self.clear_curves()
@@ -367,11 +379,21 @@ class MultiWaveform(PlotBase):
            if self.crosshair:
                self.crosshair.clear_markers()

-        # Always create a new curve and add it
-        curve = pg.PlotDataItem()
-        curve.setData(data)
-        self.plot_item.addItem(curve)
-        self.curves.append(curve)
+        last_ordinal = self._last_ordinal
+        new_traces = [
+            (ordinal, data)
+            for ordinal, data in zip(source.ordinals, source.values)
+            if last_ordinal is None or ordinal > last_ordinal
+        ]
+        if not new_traces:
+            return
+
+        for _, data in new_traces:
+            curve = pg.PlotDataItem()
+            curve.setData(data)
+            self.plot_item.addItem(curve)
+            self.curves.append(curve)
+        self._last_ordinal = new_traces[-1][0]

        # Max Trace and scale colors
        self.set_curve_limit(self.config.curve_limit, self.config.flush_buffer)
@@ -413,28 +435,36 @@ class MultiWaveform(PlotBase):

        self.highlighted_curve_index_changed.emit(self._current_highlight_index)

-    def _disconnect_monitor(self):
+    def _setup_data_api_subscription(self):
+        """(Re)create the scan-less DataAPI subscription for the configured monitor."""
+        self._cleanup_data_api_subscription()
+        monitor = self.config.monitor
+        if not monitor:
+            return
        try:
-            previous_monitor = self.config.monitor
-        except AttributeError:
-            previous_monitor = None
-
-        if previous_monitor and self.connected is True:
-            self.bec_dispatcher.disconnect_slot(
-                self.on_monitor_1d_update, MessageEndpoints.device_monitor_1d(previous_monitor)
-            )
-            self.connected = False
-
-    def _connect_monitor(self):
-        """
-        Connect the monitor to the plot widget.
-        """
-
-        if self.config.monitor and self.connected is False:
-            self.bec_dispatcher.connect_slot(
-                self.on_monitor_1d_update, MessageEndpoints.device_monitor_1d(self.config.monitor)
+            self._data_bridge = QtDataSubscription(
+                self.client,
+                sources=[(monitor, "monitor_1d")],
+                scan=None,
+                parent=self,
+                min_emit_interval=0.1,
+                max_points=self.config.curve_limit,
            )
+            self._data_bridge.updated.connect(self._on_data_update)
            self.connected = True
+        except Exception as exc:
+            logger.warning(f"Failed to configure multi waveform data subscription: {exc}")
+            self._cleanup_data_api_subscription()
+
+    def _cleanup_data_api_subscription(self):
+        self._last_ordinal = None
+        self.connected = False
+        if self._data_bridge is None:
+            return
+        try:
+            self._data_bridge.close()
+        finally:
+            self._data_bridge = None

    ################################################################################
    # Utility Methods
@@ -498,6 +528,6 @@ class MultiWaveform(PlotBase):
            cmap_widget.blockSignals(False)

    def cleanup(self):
-        self._disconnect_monitor()
+        self._cleanup_data_api_subscription()
        self.clear_curves()
        super().cleanup()
@@ -1,3 +1,6 @@
+from unittest.mock import MagicMock
+
+from bec_lib.data_api.models import SourceData, SubscriptionUpdate
 from qtpy.QtTest import QSignalSpy

 from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
@@ -5,6 +8,60 @@ from tests.unit_tests.client_mocks import mocked_client

 from .conftest import create_widget

+##################################################
+# Test helpers (DataAPI fake bridge + updates)
+##################################################
+
+
+def _fake_bridge_factory(monkeypatch):
+    created = []
+
+    class _FakeBridge:
+        def __init__(
+            self, client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None
+        ):
+            self.client = client
+            self.sources = list(sources)
+            self.scan = scan
+            self.max_points = max_points
+            self.healthy = True
+            self.closed = False
+            self.updated = MagicMock()
+
+        def close(self):
+            self.closed = True
+
+    def factory(client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None):
+        bridge = _FakeBridge(client, sources, scan=scan, max_points=max_points)
+        created.append(bridge)
+        return bridge
+
+    monkeypatch.setattr("bec_widgets.widgets.plots.motor_map.motor_map.QtDataSubscription", factory)
+    return created
+
+
+def _readback_update(motor, values, start=0):
+    """Build a full-state readback snapshot for one motor (standalone group)."""
+    ordinals = tuple(range(start, start + len(values)))
+    source = SourceData(
+        device=motor,
+        entry=motor,
+        kind="unindexed",
+        ordinals=ordinals,
+        values=tuple(values),
+        timestamps=tuple(float(ordinal) for ordinal in ordinals),
+        complete=True,
+        metadata={"stream": "readback"},
+    )
+    return SubscriptionUpdate(
+        scan_id="",
+        reason="live",
+        sources={(motor, motor): source},
+        aligned_ordinals=ordinals,
+        complete=True,
+        metadata={"group": f"standalone:{motor}/{motor}"},
+    )
+

 def test_motor_map_initialization(qtbot, mocked_client):
    """Test the initialization of the MotorMap widget."""
@@ -130,34 +187,83 @@ def test_motor_map_reset_history(qtbot, mocked_client):
    assert mm._buffer["y"][0] == 8.0


-def test_motor_map_on_device_readback(qtbot, mocked_client):
-    """Test the motor map updates when receiving device readback."""
+def test_motor_map_on_data_update(qtbot, mocked_client):
+    """Test the motor map updates when receiving one motor's readback stream update."""
    mm = create_widget(qtbot, MotorMap, client=mocked_client)
    mm.map(device_x="samx", device_y="samy")

    # Clear the buffer and add initial position
    mm._buffer = {"x": [1.0], "y": [2.0]}
+    mm._last_ordinals = {"x": None, "y": None}

-    # Simulate device readback for x motor
-    msg_x = {"signals": {"samx": {"value": 3.0}}}
-    mm.on_device_readback(msg_x, {})
+    # Simulate a readback stream update for the x motor (standalone group ->
+    # each update carries exactly one motor's series)
+    mm._on_data_update(_readback_update("samx", [3.0]))
    qtbot.wait(200)  # Allow time for the update to process

    assert len(mm._buffer["x"]) == 2
    assert len(mm._buffer["y"]) == 2
    assert mm._buffer["x"][1] == 3.0
-    assert mm._buffer["y"][1] == 2.0  # Y should remain the same
+    assert mm._buffer["y"][1] == 2.0  # Y should remain the same (forward fill)

-    # Simulate device readback for y motor
-    msg_y = {"signals": {"samy": {"value": 4.0}}}
-    mm.on_device_readback(msg_y, {})
+    # Simulate a readback stream update for the y motor
+    mm._on_data_update(_readback_update("samy", [4.0]))

    assert len(mm._buffer["x"]) == 3
    assert len(mm._buffer["y"]) == 3
-    assert mm._buffer["x"][2] == 3.0  # X should remain the same
+    assert mm._buffer["x"][2] == 3.0  # X should remain the same (forward fill)
    assert mm._buffer["y"][2] == 4.0


+def test_motor_map_snapshot_ordinal_filtering(qtbot, mocked_client):
+    """Full-state snapshots must not duplicate already-consumed positions."""
+    mm = create_widget(qtbot, MotorMap, client=mocked_client)
+    mm.map(device_x="samx", device_y="samy")
+
+    mm._buffer = {"x": [1.0], "y": [2.0]}
+    mm._last_ordinals = {"x": None, "y": None}
+
+    mm._on_data_update(_readback_update("samx", [3.0]))
+    assert mm._buffer["x"] == [1.0, 3.0]
+
+    # Same snapshot delivered again (trailing coalesced emission): no change.
+    mm._on_data_update(_readback_update("samx", [3.0]))
+    assert mm._buffer["x"] == [1.0, 3.0]
+    assert mm._buffer["y"] == [2.0, 2.0]
+
+    # Snapshot grows: only the new position is appended.
+    mm._on_data_update(_readback_update("samx", [3.0, 3.5]))
+    assert mm._buffer["x"] == [1.0, 3.0, 3.5]
+    assert mm._buffer["y"] == [2.0, 2.0, 2.0]
+
+    # Retention window slid (oldest dropped): still only newer ordinals count.
+    mm._on_data_update(_readback_update("samx", [3.5, 4.0], start=1))
+    assert mm._buffer["x"] == [1.0, 3.0, 3.5, 4.0]
+    assert mm._buffer["y"] == [2.0, 2.0, 2.0, 2.0]
+
+
+def test_motor_map_bridge_lifecycle(qtbot, mocked_client, monkeypatch):
+    """map() creates one scan-less readback bridge; re-map replaces it."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mm = create_widget(qtbot, MotorMap, client=mocked_client)
+
+    mm.map(device_x="samx", device_y="samy")
+    assert len(bridges) == 1
+    assert bridges[-1].sources == [("samx", "samx"), ("samy", "samy")]
+    assert bridges[-1].scan is None
+    assert bridges[-1].max_points == mm.config.max_points
+
+    first = bridges[-1]
+    mm.map(device_x="samx", device_y="samz")
+    assert first.closed is True
+    assert bridges[-1].sources == [("samx", "samx"), ("samz", "samz")]
+
+    mm._cleanup_data_api_subscription()
+    assert bridges[-1].closed is True
+    assert mm._data_bridge is None
+    assert mm._last_ordinals == {"x": None, "y": None}
+
+
 def test_motor_map_max_points_limit(qtbot, mocked_client):
    """Test that the buffer doesn't exceed max_points."""
    mm = create_widget(qtbot, MotorMap, client=mocked_client)
@@ -380,3 +486,18 @@ def test_motor_map_settings_dialog(qtbot, mocked_client):
    mm.motor_map_settings.close()
    qtbot.wait(200)
    assert mm.motor_map_settings is None
+
+
+def test_motor_switch_disconnects_old_limit_endpoints(qtbot, mocked_client):
+    """Switching motors must disconnect the previously connected limits
+    endpoints, not the ones derived from the already-updated config."""
+    from unittest import mock
+
+    mm = create_widget(qtbot, MotorMap, client=mocked_client)
+    mm.map("samx", "samy")
+    first_endpoints = list(mm._connected_limit_endpoints)
+    assert first_endpoints
+
+    with mock.patch.object(mm.bec_dispatcher, "disconnect_slot") as disc:
+        mm.map("samx", "samz")
+    disc.assert_called_once_with(mm.on_device_limits, first_endpoints)
@@ -1,10 +1,70 @@
+from unittest.mock import MagicMock
+
 import numpy as np
+from bec_lib.data_api.models import SourceData, SubscriptionUpdate

 from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform
 from tests.unit_tests.client_mocks import mocked_client

 from .conftest import create_widget

+##################################################
+# Test helpers (DataAPI fake bridge + updates)
+##################################################
+
+
+def _fake_bridge_factory(monkeypatch):
+    created = []
+
+    class _FakeBridge:
+        def __init__(
+            self, client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None
+        ):
+            self.client = client
+            self.sources = list(sources)
+            self.scan = scan
+            self.max_points = max_points
+            self.healthy = True
+            self.closed = False
+            self.updated = MagicMock()
+
+        def close(self):
+            self.closed = True
+
+    def factory(client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None):
+        bridge = _FakeBridge(client, sources, scan=scan, max_points=max_points)
+        created.append(bridge)
+        return bridge
+
+    monkeypatch.setattr(
+        "bec_widgets.widgets.plots.multi_waveform.multi_waveform.QtDataSubscription", factory
+    )
+    return created
+
+
+def _monitor_update(traces, scan_id="scan_1", monitor="waveform1d", start=0):
+    """Build a full-state monitor_1d snapshot: one 1-D trace per value, newest last."""
+    ordinals = tuple(range(start, start + len(traces)))
+    source = SourceData(
+        device=monitor,
+        entry="monitor_1d",
+        kind="unindexed",
+        ordinals=ordinals,
+        values=tuple(traces),
+        timestamps=tuple(float(ordinal) for ordinal in ordinals),
+        complete=True,
+        metadata={"stream": "monitor_1d", "scan_id": scan_id},
+    )
+    return SubscriptionUpdate(
+        scan_id="",
+        reason="live",
+        sources={(monitor, "monitor_1d"): source},
+        aligned_ordinals=ordinals,
+        complete=True,
+        metadata={"group": f"standalone:{monitor}/monitor_1d"},
+    )
+
+
 ##################################################
 # MultiWaveform widget base functionality tests
 ##################################################
@@ -32,17 +92,44 @@ def test_multiwaveform_initialization(qtbot, mocked_client):
    assert mw.opacity == 50
    assert mw.scan_id is None
    assert mw.highlighted_index == 0
+    assert mw._data_bridge is None


 def test_multiwaveform_set_monitor(qtbot, mocked_client):
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
    assert mw.monitor is None

-    # Set a monitor
+    # Set a monitor; data flows through a scan-less DataAPI subscription.
    mw.plot("waveform1d")
    assert mw.monitor == "waveform1d"
    assert mw.config.monitor == "waveform1d"
    assert mw.connected is True
+    assert mw._data_bridge is not None
+    assert mw._data_bridge.sources == [("waveform1d", "monitor_1d")]
+    assert mw._data_bridge.scan_id == ""  # device scope
+
+
+def test_multiwaveform_bridge_lifecycle(qtbot, mocked_client, monkeypatch):
+    """plot() creates a scan-less bridge bounded by the curve limit; re-plot replaces it."""
+    bridges = _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+
+    mw.plot("waveform1d")
+    assert len(bridges) == 1
+    assert bridges[-1].sources == [("waveform1d", "monitor_1d")]
+    assert bridges[-1].scan is None
+    assert bridges[-1].max_points == mw.config.curve_limit
+    assert mw.connected is True
+
+    first = bridges[-1]
+    mw.plot("bpm4i")
+    assert first.closed is True
+    assert bridges[-1].sources == [("bpm4i", "monitor_1d")]
+
+    mw._cleanup_data_api_subscription()
+    assert bridges[-1].closed is True
+    assert mw.connected is False
+    assert mw._data_bridge is None


 def test_multiwaveform_set_properties(qtbot, mocked_client):
@@ -71,16 +158,19 @@ def test_multiwaveform_set_properties(qtbot, mocked_client):
    assert mw.opacity == 75


-def test_multiwaveform_curve_limit_no_flush(qtbot, mocked_client):
+def test_multiwaveform_curve_limit_no_flush(qtbot, mocked_client, monkeypatch):
    """Check that limiting the number of curves without flush simply hides older ones."""
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
    mw.max_trace = 3
    mw.flush_buffer = False

-    # Simulate updates that create multiple curves
+    # Simulate updates that create multiple curves (snapshots grow, newest last)
+    traces = []
    for i in range(5):
-        msg_data = {"data": np.array([i, i + 0.5, i + 1])}
-        mw.on_monitor_1d_update(msg_data, metadata={"scan_id": "scan_1"})
+        traces.append(np.array([i, i + 0.5, i + 1]))
+        mw._on_data_update(_monitor_update(list(traces)))

    # There should be 5 curves in total, but only the last 3 are visible
    assert len(mw.curves) == 5
@@ -88,16 +178,19 @@ def test_multiwaveform_curve_limit_no_flush(qtbot, mocked_client):
    assert len(visible_curves) == 3


-def test_multiwaveform_curve_limit_flush(qtbot, mocked_client):
+def test_multiwaveform_curve_limit_flush(qtbot, mocked_client, monkeypatch):
    """Check that limiting the number of curves with flush removes older ones."""
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
    mw.max_trace = 3
    mw.flush_buffer = True

    # Simulate adding multiple curves
+    traces = []
    for i in range(5):
-        msg_data = {"data": np.array([i, i + 0.5, i + 1])}
-        mw.on_monitor_1d_update(msg_data, metadata={"scan_id": "scan_1"})
+        traces.append(np.array([i, i + 0.5, i + 1]))
+        mw._on_data_update(_monitor_update(list(traces)))

    # Only 3 curves remain after flush
    assert len(mw.curves) == 3
@@ -110,16 +203,69 @@ def test_multiwaveform_curve_limit_flush(qtbot, mocked_client):
    assert np.array_equal(y_data, [4, 4.5, 5])


-def test_multiwaveform_highlight_last_curve(qtbot, mocked_client):
-    """Check highlight_last_curve behavior."""
+def test_multiwaveform_snapshot_ordinal_filtering(qtbot, mocked_client, monkeypatch):
+    """Full-state snapshots must not duplicate already-rendered traces."""
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
+
+    first = np.array([1, 2, 3])
+    second = np.array([4, 5, 6])
+    mw._on_data_update(_monitor_update([first]))
+    assert len(mw.curves) == 1
+
+    # Same snapshot delivered again (e.g. trailing coalesced emission): no new curves.
+    mw._on_data_update(_monitor_update([first]))
+    assert len(mw.curves) == 1
+
+    # Snapshot grows by one trace: exactly one curve appended.
+    mw._on_data_update(_monitor_update([first, second]))
+    assert len(mw.curves) == 2
+    _, y_data = mw.curves[-1].getData()
+    assert np.array_equal(y_data, second)
+
+    # Retention window slid (oldest dropped): only newer ordinals are added.
+    third = np.array([7, 8, 9])
+    mw._on_data_update(_monitor_update([second, third], start=1))
+    assert len(mw.curves) == 3
+    _, y_data = mw.curves[-1].getData()
+    assert np.array_equal(y_data, third)
+
+
+def test_multiwaveform_scan_change_clears_curves(qtbot, mocked_client, monkeypatch):
+    """A new scan_id in the source metadata clears the previous scan's curves."""
+    _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
+
+    old_trace = np.array([1, 2, 3])
+    new_trace = np.array([4, 5, 6])
+    mw._on_data_update(_monitor_update([old_trace], scan_id="scan_1"))
+    assert len(mw.curves) == 1
+    assert mw.scan_id == "scan_1"
+
+    # New scan: the retained window still contains the old-scan trace, but
+    # only the not-yet-consumed ordinal is rendered after the clear.
+    mw._on_data_update(_monitor_update([old_trace, new_trace], scan_id="scan_2"))
+    assert mw.scan_id == "scan_2"
+    assert len(mw.curves) == 1
+    _, y_data = mw.curves[-1].getData()
+    assert np.array_equal(y_data, new_trace)
+
+
+def test_multiwaveform_highlight_last_curve(qtbot, mocked_client, monkeypatch):
+    """Check highlight_last_curve behavior."""
+    _fake_bridge_factory(monkeypatch)
+    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
    mw.max_trace = 5
    mw.flush_buffer = False

    # Simulate adding multiple curves
+    traces = []
    for i in range(3):
-        msg_data = {"data": np.array([i, i + 1, i + 2])}
-        mw.on_monitor_1d_update(msg_data, metadata={"scan_id": "scan_1"})
+        traces.append(np.array([i, i + 1, i + 2]))
+        mw._on_data_update(_monitor_update(list(traces)))

    # Initially highlight_last_curve is True, so the last visible curve is highlighted
    # The highlight index should be -1 in the code's logic
@@ -133,14 +279,14 @@ def test_multiwaveform_highlight_last_curve(qtbot, mocked_client):
    assert mw.highlighted_index == 0


-def test_multiwaveform_opacity_changes(qtbot, mocked_client):
+def test_multiwaveform_opacity_changes(qtbot, mocked_client, monkeypatch):
    """Check changing opacity affects existing curves."""
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
    mw.plot("waveform1d")

    # Add one curve
-    msg_data = {"data": np.array([10, 20, 30])}
-    mw.on_monitor_1d_update(msg_data, metadata={"scan_id": "scan_1"})
+    mw._on_data_update(_monitor_update([np.array([10, 20, 30])]))
    assert len(mw.curves) == 1

    # Default opacity is 50
@@ -151,15 +297,17 @@ def test_multiwaveform_opacity_changes(qtbot, mocked_client):
    assert mw.opacity == 80


-def test_multiwaveform_set_colormap(qtbot, mocked_client):
+def test_multiwaveform_set_colormap(qtbot, mocked_client, monkeypatch):
    """Check that setting a new colormap updates curve colors."""
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
    mw.plot("waveform1d")

    # Simulate multiple curve updates
+    traces = []
    for i in range(3):
-        msg_data = {"data": np.array([i, i + 1, i + 2])}
-        mw.on_monitor_1d_update(msg_data, metadata={"scan_id": "scan_1"})
+        traces.append(np.array([i, i + 1, i + 2]))
+        mw._on_data_update(_monitor_update(list(traces)))

    # Default color_palette is "magma"
    assert mw.color_palette == "plasma"
@@ -168,15 +316,17 @@ def test_multiwaveform_set_colormap(qtbot, mocked_client):
    assert mw.color_palette == "viridis"


-def test_multiwaveform_simulate_updates(qtbot, mocked_client):
+def test_multiwaveform_simulate_updates(qtbot, mocked_client, monkeypatch):
    """Simulate a series of 1D updates to ensure the data is appended and the correct number of curves appear."""
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
    mw.plot("waveform1d")

    data_series = [np.random.rand(5), np.random.rand(5), np.random.rand(5)]
+    traces = []
    for idx, arr in enumerate(data_series):
-        msg_data = {"data": arr}
-        mw.on_monitor_1d_update(msg_data, metadata={"scan_id": "scan_99"})
+        traces.append(arr)
+        mw._on_data_update(_monitor_update(list(traces), scan_id="scan_99"))
        # Each update should add a new curve
        assert len(mw.curves) == idx + 1
        x_data, y_data = mw.curves[-1].getData()
@@ -286,13 +436,15 @@ def test_control_panel_opacity_slider_spinbox(qtbot, mocked_client):
    assert spinbox_opacity.value() == 95


-def test_control_panel_highlight_slider_spinbox(qtbot, mocked_client):
+def test_control_panel_highlight_slider_spinbox(qtbot, mocked_client, monkeypatch):
    """
    Test that the slider and spinbox for curve highlighting update
    the widget's highlighted_index property, and are disabled if
    highlight_last_curve is True.
    """
+    _fake_bridge_factory(monkeypatch)
    mw = create_widget(qtbot, MultiWaveform, client=mocked_client)
+    mw.plot("waveform1d")
    slider_index = mw.controls.ui.highlighted_index
    spinbox_index = mw.controls.ui.spinbox_index
    checkbox_highlight_last = mw.controls.ui.highlight_last_curve
@@ -310,8 +462,10 @@ def test_control_panel_highlight_slider_spinbox(qtbot, mocked_client):

    # Simulate a few curves so there's something to highlight
    data_arrays = [np.array([0, 1, 2]), np.array([3, 4, 5]), np.array([6, 7, 8])]
+    traces = []
    for arr in data_arrays:
-        mw.on_monitor_1d_update({"data": arr}, {"scan_id": "scan_123"})
+        traces.append(arr)
+        mw._on_data_update(_monitor_update(list(traces), scan_id="scan_123"))

    # The number_of_visible_curves == 3 now
    max_index = mw.number_of_visible_curves - 1