From f77e5e108e8ceb16e5882c8346b3fcd061cce3a9 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 12 Aug 2026 21:20:37 +0200 Subject: [PATCH] feat(waveform): decimate oversized curves with zoom re-rendering and load progress --- bec_widgets/utils/qt_data_subscription.py | 14 ++ .../widgets/plots/waveform/waveform.py | 157 +++++++++++++++++- tests/unit_tests/test_waveform.py | 84 ++++++++++ 3 files changed, 252 insertions(+), 3 deletions(-) diff --git a/bec_widgets/utils/qt_data_subscription.py b/bec_widgets/utils/qt_data_subscription.py index c3e0ffb3..dd5ecc33 100644 --- a/bec_widgets/utils/qt_data_subscription.py +++ b/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 diff --git a/bec_widgets/widgets/plots/waveform/waveform.py b/bec_widgets/widgets/plots/waveform/waveform.py index 6a7a1bc5..f87f8c77 100644 --- a/bec_widgets/widgets/plots/waveform/waveform.py +++ b/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, diff --git a/tests/unit_tests/test_waveform.py b/tests/unit_tests/test_waveform.py index 14ff9fa2..d36b4514 100644 --- a/tests/unit_tests/test_waveform.py +++ b/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"