feat(waveform): decimate oversized curves with zoom re-rendering and load progress

bec_widgets/utils/qt_data_subscription.py

@@ -25,8 +25,12 @@ class QtDataSubscription(QObject):
 
     #: Emitted on the Qt thread with each SubscriptionUpdate.
     updated = Signal(object)
+    #: Emitted on the Qt thread with the bulk-load fraction (0..1) while the
+    #: backend reads history data.
+    progress = Signal(float)
 
     _raw = Signal(object)
+    _raw_progress = Signal(float)
 
     def __init__(
         self,
@@ -73,6 +77,7 @@ class QtDataSubscription(QObject):
         # auto-connection would then invoke _filter before _subscription is
         # assigned and before the widget had a chance to connect `updated`.
         self._raw.connect(self._filter, Qt.QueuedConnection)
+        self._raw_progress.connect(self._emit_progress, Qt.QueuedConnection)
         self._api = DataAPI(client)
         self._subscription = self._api.subscribe(
             sources=sources,
@@ -81,6 +86,7 @@ class QtDataSubscription(QObject):
             min_emit_interval=min_emit_interval,
             max_points=max_points,
             size_limit_bytes=size_limit_bytes,
+            progress_callback=self._deliver_progress,
         )
         self.destroyed.connect(lambda: self.close())
 
@@ -90,8 +96,16 @@ class QtDataSubscription(QObject):
         if not self._closed:
             self._raw.emit(update)
 
+    def _deliver_progress(self, fraction: float) -> None:
+        if not self._closed:
+            self._raw_progress.emit(float(fraction))
+
     # --- qt-thread side ------------------------------------------------------
 
+    def _emit_progress(self, fraction: float) -> None:
+        if not self._closed:
+            self.progress.emit(fraction)
+
     def _filter(self, update: SubscriptionUpdate) -> None:
         if self._closed or self._subscription is None:
             return

bec_widgets/widgets/plots/waveform/waveform.py

@@ -22,6 +22,7 @@ from qtpy.QtWidgets import (
     QHBoxLayout,
     QLabel,
     QMainWindow,
+    QProgressBar,
     QVBoxLayout,
     QWidget,
 )
@@ -75,6 +76,55 @@ class WaveformConfig(ConnectionConfig):
     _validate_color_palette = field_validator("color_palette")(Colors.validate_color_map)
 
 
+#: Curves above this many points are decimated to a min/max envelope before
+#: rendering: pyqtgraph's per-paint work is O(points) with a large constant,
+#: and a pixel column can only ever show a vertical range.
+DISPLAY_POINT_LIMIT = 1_000_000
+#: Envelope bins of a decimated display (output length = 2 * bins).
+DISPLAY_ENVELOPE_BINS = 250_000
+
+
+def _decimate_envelope(
+    y_data: np.ndarray, x_data: np.ndarray | None = None, offset: int = 0
+) -> tuple[np.ndarray, np.ndarray]:
+    """
+    Reduce a long series to an interleaved per-bin (min, max) envelope.
+
+    Every bin contributes its true extremes, so outliers stay visible at any
+    zoom level — unlike point skipping. The raw data is NOT modified; callers
+    keep it for zoom-window re-decimation.
+
+    Args:
+        y_data (np.ndarray): Raw values.
+        x_data (np.ndarray | None): Optional x column parallel to ``y_data``;
+            ``None`` means sample-index x (offset by ``offset``).
+        offset (int): Index of ``y_data[0]`` in the full series (window mode).
+
+    Returns:
+        tuple[np.ndarray, np.ndarray]: Decimated (x, y) columns.
+    """
+    n = len(y_data)
+    bin_size = int(np.ceil(n / DISPLAY_ENVELOPE_BINS))
+    n_bins = n // bin_size
+    body = y_data[: n_bins * bin_size].reshape(n_bins, bin_size)
+    mins = body.min(axis=1)
+    maxs = body.max(axis=1)
+    tail = y_data[n_bins * bin_size :]
+    if tail.size:
+        mins = np.append(mins, tail.min())
+        maxs = np.append(maxs, tail.max())
+        n_bins += 1
+    y_out = np.empty(2 * n_bins, dtype=mins.dtype)
+    y_out[0::2] = mins
+    y_out[1::2] = maxs
+    center_idx = np.minimum(np.arange(n_bins, dtype=np.int64) * bin_size + bin_size // 2, n - 1)
+    if x_data is None:
+        centers = (center_idx + offset).astype(np.float64)
+    else:
+        centers = np.asarray(x_data)[center_idx]
+    return np.repeat(centers, 2), y_out
+
+
 class Waveform(PlotBase):
     #: 15 Hz: above typical device message rates while leaving paint headroom
     #: for multi-million-point async curves (benchmarked).
@@ -1759,6 +1809,7 @@ class Waveform(PlotBase):
                 min_emit_interval=self.update_interval_s,
             )
             self._data_bridge.updated.connect(self._on_data_update)
+            self._data_bridge.progress.connect(self._on_data_load_progress)
         except Exception as exc:
             logger.warning(f"Failed to configure waveform data subscription: {exc}")
             self._cleanup_data_api_subscription()
@@ -1803,6 +1854,7 @@ class Waveform(PlotBase):
                     min_emit_interval=self.update_interval_s,
                 )
                 bridge.updated.connect(self._on_data_update)
