refactor: set downsampling to auto=True, method 'peak', activate clipToView for (Async)-Curves and fix ViewAll hook from pg.view_box menu

bec_widgets/cli/client.py

@@ -498,6 +498,15 @@ class Curve(RPCBase):
             dict: The configuration of the widget.
         """
 
+    @rpc_call
+    def _get_displayed_data(self) -> "tuple[np.ndarray, np.ndarray]":
+        """
+        Get the displayed data of the curve.
+
+        Returns:
+            tuple[np.ndarray, np.ndarray]: The x and y data of the curve.
+        """
+
     @rpc_call
     def set(self, **kwargs):
         """
@@ -593,7 +602,7 @@ class Curve(RPCBase):
         """
 
     @rpc_call
-    def get_data(self) -> "tuple[np.ndarray, np.ndarray]":
+    def get_data(self) -> "tuple[np.ndarray | None, np.ndarray | None]":
         """
         Get the data of the curve.
         Returns:

bec_widgets/widgets/plots/waveform/curve.py

@@ -61,6 +61,7 @@ class Curve(BECConnector, pg.PlotDataItem):
         "remove",
         "_rpc_id",
         "_config_dict",
+        "_get_displayed_data",
         "set",
         "set_data",
         "set_color",
@@ -100,6 +101,8 @@ class Curve(BECConnector, pg.PlotDataItem):
         self.slice_index = None
         if kwargs:
             self.set(**kwargs)
