From 6f5bb4a8aa73497838cb0a76bf08c5f6138871b0 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Thu, 6 Aug 2026 14:15:22 +0200 Subject: [PATCH] fix(waveform): repair async signal streaming for BEC async signal curves --- bec_widgets/tests/utils.py | 8 ++- .../settings/curve_settings/curve_tree.py | 2 +- .../widgets/plots/waveform/waveform.py | 35 +++++++++++-- tests/unit_tests/test_waveform.py | 50 +++++++++++++++++++ 4 files changed, 89 insertions(+), 6 deletions(-) diff --git a/bec_widgets/tests/utils.py b/bec_widgets/tests/utils.py index 7994c499..71024ff3 100644 --- a/bec_widgets/tests/utils.py +++ b/bec_widgets/tests/utils.py @@ -259,7 +259,13 @@ class DMMock: API shape used by Waveform._check_async_signal_found. """ signals: list[tuple[str, str, dict]] = [] - if signal_class_name != "AsyncSignal": + # The real API accepts a single class name or a list of class names + # (e.g. Waveform passes sorted(ASYNC_SIGNAL_CLASSES)). + if isinstance(signal_class_name, str): + signal_class_names = [signal_class_name] + else: + signal_class_names = list(signal_class_name) + if "AsyncSignal" not in signal_class_names: return signals for device in self.devices.values(): diff --git a/bec_widgets/widgets/plots/waveform/settings/curve_settings/curve_tree.py b/bec_widgets/widgets/plots/waveform/settings/curve_settings/curve_tree.py index af020e51..2957446d 100644 --- a/bec_widgets/widgets/plots/waveform/settings/curve_settings/curve_tree.py +++ b/bec_widgets/widgets/plots/waveform/settings/curve_settings/curve_tree.py @@ -449,7 +449,7 @@ class CurveRow(QTreeWidgetItem): device_entry = device_entry_info.get("obj_name", device_entry) else: device_entry = self.entry_validator.validate_signal( - device=device_name, signal=device_entry + name=device_name, entry=device_entry ) self.config.signal = DeviceSignal(device=device_name, signal=device_entry) diff --git a/bec_widgets/widgets/plots/waveform/waveform.py b/bec_widgets/widgets/plots/waveform/waveform.py index afc4398b..fe211d25 100644 --- a/bec_widgets/widgets/plots/waveform/waveform.py +++ b/bec_widgets/widgets/plots/waveform/waveform.py @@ -1510,13 +1510,30 @@ class Waveform(PlotBase): """ Clean up the curve by disconnecting the async update signal (even for sync curves). + The endpoint is resolved the same way _setup_async_curve resolved it + (device_async_signal for BEC async signals, legacy device_async_readback + otherwise) and the subscription is only dropped once no other curve + shares the stream. + Args: curve(Curve): The curve to clean up. """ - self.bec_dispatcher.disconnect_slot( - self.on_async_readback, - MessageEndpoints.device_async_readback(self.scan_id, curve.name()), - ) + name = getattr(curve.config.signal, "device", None) if curve.config.signal else None + signal = getattr(curve.config.signal, "signal", None) if curve.config.signal else None + if name is not None: + async_signal_found, stream = self._check_async_signal_found(name, signal) + if async_signal_found: + stream_key = (name, stream) + endpoint = MessageEndpoints.device_async_signal(self.scan_id, name, stream) + else: + stream_key = (name, None) + endpoint = MessageEndpoints.device_async_readback(self.scan_id, name) + subscribers = self._async_streams_setup.get(stream_key) + if subscribers is not None and signal in subscribers: + subscribers.remove(signal) + if not subscribers: + self._async_streams_setup.pop(stream_key, None) + self.bec_dispatcher.disconnect_slot(self.on_async_readback, endpoint) curve.rpc_register.remove_rpc(curve) # Remove itself from the DAP summary only for side panels @@ -1700,6 +1717,15 @@ class Waveform(PlotBase): device_entry = curve.config.signal.signal if access_key == "val": # live access device_data = data.get(device_name, {}).get(device_entry, {}).get(access_key, None) + if device_data is None: + # Live async data never passes through scan_item.live_data; it + # arrives via the Redis stream subscription (on_async_readback). + # Nothing to redraw here — and nothing to warn about. + logger.debug( + f"No live_data entry for async curve {curve.name()}; " + "updates come from the async stream subscription." + ) + continue else: # history access dataset_obj = data.get(device_name, {}) if self._skip_large_dataset_check is False: @@ -1844,6 +1870,7 @@ class Waveform(PlotBase): scan_id = sender.cb_info.get("scan_id", None) if scan_id != self.scan_id: logger.info("Scan ID mismatch, ignoring async readback.") + return instruction = metadata.get("async_update", {}).get("type") if instruction not in ["add", "add_slice", "replace"]: diff --git a/tests/unit_tests/test_waveform.py b/tests/unit_tests/test_waveform.py index 35aedeb4..b4e1e342 100644 --- a/tests/unit_tests/test_waveform.py +++ b/tests/unit_tests/test_waveform.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock import numpy as np import pyqtgraph as pg import pytest +from bec_lib.endpoints import MessageEndpoints from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem from qtpy.QtCore import QTimer from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox @@ -913,6 +914,55 @@ def test_setup_async_curve(qtbot, mocked_client, monkeypatch): connect_spy.assert_called_once() +def test_setup_async_curve_uses_device_async_signal_for_bec_signals( + qtbot, mocked_client, monkeypatch +): + """ + A curve backed by a declared BEC async signal (e.g. a DynamicSignal group + sub-signal) must subscribe to the per-signal device_async_signal endpoint, + not the legacy per-device device_async_readback stream. + """ + wf = create_widget(qtbot, Waveform, client=mocked_client) + wf.old_scan_id = "111" + wf.scan_id = "222" + + c = wf.plot(arg1="async_device", label="async_device-async_device") + wf._async_curves = [c] + c.config.signal.signal = "async_device_mean_ai0" + + monkeypatch.setattr( + wf.client.device_manager, + "get_bec_signals", + lambda classes: [ + ( + "async_device", + "mean.ai0", + { + "obj_name": "async_device_mean_ai0", + "storage_name": "async_device_mean", + "signal_class": "DynamicSignal", + }, + ) + ], + ) + connect_spy = MagicMock() + monkeypatch.setattr(wf.bec_dispatcher, "connect_slot", connect_spy) + + wf._async_streams_setup = {} + wf._setup_async_curve(c) + connect_spy.assert_called_once() + endpoint_called = connect_spy.call_args[0][1].endpoint + expected = MessageEndpoints.device_async_signal("222", "async_device", "async_device_mean") + assert endpoint_called == expected.endpoint + + # cleanup drops the same endpoint once the last subscriber is removed + disconnect_spy = MagicMock() + monkeypatch.setattr(wf.bec_dispatcher, "disconnect_slot", disconnect_spy) + wf._curve_clean_up(c) + assert disconnect_spy.call_args[0][1].endpoint == expected.endpoint + assert (("async_device", "async_device_mean")) not in wf._async_streams_setup + + def test_on_async_readback_add_update(qtbot, mocked_client): """ Test that on_async_readback extends or replaces async data depending on metadata instruction.