refactor(waveform): consume the data api for live and history curves

bec_widgets/utils/qt_data_subscription.py

@@ -36,6 +36,7 @@ class QtDataSubscription(QObject):
         parent: QObject | None = None,
         min_emit_interval: float = 0.1,
         max_points: int | None = None,
+        size_limit_bytes: int | None = None,
     ):
         """
         Subscribe to data for the given sources.
@@ -54,6 +55,11 @@ class QtDataSubscription(QObject):
             max_points (int | None): Per-source retention cap; oldest points
                 are dropped beyond it. Recommended for endless device-stream
                 subscriptions (``scan=None``).
+            size_limit_bytes (int | None): Withhold the load when the backend
+                can estimate the payload up front (history scans) and the
+                estimate exceeds this limit. Nothing is read; the bridge
+                reports :attr:`size_gated` with :attr:`estimated_bytes` and
+                waits for :meth:`confirm_size`.
 
         Raises:
             ValueError: If a concrete scan id cannot be served.
@@ -74,6 +80,7 @@ class QtDataSubscription(QObject):
             callback=self._deliver,
             min_emit_interval=min_emit_interval,
             max_points=max_points,
+            size_limit_bytes=size_limit_bytes,
         )
         self.destroyed.connect(lambda: self.close())
 
@@ -112,6 +119,25 @@ class QtDataSubscription(QObject):
         """Whether every declared source is currently delivering."""
         return not self._subscription.unbound_sources
 
+    @property
+    def size_gated(self) -> bool:
+        """Whether delivery is withheld pending :meth:`confirm_size`."""
+        return bool(self._subscription.size_gated)
+
+    @property
+    def estimated_bytes(self) -> int | None:
+        """Estimated payload size of the bound scan, if the backend knows it."""
+        return self._subscription.estimated_bytes
+
+    def confirm_size(self) -> None:
+        """
+        Release a load withheld by ``size_limit_bytes``.
+
+        Returns immediately: the file read runs on the backend's worker
+        thread and the data arrives later through :attr:`updated`.
+        """
+        self._subscription.confirm_size()
+
     def set_sources(self, sources: list[SourceKey]) -> None:
         """Atomically replace the source set."""
         self._subscription.set_sources(sources)

bec_widgets/widgets/plots/waveform/curve.py

@@ -105,7 +105,6 @@ class Curve(BECConnector, pg.PlotDataItem):
         self.apply_config()
         self.dap_params = None
         self.dap_summary = None
-        self.slice_index = None
         self._data_version = 0
         self._last_dap_request_fingerprint: tuple | None = None
         if kwargs:

tests/unit_tests/test_qt_data_subscription.py

@@ -73,6 +73,41 @@ def test_health_and_source_delegation(fake_api):
     assert subscription.close.call_count == 1
 
 
+def test_size_limit_is_forwarded_to_the_backend(fake_api):
+    api, subscription = fake_api
+    bridge = QtDataSubscription(
+        mock.MagicMock(), sources=[("samx", "samx")], scan="scan_1", size_limit_bytes=1024
+    )
+    assert api.subscribe.call_args.kwargs["size_limit_bytes"] == 1024
+    bridge.close()
+
+
+def test_size_limit_defaults_to_none(fake_api):
+    api, _ = fake_api
+    bridge = QtDataSubscription(mock.MagicMock(), sources=[("samx", "samx")])
+    assert api.subscribe.call_args.kwargs["size_limit_bytes"] is None
+    bridge.close()
+
+
+def test_size_gate_properties_and_confirm(fake_api):
+    _, subscription = fake_api
+    subscription.size_gated = True
+    subscription.estimated_bytes = 4096
+    bridge = QtDataSubscription(
+        mock.MagicMock(), sources=[("samx", "samx")], scan="scan_1", size_limit_bytes=1024
+    )
+
+    assert bridge.size_gated is True
+    assert bridge.estimated_bytes == 4096
+
+    bridge.confirm_size()
+    subscription.confirm_size.assert_called_once_with()
+
+    subscription.size_gated = False
+    assert bridge.size_gated is False
+    bridge.close()
+
+
 def test_synchronous_initial_delivery_is_queued(qtbot, monkeypatch):
     """The backend delivers the initial backfill synchronously inside
     subscribe() on the Qt thread; the bridge must neither crash on its