+        # Activate setClipToView, to boost performance for large datasets per default
+        self.setClipToView(True)
 
     def apply_config(self, config: dict | CurveConfig | None = None, **kwargs) -> None:
         """
@@ -327,3 +330,15 @@ class Curve(BECConnector, pg.PlotDataItem):
         # self.parent_item.removeItem(self)
         self.parent_item.remove_curve(self.name())
         super().remove()
+
+    def _get_displayed_data(self) -> tuple[np.ndarray, np.ndarray]:
+        """
+        Get the displayed data of the curve.
+
+        Returns:
+            tuple[np.ndarray, np.ndarray]: The x and y data of the curve.
+        """
+        x_data, y_data = self.getData()
+        if x_data is None or y_data is None:
+            return np.array([]), np.array([])
+        return x_data, y_data

bec_widgets/widgets/plots/waveform/waveform.py

@@ -9,7 +9,7 @@ import pyqtgraph as pg
 from bec_lib import bec_logger, messages
 from bec_lib.endpoints import MessageEndpoints
 from pydantic import Field, ValidationError, field_validator
-from qtpy.QtCore import QTimer, Signal
+from qtpy.QtCore import QTimer, Signal, Slot
 from qtpy.QtWidgets import QDialog, QHBoxLayout, QMainWindow, QVBoxLayout, QWidget
 
 from bec_widgets.utils import ConnectionConfig
@@ -179,6 +179,20 @@ class Waveform(PlotBase):
 
         # for updating a color scheme of curves
         self._connect_to_theme_change()
+        # To fix the ViewAll action with clipToView activated
+        self._connect_viewbox_menu_actions()
+
+    def _connect_viewbox_menu_actions(self):
+        """Connect the viewbox menu action ViewAll to the custom reset_view method."""
+        menu = self.plot_item.vb.menu
+        # Find and replace "View All" action
+        for action in menu.actions():
+            if action.text() == "View All":
+                # Disconnect the default autoRange action
+                action.triggered.disconnect()
+                # Connect to the custom reset_view method
+                action.triggered.connect(self._reset_view)
+                break
 
     def __getitem__(self, key: int | str):
         return self.get_curve(key)
@@ -220,6 +234,23 @@ class Waveform(PlotBase):
         )
         self.toolbar.widgets["fit_params"].action.triggered.connect(self.show_dap_summary_popup)
 
+    @SafeSlot()
+    def _reset_view(self):
+        """
+        Custom _reset_view method to fix ViewAll action in toolbar.
+        Due to setting clipToView to True on the curves, the autoRange() method
+        of the ViewBox does no longer work as expected. This method deactivates the
+        setClipToView for all curves, calls autoRange() to circumvent that issue.
+        Afterwards, it re-enables the setClipToView for all curves again.
+
+        It is hooked to the ViewAll action in the right-click menu of the pg.PlotItem ViewBox.
+        """
+        for curve in self._async_curves + self._sync_curves:
+            curve.setClipToView(False)
+        self.plot_item.vb.autoRange()
+        for curve in self._async_curves + self._sync_curves:
+            curve.setClipToView(True)
+
     ################################################################################
     # Roi manager
 
@@ -1072,10 +1103,12 @@ class Waveform(PlotBase):
             # If there's actual data, set it
             if device_data is not None:
                 self._auto_adjust_async_curve_settings(curve, len(device_data))
-                if x_data is not None:
+                if x_data is not None or self.x_axis_mode["name"] == "timestamp":
                     curve.setData(x_data, device_data)
                 else:
-                    curve.setData(device_data)
+                    curve.setData(
+                        np.linspace(0, len(device_data) - 1, len(device_data)), device_data
+                    )
         self.request_dap_update.emit()
 
     def _setup_async_curve(self, curve: Curve):
@@ -1116,57 +1149,57 @@ class Waveform(PlotBase):
             new_data = None
             y_data = None
             x_data = None
-            y_entry = curve.config.signal.entry
             x_name = self.x_axis_mode["name"]
-            for device, async_data in msg["signals"].items():
+            async_data = msg["signals"].get(curve.config.signal.entry, None)
-                if device == y_entry:
+            if async_data is None:
-                    data_plot = async_data["value"]
+                continue
-                    # Add
+            data_plot = async_data["value"]
-                    if instruction == "add":
+            # Add
-                        if len(max_shape) > 1:
+            if instruction == "add":
-                            if len(data_plot.shape) > 1:
+                if len(max_shape) > 1:
-                                data_plot = data_plot[-1, :]
+                    if len(data_plot.shape) > 1:
-                        else:
+                        data_plot = data_plot[-1, :]
-                            x_data, y_data = curve.get_data()
+                else:
+                    x_data, y_data = curve.get_data()
 
-                        if y_data is not None:
+                if y_data is not None:
-                            new_data = np.hstack((y_data, data_plot))  # TODO check performance
+                    new_data = np.hstack((y_data, data_plot))  # TODO check performance
-                        else:
+                else:
-                            new_data = data_plot
+                    new_data = data_plot
-                        if x_name == "timestamp":
+                if x_name == "timestamp":
-                            if x_data is not None:
+                    if x_data is not None:
-                                x_data = np.hstack((x_data, async_data["timestamp"]))
+                        x_data = np.hstack((x_data, async_data["timestamp"]))
-                            else:
-                                x_data = async_data["timestamp"]
-                                # FIXME x axis wrong if timestamp switched during scan
-                    # Add slice
-                    elif instruction == "add_slice":
-                        current_slice_id = metadata.get("async_update", {}).get("index")
-                        data_plot = async_data["value"]
-                        if current_slice_id != curve.slice_index:
-                            curve.slice_index = current_slice_id
-                        else:
-                            x_data, y_data = curve.get_data()
-                        if y_data is not None:
-                            new_data = np.hstack((y_data, data_plot))
-                        else:
-                            new_data = data_plot
-                    # Replace
-                    elif instruction == "replace":
-                        if x_name == "timestamp":
-                            x_data = async_data["timestamp"]
-                        new_data = data_plot
-
-                    # If update is not add, add_slice or replace, continue.
-                    if new_data is None:
-                        continue
-                    # Hide symbol, activate downsampling if data >1000
-                    self._auto_adjust_async_curve_settings(curve, len(new_data))
-                    # Set data on the curve
-                    if x_name == "timestamp" and instruction != "add_slice":
-                        curve.setData(x_data, new_data)
                     else:
-                        curve.setData(np.linspace(0, len(new_data) - 1, len(new_data)), new_data)
+                        x_data = async_data["timestamp"]
+                        # FIXME x axis wrong if timestamp switched during scan
+            # Add slice
+            elif instruction == "add_slice":
+                current_slice_id = metadata.get("async_update", {}).get("index")
+                data_plot = async_data["value"]
+                if current_slice_id != curve.slice_index:
+                    curve.slice_index = current_slice_id
+                else:
+                    x_data, y_data = curve.get_data()
+                if y_data is not None:
+                    new_data = np.hstack((y_data, data_plot))
+                else:
+                    new_data = data_plot
+            # Replace
+            elif instruction == "replace":
+                if x_name == "timestamp":
+                    x_data = async_data["timestamp"]
+                new_data = data_plot
+
+            # If update is not add, add_slice or replace, continue.
+            if new_data is None:
+                continue
+            # Hide symbol, activate downsampling if data >1000
+            self._auto_adjust_async_curve_settings(curve, len(new_data))
+            # Set data on the curve
+            if x_name == "timestamp" and instruction != "add_slice":
+                curve.setData(x_data, new_data)
+            else:
+                curve.setData(np.linspace(0, len(new_data) - 1, len(new_data)), new_data)
 
         self.request_dap_update.emit()
 
@@ -1175,7 +1208,7 @@ class Waveform(PlotBase):
         curve: Curve,
         data_length: int,
         limit: int = 1000,
-        method: Literal["subsample", "mean", "peak"] | None = "mean",
+        method: Literal["subsample", "mean", "peak"] | None = "peak",
     ) -> None:
         """
         Based on the length of the data this method will adjust the plotting settings of