+                bridge.progress.connect(self._on_data_load_progress)
                 self._history_bridges[scan_id] = bridge
             except Exception as exc:
                 logger.warning(
@@ -2073,10 +2125,24 @@ class Waveform(PlotBase):
                     f"Async data for curve {curve.name()} and x_axis {x_key} is not of equal "
                     "length. Falling back to 'index' plotting."
                 )
+        n_points = len(y_data)
+        self._auto_adjust_async_curve_settings(curve, n_points)
+        y_arr = np.asarray(y_data)
+        if n_points > DISPLAY_POINT_LIMIT and y_arr.dtype != object:
+            # min/max envelope: bounded render cost, extremes stay visible.
+            # The raw series is kept on the curve so zooming re-decimates the
+            # visible window at full fidelity (see _apply_lod_windows).
+            x_arr = None if x_data is None else np.asarray(x_data)
+            curve._lod_raw = (x_arr, y_arr)
+            curve._lod_window = None
+            self._ensure_lod_hooks()
+            x_dec, y_dec = _decimate_envelope(y_arr, x_arr)
+            curve.setData(x_dec, y_dec)
+            return True
+        curve._lod_raw = None
         if x_data is None:
-            x_data = np.arange(len(y_data))
-        self._auto_adjust_async_curve_settings(curve, len(y_data))
-        curve.setData(x_data, np.asarray(y_data))
+            x_data = np.arange(n_points)
+        curve.setData(np.asarray(x_data), y_arr)
         return True
 
     @staticmethod
@@ -2189,6 +2255,91 @@ class Waveform(PlotBase):
         }
         return buffer
 
