From 06523df66b82c441031d9aaceb7f3d013204c334 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Fri, 7 Aug 2026 12:49:21 +0200 Subject: [PATCH] perf(widgets): render async data incrementally and match main update rates --- bec_widgets/widgets/plots/image/image.py | 82 ++++++++- .../widgets/plots/motor_map/motor_map.py | 2 +- .../plots/multi_waveform/multi_waveform.py | 4 +- .../scatter_waveform/scatter_waveform.py | 2 +- .../widgets/plots/waveform/waveform.py | 79 ++++++++- .../test_data_api_widget_throughput.py | 161 ++++++++++++++++++ tests/unit_tests/test_image_view_next_gen.py | 63 +++++++ tests/unit_tests/test_waveform.py | 107 +++++++++++- 8 files changed, 481 insertions(+), 19 deletions(-) create mode 100644 tests/unit_tests/benchmarks/test_data_api_widget_throughput.py diff --git a/bec_widgets/widgets/plots/image/image.py b/bec_widgets/widgets/plots/image/image.py index a9e2f077..4affc4c5 100644 --- a/bec_widgets/widgets/plots/image/image.py +++ b/bec_widgets/widgets/plots/image/image.py @@ -130,6 +130,7 @@ class Image(ImageBase): self._data_bridge: QtDataSubscription | None = None self._source_key: tuple[str, str] | None = None self._min_display_ordinal: int | None = None + self._waterfall_cache: dict | None = None self.old_scan_id = None self.scan_id = None self.async_update = False @@ -451,7 +452,7 @@ class Image(ImageBase): sources=[(self._config.device, entry)], scan=scan, parent=self, - min_emit_interval=0.1, + min_emit_interval=0.04, max_points=max_points, ) self._data_bridge.updated.connect(self._on_data_update) @@ -466,6 +467,7 @@ class Image(ImageBase): self._source_key = (self._config.device, entry) self._min_display_ordinal = None + self._waterfall_cache = None self._set_connection_status("connected") logger.info( f"Connected to {self._config.device}.{self._config.signal} with type {config.monitor_type}" @@ -476,6 +478,7 @@ class Image(ImageBase): """Close the active DataAPI bridge, if any.""" self._source_key = None self._min_display_ordinal = None + self._waterfall_cache = None if self._data_bridge is None: return try: @@ -752,7 +755,7 @@ class Image(ImageBase): if self.subscriptions["main"].monitor_type == "2d": data = np.asarray(source.values[-1]) else: - data = self._build_1d_buffer(source) + data = self._build_1d_buffer(source, reason=update.reason) if data is None: return self._render_image_data(data) @@ -801,14 +804,85 @@ class Image(ImageBase): if self.crosshair is not None: self.crosshair.reset() - def _build_1d_buffer(self, source) -> np.ndarray | None: + def _build_1d_buffer(self, source, reason: str = "live") -> np.ndarray | None: """ - Rebuild the 2-D waterfall buffer from the 1-D columnar fragments of a + Build the 2-D waterfall buffer from the 1-D columnar fragments of a source: one row per ordinal, rows zero-padded to the longest row, newest row last. Covers async 'add' (one fragment per ordinal), 'add_slice' (accumulated row per ordinal), 'replace' (single current state) and preview streams (one waveform per arrival) alike. + The padded buffer and the consumed ordinal frontier are cached, so a + live append-only emission stacks only the new rows (padded to the + cached width) — O(new data) per emission instead of O(total). The + buffer is rebuilt from all fragments when the emission cannot be a + pure append: a non-live reason, a scan or display-window change, new + data at or below the frontier (late hole-fills, retention drops) or a + new row wider than the cached buffer. Every full rebuild reseeds the + cache. + + Args: + source (SourceData): The 1-D source snapshot. + reason (str): The update reason ("live", "backfill", ...). + + Returns: + np.ndarray | None: The (n_rows, max_len) buffer, or None if no + displayable rows remain. + """ + ordinals = source.ordinals + cache = self._waterfall_cache + if ( + cache is not None + and reason == "live" + and ordinals + and cache["scan_id"] == self.scan_id + and cache["min_display_ordinal"] == self._min_display_ordinal + ): + n_seen = cache["n_seen"] + frontier_intact = ( + len(ordinals) >= n_seen and ordinals[n_seen - 1] == cache["last_ordinal"] + ) + if frontier_intact and len(ordinals) == n_seen: + # Unchanged snapshot (the backend reuses source snapshots). + return cache["buffer"] + if frontier_intact and ordinals[n_seen] > cache["last_ordinal"]: + new_rows = [np.atleast_1d(np.asarray(value)) for value in source.values[n_seen:]] + new_rows = [row for row in new_rows if row.ndim == 1] + width = cache["width"] + if all(row.shape[0] <= width for row in new_rows): + buffer = cache["buffer"] + if new_rows: + padded = [ + np.pad( + row, (0, width - row.shape[0]), mode="constant", constant_values=0 + ) + for row in new_rows + ] + buffer = np.vstack([buffer, *padded]) + cache["buffer"] = buffer + cache["n_seen"] = len(ordinals) + cache["last_ordinal"] = ordinals[-1] + return buffer + buffer = self._rebuild_1d_buffer(source) + if ordinals: + self._waterfall_cache = { + "scan_id": self.scan_id, + "min_display_ordinal": self._min_display_ordinal, + "n_seen": len(ordinals), + "last_ordinal": ordinals[-1], + "buffer": buffer, + "width": 0 if buffer is None else buffer.shape[1], + } + else: + self._waterfall_cache = None + return buffer + + def _rebuild_1d_buffer(self, source) -> np.ndarray | None: + """ + From-scratch reference construction of the waterfall buffer (see + :meth:`_build_1d_buffer`): all fragments, zero-padded to the longest + row, restricted to the current display window. + Args: source (SourceData): The 1-D source snapshot. diff --git a/bec_widgets/widgets/plots/motor_map/motor_map.py b/bec_widgets/widgets/plots/motor_map/motor_map.py index 99fe0dfc..c346b08f 100644 --- a/bec_widgets/widgets/plots/motor_map/motor_map.py +++ b/bec_widgets/widgets/plots/motor_map/motor_map.py @@ -672,7 +672,7 @@ class MotorMap(PlotBase): sources=[(device_x, device_x), (device_y, device_y)], scan=None, parent=self, - min_emit_interval=0.1, + min_emit_interval=0.04, max_points=self.config.max_points, ) self._data_bridge.updated.connect(self._on_data_update) diff --git a/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py b/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py index ba553c9f..ce848bc0 100644 --- a/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py +++ b/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py @@ -616,7 +616,7 @@ class MultiWaveform(PlotBase): if self._source_key is None: return source = update.get(*self._source_key) - if source is None or not source.values: + if source is None or source.values is None or len(source.values) == 0: return current_scan_id = self._effective_scan_id(update, source) @@ -835,7 +835,7 @@ class MultiWaveform(PlotBase): sources=[(device, entry)], scan=scan, parent=self, - min_emit_interval=0.1, + min_emit_interval=0.04, max_points=max_points, ) self._data_bridge.updated.connect(self._on_data_update) diff --git a/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py b/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py index 66350b21..f9e3dec7 100644 --- a/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py +++ b/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py @@ -345,7 +345,7 @@ class ScatterWaveform(PlotBase): return try: self._data_bridge = QtDataSubscription( - self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.1 + self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.04 ) self._data_bridge.updated.connect(self._on_data_update) except Exception as exc: diff --git a/bec_widgets/widgets/plots/waveform/waveform.py b/bec_widgets/widgets/plots/waveform/waveform.py index d5c9f662..73f44907 100644 --- a/bec_widgets/widgets/plots/waveform/waveform.py +++ b/bec_widgets/widgets/plots/waveform/waveform.py @@ -1748,8 +1748,10 @@ class Waveform(PlotBase): if not sources: return try: + # 15 Hz render coalescing: above typical device message rates while + # leaving paint headroom for multi-million-point curves. self._data_bridge = QtDataSubscription( - self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.1 + self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.0667 ) self._data_bridge.updated.connect(self._on_data_update) except Exception as exc: @@ -1793,7 +1795,7 @@ class Waveform(PlotBase): sources=list(dict.fromkeys(sources)), scan=scan_id, parent=self, - min_emit_interval=0.1, + min_emit_interval=0.0667, ) bridge.updated.connect(self._on_data_update) self._history_bridges[scan_id] = bridge @@ -2040,7 +2042,7 @@ class Waveform(PlotBase): lengths only) or a same-length x device column, with an index fallback on any length mismatch. """ - y_data = self._async_display_values(source) + y_data = self._async_display_values_cached(curve, update, source) if y_data is None or len(y_data) == 0: return False mode = self.x_axis_mode["name"] or "auto" @@ -2093,7 +2095,7 @@ class Waveform(PlotBase): np.ndarray | None: The displayed y data. """ values = source.values - if not values: + if values is None or len(values) == 0: return None update_type = source.metadata.get("async_update_type") max_shape = source.metadata.get("max_shape") or [] @@ -2113,6 +2115,75 @@ class Waveform(PlotBase): return np.asarray(values) return np.atleast_1d(np.asarray(values[-1])) + def _async_display_values_cached(self, curve: Curve, update, source) -> np.ndarray | None: + """ + Incremental variant of :meth:`_async_display_values` for the one + display mode whose from-scratch cost grows with the scan — 1-D 'add' + concatenation. The concatenated buffer and the consumed ordinal + frontier are cached on the curve; per emission only the fragments + beyond the frontier are appended, keeping the per-message cost + O(new data) instead of O(total). + + The cache is dropped and the series rebuilt from all fragments when + the emission cannot be a pure append: a non-live reason (backfill, + history, rebind), a scan change, a source-key change, or new data at + or below the frontier (late hole-fills, retention drops). Every full + rebuild reseeds the cache, so a live stream resumes incrementally + after it. All other display modes are already O(new data) and are + delegated unchanged. + + Args: + curve(Curve): The rendered curve (cache carrier). + update(SubscriptionUpdate): The update snapshot. + source(SourceData): The async source snapshot. + + Returns: + np.ndarray | None: The displayed y data (identical to + :meth:`_async_display_values`). + """ + values = source.values + if values is None or len(values) == 0: + return None + update_type = source.metadata.get("async_update_type") + max_shape = source.metadata.get("max_shape") or [] + if update_type != "add" or len(max_shape) > 1: + # Last-fragment / last-row display modes: O(new data) already. + curve._data_api_async_cache = None + return self._async_display_values(source) + ordinals = source.ordinals + if ordinals is None or len(ordinals) == 0: + return self._async_display_values(source) + cache = getattr(curve, "_data_api_async_cache", None) + if ( + cache is not None + and update.reason == "live" + and cache["scan_id"] == update.scan_id + and cache["source_key"] == source.key + ): + n_seen = cache["n_seen"] + frontier_intact = ( + len(ordinals) >= n_seen and ordinals[n_seen - 1] == cache["last_ordinal"] + ) + if frontier_intact and len(ordinals) == n_seen: + # Unchanged snapshot (the backend reuses source snapshots). + return cache["buffer"] + if frontier_intact and ordinals[n_seen] > cache["last_ordinal"]: + new_fragments = [np.atleast_1d(np.asarray(value)) for value in values[n_seen:]] + buffer = np.concatenate([cache["buffer"], *new_fragments]) + cache["buffer"] = buffer + cache["n_seen"] = len(ordinals) + cache["last_ordinal"] = ordinals[-1] + return buffer + buffer = self._async_display_values(source) + curve._data_api_async_cache = { + "scan_id": update.scan_id, + "source_key": source.key, + "n_seen": len(ordinals), + "last_ordinal": ordinals[-1], + "buffer": buffer, + } + return buffer + def _auto_adjust_async_curve_settings( self, curve: Curve, diff --git a/tests/unit_tests/benchmarks/test_data_api_widget_throughput.py b/tests/unit_tests/benchmarks/test_data_api_widget_throughput.py new file mode 100644 index 00000000..e5286647 --- /dev/null +++ b/tests/unit_tests/benchmarks/test_data_api_widget_throughput.py @@ -0,0 +1,161 @@ +""" +Comparative throughput benchmark of the DataAPI widget rendering path. + +Feeds one async 'add' source through the waveform's ``_on_data_update`` +(``curve.setData`` mocked out, isolating the data-path cost) and compares it +with a main-style baseline: one ``np.hstack`` of (buffer, fragment) per +readback message for the same total data. The DataAPI path receives the data +in coalesced emissions (10 fragments per emission) and renders each one +incrementally, so it must not be slower than the per-message baseline. +""" + +from __future__ import annotations + +import sys +import time +from unittest.mock import MagicMock + +import numpy as np +import pytest +from bec_lib.data_api.models import SourceData, SubscriptionUpdate + +from bec_widgets.widgets.plots.waveform.waveform import Waveform +from tests.unit_tests.client_mocks import mocked_client +from tests.unit_tests.conftest import create_widget + +N_EMISSIONS = 150 +FRAGMENTS_PER_EMISSION = 10 +SAMPLES_PER_FRAGMENT = 500 +#: Timing rounds per path; the best round is compared (filters scheduler noise). +ROUNDS = 2 + + +@pytest.fixture +def waveform_widget(qtbot, mocked_client, monkeypatch): + """A Waveform with a stubbed data bridge and one async curve.""" + + def factory(client, sources, scan="live", parent=None, **kwargs): + bridge = MagicMock() + bridge.sources = list(sources) + bridge.scan_id = None if scan == "live" else scan + bridge.healthy = True + return bridge + + monkeypatch.setattr("bec_widgets.widgets.plots.waveform.waveform.QtDataSubscription", factory) + wf = create_widget(qtbot, Waveform, client=mocked_client) + yield wf + + +def _build_updates(fragments: list[np.ndarray]) -> list[SubscriptionUpdate]: + """ + One SubscriptionUpdate per emission, each snapshot extending the previous + one by FRAGMENTS_PER_EMISSION fragments. The fragment objects are shared + between snapshots, mimicking the backend's snapshot reuse. + """ + updates = [] + metadata = {"async_update_type": "add", "max_shape": [None], "acquisition_group": None} + for emission in range(N_EMISSIONS): + n_fragments = (emission + 1) * FRAGMENTS_PER_EMISSION + ordinals = tuple(range(n_fragments)) + source = SourceData( + device="async_device", + entry="async_device", + kind="async", + ordinals=ordinals, + values=tuple(fragments[:n_fragments]), + timestamps=tuple(float(i) for i in ordinals), + complete=True, + metadata=metadata, + ) + updates.append( + SubscriptionUpdate( + scan_id="benchmark_scan", + reason="live", + sources={source.key: source}, + aligned_ordinals=ordinals, + complete=True, + metadata={"group": "scan"}, + ) + ) + return updates + + +def _measure_dataapi(wf: Waveform, curve, updates: list[SubscriptionUpdate]) -> float: + """Time one full pass of the DataAPI path over all emissions.""" + curve._data_api_async_cache = None # each pass starts from an empty cache + start = time.perf_counter() + for update in updates: + wf._on_data_update(update) + return time.perf_counter() - start + + +def _measure_baseline(fragments: list[np.ndarray]) -> tuple[float, np.ndarray]: + """Time main's per-message loop: one np.hstack of (buffer, fragment) each.""" + start = time.perf_counter() + buffer = np.empty(0) + for fragment in fragments: + buffer = np.hstack((buffer, fragment)) + return time.perf_counter() - start, buffer + + +def test_data_api_waveform_throughput(waveform_widget, monkeypatch): + """ + The DataAPI rendering path (coalesced emissions, incremental append) must + match the per-message cost of main's incremental hstack loop for the same + total data. + """ + wf = waveform_widget + curve = wf.plot(arg1="async_device", label="async_device-async_device") + wf.scan_id = "benchmark_scan" + wf.x_axis_mode["name"] = "index" + monkeypatch.setattr(curve, "setData", lambda *args, **kwargs: None) + + rng = np.random.default_rng(42) + total_fragments = N_EMISSIONS * FRAGMENTS_PER_EMISSION + fragments = [rng.random(SAMPLES_PER_FRAGMENT) for _ in range(total_fragments)] + updates = _build_updates(fragments) + + # The mocked BEC client keeps polling threads alive; every numpy GIL + # release then stalls the timed loop for up to the thread switch interval + # (5 ms by default), swamping the actual computation. A short interval + # during the measurement removes that scheduler noise for both paths. + switch_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-4) + try: + # Warm-up: fault in pages and settle CPU scheduling before timing. + _measure_baseline(fragments[: total_fragments // 5]) + _measure_dataapi(wf, curve, updates[: N_EMISSIONS // 5]) + + # Alternate the measurement order between rounds and keep the best + # round of each path. + dataapi_times: list[float] = [] + baseline_times: list[float] = [] + baseline_buffer = None + for round_index in range(ROUNDS): + if round_index % 2 == 0: + baseline_time, baseline_buffer = _measure_baseline(fragments) + baseline_times.append(baseline_time) + dataapi_times.append(_measure_dataapi(wf, curve, updates)) + else: + dataapi_times.append(_measure_dataapi(wf, curve, updates)) + baseline_time, baseline_buffer = _measure_baseline(fragments) + baseline_times.append(baseline_time) + finally: + sys.setswitchinterval(switch_interval) + dataapi_time = min(dataapi_times) + baseline_time = min(baseline_times) + + # Both paths must have produced the identical series. + rendered = wf._async_display_values_cached( + curve, updates[-1], updates[-1].sources[("async_device", "async_device")] + ) + np.testing.assert_array_equal(rendered, baseline_buffer) + + print( + f"\nDataAPI path: {dataapi_time * 1000:.1f} ms for {N_EMISSIONS} emissions " + f"({total_fragments} fragments, {baseline_buffer.size} samples); " + f"main-style per-message baseline: {baseline_time * 1000:.1f} ms" + ) + assert dataapi_time <= baseline_time * 1.5, ( + f"DataAPI rendering path too slow: {dataapi_time:.3f}s vs " f"baseline {baseline_time:.3f}s" + ) diff --git a/tests/unit_tests/test_image_view_next_gen.py b/tests/unit_tests/test_image_view_next_gen.py index 0f892a4a..83d6566f 100644 --- a/tests/unit_tests/test_image_view_next_gen.py +++ b/tests/unit_tests/test_image_view_next_gen.py @@ -741,6 +741,69 @@ def test_scan_rollover_same_scan_noop(qtbot, mocked_client, monkeypatch): assert view.main_image.raw_data.shape == (2, 3) +def test_build_1d_buffer_incremental_matches_full_rebuild(qtbot, mocked_client, monkeypatch): + """ + The incremental waterfall buffer must equal the from-scratch construction + across live appends (contiguous and gapped), an unchanged reused + snapshot, an out-of-order hole-fill, a wider new row, a non-live reason + and a scan change — and must only fall back to the full rebuild for the + emissions that cannot be pure appends. + """ + _fake_bridge_factory(monkeypatch) + view = create_widget(qtbot, Image, client=mocked_client) + _set_signal_config( + mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj" + ) + view.image(device="eiger", signal="img") + + from_scratch = Image._rebuild_1d_buffer + rebuilds = [] + + def counting(self, source): + rebuilds.append(source) + return from_scratch(self, source) + + monkeypatch.setattr(Image, "_rebuild_1d_buffer", counting) + + def reference(values): + rows = [np.atleast_1d(np.asarray(value)) for value in values] + width = max(row.shape[0] for row in rows) + return np.vstack([np.pad(row, (0, width - row.shape[0])) for row in rows]) + + steps = [ + # (scan_id, reason, values, ordinals, expected rebuild count so far) + ("scan_1", "live", [[1.0, 2.0]], (0,), 1), # first emission seeds the cache + ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0]], (0, 1), 1), # append + ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0]], (0, 1), 1), # unchanged reused snapshot + ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [5.0]], (0, 1, 3), 1), # gapped short append + # late hole-fill below the frontier -> full rebuild + ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0]], (0, 1, 2, 3), 2), + # new row wider than the cached buffer -> full rebuild + ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0], [6.0] * 4], (0, 1, 2, 3, 4), 3), + # non-live reason -> full rebuild + ( + "scan_1", + "backfill", + [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0], [6.0] * 4], + (0, 1, 2, 3, 4), + 4, + ), + ("scan_2", "live", [[9.0, 9.0]], (0,), 5), # scan change -> full rebuild + ("scan_2", "live", [[9.0, 9.0], [10.0, 11.0]], (0, 1), 5), # incremental resumes + ] + for scan_id, reason, values, ordinals, expected_rebuilds in steps: + source = _make_source( + "eiger", + "async_obj", + values, + ordinals=ordinals, + metadata={"async_update_type": "add", "max_shape": [None]}, + ) + view._on_data_update(_make_update(source, scan_id=scan_id, reason=reason)) + np.testing.assert_array_equal(view.main_image.raw_data, reference(values)) + assert len(rebuilds) == expected_rebuilds + + def test_image_data_update_2d(qtbot, mocked_client, monkeypatch): _fake_bridge_factory(monkeypatch) bec_image_view = create_widget(qtbot, Image, client=mocked_client) diff --git a/tests/unit_tests/test_waveform.py b/tests/unit_tests/test_waveform.py index cd127b49..7ded46a5 100644 --- a/tests/unit_tests/test_waveform.py +++ b/tests/unit_tests/test_waveform.py @@ -113,20 +113,21 @@ def _fake_bridge_factory(monkeypatch, gated_bytes: int | None = None): return created -def _monitored_source(device, values, entry=None, timestamps=None, ordinals=None): +def _monitored_source(device, values, entry=None, timestamps=None, ordinals=None, as_numpy=False): from bec_lib.data_api.models import SourceData entry = entry or device ordinals = tuple(range(len(values))) if ordinals is None else tuple(ordinals) if timestamps is None: timestamps = tuple(float(i) for i in ordinals) + wrap = (lambda seq: np.asarray(seq)) if as_numpy else tuple return SourceData( device=device, entry=entry, kind="monitored", - ordinals=ordinals, - values=tuple(values), - timestamps=tuple(timestamps), + ordinals=wrap(ordinals), + values=wrap(values), + timestamps=wrap(timestamps), complete=True, ) @@ -140,6 +141,7 @@ def _async_source( max_shape=(None,), kind="async", ordinals=None, + as_numpy=False, ): from bec_lib.data_api.models import SourceData @@ -147,13 +149,14 @@ def _async_source( ordinals = tuple(range(len(values))) if ordinals is None else tuple(ordinals) if timestamps is None: timestamps = tuple(float(i) for i in ordinals) + wrap = (lambda seq: np.asarray(seq)) if as_numpy else tuple return SourceData( device=device, entry=entry, kind=kind, - ordinals=ordinals, - values=tuple(values), - timestamps=tuple(timestamps), + ordinals=wrap(ordinals), + values=wrap(values), + timestamps=wrap(timestamps), complete=True, metadata={ "async_update_type": update_type, @@ -1324,6 +1327,55 @@ def test_on_data_update_async_history_rows(qtbot, mocked_client, monkeypatch): np.testing.assert_array_equal(y_data, [4, 5, 6]) +def test_on_data_update_async_add_incremental_matches_full_rebuild( + qtbot, mocked_client, monkeypatch +): + """ + The incremental 1-D 'add' render path must yield exactly the series the + from-scratch concatenation yields, across live appends (contiguous and + gapped), an unchanged reused snapshot, an out-of-order hole-fill, a + non-live reason and a scan change — and must only fall back to the full + rebuild for the emissions that cannot be pure appends. + """ + _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 = None # render updates of any scan + wf.x_axis_mode["name"] = "index" + + from_scratch = Waveform._async_display_values + rebuilds = [] + + def counting(source): + rebuilds.append(source) + return from_scratch(source) + + monkeypatch.setattr(Waveform, "_async_display_values", staticmethod(counting)) + + steps = [ + # (scan_id, reason, values, ordinals, expected rebuild count so far) + ("scan_1", "live", ([0.0, 1.0],), (0,), 1), # first emission seeds the cache + ("scan_1", "live", ([0.0, 1.0], [2.0]), (0, 1), 1), # append + ("scan_1", "live", ([0.0, 1.0], [2.0]), (0, 1), 1), # unchanged reused snapshot + ("scan_1", "live", ([0.0, 1.0], [2.0], [4.0, 5.0]), (0, 1, 3), 1), # gapped append + # late hole-fill below the frontier -> full rebuild + ("scan_1", "live", ([0.0, 1.0], [2.0], [3.0], [4.0, 5.0]), (0, 1, 2, 3), 2), + ("scan_1", "live", ([0.0, 1.0], [2.0], [3.0], [4.0, 5.0], [6.0]), (0, 1, 2, 3, 4), 2), + # non-live reason -> full rebuild + ("scan_1", "backfill", ([0.0, 1.0], [2.0], [3.0], [4.0, 5.0], [6.0]), tuple(range(5)), 3), + ("scan_2", "live", ([7.0],), (0,), 4), # scan change -> full rebuild + ("scan_2", "live", ([7.0], [8.0, 9.0]), (0, 1), 4), # incremental resumes + ] + for scan_id, reason, values, ordinals, expected_rebuilds in steps: + source = _async_source("async_device", values=values, ordinals=ordinals, update_type="add") + wf._on_data_update(_make_update([source], scan_id=scan_id, reason=reason)) + x_data, y_data = c.get_data() + expected = from_scratch(source) + np.testing.assert_array_equal(y_data, expected) + np.testing.assert_array_equal(x_data, np.arange(len(expected))) + assert len(rebuilds) == expected_rebuilds + + ################################################## # The following tests are for the Curve class ################################################## @@ -2125,3 +2177,44 @@ def test_detector_shaped_history_curve_is_not_hidden(qtbot, mocked_client, monke scan_item._msg.num_monitored_readouts = 37 scan_item._msg.num_points = 37 assert wf._history_curve_compatible(curve_for("bpm4i", "bpm4i")) is False + + +def test_history_source_with_numpy_columns_renders(qtbot, mocked_client, monkeypatch): + """Regression: the history plugin delivers numpy-array columns (bulk + ingest keeps the file arrays intact). The render must accept them; a + ``if not source.values`` truth-test raised ValueError on multi-element + arrays and SafeSlot swallowed it, so the curve showed no data.""" + _fake_bridge_factory(monkeypatch) + wf = create_widget(qtbot, Waveform, client=mocked_client) + wf.x_axis_mode["name"] = "index" + + # monitored history column, numpy-valued (what bec_lib now emits) + c = wf.plot(arg1="bpm4i", label="bpm4i-bpm4i") + wf.scan_id = "dummy" + src = _monitored_source("bpm4i", values=[5.0, 6.0, 7.0, 8.0], as_numpy=True) + wf._on_data_update(_make_update([src], reason="history")) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(y_data, [5.0, 6.0, 7.0, 8.0]) + + # async history waveform, flat numpy value column, no async_update_type + ca = wf.plot(arg1="async_device", label="async_device-async_device") + src_a = _async_source( + "async_device", values=[1.0, 2.0, 3.0, 4.0, 5.0], update_type=None, as_numpy=True + ) + # history reads carry no async_update_type + src_a.metadata.pop("async_update_type", None) + wf._on_data_update(_make_update([src_a], reason="history")) + x_data, y_data = ca.get_data() + assert y_data is not None and len(y_data) == 5 + + +def test_async_display_values_accepts_numpy(qtbot, mocked_client): + """Both display helpers must handle numpy-valued sources (bulk history).""" + from types import SimpleNamespace + + src = _async_source("async_device", values=[1, 2, 3], update_type="add", as_numpy=True) + assert list(Waveform._async_display_values(src)) == [1, 2, 3] + wf = create_widget(qtbot, Waveform, client=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]