@@ -1196,12 +1229,15 @@ class Waveform(PlotBase):
         if data_length > limit:
             if curve.config.symbol is not None:
                 curve.set_symbol(None)
-            sampling_factor = int(data_length / (5 * limit))  # increase by limit 5x
+            if curve.config.pen_width > 3:
-            curve.setDownsampling(ds=sampling_factor, auto=None, method=method)
+                curve.set_pen_width(3)
+            curve.setDownsampling(ds=None, auto=True, method=method)
+            curve.setClipToView(True)
         elif data_length <= limit:
             curve.set_symbol("o")
-            sampling_factor = 1
+            curve.set_pen_width(4)
-            curve.setDownsampling(ds=sampling_factor, auto=None, method=method)
+            curve.setDownsampling(ds=1, auto=None, method=method)
+            curve.setClipToView(True)
 
     def setup_dap_for_scan(self):
         """Setup DAP updates for the new scan."""

tests/end-2-end/test_plotting_framework_e2e.py

@@ -113,6 +113,51 @@ def test_rpc_waveform_scan(qtbot, bec_client_lib, connected_client_gui_obj):
     assert plt_data["bpm4d-bpm4d"]["y"] == last_scan_data["bpm4d"]["bpm4d"].val
 
 
+def test_async_plotting(qtbot, bec_client_lib, connected_client_gui_obj):
+    gui = connected_client_gui_obj
+    dock = gui.bec
+
+    client = bec_client_lib
+    dev = client.device_manager.devices
+    scans = client.scans
+    queue = client.queue
+
+    # Test add
+    dev.waveform.sim.select_model("GaussianModel")
+    dev.waveform.sim.params = {"amplitude": 1000, "center": 4000, "sigma": 300}
+    dev.waveform.async_update.put("add")
+    dev.waveform.waveform_shape.put(10000)
+    wf = dock.new("wf_dock").new("Waveform")
+    curve = wf.plot(y_name="waveform")
+
+    status = scans.line_scan(dev.samx, -5, 5, steps=10, exp_time=0.05, relative=False)
+    status.wait()
+
+    # Wait for the scan to finish and the data to be available in history
+    # Wait until scan_id is in history
+    def _wait_for_scan_in_hisotry():
+        if len(client.history) == 0:
+            return False
+        # Once items appear in storage, the last one hast to be the one we just scanned
+        return client.history[-1].metadata.bec["scan_id"] == status.scan.scan_id
+
+    qtbot.waitUntil(_wait_for_scan_in_hisotry, timeout=10000)
+    last_scan_data = client.history[-1]
+    # check plotted data
+    x_data, y_data = curve.get_data()
+    assert np.array_equal(x_data, np.linspace(0, len(y_data) - 1, len(y_data)))
+    assert np.array_equal(
+        y_data, last_scan_data.devices.waveform.get("waveform_waveform", {}).read().get("value", [])
+    )
+
+    # Check displayed data
+    x_data_display, y_data_display = curve._get_displayed_data()
+    # Should be not more than 1% difference, actually be closer but this might be flaky
+    assert np.isclose(x_data_display[-1], x_data[-1], rtol=0.01)
+    # Downsampled data should be smaller than original data
+    assert len(y_data_display) < len(y_data)
+
+
 def test_rpc_image(qtbot, bec_client_lib, connected_client_gui_obj):
     gui = connected_client_gui_obj
     dock = gui.bec

tests/unit_tests/test_waveform_next_gen.py

@@ -595,13 +595,15 @@ def test_on_async_readback_add_update(qtbot, mocked_client, x_mode):
     c.setData([], [])
 
     # Test large updates, limit 1000 to deactivate symbols, downsampling for 8000 should be factor 2.
-    waveform_shape = 12000
+    waveform_shape = 100000
-    for ii in range(12):
+    n_cycles = 10
+    for ii in range(n_cycles):
         msg = {
             "signals": {
                 "async_device": {
-                    "value": np.array(range(1000)),
+                    "value": np.array(range(waveform_shape // n_cycles)),
-                    "timestamp": (ii + 1) * np.linspace(0, 1000 - 1, 1000),
+                    "timestamp": (ii + 1)
+                    * np.linspace(0, waveform_shape // n_cycles - 1, waveform_shape // n_cycles),
                 }
             }
         }
@@ -615,9 +617,7 @@ def test_on_async_readback_add_update(qtbot, mocked_client, x_mode):
     assert c.opts["symbol"] == None
     # Get displayed data
     displayed_x, displayed_y = c.getData()
-    assert len(displayed_y) == waveform_shape / 2
+    assert len(displayed_y) == len(displayed_x)
-    assert len(displayed_x) == waveform_shape / 2
-    assert displayed_x[-1] == waveform_shape - 1  # Should be the correct index stil.
 
     ############# Test replace ################
     waveform_shape = 10