+    def _ensure_lod_hooks(self):
+        """Re-decimate oversized curves for the visible window on zoom/pan."""
+        if getattr(self, "_lod_timer", None) is not None:
+            return
+        self._lod_timer = QTimer(self)
+        self._lod_timer.setSingleShot(True)
+        self._lod_timer.setInterval(150)
+        self._lod_timer.timeout.connect(self._apply_lod_windows)
+        self._lod_applying = False
+        self.plot_item.vb.sigXRangeChanged.connect(self._schedule_lod)
+
+    @SafeSlot()
+    def _schedule_lod(self, *_args):
+        """Debounce zoom/pan events; sigXRangeChanged passes (viewbox, range)."""
+        if not getattr(self, "_lod_applying", False):
+            self._lod_timer.start()
+
+    @SafeSlot()
+    def _apply_lod_windows(self):
+        """Render the visible slice of every oversized curve.
+
+        Zoomed out, the min/max envelope of all data is shown; zooming in
+        narrows the window until the raw samples themselves are rendered —
+        no information is lost at any zoom level.
+        """
+        x_min, x_max = self.plot_item.vb.viewRange()[0]
+        self._lod_applying = True
+        try:
+            for curve in self.curves + self._history_curves:
+                raw = getattr(curve, "_lod_raw", None)
+                if raw is None:
+                    continue
+                x_arr, y_arr = raw
+                n = len(y_arr)
+                if x_arr is None:
+                    i0 = max(0, int(np.floor(x_min)))
+                    i1 = min(n, int(np.ceil(x_max)) + 1)
+                else:
+                    i0, i1 = np.searchsorted(x_arr, [x_min, x_max])
+                    i0, i1 = max(0, i0 - 1), min(n, i1 + 1)
+                if i1 <= i0:
+                    continue
+                window = (i0, i1)
+                if getattr(curve, "_lod_window", None) == window:
+                    continue
+                curve._lod_window = window
+                y_win = y_arr[i0:i1]
+                x_win = None if x_arr is None else x_arr[i0:i1]
+                if len(y_win) > DISPLAY_POINT_LIMIT:
+                    x_dec, y_dec = _decimate_envelope(y_win, x_win, offset=i0)
+                    curve.setData(x_dec, y_dec)
+                else:
+                    curve.setData(np.arange(i0, i1) if x_win is None else x_win, y_win)
+        finally:
+            self._lod_applying = False
+
+    def _ensure_load_progress_bar(self) -> QProgressBar:
+        """Create (once) a full-width progress row below the plot."""
+        if getattr(self, "_load_progress_row", None) is None:
+            row = QWidget(self)
+            row_layout = QHBoxLayout(row)
+            row_layout.setContentsMargins(8, 2, 8, 2)
+            row_layout.setSpacing(8)
+            label = QLabel("Loading large dataset...", row)
+            bar = QProgressBar(row)
+            bar.setRange(0, 100)
+            bar.setTextVisible(True)
+            row_layout.addWidget(label)
+            row_layout.addWidget(bar, 1)
+            row.hide()
+            self.layout.addWidget(row)
+            self._load_progress_row = row
+            self._load_progress_bar = bar
+        return self._load_progress_bar
+
+    @SafeSlot(float)
+    def _on_data_load_progress(self, fraction: float):
+        """Show bulk-load progress so a long history read is visibly alive."""
+        bar = self._ensure_load_progress_bar()
+        if fraction >= 1.0:
+            self._load_progress_row.hide()
+            return
+        bar.setValue(int(fraction * 100))
+        self._load_progress_row.show()
+
     def _auto_adjust_async_curve_settings(
         self,
         curve: Curve,

tests/unit_tests/test_waveform.py

@@ -86,6 +86,7 @@ def _fake_bridge_factory(monkeypatch, gated_bytes: int | None = None):
             self.healthy = True
             self.closed = False
             self.updated = MagicMock()
+            self.progress = MagicMock()
             self.size_limit_bytes = size_limit_bytes
             self.min_emit_interval = min_emit_interval
             self.estimated_bytes = gated_bytes
@@ -2235,3 +2236,86 @@ def test_async_display_values_accepts_numpy(qtbot, mocked_client):
     carrier = SimpleNamespace(_data_api_async_cache=None)
     cached = wf._async_display_values_cached(carrier, _make_update([src]), src)
     assert list(cached) == [1, 2, 3]
+
+
+def test_decimate_envelope_preserves_extremes():
+    """The envelope must contain the global min/max (spikes stay visible)."""
+    from bec_widgets.widgets.plots.waveform.waveform import _decimate_envelope
+
+    rng = np.random.default_rng(1)
+    y = rng.random(3_000_000)
+    y[1_234_567] = 5.0  # single-sample spike
+    y[2_222_222] = -5.0
+    x_dec, y_dec = _decimate_envelope(y)
+    assert len(y_dec) <= 2 * 250_000 + 2
+    assert len(x_dec) == len(y_dec)
+    assert y_dec.max() == 5.0
+    assert y_dec.min() == -5.0
+    assert x_dec[0] >= 0 and x_dec[-1] <= len(y) - 1
+
+    # custom x column keeps positional correspondence
+    x = np.linspace(100.0, 200.0, len(y))
+    x_dec2, y_dec2 = _decimate_envelope(y, x)
+    assert 100.0 <= x_dec2[0] <= x_dec2[-1] <= 200.0
+
+
+def test_oversized_async_curve_is_decimated_and_zoomable(qtbot, mocked_client, monkeypatch):
+    """An oversized history curve renders a bounded envelope; zooming into a
+    small window re-renders the raw samples (no data loss on investigation)."""
+    from bec_widgets.widgets.plots.waveform.waveform import DISPLAY_POINT_LIMIT
+
+    _fake_bridge_factory(monkeypatch)
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    c = wf.plot(arg1="async_device", label="async_device-async_device")
+    wf.scan_id = "dummy"
+    wf.x_axis_mode["name"] = "index"
+
+    n = DISPLAY_POINT_LIMIT + 500_000
+    raw = np.random.default_rng(2).random(n)
+    src = _async_source("async_device", values=raw, update_type=None, as_numpy=True)
+    src.metadata.pop("async_update_type", None)
+    wf._on_data_update(_make_update([src], reason="history"))
+
+    x_data, y_data = c.get_data()
+    assert len(y_data) <= 2 * 250_000 + 2  # bounded display
+    assert c._lod_raw is not None  # raw kept for zooming
+
+    # zoom into a 1000-sample window -> raw samples rendered exactly
+    wf.plot_item.vb.setXRange(1000, 2000, padding=0)
+    wf._apply_lod_windows()
+    x_data, y_data = c.get_data()
+    assert len(y_data) <= 1200
+    i0 = int(x_data[0])
+    np.testing.assert_array_almost_equal(y_data[:10], raw[i0 : i0 + 10])
+
+
+def test_data_load_progress_bar_shows_and_hides(qtbot, mocked_client, monkeypatch):
+    """The load progress bar appears with fractions < 1 and hides at 1.0."""
+    _fake_bridge_factory(monkeypatch)
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    wf._on_data_load_progress(0.3)
+    assert not wf._load_progress_row.isHidden()
+    assert wf._load_progress_bar.value() == 30
+    wf._on_data_load_progress(1.0)
+    assert wf._load_progress_row.isHidden()
+
+
+def test_zoom_signal_schedules_lod_redecimation(qtbot, mocked_client, monkeypatch):
+    """Regression: sigXRangeChanged passes (viewbox, range); the debounce slot
+    must accept them — a TypeError here was swallowed by SafeSlot on every
+    plot interaction and LOD re-decimation never ran."""
+    from bec_widgets.widgets.plots.waveform.waveform import DISPLAY_POINT_LIMIT
+
+    _fake_bridge_factory(monkeypatch)
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    c = wf.plot(arg1="async_device", label="async_device-async_device")
+    wf.scan_id = "dummy"
+    wf.x_axis_mode["name"] = "index"
+    raw = np.zeros(DISPLAY_POINT_LIMIT + 10)
+    src = _async_source("async_device", values=raw, update_type=None, as_numpy=True)
+    src.metadata.pop("async_update_type", None)
+    wf._on_data_update(_make_update([src], reason="history"))
+
+    wf._lod_timer.stop()
+    wf.plot_item.vb.setXRange(10, 20, padding=0)  # emits sigXRangeChanged
+    assert wf._lod_timer.isActive(), "zoom did not schedule LOD re-decimation"