From 970c73e7678a5669dbe3e6726b71083873022211 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Fri, 7 Aug 2026 12:49:01 +0200 Subject: [PATCH] refactor(waveform): consume the data api for live and history curves --- bec_widgets/utils/qt_data_subscription.py | 26 + bec_widgets/widgets/plots/waveform/curve.py | 1 - .../widgets/plots/waveform/waveform.py | 1323 +++++++++-------- tests/unit_tests/test_qt_data_subscription.py | 35 + tests/unit_tests/test_waveform.py | 998 ++++++++----- 5 files changed, 1408 insertions(+), 975 deletions(-) diff --git a/bec_widgets/utils/qt_data_subscription.py b/bec_widgets/utils/qt_data_subscription.py index b21fdc9b..5c0ee5f0 100644 --- a/bec_widgets/utils/qt_data_subscription.py +++ b/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) diff --git a/bec_widgets/widgets/plots/waveform/curve.py b/bec_widgets/widgets/plots/waveform/curve.py index 6644b074..efc61504 100644 --- a/bec_widgets/widgets/plots/waveform/curve.py +++ b/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: diff --git a/bec_widgets/widgets/plots/waveform/waveform.py b/bec_widgets/widgets/plots/waveform/waveform.py index 123c5295..d5c9f662 100644 --- a/bec_widgets/widgets/plots/waveform/waveform.py +++ b/bec_widgets/widgets/plots/waveform/waveform.py @@ -31,13 +31,10 @@ from bec_widgets.utils.bec_signal_proxy import BECSignalProxy from bec_widgets.utils.colors import Colors, apply_theme from bec_widgets.utils.container_utils import WidgetContainerUtils from bec_widgets.utils.error_popups import SafeProperty, SafeSlot +from bec_widgets.utils.qt_data_subscription import QtDataSubscription from bec_widgets.utils.settings_dialog import SettingsDialog from bec_widgets.utils.side_panel import SidePanel -from bec_widgets.utils.signal_classification import ( - ASYNC_SIGNAL_CLASSES, - SignalCategory, - classify_device_signal, -) +from bec_widgets.utils.signal_classification import SignalCategory, classify_device_signal from bec_widgets.utils.toolbars.bundles import ToolbarBundle from bec_widgets.utils.toolbars.toolbar import MaterialIconAction from bec_widgets.widgets.dap.lmfit_dialog.lmfit_dialog import LMFitDialog @@ -115,8 +112,6 @@ class Waveform(PlotBase): "clear_all", ] - sync_signal_update = Signal() - async_signal_update = Signal() request_dap_update = Signal() unblock_dap_proxy = Signal() dap_params_update = Signal(dict, dict) @@ -146,17 +141,20 @@ class Waveform(PlotBase): # Curve data self._sync_curves = [] self._async_curves = [] - # Async streams already subscribed in the current scan. Keyed by - # (device, stream) (stream is None for the old per-device endpoint); the value - # is the list of signal entries that share that single subscription. - self._async_streams_setup: dict = {} self._history_curves = [] - self._slice_index = None self._dap_curves = [] self._mode = None + # Data delivery through the DataAPI: one bridge follows the widget's + # scan (live or a history scan pinned via update_with_scan_history), + # plus one bridge per scan pinned by individual history curves. + self._data_bridge: QtDataSubscription | None = None + self._data_api_scan: str = "live" + self._history_bridges: dict[str, QtDataSubscription] = {} + self._history_x_keys: dict[str, tuple[str, str] | None] = {} + self._shutting_down = False + # Scan data - self._scan_done = True # means scan is not running self.old_scan_id = None self.scan_id = None self.scan_item = None @@ -185,21 +183,16 @@ class Waveform(PlotBase): self._init_alignment_mode() self.curve_settings_dialog = None - # Large‑dataset guard + # Large-dataset guard self._skip_large_dataset_warning = False # session flag + self._size_confirmed_sources: set = set() # (scan_id, source) confirm memo self._skip_large_dataset_check = False # per-plot flag, to skip the warning for this plot - # Scan status update loop + # Scan status is only used for per-scan bookkeeping (reset, scan_id, + # categorisation, DAP triggering); the data flows through the DataAPI. self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status()) - self.bec_dispatcher.connect_slot(self.on_scan_progress, MessageEndpoints.scan_progress()) - # Curve update loop - self.proxy_update_sync = pg.SignalProxy( - self.sync_signal_update, rateLimit=25, slot=self.update_sync_curves - ) - self.proxy_update_async = pg.SignalProxy( - self.async_signal_update, rateLimit=25, slot=self.update_async_curves - ) + # DAP update loop self.proxy_dap_request = BECSignalProxy( self.request_dap_update, rateLimit=25, slot=self.request_dap, timeout=10.0 ) @@ -717,8 +710,8 @@ class Waveform(PlotBase): self._current_x_device = None self._refresh_history_curves() self._update_curve_visibility() - self.async_signal_update.emit() - self.sync_signal_update.emit() + # Rebuild the DataAPI subscription so the x source follows the new mode. + self._setup_data_api_subscription(scan=self._data_api_scan) self.plot_item.enableAutoRange(x=True) self.round_plot_widget.apply_plot_widget_style() # To keep the correct theme self._refresh_alignment_state(force_readback=True) @@ -747,8 +740,8 @@ class Waveform(PlotBase): self._switch_x_axis_item(mode="device") self._refresh_history_curves() self._update_curve_visibility() - self.async_signal_update.emit() - self.sync_signal_update.emit() + # Rebuild the DataAPI subscription so the x source follows the new signal. + self._setup_data_api_subscription(scan=self._data_api_scan) self.plot_item.enableAutoRange(x=True) self.round_plot_widget.apply_plot_widget_style() self._refresh_alignment_state(force_readback=True) @@ -1160,25 +1153,27 @@ class Waveform(PlotBase): raise ValueError("For 'custom' curves, x_data and y_data must be provided.") # Actually create the Curve item - curve = self._add_curve_object(name=label, config=config, scan_item=scan_item) + curve = self._add_curve_object(name=label, config=config) # If custom => set initial data if config.source == "custom" and x_data is not None and y_data is not None: curve.setData(x_data, y_data) - # If device => schedule BEC updates + # If device => let the DataAPI subscription deliver the data if config.source == "device": if self.scan_item is None: self.update_with_scan_history(-1) - self.async_signal_update.emit() - self.sync_signal_update.emit() + self._setup_data_api_subscription(scan=self._data_api_scan) if config.source == "dap": self._dap_curves.append(curve) self.setup_dap_for_scan() self.roi_enable.emit(True) # Enable the ROI toolbar action self.request_dap() # Request DAP update directly without blocking proxy if config.source == "history": + self._validate_history_curve(curve, scan_item) self._history_curves.append(curve) + self._sync_history_curve_state(curve) + self._setup_history_curve_subscriptions() QTimer.singleShot( 150, self.auto_range @@ -1187,174 +1182,158 @@ class Waveform(PlotBase): return curve - def _add_curve_object( - self, name: str, config: CurveConfig, scan_item: ScanDataContainer | None = None - ) -> Curve | None: + def _add_curve_object(self, name: str, config: CurveConfig) -> Curve: """ Low-level creation of the PlotDataItem (Curve) from a `CurveConfig`. Args: name (str): The name/label of the curve. config (CurveConfig): Configuration model describing the curve. - scan_item (ScanDataContainer | None): Optional scan item for history curves. Returns: Curve: The newly created curve object, added to the plot. """ curve = Curve(config=config, name=name, parent_item=self) self.plot_item.addItem(curve) - if scan_item is not None: - self._fetch_history_data_for_curve(curve, scan_item) self._categorise_device_curves() curve.visibleChanged.connect(self._refresh_crosshair_markers) curve.visibleChanged.connect(self.auto_range) return curve - def _fetch_history_data_for_curve( - self, curve: Curve, scan_item: ScanDataContainer - ) -> Curve | None: - # Check if the data are already set - device = curve.config.signal.device - entry = curve.config.signal.signal + def _validate_history_curve(self, curve: Curve, scan_item: ScanDataContainer) -> None: + """ + Metadata-level validation of a new history curve (no file I/O). - all_devices_used = getattr( + Args: + curve(Curve): The freshly created history curve. + scan_item(ScanDataContainer): The resolved scan item. + + Raises: + ValueError: If the scan item carries no stored-data info or the + requested device/entry is not part of the scan. The curve is + removed from the plot before raising. + """ + stored_data_info = getattr( getattr(scan_item, "_msg", None), "stored_data_info", None ) or getattr(scan_item, "stored_data_info", None) - if all_devices_used is None: + device = curve.config.signal.device + entry = curve.config.signal.signal + if stored_data_info is None: curve.remove() raise ValueError( f"No stored data info found in scan item ID:{curve.config.scan_id} for curve '{curve.name()}'. " f"Upgrade BEC to the latest version." ) - - # 1. get y data - x_data, y_data = None, None - if device not in all_devices_used: + if device not in stored_data_info: + curve.remove() raise ValueError(f"Device '{device}' not found in scan item ID:{curve.config.scan_id}.") - if entry not in all_devices_used[device]: + if entry not in stored_data_info[device]: + curve.remove() raise ValueError( f"Entry '{entry}' not found in device '{device}' in scan item ID:{curve.config.scan_id}." ) - y_shape = all_devices_used.get(device).get(entry).shape[0] - # Determine X-axis data - if self.x_axis_mode["name"] == "index": - x_data = np.arange(y_shape) - curve.config.current_x_mode = "index" - self._update_x_label_suffix(" (index)") - elif self.x_axis_mode["name"] == "timestamp": - y_device = scan_item.devices.get(device) - x_data = y_device.get(entry).read().get("timestamp") - curve.config.current_x_mode = "timestamp" - self._update_x_label_suffix(" (timestamp)") - elif self.x_axis_mode["name"] not in ("index", "timestamp", "auto"): # Custom device mode - if self.x_axis_mode["name"] not in all_devices_used: - logger.warning( - f"Custom device '{self.x_axis_mode['name']}' not found in scan item of history curve '{curve.name()}'; scan ID: {curve.config.scan_id}." - ) - curve.setVisible(False) - return - signal_x_custom = self.x_axis_mode.get("entry") - if signal_x_custom is None: - signal_x_custom = self.entry_validator.validate_signal( - self.x_axis_mode["name"], None - ) - if signal_x_custom not in all_devices_used[self.x_axis_mode["name"]]: - logger.warning( - f"Custom entry '{signal_x_custom}' for device '{self.x_axis_mode['name']}' not found in scan item of history curve '{curve.name()}'; scan ID: {curve.config.scan_id}." - ) - curve.setVisible(False) - return - x_shape = ( - scan_item._msg.stored_data_info.get(self.x_axis_mode["name"]) - .get(signal_x_custom) - .shape[0] - ) - if x_shape != y_shape: - logger.warning( - f"Shape mismatch for x data '{x_shape}' and y data '{y_shape}' in history curve '{curve.name()}'; scan ID: {curve.config.scan_id}." - ) - curve.setVisible(False) - return - x_device = scan_item.devices.get(self.x_axis_mode["name"]) - x_data = x_device.get(signal_x_custom).read().get("value") - curve.config.current_x_mode = self.x_axis_mode["name"] - self._update_x_label_suffix(f" (custom: {self.x_axis_mode['name']}-{signal_x_custom})") - elif self.x_axis_mode["name"] == "auto": - if ( - self._current_x_device is None - ): # Scenario where no x device is set yet, because there was no live scan done in this widget yet - # If no current x device, use the first motor from scan item - scan_motors = self._ensure_str_list( - scan_item.metadata.get("bec").get("scan_report_devices") - ) - if not scan_motors: # scan was done without reported motor from whatever reason - x_data = np.arange(y_shape) # Fallback to index - y_data = scan_item.devices.get(device).get(entry).read().get("value") - curve.set_data(x=x_data, y=y_data) - self._update_x_label_suffix(" (auto: index)") - return curve - signal_x = self.entry_validator.validate_signal(scan_motors[0], None) - if signal_x not in all_devices_used.get(scan_motors[0], {}): - logger.warning( - f"Auto x entry '{signal_x}' for device '{scan_motors[0]}' not found in scan item of history curve '{curve.name()}'; scan ID: {curve.config.scan_id}." - ) - curve.setVisible(False) - return - if y_shape != all_devices_used.get(scan_motors[0]).get(signal_x, {}).shape[0]: - logger.warning( - f"Shape mismatch for x data '{all_devices_used.get(scan_motors[0]).get(signal_x, {}).get('shape', [0])[0]}' and y data '{y_shape}' in history curve '{curve.name()}'; scan ID: {curve.config.scan_id}." - ) - curve.setVisible(False) - return - x_data = scan_item.devices.get(scan_motors[0]).get(signal_x).read().get("value") - self._current_x_device = (scan_motors[0], signal_x) - self._update_x_label_suffix(f" (auto: {scan_motors[0]}-{signal_x})") - curve.config.current_x_mode = "auto" - self._update_x_label_suffix(f" (auto: {scan_motors[0]}-{signal_x})") - else: # Scan in auto mode was done and live scan already set the current x device - if self._current_x_device[0] not in all_devices_used: - logger.warning( - f"Auto x data for device '{self._current_x_device[0]}' " - f"and entry '{self._current_x_device[1]}'" - f" not found in scan item of the history curve {curve.name()}." - ) - curve.setVisible(False) - return - x_device = scan_item.devices.get(self._current_x_device[0]) - x_data = x_device.get(self._current_x_device[1]).read().get("value") - curve.config.current_x_mode = "auto" - self._update_x_label_suffix( - f" (auto: {self._current_x_device[0]}-{self._current_x_device[1]})" - ) - if x_data is None: + @staticmethod + def _history_num_points(scan_item) -> int | None: + """Number of monitored readouts of a history scan, if known.""" + msg = getattr(scan_item, "_msg", None) + return getattr(msg, "num_monitored_readouts", None) or getattr(msg, "num_points", None) + + def _is_monitored_shaped(self, scan_item, info) -> bool: + """ + Whether a stored dataset holds exactly one value per scan point. + + Only such datasets can be plotted against a monitored x device; every + other shape is detector data plotted against its sample index. + """ + shape = tuple(getattr(info, "shape", None) or ()) + num_points = self._history_num_points(scan_item) + return len(shape) == 1 and num_points is not None and shape[0] == num_points + + def _history_curve_compatible(self, curve: Curve) -> bool: + """ + Whether a history curve can be shown with the current x-axis mode. + + Answered from the scan's stored-data info (no file data reads): the + x device/entry must exist in the scan and match the y shape. Index and + timestamp modes are always compatible; auto falls back to index when + no x device can be resolved. + + Args: + curve(Curve): The history curve to check. + + Returns: + bool: True if the curve is compatible with the current x mode. + """ + mode = self.x_axis_mode.get("name") or "auto" + if mode in ("index", "timestamp"): + return True + scan_item = self.get_history_scan_item( + scan_id=curve.config.scan_id, scan_index=curve.config.scan_number + ) + if scan_item is None: + logger.warning(f"Scan item for curve {curve.name()} not found.") + return False + stored_data_info = getattr( + getattr(scan_item, "_msg", None), "stored_data_info", None + ) or getattr(scan_item, "stored_data_info", None) + if not stored_data_info: + return True # cannot verify; let the delivered data decide + signal = curve.config.signal + y_info = stored_data_info.get(signal.device, {}).get(signal.signal) if signal else None + if y_info is None: + return False + if not self._is_monitored_shaped(scan_item, y_info): + # Detector/async data: one row per acquisition, not per scan point. + # It is rendered against the sample index (with an index fallback + # when an x device has a different length), so an x/y row-count + # difference is expected and must not hide the curve. + return True + x_key = self._history_x_source_key(scan_item) + if x_key is None: + # Auto mode falls back to index plotting; an unresolvable custom + # device hides the curve (mirrors the legacy behaviour). + return mode == "auto" + x_info = stored_data_info.get(x_key[0], {}).get(x_key[1]) + if x_info is None: logger.warning( - f"X data for curve '{curve.name()}' could not be determined. " - f"Check if the x_mode '{self.x_axis_mode['name']}' is valid for the scan item." + f"X device '{x_key[0]}-{x_key[1]}' not found in scan item of history curve " + f"'{curve.name()}'; scan ID: {curve.config.scan_id}." ) - curve.setVisible(False) - return - if y_data is None: - y_data = scan_item.devices.get(device).get(entry).read().get("value") - if y_data is None: + return False + try: + if tuple(x_info.shape)[0] != tuple(y_info.shape)[0]: logger.warning( - f"Y data for curve '{curve.name()}' could not be determined. " - f"Check if the device '{device}' and entry '{entry}' are valid for the scan item." + f"Shape mismatch for x data '{x_info.shape[0]}' and y data '{y_info.shape[0]}' " + f"in history curve '{curve.name()}'; scan ID: {curve.config.scan_id}." ) - curve.setVisible(False) - return - curve.set_data(x=x_data, y=y_data) - return curve + return False + except (TypeError, IndexError): + return True + return True + + def _sync_history_curve_state(self, curve: Curve) -> None: + """ + Sync visibility and the recorded x mode of one history curve with the + current x-axis mode. + + Args: + curve(Curve): The history curve to sync. + """ + compatible = self._history_curve_compatible(curve) + curve.setVisible(compatible) + if compatible: + curve.config.current_x_mode = self.x_axis_mode["name"] def _refresh_history_curves(self): + """ + Re-sync history curves after an x-mode change: visibility comes from + the stored-data metadata, the data itself is re-delivered through + fresh scan-bound DataAPI subscriptions. + """ for curve in self._history_curves: - scan_item = self.get_history_scan_item( - scan_id=curve.config.scan_id, scan_index=curve.config.scan_number - ) - if scan_item is not None: - self._fetch_history_data_for_curve(curve, scan_item) - else: - logger.warning(f"Scan item for curve {curve.name()} not found.") + self._sync_history_curve_state(curve) + self._setup_history_curve_subscriptions() def _refresh_crosshair_markers(self): """ @@ -1399,27 +1378,19 @@ class Waveform(PlotBase): Return True when *curve* can be shown with the current x-axis mode. - ‘index’, ‘timestamp’ are always compatible. - - For history curves we check whether the requested motor - (self.x_axis_mode["name"]) exists in the cached - history_data_buffer["x"] dictionary. + - History curves are checked against their scan's stored-data info. - DAP is done by checking if the parent curve is visible. - - Device curves are fetched by update sync/async curves, which solves the compatibility there. + - Device curves are rendered from the DataAPI updates, which resolve + the compatibility (index fallback) there. """ mode = self.x_axis_mode.get("name", "index") if mode in ("index", "timestamp"): # always compatible - wild west mode return True if curve.config.source == "history": - scan_item = self.get_history_scan_item( - scan_id=curve.config.scan_id, scan_index=curve.config.scan_number - ) - curve = self._fetch_history_data_for_curve(curve, scan_item) - if curve is None: - return False + return self._history_curve_compatible(curve) if curve.config.source == "dap": parent_curve = self._find_curve_by_label(curve.config.parent_label) - if parent_curve.isVisible(): - return True - return False # DAP curve is not compatible if parent curve is not visible + return bool(parent_curve is not None and parent_curve.isVisible()) return True def _update_curve_visibility(self) -> None: @@ -1435,6 +1406,7 @@ class Waveform(PlotBase): self._dap_curves = [] self._sync_curves = [] self._async_curves = [] + self._history_curves = [] for curve in curve_list: self.remove_curve(curve.name()) if self.crosshair is not None: @@ -1475,6 +1447,7 @@ class Waveform(PlotBase): self._refresh_colors() self._categorise_device_curves() + self._refresh_data_subscriptions() self._refresh_alignment_state() def _remove_curve_by_name(self, name: str): @@ -1508,15 +1481,13 @@ class Waveform(PlotBase): def _curve_clean_up(self, curve: Curve): """ - Clean up the curve by disconnecting the async update signal (even for sync curves). + Clean up the curve bookkeeping and RPC registration. Args: curve(Curve): The curve to clean up. """ - self.bec_dispatcher.disconnect_slot( - self.on_async_readback, - MessageEndpoints.device_async_readback(self.scan_id, curve.name()), - ) + if curve in self._history_curves: + self._history_curves.remove(curve) curve.rpc_register.remove_rpc(curve) # Remove itself from the DAP summary only for side panels @@ -1572,7 +1543,8 @@ class Waveform(PlotBase): def on_scan_status(self, msg: dict, meta: dict): """ Initial scan status message handler, which is triggered at the begging and end of scan. - Used for triggering the update of the sync and async curves. + Used only for per-scan bookkeeping (reset, scan id tracking, curve + categorisation, DAP triggering); the data flows through the DataAPI. Args: msg(dict): The message content. @@ -1594,336 +1566,552 @@ class Waveform(PlotBase): self.old_scan_id = self.scan_id self.scan_id = current_scan_id self.scan_item = self.queue.scan_storage.find_scan_by_ID(self.scan_id) # live scan - self._slice_index = None # Reset the slice index self._update_curve_visibility() self._mode = self._categorise_device_curves() - - # First trigger to sync and async data. Async curve subscriptions - # were already set up by _categorise_device_curves above (it calls - # _setup_async_curve per async curve for live scans); repeating it - # here cleared each curve's data a second time and re-attempted - # the stream registration for every new scan. - if self._mode == "sync": - self.sync_signal_update.emit() - logger.info("Scan status: Sync mode") - elif self._mode == "async": - self.async_signal_update.emit() - logger.info("Scan status: Async mode") - else: - self.sync_signal_update.emit() - self.async_signal_update.emit() - logger.info("Scan status: Mixed mode") - logger.warning("Mixed mode - integrity of x axis cannot be guaranteed.") + # Rebuild the subscription so the per-scan source resolution + # (e.g. the auto x device) follows the new scan. + self._setup_data_api_subscription(scan="live") self.setup_dap_for_scan() - @SafeSlot(dict, dict) - def on_scan_progress(self, msg: dict, meta: dict): - """ - Slot for handling scan progress messages. Used for triggering the update of the sync curves. + ################################################################################ + # DataAPI subscription handling + ################################################################################ - Args: - msg(dict): The message content. - meta(dict): The message metadata. + def _config_sources(self) -> list[tuple[str, str]] | None: """ - self.sync_signal_update.emit() - self._scan_done = msg.get("done") - if self._scan_done: - QTimer.singleShot(100, self.update_sync_curves) - QTimer.singleShot(300, self.update_sync_curves) - - def _fetch_scan_data_and_access(self) -> tuple[dict, str] | tuple[None, None]: - """ - Decide whether the widget is in live or historical mode - and return the appropriate data dict and access key. + Return the deduplicated (device, entry) sources of all device curves, + including the device supplying the x axis (custom device mode, or the + device the auto mode resolves to). Returns: - data_dict (dict): The data structure for the current scan. - access_key (str): Either 'val' (live) or 'value' (history). + list[tuple[str, str]] | None: The source list, or None if no + device curve is configured. """ - if self.scan_item is None: - # Optionally fetch the latest from history if nothing is set - self.update_with_scan_history(-1) - if self.scan_item is None: - logger.info("No scan executed so far; skipping device curves categorisation.") - return None, None - - if hasattr(self.scan_item, "live_data"): - # Live scan - return self.scan_item.live_data, "val" - else: - # Historical - scan_devices = self.scan_item.devices - return (scan_devices, "value") - - def update_sync_curves(self): - """ - Update the sync curves with the latest data from the scan. - """ - if self.scan_item is None: - logger.info("No scan executed so far; skipping device curves categorisation.") - return - data, access_key = self._fetch_scan_data_and_access() - for curve in self._sync_curves: - device_name = curve.config.signal.device - device_entry = curve.config.signal.signal - if access_key == "val": - device_data = data.get(device_name, {}).get(device_entry, {}).get(access_key, None) - else: - entry_obj = data.get(device_name, {}).get(device_entry) - device_data = entry_obj.read()["value"] if entry_obj else None - x_data = self._get_x_data(device_name, device_entry) - if x_data is not None: - if np.isscalar(x_data): - self.clear_data() - return - if device_data is not None and x_data is not None: - curve.setData(x_data, device_data) - if device_data is not None and x_data is None: - curve.setData(device_data) - self.request_dap_update.emit() - - def update_async_curves(self): - """ - Updates asynchronously displayed curves with the latest scan data. - - Fetches the scan data and access key to update each curve in `_async_curves` with - new values. If the data is available for a specific curve, it sets the x and y - data for the curve. Emits a signal to request an update once all curves are updated. - - Raises: - The raised errors are dependent on the internal methods such as - `_fetch_scan_data_and_access`, `_get_x_data`, or `setData` used in this - function. - - """ - data, access_key = self._fetch_scan_data_and_access() - - for curve in self._async_curves: - device_name = curve.config.signal.device - device_entry = curve.config.signal.signal - if access_key == "val": # live access - device_data = data.get(device_name, {}).get(device_entry, {}).get(access_key, None) - else: # history access - dataset_obj = data.get(device_name, {}) - if self._skip_large_dataset_check is False: - if not self._check_dataset_size_and_confirm(dataset_obj, device_entry): - continue # user declined to load; skip this curve - entry_obj = dataset_obj.get(device_entry, None) - device_data = entry_obj.read()["value"] if entry_obj else None - - # if shape is 2D cast it into 1D and take the last waveform - if len(np.shape(device_data)) > 1: - device_data = device_data[-1, :] - - if device_data is None: - logger.warning(f"Async data for curve {curve.name()} is None.") + sources: list[tuple[str, str]] = [] + for curve in self.curves: + if curve.config.source != "device" or curve.config.signal is None: continue + sources.append((curve.config.signal.device, curve.config.signal.signal)) + if not sources: + return None + x_key = self._x_source_key() + if x_key is not None: + sources.append(x_key) + return list(dict.fromkeys(sources)) - # Async curves only support plotting vs index or other device - if self.x_axis_mode["name"] in ["timestamp", "index", "auto"]: - device_data_x = np.linspace(0, len(device_data) - 1, len(device_data)) - else: - # Fetch data from signal instead - device_data_x = self._get_x_data(device_name, device_entry) - - # Fallback to 'index' in case data is not of equal length - if len(device_data_x) != len(device_data): - logger.warning( - f"Async data for curve {curve.name()} and x_axis {device_entry} is not of equal length. Falling back to 'index' plotting." - ) - device_data_x = np.linspace(0, len(device_data) - 1, len(device_data)) - - self._auto_adjust_async_curve_settings(curve, len(device_data)) - curve.setData(device_data_x, device_data) - - self.request_dap_update.emit() - - def _check_async_signal_found(self, name: str, signal: str) -> tuple[bool, str]: + def _report_devices_no_file_io(self, scan_item) -> list[str]: """ - Check if the async signal is found in the BEC device manager. + Resolve the scan report devices without touching the data file. + + ``ScanDataContainer.metadata`` is lazy: reading it opens the HDF5 file + synchronously, which must not happen on the GUI thread (and would even + precede the large-dataset gate). Live scans carry the report devices on + their status message, finished scans carry the scan request in the + (Redis-backed) scan history message. Args: - name(str): The name of the async signal. - signal(str): The entry of the async signal. + scan_item: Live scan item or history data container. Returns: - tuple[bool, str]: A tuple where the first element is True if the async signal is found (False otherwise), - and the second element is the signal name (either the original signal or the storage_name for AsyncMultiSignal). + list[str]: Report device names, empty when only the file knows. """ - bec_async_signals = self.client.device_manager.get_bec_signals(sorted(ASYNC_SIGNAL_CLASSES)) - for signal_name, _, entry_data in bec_async_signals: - if signal_name == name and entry_data.get("obj_name") == signal: - return True, entry_data.get("storage_name") - return False, signal - - def _setup_async_curve(self, curve: Curve): - """ - Setup async curve. - - Several curves can resolve to the same underlying async stream -- most - commonly the sub-signals of one ``AsyncMultiSignal``, which share a - ``storage_name`` and are published on a single endpoint. Such curves reuse one - subscription instead of subscribing to the same endpoint repeatedly; - ``on_async_readback`` then feeds every curve that maps to it. - - Args: - curve(Curve): The curve to set up. - """ - name = curve.config.signal.device - signal = curve.config.signal.signal - async_signal_found, stream = self._check_async_signal_found(name, signal) - - try: - curve.clear_data() - except KeyError: - logger.warning(f"Curve {name} not found in plot item.") - - if async_signal_found: - stream_key = (name, stream) # AsyncMultiSignal sub-signals share `stream` - new_endpoint = MessageEndpoints.device_async_signal(self.scan_id, name, stream) - old_endpoint = MessageEndpoints.device_async_signal(self.old_scan_id, name, stream) - legacy_endpoint = False - else: - # No BEC async signal (AsyncSignal/AsyncMultiSignal/DynamicSignal) was found - # for this curve, so it must be served by the deprecated per-device - # device_async_readback endpoint. - stream_key = (name, None) # old endpoint is keyed by device only - new_endpoint = MessageEndpoints.device_async_readback(self.scan_id, name) - old_endpoint = MessageEndpoints.device_async_readback(self.old_scan_id, name) - legacy_endpoint = True - - endpoint_str = getattr(new_endpoint, "endpoint", new_endpoint) - - shared = self._async_streams_setup.get(stream_key) - if shared is not None: - # Another curve already subscribed to this exact stream this scan. - shared.append(signal) - logger.info( - f"Async signals {shared} share a single subscription on endpoint " - f"'{endpoint_str}'; reusing it instead of subscribing again." + status_message = getattr(scan_item, "status_message", None) + if status_message is not None: + raw = getattr(status_message, "scan_report_devices", None) + devices = ( + self._ensure_str_list(raw) if isinstance(raw, (list, tuple, np.ndarray)) else [] ) - return + if devices: + return devices + info = getattr(status_message, "info", None) + raw = info.get("scan_report_devices") if isinstance(info, dict) else None + devices = ( + self._ensure_str_list(raw) if isinstance(raw, (list, tuple, np.ndarray)) else [] + ) + if devices: + return devices - # TODO implement deprecation warning when simulations are migrated - # if legacy_endpoint: - # logger.warning( - # f"Deprecated async endpoint in use: device '{name}' provides async data via the " - # f"legacy 'device_async_readback' endpoint ('{endpoint_str}'). Migrate the device " - # "to an AsyncSignal/AsyncMultiSignal/DynamicSignal so it publishes on the " - # "'device_async_signal' endpoint; support for 'device_async_readback' in the " - # "waveform widget will be removed in a future release." - # ) + # History: the scan request of the scan history message names the + # scanned devices; its first entry is the first report device. + history_msg = getattr(scan_item, "_msg", None) + request_inputs = getattr(history_msg, "request_inputs", None) + arg_bundle = request_inputs.get("arg_bundle") if isinstance(request_inputs, dict) else None + if isinstance(arg_bundle, (list, tuple)): + devices = [item for item in arg_bundle if isinstance(item, str)] + if devices: + return devices[:1] + return [] - self._async_streams_setup[stream_key] = [signal] - self.bec_dispatcher.disconnect_slot(self.on_async_readback, old_endpoint) - self.bec_dispatcher.connect_slot( - self.on_async_readback, new_endpoint, from_start=True, cb_info={"scan_id": self.scan_id} - ) - logger.info(f"Setup async curve {name} (signal '{signal}', endpoint '{endpoint_str}')") - - @SafeSlot(dict, dict, verify_sender=True) - def on_async_readback(self, msg, metadata): + def _x_source_key(self) -> tuple[str, str] | None: """ - Get async data readback. This code needs to be fast, therefor we try - to reduce the number of copies in between cycles. Be careful when refactoring - this part as it will affect the performance of the async readback. + Resolve the (device, entry) supplying the x axis for the current + x-axis mode; None for index/timestamp or when auto resolves to index. - Async curves support plotting against 'index' or other 'device_signal'. No 'auto' or 'timestamp'. - The fallback mechanism for 'auto' and 'timestamp' is to use the 'index'. + Returns: + tuple[str, str] | None: The x source key, or None. + """ + mode = self.x_axis_mode["name"] or "auto" + if mode in ("timestamp", "index"): + return None + if mode != "auto": # custom device mode + entry = self.x_axis_mode.get("entry") + if entry is None: + try: + entry = self.entry_validator.validate_signal(mode, None) + except Exception: + return None + return (mode, entry) + # Auto mode: index when async curves are present, otherwise the first + # device from the scan report. + if self._async_curves: + return None + if self.scan_item is None: + return None + report_devices = self._report_devices_no_file_io(self.scan_item) + if not report_devices: + # Last resort only: this opens the data file synchronously. + try: + report_devices = self._ensure_str_list( + self.scan_item.metadata["bec"]["scan_report_devices"] + ) + except Exception: + return None + if not report_devices: + return None + device_x = report_devices[0] + try: + signal_x = self.entry_validator.validate_signal(device_x, None) + except Exception: + return None + return (device_x, signal_x) - Note: - We create data_plot_x and data_plot_y and modify them within this function - to avoid creating new arrays. This is important for performance. - Support update instructions are 'add', 'add_slice', and 'replace'. + def _history_x_source_key(self, scan_item) -> tuple[str, str] | None: + """ + Resolve the x source key for a pinned history scan. + + Auto mode prefers the widget's current x device (set by live scans) + and falls back to the scan's first report device. Args: - msg(dict): Message with the async data. - metadata(dict): Metadata of the message. + scan_item: The scan item of the pinned history scan. + + Returns: + tuple[str, str] | None: The x source key, or None for index-like + modes. """ - sender = self.sender() - if not hasattr(sender, "cb_info"): - logger.info(f"Sender {sender} has no cb_info.") - return - scan_id = sender.cb_info.get("scan_id", None) - if scan_id != self.scan_id: - logger.info("Scan ID mismatch, ignoring async readback.") - - instruction = metadata.get("async_update", {}).get("type") - if instruction not in ["add", "add_slice", "replace"]: - logger.warning(f"Invalid async update instruction: {instruction}") - return - max_shape = metadata.get("async_update", {}).get("max_shape", []) - plot_mode = self.x_axis_mode["name"] - for curve in self._async_curves: - x_data = None # Reset x_data - y_data = None # Reset y_data - # Get the curve data - async_data = msg["signals"].get(curve.config.signal.signal, None) - if async_data is None: - continue - # y-data - data_plot_y = async_data["value"] - if data_plot_y is None: - logger.warning(f"Async data for curve {curve.name()} is None.") - continue - # Ensure we have numpy array for data_plot_y - data_plot_y = np.asarray(data_plot_y) - if data_plot_y.ndim == 0: - # Convert scalars/0d arrays to 1d so len() and stacking work - data_plot_y = data_plot_y.reshape(1) - # Add - if instruction == "add": - if len(max_shape) > 1: - if len(data_plot_y.shape) > 1: - data_plot_y = data_plot_y[-1, :] - else: - x_data, y_data = curve.get_data() - if y_data is not None: - data_plot_y = np.hstack((y_data, data_plot_y)) - # Add slice - if instruction == "add_slice": - current_slice_id = metadata.get("async_update", {}).get("index") - 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: - data_plot_y = np.hstack((y_data, data_plot_y)) - - # Replace is trivial, no need to modify data_plot_y - - # Get x data for plotting - if plot_mode in ["index", "auto", "timestamp"]: - data_plot_x = np.linspace(0, len(data_plot_y) - 1, len(data_plot_y)) - self._auto_adjust_async_curve_settings(curve, len(data_plot_y)) - curve.setData(data_plot_x, data_plot_y) - # Move on in the loop - continue - - # x_axis_mode is device signal - # Only consider device signals that are async for now, fallback is index - signal_x = self.x_axis_mode["entry"] - async_data = msg["signals"].get(signal_x, None) - # Make sure the signal exists, otherwise fall back to index - if async_data is None: - # Try to grab the data from device signals - data_plot_x = self._get_x_data(plot_mode, signal_x) - else: - data_plot_x = np.asarray(async_data["value"]) - if x_data is not None: - data_plot_x = np.hstack((x_data, data_plot_x)) - # Fallback incase data is not of equal length - if len(data_plot_x) != len(data_plot_y): - logger.warning( - f"Async data for curve {curve.name()} and x_axis {signal_x} is not of equal length. Falling back to 'index' plotting." + mode = self.x_axis_mode["name"] or "auto" + if mode in ("timestamp", "index"): + return None + if mode != "auto": # custom device mode + entry = self.x_axis_mode.get("entry") + if entry is None: + try: + entry = self.entry_validator.validate_signal(mode, None) + except Exception: + return None + return (mode, entry) + if self._current_x_device is not None: + return self._current_x_device + report_devices = self._report_devices_no_file_io(scan_item) + if not report_devices: + # Last resort only: this opens the data file synchronously. + try: + report_devices = self._ensure_str_list( + scan_item.metadata.get("bec", {}).get("scan_report_devices") or [] ) - data_plot_x = np.linspace(0, len(data_plot_y) - 1, len(data_plot_y)) + except Exception: + return None + if not report_devices: + return None + device_x = report_devices[0] + try: + signal_x = self.entry_validator.validate_signal(device_x, None) + except Exception: + return None + return (device_x, signal_x) - # Plot the data - self._auto_adjust_async_curve_settings(curve, len(data_plot_y)) - curve.setData(data_plot_x, data_plot_y) + def _setup_data_api_subscription(self, scan: str = "live"): + """ + (Re)create the DataAPI subscription for the configured device curves. - self.request_dap_update.emit() + Args: + scan(str): "live" to follow the active scan, or a terminal scan id. + """ + self._cleanup_data_api_subscription() + self._data_api_scan = scan + if self._shutting_down: + return + sources = self._config_sources() + if sources is None: + return + sources = self._filter_oversized_sources(sources, scan) + if not sources: + return + try: + self._data_bridge = QtDataSubscription( + self.client, sources=sources, scan=scan, parent=self, min_emit_interval=0.1 + ) + self._data_bridge.updated.connect(self._on_data_update) + except Exception as exc: + logger.warning(f"Failed to configure waveform data subscription: {exc}") + self._cleanup_data_api_subscription() + + def _cleanup_data_api_subscription(self): + if self._data_bridge is None: + return + try: + self._data_bridge.close() + finally: + self._data_bridge = None + + def _setup_history_curve_subscriptions(self): + """ + (Re)create one DataAPI subscription per scan pinned by history curves. + """ + self._cleanup_history_subscriptions() + if self._shutting_down: + return + sources_by_scan: dict[str, list[tuple[str, str]]] = {} + for curve in self._history_curves: + scan_id = curve.config.scan_id + signal = curve.config.signal + if not scan_id or signal is None: + continue + sources_by_scan.setdefault(scan_id, []).append((signal.device, signal.signal)) + for scan_id, sources in sources_by_scan.items(): + sources = self._filter_oversized_sources(sources, scan_id) + if not sources: + continue + scan_item = self.get_history_scan_item(scan_id=scan_id) + x_key = self._history_x_source_key(scan_item) if scan_item is not None else None + if x_key is not None: + sources.append(x_key) + self._history_x_keys[scan_id] = x_key + try: + bridge = QtDataSubscription( + self.client, + sources=list(dict.fromkeys(sources)), + scan=scan_id, + parent=self, + min_emit_interval=0.1, + ) + bridge.updated.connect(self._on_data_update) + self._history_bridges[scan_id] = bridge + except Exception as exc: + logger.warning( + f"Failed to configure history data subscription for scan {scan_id}: {exc}" + ) + + def _cleanup_history_subscriptions(self): + for bridge in self._history_bridges.values(): + try: + bridge.close() + except Exception: # pylint: disable=broad-except + pass + self._history_bridges = {} + self._history_x_keys = {} + + def _estimate_source_bytes(self, scan: str, source: tuple[str, str]) -> int | None: + """ + Estimated stored size of one source, from the scan-history metadata. + + The estimate costs no file I/O — the decision to load is taken before + anything is read. + + Args: + scan(str): The scan id the source belongs to. + source(tuple[str, str]): (device, entry). + + Returns: + int | None: Size in bytes, or None when it cannot be estimated. + """ + try: + estimate = self.client.data_api.estimate_bytes([source], scan) + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"Could not estimate the size of {source[0]}-{source[1]}: {exc}") + return None + return estimate if isinstance(estimate, int) else None + + def _stored_shape(self, scan: str, device: str, entry: str) -> tuple[int, ...] | None: + """Stored dataset shape of one source, for the confirmation dialog.""" + scan_item = self.get_history_scan_item(scan_id=scan) + stored = getattr(getattr(scan_item, "_msg", None), "stored_data_info", None) + if not stored: + return None + device_info = stored.get(device) or {} + info = device_info.get(entry) + if info is None: + prefix = f"{device}_" + if entry.startswith(prefix): + # Async datasets are stored under the signal's storage name. + info = device_info.get(entry[len(prefix) :]) + shape = getattr(info, "shape", None) if info is not None else None + return tuple(shape) if shape else None + + def _filter_oversized_sources(self, sources: list, scan: str | None) -> list: + """ + Drop sources the user does not want to load, asking per dataset. + + Mirrors the pre-DataAPI guard: each oversized dataset is offered + individually, a declined dataset is simply not loaded (its curve stays + visible but empty), and a confirmed dataset is remembered for the rest + of the session so subscription rebuilds do not re-prompt. + + Args: + sources(list): Candidate (device, entry) sources. + scan(str | None): The scan the subscription binds to. + + Returns: + list: The sources that may be loaded. + """ + if scan is None or scan == "live" or self._skip_large_dataset_check: + return list(sources) + limit_mb = self.config.max_dataset_size_mb + limit_bytes = int(limit_mb * 1024 * 1024) + kept = [] + for source in sources: + if (scan, tuple(source)) in self._size_confirmed_sources: + kept.append(source) + continue + size = self._estimate_source_bytes(scan, tuple(source)) + if size is None or size <= limit_bytes: + kept.append(source) + continue + size_mb = size / (1024 * 1024) + logger.warning( + f"Attempt to load large dataset {source[0]}-{source[1]}: {size_mb:.1f} MB " + f"(limit {limit_mb} MB)" + ) + if self._skip_large_dataset_warning: + logger.info("Skipping large dataset warning dialog; dataset not loaded.") + continue + if not self._confirm_large_dataset( + size_mb, source=tuple(source), shape=self._stored_shape(scan, *source) + ): + continue + self._size_confirmed_sources.add((scan, tuple(source))) + kept.append(source) + return kept + + def _refresh_data_subscriptions(self): + """Rebuild all DataAPI subscriptions after a source-set change.""" + if self._shutting_down: + return + self._setup_data_api_subscription(scan=self._data_api_scan) + self._setup_history_curve_subscriptions() + + def _resolve_x_axis(self) -> tuple[str, str] | None: + """ + Resolve the device x source for the current x mode and sync the label + suffix, the current-x-device bookkeeping and the alignment state. + + Returns: + tuple[str, str] | None: The x source key for device/auto modes, + None when the x axis is index- or timestamp-based. + """ + mode = self.x_axis_mode["name"] or "auto" + previous_x_device = self._current_x_device + x_key = None + if mode == "timestamp": + new_suffix = " (timestamp)" + self._current_x_device = None + elif mode == "index": + new_suffix = " (index)" + self._current_x_device = None + elif mode == "auto": + x_key = self._x_source_key() + if x_key is None: + new_suffix = " (auto: index)" + self._current_x_device = None + else: + new_suffix = f" (auto: {x_key[0]}-{x_key[1]})" + self._current_x_device = x_key + else: # custom device mode + x_key = self._x_source_key() + if x_key is None: + new_suffix = " (index)" + self._current_x_device = None + else: + new_suffix = f" (custom: {x_key[0]}-{x_key[1]})" + self._current_x_device = x_key + self._update_x_label_suffix(new_suffix) + if previous_x_device != self._current_x_device: + self._refresh_alignment_state(force_readback=True) + return x_key + + @SafeSlot(object) + def _on_data_update(self, update) -> None: + """ + Render one columnar DataAPI update (live, backfill or history). + + Monitored sources are drawn from the aligned columns; async sources + are reconstructed from their columnar fragments according to the + async update type. History curves are routed by their pinned scan id. + + Args: + update (SubscriptionUpdate): Full-state columnar snapshot. + """ + updated_any = False + + if self.scan_id is None or update.scan_id == self.scan_id: + x_key = self._resolve_x_axis() + aligned_columns = update.aligned() + for curve in self.curves: + if curve.config.source != "device": + continue + if self._render_curve_from_update(curve, update, aligned_columns, x_key): + updated_any = True + + history_curves = [c for c in self._history_curves if c.config.scan_id == update.scan_id] + if history_curves: + x_key = self._history_x_keys.get(update.scan_id) + aligned_columns = update.aligned() + for curve in history_curves: + if self._render_curve_from_update(curve, update, aligned_columns, x_key): + updated_any = True + + if updated_any: + self.request_dap_update.emit() + + def _render_curve_from_update( + self, curve: Curve, update, aligned_columns: dict, x_key: tuple[str, str] | None + ) -> bool: + """ + Render one curve from a DataAPI update. + + Args: + curve(Curve): The curve to render. + update(SubscriptionUpdate): The update snapshot. + aligned_columns(dict): Cached `update.aligned()` columns. + x_key(tuple[str, str] | None): The resolved x source key. + + Returns: + bool: True if the curve data was set. + """ + signal = curve.config.signal + if signal is None: + return False + key = (signal.device, signal.signal) + source = update.sources.get(key) + if source is None: + return False + if source.kind == "monitored": + return self._render_monitored_curve(curve, update, key, aligned_columns, x_key) + return self._render_async_curve(curve, update, source, x_key) + + def _render_monitored_curve( + self, + curve: Curve, + update, + key: tuple[str, str], + aligned_columns: dict, + x_key: tuple[str, str] | None, + ) -> bool: + """ + Render a monitored (sync) curve from the aligned columns of an update. + + The x column follows the x-axis mode: aligned ordinals for index, the + source timestamps for timestamp mode, and the aligned values of the x + device for device/auto modes with an index fallback when the x source + is not part of the update. + """ + y_data = aligned_columns.get(key) + if y_data is None or len(y_data) == 0: + return False + mode = self.x_axis_mode["name"] or "auto" + x_data = None + if mode == "timestamp": + x_data = update.axis("timestamp", key) + elif mode != "index" and x_key is not None and x_key in update.sources: + x_data = update.axis("device", x_key) + if x_data is None: + x_data = update.aligned_ordinals + curve.setData(np.asarray(x_data), np.asarray(y_data)) + return True + + def _render_async_curve( + self, curve: Curve, update, source, x_key: tuple[str, str] | None + ) -> bool: + """ + Render an async (or unindexed legacy) curve from its columnar source. + + The y data is reconstructed according to the async update type; x is + the sample index, the source timestamps (timestamp mode, matching + lengths only) or a same-length x device column, with an index + fallback on any length mismatch. + """ + y_data = self._async_display_values(source) + if y_data is None or len(y_data) == 0: + return False + mode = self.x_axis_mode["name"] or "auto" + x_data = None + if mode == "timestamp": + timestamps = source.timestamps + if timestamps is not None and len(timestamps) == len(y_data): + try: + x_data = np.asarray(timestamps, dtype=float) + except (TypeError, ValueError): + x_data = None + elif mode not in ("index", "auto") and x_key is not None: + x_source = update.sources.get(x_key) + if x_source is not None: + if x_source.kind == "monitored": + x_candidate = np.asarray(x_source.values) + else: + x_candidate = self._async_display_values(x_source) + if x_candidate is not None and len(x_candidate) == len(y_data): + x_data = x_candidate + if x_data is None: + logger.warning( + f"Async data for curve {curve.name()} and x_axis {x_key} is not of equal " + "length. Falling back to 'index' plotting." + ) + 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)) + return True + + @staticmethod + def _async_display_values(source) -> np.ndarray | None: + """ + Reconstruct the displayed y data of an async source from its columnar + fragments. + + - 'add' with a 2-D max_shape displays the latest waveform (last row); + 1-D 'add' concatenates all fragments. + - 'add_slice' displays the last accumulated row. + - 'replace' (and unindexed legacy sources without further metadata) + display the current full state, i.e. the last element. + - Without an update type (history reads), scalar rows form the full + series and array rows mirror the legacy "display the last row". + + Args: + source(SourceData): The async source snapshot. + + Returns: + np.ndarray | None: The displayed y data. + """ + values = source.values + if not values: + return None + update_type = source.metadata.get("async_update_type") + max_shape = source.metadata.get("max_shape") or [] + if update_type == "add": + if len(max_shape) > 1: + y_data = np.asarray(values[-1]) + if y_data.ndim > 1: + y_data = y_data[-1, :] + return np.atleast_1d(y_data) + return np.concatenate([np.atleast_1d(np.asarray(value)) for value in values]) + if update_type in ("add_slice", "replace"): + return np.atleast_1d(np.asarray(values[-1])) + # No async-update metadata (e.g. history file reads): rows are the + # dataset rows. + first = np.asarray(values[0]) + if first.ndim == 0: + return np.asarray(values) + return np.atleast_1d(np.asarray(values[-1])) def _auto_adjust_async_curve_settings( self, @@ -2218,100 +2406,6 @@ class Waveform(PlotBase): self.dap_params_update.emit(curve.dap_params, {"curve_id": curve.name()}) self.dap_summary_update.emit(curve.dap_summary, {"curve_id": curve.name()}) - def _get_x_data(self, device_name: str, device_entry: str) -> list | np.ndarray | None: - """ - Get the x data for the curves with the decision logic based on the widget x mode configuration: - - If x is called 'timestamp', use the timestamp data from the scan item. - - If x is called 'index', use the rolling index. - - If x is a custom signal, use the data from the scan item. - - If x is not specified, use the first device from the scan report. - - Additionally, checks and updates the x label suffix. - - Args: - device_name(str): The name of the device. - device_entry(str): The entry of the device - - Returns: - list|np.ndarray|None: X data for the curve. - """ - x_data = None - new_suffix = None - previous_x_device = self._current_x_device - data, access_key = self._fetch_scan_data_and_access() - - # 1 User wants custom signal - if self.x_axis_mode["name"] not in ["timestamp", "index", "auto"]: - device_x = self.x_axis_mode["name"] - signal_x = self.x_axis_mode.get("entry", None) - if signal_x is None: - signal_x = self.entry_validator.validate_signal(device_x, None) - # if the motor was not scanned, an empty list is returned and curves are not updated - if access_key == "val": # live data - x_data = data.get(device_x, {}).get(signal_x, {}).get(access_key, [0]) - else: # history data - entry_obj = data.get(device_x, {}).get(signal_x) - x_data = entry_obj.read()["value"] if entry_obj else [0] - new_suffix = f" (custom: {device_x}-{signal_x})" - self._current_x_device = (device_x, signal_x) - - # 2 User wants timestamp - if self.x_axis_mode["name"] == "timestamp": - if access_key == "val": # live - x_data = data.get(device_name, {}).get(device_entry, None) - if x_data is None: - return None - else: - timestamps = x_data.timestamps - else: # history data - entry_obj = data.get(device_name, {}).get(device_entry) - timestamps = entry_obj.read()["timestamp"] if entry_obj else [0] - x_data = timestamps - new_suffix = " (timestamp)" - self._current_x_device = None - - # 3 User wants index - if self.x_axis_mode["name"] == "index": - x_data = None - new_suffix = " (index)" - self._current_x_device = None - - # 4 Best effort automatic mode - if self.x_axis_mode["name"] is None or self.x_axis_mode["name"] == "auto": - # 4.1 If there are async curves, use index - if len(self._async_curves) > 0: - x_data = None - new_suffix = " (auto: index)" - self._current_x_device = None - # 4.2 If there are sync curves, use the first device from the scan report - else: - try: - scan_report_devices = self._ensure_str_list( - self.scan_item.metadata["bec"]["scan_report_devices"] - ) - except Exception: - scan_report_devices = self.scan_item.status_message.info.get( - "scan_report_devices", [] - ) - if not scan_report_devices: - x_data = None - new_suffix = " (auto: index)" - self._current_x_device = None - else: - device_x = scan_report_devices[0] - signal_x = self.entry_validator.validate_signal(device_x, None) - if access_key == "val": - x_data = data.get(device_x, {}).get(signal_x, {}).get(access_key, None) - else: - entry_obj = data.get(device_x, {}).get(signal_x) - x_data = entry_obj.read()["value"] if entry_obj else None - new_suffix = f" (auto: {device_x}-{signal_x})" - self._current_x_device = (device_x, signal_x) - self._update_x_label_suffix(new_suffix) - if previous_x_device != self._current_x_device: - self._refresh_alignment_state(force_readback=True) - return x_data - def _update_x_label_suffix(self, new_suffix: str): """ Update x_label so it ends with `new_suffix`, removing any old suffix. @@ -2375,7 +2469,6 @@ class Waveform(PlotBase): # Reset sync/async curve lists self._async_curves.clear() self._sync_curves.clear() - self._async_streams_setup = {} found_async = False found_sync = False mode = "sync" @@ -2391,8 +2484,6 @@ class Waveform(PlotBase): category = classify_device_signal(self.dev.get(dev_name), entry) if category == SignalCategory.ASYNC or dev_name in readout_priority_async: self._async_curves.append(curve) - if hasattr(self.scan_item, "live_data"): - self._setup_async_curve(curve) found_async = True elif category == SignalCategory.SYNC or dev_name in readout_priority_sync: self._sync_curves.append(curve) @@ -2495,57 +2586,49 @@ class Waveform(PlotBase): else: self.scan_id = self.scan_item.scan_id - self._emit_signal_update() - - def _emit_signal_update(self): - self._categorise_device_curves() - + self._mode = self._categorise_device_curves() self.setup_dap_for_scan() - self.sync_signal_update.emit() - self.async_signal_update.emit() + # The data flows through the DataAPI: a still-running scan is followed + # live, a terminal scan is served by the history plugin. + if hasattr(self.scan_item, "live_data"): + self._setup_data_api_subscription(scan="live") + else: + self._setup_data_api_subscription(scan=self.scan_id) ################################################################################ # Utility Methods ################################################################################ # Large dataset handling helpers - def _check_dataset_size_and_confirm(self, dataset_obj, device_entry: str) -> bool: + @staticmethod + def _describe_dataset(source: tuple[str, str] | None, shape: tuple[int, ...] | None) -> str: """ - Check the size of the dataset and confirm with the user if it exceeds the limit. + Human-readable identity and extent of a dataset for the dialog. Args: - dataset_obj: The dataset object containing the information. - device_entry( str): The specific device entry to check. + source(tuple[str, str] | None): (device, entry) of the dataset. + shape(tuple[int, ...] | None): Stored dataset shape. Returns: - bool: True if the dataset is within the size limit or user confirmed to load it, - False if the dataset exceeds the size limit and user declined to load it. + str: A sentence naming the dataset and its number of points. """ - try: - info = dataset_obj._info - mem_bytes = info.get(device_entry, {}).get("value", {}).get("mem_size", 0) - # Fallback – grab first entry if lookup failed - if mem_bytes == 0 and info: - first_key = next(iter(info)) - mem_bytes = info[first_key]["value"]["mem_size"] - size_mb = mem_bytes / (1024 * 1024) - logger.info(f"Dataset size: {size_mb:.1f} MB") - except Exception as exc: # noqa: BLE001 - logger.error(f"Unable to evaluate dataset size: {exc}") - return True - - if size_mb <= self.config.max_dataset_size_mb: - return True - logger.warning( - f"Attempt to load large dataset: {size_mb:.1f} MB " - f"(limit {self.config.max_dataset_size_mb} MB)" + name = f"'{source[0]}-{source[1]}'" if source else "The selected dataset" + if not shape: + return f"{name} (size unknown)" + if len(shape) == 1: + return f"{name} holds {shape[0]:,} points" + per_point = int(np.prod(shape[1:])) + return ( + f"{name} holds {shape[0]:,} points x {per_point:,} samples " + f"({int(np.prod(shape)):,} values, shape {tuple(shape)})" ) - if self._skip_large_dataset_warning: - logger.info("Skipping large dataset warning dialog.") - return False - return self._confirm_large_dataset(size_mb) - def _confirm_large_dataset(self, size_mb: float) -> bool: + def _confirm_large_dataset( + self, + size_mb: float, + source: tuple[str, str] | None = None, + shape: tuple[int, ...] | None = None, + ) -> bool: """ Confirm with the user whether to load a large dataset with dialog popup. Also allows the user to adjust the maximum dataset size limit and if user @@ -2553,6 +2636,9 @@ class Waveform(PlotBase): Args: size_mb(float): Size of the dataset in MB. + source(tuple[str, str] | None): (device, entry) of the dataset, shown + in the dialog so the user knows which curve is affected. + shape(tuple[int, ...] | None): Stored shape, shown as a point count. Returns: bool: True if the user confirmed to load the dataset, False otherwise. @@ -2587,8 +2673,8 @@ class Waveform(PlotBase): # widget layout main_dialog_layout.addWidget( QLabel( - f"The selected dataset is {size_mb:.1f} MB which exceeds the " - f"current limit of {self.config.max_dataset_size_mb} MB.\n" + f"{self._describe_dataset(source, shape)} and is {size_mb:.1f} MB, " + f"which exceeds the current limit of {self.config.max_dataset_size_mb} MB.\n" ) ) main_dialog_layout.addLayout(limit_adjustment_layout) @@ -2716,6 +2802,9 @@ class Waveform(PlotBase): """ Cleanup the widget by disconnecting signals and closing dialogs. """ + self._shutting_down = True + self._cleanup_data_api_subscription() + self._cleanup_history_subscriptions() self.proxy_dap_request.cleanup() if self._alignment_controller is not None: self._alignment_controller.cleanup() diff --git a/tests/unit_tests/test_qt_data_subscription.py b/tests/unit_tests/test_qt_data_subscription.py index 94171391..d384701b 100644 --- a/tests/unit_tests/test_qt_data_subscription.py +++ b/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 diff --git a/tests/unit_tests/test_waveform.py b/tests/unit_tests/test_waveform.py index bf50a022..cd127b49 100644 --- a/tests/unit_tests/test_waveform.py +++ b/tests/unit_tests/test_waveform.py @@ -19,7 +19,6 @@ from bec_widgets.widgets.services.scan_history_browser.scan_history_browser impo ScanHistoryBrowser, ) from tests.unit_tests.client_mocks import ( - DummyData, create_dummy_scan_item, dap_plugin_message, inject_scan_history, @@ -52,6 +51,134 @@ def make_alignment_fit_summary(center: float | None = None) -> dict: } +################################################## +# DataAPI test helpers +################################################## + + +def _fake_bridge_factory(monkeypatch, gated_bytes: int | None = None): + """ + Patch QtDataSubscription in the waveform module with a lightweight + stand-in. Returns the list of created bridges (newest last). + + Args: + gated_bytes(int | None): When given, every bridge created with a + ``size_limit_bytes`` smaller than this value reports itself as + size-gated with ``estimated_bytes = gated_bytes`` (mirrors the + backend gate without any file I/O). + """ + created = [] + + class _FakeBridge: + def __init__( + self, + client, + sources, + scan="live", + parent=None, + min_emit_interval=0.1, + size_limit_bytes=None, + ): + self.client = client + self.sources = list(sources) + self.scan = scan + self.scan_id = None if scan == "live" else scan + self.healthy = True + self.closed = False + self.updated = MagicMock() + self.size_limit_bytes = size_limit_bytes + self.estimated_bytes = gated_bytes + self.size_gated = ( + gated_bytes is not None + and size_limit_bytes is not None + and gated_bytes > size_limit_bytes + ) + self.confirmed = False + + def confirm_size(self): + self.confirmed = True + self.size_gated = False + + def close(self): + self.closed = True + + def factory( + client, sources, scan="live", parent=None, min_emit_interval=0.1, size_limit_bytes=None + ): + bridge = _FakeBridge(client, sources, scan=scan, size_limit_bytes=size_limit_bytes) + created.append(bridge) + return bridge + + monkeypatch.setattr("bec_widgets.widgets.plots.waveform.waveform.QtDataSubscription", factory) + return created + + +def _monitored_source(device, values, entry=None, timestamps=None, ordinals=None): + 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) + return SourceData( + device=device, + entry=entry, + kind="monitored", + ordinals=ordinals, + values=tuple(values), + timestamps=tuple(timestamps), + complete=True, + ) + + +def _async_source( + device, + values, + entry=None, + timestamps=None, + update_type="add", + max_shape=(None,), + kind="async", + ordinals=None, +): + 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) + return SourceData( + device=device, + entry=entry, + kind=kind, + ordinals=ordinals, + values=tuple(values), + timestamps=tuple(timestamps), + complete=True, + metadata={ + "async_update_type": update_type, + "max_shape": list(max_shape), + "acquisition_group": None, + }, + ) + + +def _make_update(sources, scan_id="dummy", reason="live", group="scan"): + from bec_lib.data_api.models import SubscriptionUpdate + + source_map = {source.key: source for source in sources} + ordinal_sets = [set(source.ordinals) for source in source_map.values()] + aligned = tuple(sorted(set.intersection(*ordinal_sets))) if ordinal_sets else () + return SubscriptionUpdate( + scan_id=scan_id, + reason=reason, + sources=source_map, + aligned_ordinals=aligned, + complete=True, + metadata={"group": group}, + ) + + def test_waveform_initialization(qtbot, mocked_client): """ Test that a new Waveform widget initializes with the correct defaults. @@ -344,80 +471,105 @@ def test_curve_json_setter_ignores_custom(qtbot, mocked_client): ################################################## -def test_update_sync_curves(monkeypatch, qtbot, mocked_client): +def test_on_data_update_sync_timestamp_mode(monkeypatch, qtbot, mocked_client): """ - Test that update_sync_curves retrieves live data correctly and calls setData on sync curves. + A monitored curve in timestamp mode is rendered from the aligned columns + against the source timestamps. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) c = wf.plot(arg1="bpm4i") - wf._sync_curves = [c] wf.x_mode = "timestamp" - dummy_scan = create_dummy_scan_item() - wf.scan_item = dummy_scan + wf.scan_id = "dummy" - recorded = {} + update = _make_update( + [_monitored_source("bpm4i", values=(5, 6, 7), timestamps=(101, 201, 301))] + ) + wf._on_data_update(update) - def fake_setData(x, y): - recorded["x"] = x - recorded["y"] = y - - monkeypatch.setattr(c, "setData", fake_setData) - - wf.update_sync_curves() - np.testing.assert_array_equal(recorded.get("x"), [101, 201, 301]) - np.testing.assert_array_equal(recorded.get("y"), [5, 6, 7]) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [101, 201, 301]) + np.testing.assert_array_equal(y_data, [5, 6, 7]) -def test_update_async_curves(monkeypatch, qtbot, mocked_client): +def test_on_data_update_sync_index_mode(monkeypatch, qtbot, mocked_client): """ - Test that update_async_curves retrieves live data correctly and calls setData on async curves. + A monitored curve in index mode is rendered against the aligned ordinals. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - c = wf.plot(arg1="async_device", label="async_device-async_device") - wf._async_curves = [c] - wf.x_mode = "timestamp" # Timestamp is not supported, fallback to index. - dummy_scan = create_dummy_scan_item() - wf.scan_item = dummy_scan + c = wf.plot(arg1="bpm4i") + wf.x_mode = "index" + wf.scan_id = "dummy" - recorded = {} + wf._on_data_update(_make_update([_monitored_source("bpm4i", values=(5, 6, 7))])) - def fake_setData(x, y): - recorded["x"] = x - recorded["y"] = y - - monkeypatch.setattr(c, "setData", fake_setData) - - wf.update_async_curves() - np.testing.assert_array_equal(recorded.get("x"), [0, 1, 2]) - np.testing.assert_array_equal(recorded.get("y"), [1, 2, 3]) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) + np.testing.assert_array_equal(y_data, [5, 6, 7]) -def test_get_x_data_custom(monkeypatch, qtbot, mocked_client): +def test_on_data_update_sync_custom_x_device(monkeypatch, qtbot, mocked_client): """ - Test that _get_x_data returns the correct custom signal data. + A monitored curve in custom device mode uses the aligned values of the x + device delivered in the same update; a missing x source falls back to + the index. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - # Set x_mode to a custom mode. - wf.x_axis_mode["name"] = "custom_signal" - wf.x_axis_mode["entry"] = "custom_entry" - dummy_data = DummyData(val=[50, 60, 70], timestamps=[150, 160, 170]) - dummy_live = {"custom_signal": {"custom_entry": dummy_data}} - monkeypatch.setattr(wf, "_fetch_scan_data_and_access", lambda: (dummy_live, "val")) - x_data = wf._get_x_data("irrelevant", "irrelevant") + c = wf.plot(arg1="bpm4i") + wf.x_mode = "samx" + wf.scan_id = "dummy" + + update = _make_update( + [ + _monitored_source("bpm4i", values=(5, 6, 7)), + _monitored_source("samx", values=(50, 60, 70)), + ] + ) + wf._on_data_update(update) + + x_data, y_data = c.get_data() np.testing.assert_array_equal(x_data, [50, 60, 70]) + np.testing.assert_array_equal(y_data, [5, 6, 7]) + assert wf._current_x_device == ("samx", "samx") + + # X source missing from the update -> index fallback + wf._on_data_update(_make_update([_monitored_source("bpm4i", values=(5, 6, 7))])) + x_data, _ = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) -def test_get_x_data_timestamp(monkeypatch, qtbot, mocked_client): +def test_on_data_update_sync_auto_mode(monkeypatch, qtbot, mocked_client): """ - Test that _get_x_data returns the correct timestamp data. + Auto mode resolves the x device from the scan report devices when no + async curves are present, and falls back to the index otherwise. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.x_axis_mode["name"] = "timestamp" - dummy_data = DummyData(val=[50, 60, 70], timestamps=[101, 202, 303]) - dummy_live = {"deviceX": {"entryX": dummy_data}} - monkeypatch.setattr(wf, "_fetch_scan_data_and_access", lambda: (dummy_live, "val")) - x_data = wf._get_x_data("deviceX", "entryX") - np.testing.assert_array_equal(x_data, [101, 202, 303]) + c = wf.plot(arg1="bpm4i") + wf.scan_id = "dummy" + wf.scan_item = create_dummy_scan_item() # scan_report_devices == ["samx"] + + update = _make_update( + [ + _monitored_source("bpm4i", values=(5, 6, 7)), + _monitored_source("samx", values=(10, 20, 30)), + ] + ) + wf._on_data_update(update) + + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [10, 20, 30]) + np.testing.assert_array_equal(y_data, [5, 6, 7]) + assert wf._current_x_device == ("samx", "samx") + + # With an async curve present, auto mode falls back to index. + wf._async_curves = [MagicMock()] + wf._on_data_update(update) + x_data, _ = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) + assert wf._current_x_device is None def test_categorise_device_curves(monkeypatch, qtbot, mocked_client): @@ -438,14 +590,13 @@ def test_categorise_device_curves(monkeypatch, qtbot, mocked_client): assert c_async in wf._async_curves -@pytest.mark.parametrize( - ["mode", "calls"], [("sync", (1, 0)), ("async", (0, 1)), ("mixed", (1, 1))] -) -def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls): +@pytest.mark.parametrize("mode", ["sync", "async", "mixed"]) +def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode): """ - Test that on_scan_status sets up a new scan correctly, - categorizes curves, and triggers sync/async updates as needed. + Test that on_scan_status performs the per-scan bookkeeping (scan id, + categorisation) and rebuilds the DataAPI subscription for the new scan. """ + bridges = _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) # Force creation of a couple of device curves if mode == "sync": @@ -461,23 +612,26 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls): dummy_scan.metadata["bec"]["scan_id"] = "1234" monkeypatch.setattr(wf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan) - # We'll track calls to sync_signal_update and async_signal_update - sync_spy = MagicMock() - async_spy = MagicMock() - wf.sync_signal_update.connect(sync_spy) - wf.async_signal_update.connect(async_spy) - - # Prepare fake message data - msg = {"scan_id": "1234"} - meta = {} - wf.on_scan_status(msg, meta) + n_bridges = len(bridges) + wf.on_scan_status({"scan_id": "1234"}, {}) assert wf.scan_id == "1234" assert wf.scan_item == dummy_scan assert wf._mode == mode - assert sync_spy.call_count == calls[0], "sync_signal_update should be called exactly once" - assert async_spy.call_count == calls[1], "async_signal_update should be called exactly once" + # The DataAPI subscription is rebuilt for the new scan and follows it live. + assert len(bridges) > n_bridges + assert bridges[-1].scan == "live" + expected_sources = [] + if mode in ("sync", "mixed"): + expected_sources.append(("bpm4i", "bpm4i")) + if mode in ("async", "mixed"): + expected_sources.append(("async_device", "async_device")) + for key in expected_sources: + assert key in bridges[-1].sources + if mode == "sync": + # Auto mode resolves the x device from the scan report devices. + assert ("samx", "samx") in bridges[-1].sources def test_on_scan_status_ignored_without_device_curves(qtbot, mocked_client_with_dap, monkeypatch): @@ -705,15 +859,14 @@ def test_alignment_panel_updates_when_auto_x_motor_changes( wf._current_x_device = ("samx", "samx") wf._alignment_panel.set_positioner_device("samx") wf.scan_item = create_dummy_scan_item() + # A real scan carries the report devices on its (in-memory) status + # message; the data file is only the last-resort source. + wf.scan_item.status_message.info["scan_report_devices"] = ["samy"] wf.scan_item.metadata["bec"]["scan_report_devices"] = ["samy"] - data = { - "samy": {"samy": {"val": np.array([1.0, 2.0, 3.0])}}, - "bpm4i": {"bpm4i": {"val": np.array([10.0, 20.0, 30.0])}}, - } - monkeypatch.setattr(wf, "_fetch_scan_data_and_access", lambda: (data, "val")) - - wf._get_x_data("bpm4i", "bpm4i") + # Rendering a DataAPI update re-resolves the auto x device from the scan + # report devices and refreshes the alignment state. + wf._resolve_x_axis() assert wf._current_x_device == ("samy", "samy") assert wf._alignment_positioner_name == "samy" @@ -996,217 +1149,179 @@ def test_request_dap_includes_composite_parameters_list(qtbot, mocked_client_wit assert msg.content["config"]["class_kwargs"]["model"] == ["GaussianModel", "GaussianModel"] -def test_fetch_scan_data_and_access(qtbot, mocked_client, monkeypatch): +# NOTE: the legacy pull path (_fetch_scan_data_and_access, update_sync_curves, +# update_async_curves, _setup_async_curve, on_async_readback, _get_x_data) was +# replaced by the DataAPI subscription; its behaviour is asserted through the +# _on_data_update tests below. + + +def test_async_curve_sources_in_subscription(qtbot, mocked_client, monkeypatch): """ - Test the _fetch_scan_data_and_access method returns live_data/val if in a live scan, - or device dict/value if in a historical scan. Also test fallback if no scan_item. + Async curves are served by the DataAPI subscription (the backend resolves + the async endpoints); their (device, entry) pairs are part of the bridge + sources and are not duplicated. """ + bridges = _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.scan_item = None + wf.plot(arg1="async_device", label="async_device-async_device") + first_bridge = bridges[-1] + assert ("async_device", "async_device") in first_bridge.sources - hist_mock = MagicMock() - monkeypatch.setattr(wf, "update_with_scan_history", hist_mock) - - wf._fetch_scan_data_and_access() - hist_mock.assert_called_once_with(-1) - - # Check live mode - dummy_scan = create_dummy_scan_item() - wf.scan_item = dummy_scan - data_dict, access_key = wf._fetch_scan_data_and_access() - assert data_dict == dummy_scan.live_data - assert access_key == "val" - - # Check history mode - del dummy_scan.live_data - dummy_scan.devices = {"some_device": {"some_entry": "some_value"}} - data_dict, access_key = wf._fetch_scan_data_and_access() - assert "some_device" in data_dict # from dummy_scan.devices - assert access_key == "value" + # A second curve on the same signal replaces the subscription without + # duplicating the source. + wf.plot(arg1="async_device", label="second-curve") + assert first_bridge.closed is True + assert bridges[-1].sources.count(("async_device", "async_device")) == 1 -def test_setup_async_curve(qtbot, mocked_client, monkeypatch): +def test_on_data_update_async_add(qtbot, mocked_client, monkeypatch): """ - Test that _setup_async_curve properly disconnects old signals - and re-connects the async readback for a new scan ID. + An async 'add' source concatenates its fragments (1-D max_shape); a + 'replace' source displays only the current full state. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.old_scan_id = "111" - wf.scan_id = "222" - c = wf.plot(arg1="async_device", label="async_device-async_device") - # check that it was placed in _async_curves or so - wf._async_curves = [c] - - # We'll spy on connect_slot - connect_spy = MagicMock() - monkeypatch.setattr(wf.bec_dispatcher, "connect_slot", connect_spy) - - # _setup_async_curve dedupes subscriptions per scan; reset the tracking so this - # isolated call performs the (single) subscription. - wf._async_streams_setup = {} - wf._setup_async_curve(c) - connect_spy.assert_called_once() - endpoint_called = connect_spy.call_args[0][1].endpoint - # We expect MessageEndpoints.device_async_readback('222', 'async_device') - assert "222" in endpoint_called - assert "async_device" in endpoint_called - - # A second curve that resolves to the same stream reuses the one subscription - # instead of subscribing again. - wf._setup_async_curve(c) - connect_spy.assert_called_once() - - -def test_on_async_readback_add_update(qtbot, mocked_client): - """ - Test that on_async_readback extends or replaces async data depending on metadata instruction. - 'Index' mode - """ - wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.scan_item = create_dummy_scan_item() - wf._scan_done = False # simulate a live scan - c = wf.plot(arg1="async_device", label="async_device-async_device") - wf._async_curves = [c] - # Suppose existing data - c.setData([0, 1, 2], [10, 11, 12]) - - # Set the x_axis_mode + wf.scan_id = "dummy" wf.x_axis_mode["name"] = "index" - ############# Test add ################ - - msg = {"signals": {"async_device": {"value": [100, 200], "timestamp": [1001, 1002]}}} - metadata = {"async_update": {"max_shape": [None], "type": "add"}} - - cb_info_ret = {"scan_id": wf.scan_id} - - def ret_sender(): - return SimpleNamespace(cb_info={"scan_id": wf.scan_id}) - - with mock.patch.object(wf, "sender", side_effect=ret_sender): - wf.on_async_readback(msg, metadata, _override_slot_params={"verify_sender": False}) + # 'add': fragments accumulate into one displayed series + source = _async_source( + "async_device", values=([10, 11, 12], [100, 200]), update_type="add", max_shape=(None,) + ) + wf._on_data_update(_make_update([source])) x_data, y_data = c.get_data() - assert len(x_data) == 5 - # Check x_data based on x_mode np.testing.assert_array_equal(x_data, [0, 1, 2, 3, 4]) - np.testing.assert_array_equal(y_data, [10, 11, 12, 100, 200]) - # instruction='replace' - msg2 = {"signals": {"async_device": {"value": [999], "timestamp": [555]}}} - metadata2 = {"async_update": {"max_shape": [None], "type": "replace"}} - with mock.patch.object(wf, "sender", side_effect=ret_sender): - wf.on_async_readback(msg2, metadata2, _override_slot_params={"verify_sender": False}) - x_data2, y_data2 = c.get_data() - np.testing.assert_array_equal(x_data2, [0]) + # 'replace': only the last full state is shown + source = _async_source("async_device", values=([999],), update_type="replace") + wf._on_data_update(_make_update([source])) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [0]) + np.testing.assert_array_equal(y_data, [999]) - np.testing.assert_array_equal(y_data2, [999]) - ############# Test add_slice ################ +def test_on_data_update_async_add_2d(qtbot, mocked_client, monkeypatch): + """ + An async 'add' source with a 2-D max_shape displays the latest waveform. + """ + _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" - # Few updates, no downsampling, no symbol removed - waveform_shape = 10 - for ii in range(10): - msg = {"signals": {"async_device": {"value": [100], "timestamp": [1001]}}} - metadata = { - "async_update": {"max_shape": [None, waveform_shape], "index": 0, "type": "add_slice"} - } - with mock.patch.object(wf, "sender", side_effect=ret_sender): - wf.on_async_readback(msg, metadata, _override_slot_params={"verify_sender": False}) + source = _async_source( + "async_device", values=([1, 2, 3], [4, 5, 6]), update_type="add", max_shape=(None, 3) + ) + wf._on_data_update(_make_update([source])) - # Old data should be deleted since the slice_index did not match + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) + np.testing.assert_array_equal(y_data, [4, 5, 6]) + + +def test_on_data_update_async_add_slice(qtbot, mocked_client, monkeypatch): + """ + An async 'add_slice' source displays the last accumulated row. Small rows + keep the symbol, large rows activate the downsampling settings. + """ + _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" + + # Small accumulated row: symbol stays, no downsampling + source = _async_source( + "async_device", + values=([100] * 10,), + ordinals=(0,), + update_type="add_slice", + max_shape=(None, 10), + ) + wf._on_data_update(_make_update([source])) x_data, y_data = c.get_data() assert len(y_data) == 10 assert len(x_data) == 10 assert c.opts["symbol"] == "o" - # Clear data from curve - c.setData([], []) - - # Test large updates, limit 1000 to deactivate symbols, downsampling for 8000 should be factor 2. + # Large accumulated row: symbol removed, downsampling active waveform_shape = 100000 - n_cycles = 10 - for ii in range(n_cycles): - msg = { - "signals": { - "async_device": { - "value": np.array(range(waveform_shape // n_cycles)), - "timestamp": (ii + 1) - * np.linspace(0, waveform_shape // n_cycles - 1, waveform_shape // n_cycles), - } - } - } - metadata = { - "async_update": {"max_shape": [None, waveform_shape], "index": 0, "type": "add_slice"} - } - with mock.patch.object(wf, "sender", side_effect=ret_sender): - wf.on_async_readback(msg, metadata, _override_slot_params={"verify_sender": False}) + source = _async_source( + "async_device", + values=(np.arange(waveform_shape),), + ordinals=(0,), + update_type="add_slice", + max_shape=(None, waveform_shape), + ) + wf._on_data_update(_make_update([source])) x_data, y_data = c.get_data() assert len(y_data) == waveform_shape assert len(x_data) == waveform_shape - assert c.opts["symbol"] == None - # Get displayed data + assert c.opts["symbol"] is None displayed_x, displayed_y = c.getData() assert len(displayed_y) == len(displayed_x) - ############# Test replace ################ - waveform_shape = 10 - for ii in range(10): - msg = { - "signals": { - "async_device": { - "value": np.array(range(waveform_shape)), - "timestamp": np.array(range(waveform_shape)), - } - } - } - metadata = {"async_update": {"type": "replace"}} - with mock.patch.object(wf, "sender", side_effect=ret_sender): - wf.on_async_readback(msg, metadata, _override_slot_params={"verify_sender": False}) - x_data, y_data = c.get_data() - assert np.array_equal(y_data, np.array(range(waveform_shape))) - assert len(x_data) == waveform_shape - assert c.opts["symbol"] == "o" - y_displayed, x_displayed = c.getData() - assert len(y_displayed) == waveform_shape - - -def test_get_x_data(qtbot, mocked_client, monkeypatch): +def test_on_data_update_async_timestamp_mode(qtbot, mocked_client, monkeypatch): """ - Test _get_x_data logic for multiple modes: 'timestamp', 'index', 'custom', 'auto'. - Use a dummy scan_item that returns specific data for the requested signal. + Async curves in timestamp mode use the source timestamps when they match + the displayed length and fall back to index otherwise. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - dummy_scan = create_dummy_scan_item() - wf.scan_item = dummy_scan - - # 1) x_mode == 'timestamp' + c = wf.plot(arg1="async_device", label="async_device-async_device") + wf.scan_id = "dummy" wf.x_axis_mode["name"] = "timestamp" - x_data = wf._get_x_data("bpm4i", "bpm4i") - np.testing.assert_array_equal(x_data, [101, 201, 301]) - # 2) x_mode == 'index' => returns None => means use Y data indexing + # One timestamp per single-sample fragment: lengths match -> timestamps + source = _async_source( + "async_device", values=([1], [2], [3]), timestamps=(11, 21, 31), update_type="add" + ) + wf._on_data_update(_make_update([source])) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [11, 21, 31]) + np.testing.assert_array_equal(y_data, [1, 2, 3]) + + # One timestamp for a multi-sample fragment: length mismatch -> index + source = _async_source("async_device", values=([1, 2, 3],), timestamps=(11,), update_type="add") + wf._on_data_update(_make_update([source])) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) + np.testing.assert_array_equal(y_data, [1, 2, 3]) + + +def test_on_data_update_async_history_rows(qtbot, mocked_client, monkeypatch): + """ + History emissions carry no async-update metadata: scalar rows form the + full series, array rows display the last row (mirrors the legacy 2-D + behaviour). + """ + _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" - x_data2 = wf._get_x_data("bpm4i", "bpm4i") - assert x_data2 is None - # 3) custom x => e.g. "samx" - wf.x_axis_mode["name"] = "samx" - x_custom = wf._get_x_data("bpm4i", "bpm4i") - # because dummy_scan.live_data["samx"]["samx"].val => [10,20,30] - np.testing.assert_array_equal(x_custom, [10, 20, 30]) + # Scalar rows (1-D dataset) + source = _async_source("async_device", values=(1, 2, 3), update_type=None, max_shape=()) + wf._on_data_update(_make_update([source], reason="history")) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) + np.testing.assert_array_equal(y_data, [1, 2, 3]) - # 4) auto - wf._async_curves.clear() - wf._sync_curves = [MagicMock()] # pretend we have a sync device - wf.x_axis_mode["name"] = "auto" - x_auto = wf._get_x_data("bpm4i", "bpm4i") - # By default it tries the "scan_report_devices" => "samx" => same as custom above - np.testing.assert_array_equal(x_auto, [10, 20, 30]) + # Array rows (2-D dataset): last row is displayed + source = _async_source( + "async_device", values=([1, 2, 3], [4, 5, 6]), update_type=None, max_shape=() + ) + wf._on_data_update(_make_update([source], reason="history")) + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [0, 1, 2]) + np.testing.assert_array_equal(y_data, [4, 5, 6]) ################################################## @@ -1391,14 +1506,15 @@ def test_show_scan_history_popup(qtbot, mocked_client): ##################################################### -# The following tests are for the async dataset guard +# The following tests are for the dataset-size guard ##################################################### -def test_skip_large_dataset_warning_property(qtbot, mocked_client): +def test_skip_large_dataset_warning_property(qtbot, mocked_client, monkeypatch): """ Verify the getter and setter of skip_large_dataset_warning work correctly. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) # Default should be False @@ -1413,13 +1529,26 @@ def test_skip_large_dataset_warning_property(qtbot, mocked_client): assert wf.skip_large_dataset_warning is False -def test_max_dataset_size_mb_property(qtbot, mocked_client): +def test_skip_large_dataset_check_property(qtbot, mocked_client, monkeypatch): + """ + Verify the getter and setter of the per-plot skip_large_dataset_check flag. + """ + _fake_bridge_factory(monkeypatch) + wf = create_widget(qtbot, Waveform, client=mocked_client) + + assert wf.skip_large_dataset_check is False + wf.skip_large_dataset_check = True + assert wf.skip_large_dataset_check is True + + +def test_max_dataset_size_mb_property(qtbot, mocked_client, monkeypatch): """ Verify getter, setter, and validation of max_dataset_size_mb. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) - # Default from WaveformConfig is 1 MB + # Default from WaveformConfig is 10 MB assert wf.max_dataset_size_mb == 10 # Set to a valid new value @@ -1428,116 +1557,118 @@ def test_max_dataset_size_mb_property(qtbot, mocked_client): # Ensure the config is updated too assert wf.config.max_dataset_size_mb == 5.5 - -def _dummy_dataset(mem_bytes: int, entry: str = "waveform_waveform"): - """ - Return an object that mimics the BEC dataset structure: - it has exactly one attribute `_info` with the expected layout. - """ - return SimpleNamespace(_info={entry: {"value": {"mem_size": mem_bytes}}}) + # The config survives a round-trip through the widget config dict + assert wf._config_dict["max_dataset_size_mb"] == 5.5 -def test_dataset_guard_under_limit(qtbot, mocked_client, monkeypatch): - """ - Dataset below the limit should load without triggering the dialog. - """ - wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.max_dataset_size_mb = 1 # 1 MiB - - # If the dialog is called, we flip this flag – it must stay False. - called = {"dlg": False} +def _oversized(monkeypatch, wf, size_mb=50.0, shape=(100, 1000)): + """Make every source of the widget look oversized, with a known shape.""" monkeypatch.setattr( - Waveform, "_confirm_large_dataset", lambda self, size_mb: called.__setitem__("dlg", True) + Waveform, "_estimate_source_bytes", lambda self, scan, source: int(size_mb * 1024 * 1024) ) - - dataset = _dummy_dataset(mem_bytes=512_000) # ≈0.49 MiB - assert wf._check_dataset_size_and_confirm(dataset, "waveform_waveform") is True - assert called["dlg"] is False + monkeypatch.setattr(Waveform, "_stored_shape", lambda self, scan, device, entry: shape) + return wf -def test_dataset_guard_over_limit_accept(qtbot, mocked_client, monkeypatch): - """ - Dataset above the limit where user presses *Yes*. - """ +def test_small_dataset_loads_without_prompt(qtbot, mocked_client, monkeypatch): wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.max_dataset_size_mb = 1 # 1 MiB - - # Pretend the user clicked “Yes” - monkeypatch.setattr(Waveform, "_confirm_large_dataset", lambda *_: True) - - dataset = _dummy_dataset(mem_bytes=2_000_000) # ≈1.9 MiB - assert wf._check_dataset_size_and_confirm(dataset, "waveform_waveform") is True + monkeypatch.setattr(Waveform, "_estimate_source_bytes", lambda self, scan, source: 1024) + with mock.patch.object(wf, "_confirm_large_dataset") as dialog: + kept = wf._filter_oversized_sources([("bpm4i", "bpm4i")], "scan-1") + dialog.assert_not_called() + assert kept == [("bpm4i", "bpm4i")] -def test_dataset_guard_over_limit_reject(qtbot, mocked_client, monkeypatch): - """ - Dataset above the limit where user presses *No*. - """ +def test_live_and_skip_check_bypass_the_gate(qtbot, mocked_client, monkeypatch): wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.max_dataset_size_mb = 1 # 1 MiB - - # Pretend the user clicked “No” - monkeypatch.setattr(Waveform, "_confirm_large_dataset", lambda *_: False) - - dataset = _dummy_dataset(mem_bytes=2_000_000) # ≈1.9 MiB - assert wf._check_dataset_size_and_confirm(dataset, "waveform_waveform") is False + _oversized(monkeypatch, wf) + sources = [("waveform", "waveform_waveform")] + with mock.patch.object(wf, "_confirm_large_dataset") as dialog: + assert wf._filter_oversized_sources(sources, "live") == sources + assert wf._filter_oversized_sources(sources, None) == sources + wf.skip_large_dataset_check = True + assert wf._filter_oversized_sources(sources, "scan-1") == sources + dialog.assert_not_called() -################################################## -# Dialog propagation behaviour -################################################## - - -def test_dialog_accept_updates_limit(monkeypatch, qtbot, mocked_client): - """ - Simulate clicking 'Yes' in the dialog *after* changing the spinner value. - Verify max_dataset_size_mb is updated and dataset loads. - """ +def test_oversized_dataset_prompts_per_dataset_and_loads_on_accept( + qtbot, mocked_client, monkeypatch +): + """The user decides per dataset; an accepted dataset is loaded and is not + re-prompted when the subscription is rebuilt.""" wf = create_widget(qtbot, Waveform, client=mocked_client) - wf.max_dataset_size_mb = 1 # start small + _oversized(monkeypatch, wf, size_mb=50.0, shape=(100, 1000)) + sources = [("waveform", "waveform_waveform"), ("bpm4i", "bpm4i")] - def fake_confirm(self, size_mb): - # Simulate user typing '5' in the spinbox then pressing Yes - self.config.max_dataset_size_mb = 5 - return True # Yes pressed + with mock.patch.object(wf, "_confirm_large_dataset", return_value=True) as dialog: + kept = wf._filter_oversized_sources(sources, "scan-1") + assert kept == sources + # One prompt per dataset, carrying the dataset identity and its shape. + assert dialog.call_count == 2 + first = dialog.call_args_list[0] + assert first.kwargs["source"] == ("waveform", "waveform_waveform") + assert first.kwargs["shape"] == (100, 1000) + assert first.args[0] == pytest.approx(50.0) - monkeypatch.setattr(Waveform, "_confirm_large_dataset", fake_confirm) - - big_dataset = _dummy_dataset(mem_bytes=4_800_000) # ≈4.6 MiB - accepted = wf._check_dataset_size_and_confirm(big_dataset, "waveform_waveform") - - # The load should be accepted and the limit must reflect the new value - assert accepted is True - assert wf.max_dataset_size_mb == 5 - assert wf.config.max_dataset_size_mb == 5 + with mock.patch.object(wf, "_confirm_large_dataset") as dialog2: + assert wf._filter_oversized_sources(sources, "scan-1") == sources + dialog2.assert_not_called() -def test_dialog_cancel_sets_skip(monkeypatch, qtbot, mocked_client): - """ - Simulate clicking 'No' but ticking 'Don't show again'. - Verify skip_large_dataset_warning becomes True and dataset is skipped. - """ +def test_declined_dataset_is_not_loaded_but_others_are(qtbot, mocked_client, monkeypatch): + """Declining drops only that dataset — the curve stays (empty), it is not + hidden, and the remaining datasets still load.""" wf = create_widget(qtbot, Waveform, client=mocked_client) - assert wf.skip_large_dataset_warning is False - def fake_confirm(self, size_mb): - # Mimic ticking the checkbox then pressing No - self._skip_large_dataset_warning = True - return False # No pressed + def estimate(self, scan, source): + return 50 * 1024 * 1024 if source[0] == "waveform" else 1024 - monkeypatch.setattr(Waveform, "_confirm_large_dataset", fake_confirm) + monkeypatch.setattr(Waveform, "_estimate_source_bytes", estimate) + monkeypatch.setattr(Waveform, "_stored_shape", lambda self, scan, device, entry: (100, 1000)) + sources = [("waveform", "waveform_waveform"), ("bpm4i", "bpm4i")] - big_dataset = _dummy_dataset(mem_bytes=11_000_000) - accepted = wf._check_dataset_size_and_confirm(big_dataset, "waveform_waveform") - - # Dataset must not load, but future warnings are suppressed - assert accepted is False - assert wf.skip_large_dataset_warning is True + with mock.patch.object(wf, "_confirm_large_dataset", return_value=False) as dialog: + kept = wf._filter_oversized_sources(sources, "scan-1") + dialog.assert_called_once() + assert kept == [("bpm4i", "bpm4i")] -################################################## -# Live dialog interaction (no monkey‑patching) -################################################## +def test_skip_large_dataset_warning_suppresses_dialog(qtbot, mocked_client, monkeypatch): + wf = create_widget(qtbot, Waveform, client=mocked_client) + _oversized(monkeypatch, wf) + wf.skip_large_dataset_warning = True + with mock.patch.object(wf, "_confirm_large_dataset") as dialog: + kept = wf._filter_oversized_sources([("waveform", "waveform_waveform")], "scan-1") + dialog.assert_not_called() + assert kept == [] + + +def test_history_subscription_gates_before_reading(qtbot, mocked_client, monkeypatch): + """No bridge (and therefore no file read) is created for a declined + dataset.""" + bridges = _fake_bridge_factory(monkeypatch) + wf = create_widget(qtbot, Waveform, client=mocked_client) + wf.plot("bpm4i") # a configured curve, so the widget has sources to bind + _oversized(monkeypatch, wf) + bridges.clear() + + with mock.patch.object(wf, "_confirm_large_dataset", return_value=False): + wf._setup_data_api_subscription(scan="scan-1") + assert not bridges + assert wf._data_bridge is None + + with mock.patch.object(wf, "_confirm_large_dataset", return_value=True): + wf._setup_data_api_subscription(scan="scan-1") + assert bridges and bridges[-1].scan == "scan-1" + # The gate is decided up front, so the bridge itself is never size-gated. + assert getattr(bridges[-1], "size_limit_bytes", None) is None + + +def test_describe_dataset_reports_point_counts(): + assert "1,000 points" in Waveform._describe_dataset(("det", "sig"), (1000,)) + text = Waveform._describe_dataset(("det", "sig"), (100, 1000)) + assert "100 points x 1,000 samples" in text + assert "'det-sig'" in text def _open_dialog_and_click(handler): @@ -1555,48 +1686,43 @@ def _open_dialog_and_click(handler): return _cb -def test_dialog_accept_real_interaction(qtbot, mocked_client): +def test_dialog_accept_real_interaction(qtbot, mocked_client, monkeypatch): """ - End‑to‑end: user changes the limit spinner to 5 MiB, ticks + End-to-end: user changes the limit spinner to 5 MiB, ticks 'don't show again', then presses YES. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.max_dataset_size_mb = 1 - # Prepare a large dataset (≈4.6 MiB) - big_dataset = _dummy_dataset(mem_bytes=4_800_000) - def handler(dlg): spin: QDoubleSpinBox = dlg.findChild(QDoubleSpinBox) chk: QCheckBox = dlg.findChild(QCheckBox) btns: QDialogButtonBox = dlg.findChild(QDialogButtonBox) - # # Interact with widgets spin.setValue(5) chk.setChecked(True) yes_btn = btns.button(QDialogButtonBox.Yes) yes_btn.click() - # Schedule the handler right before invoking the check QTimer.singleShot(0, _open_dialog_and_click(handler)) - accepted = wf._check_dataset_size_and_confirm(big_dataset, "waveform_waveform") + accepted = wf._confirm_large_dataset(4.6) assert accepted is True assert wf.max_dataset_size_mb == 5 assert wf.skip_large_dataset_warning is True -def test_dialog_reject_real_interaction(qtbot, mocked_client): +def test_dialog_reject_real_interaction(qtbot, mocked_client, monkeypatch): """ - End‑to‑end: user leaves spinner unchanged, ticks 'don't show again', + End-to-end: user leaves spinner unchanged, ticks 'don't show again', and presses NO. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.max_dataset_size_mb = 1 - big_dataset = _dummy_dataset(mem_bytes=4_800_000) - def handler(dlg): chk: QCheckBox = dlg.findChild(QCheckBox) btns: QDialogButtonBox = dlg.findChild(QDialogButtonBox) @@ -1607,17 +1733,18 @@ def test_dialog_reject_real_interaction(qtbot, mocked_client): QTimer.singleShot(0, _open_dialog_and_click(handler)) - accepted = wf._check_dataset_size_and_confirm(big_dataset, "waveform_waveform") + accepted = wf._confirm_large_dataset(4.6) assert accepted is False assert wf.skip_large_dataset_warning is True # Limit remains unchanged assert wf.max_dataset_size_mb == 1 -def test_update_with_scan_history_by_index(qtbot, mocked_client, scan_history_factory): +def test_update_with_scan_history_by_index(qtbot, mocked_client, scan_history_factory, monkeypatch): """ Test that update_with_scan_history by index loads the correct historical scan. """ + bridges = _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) hist1, hist2 = inject_scan_history(wf, scan_history_factory, ("hist1", 1), ("hist2", 2)) @@ -1642,12 +1769,60 @@ def test_update_with_scan_history_by_index(qtbot, mocked_client, scan_history_fa assert c2.config.scan_number == 2 assert c2.name() == "bpm4i-bpm4i-scan-2" + # One scan-bound subscription per pinned history scan. + history_scans = {bridge.scan for bridge in bridges if not bridge.closed} + assert history_scans == {"hist1", "hist2"} + + +def test_history_curve_receives_data_from_scan_bound_subscription( + qtbot, mocked_client, scan_history_factory, monkeypatch +): + """ + History curve data flows through a DataAPI subscription bound to the + pinned scan id; updates are routed by scan id and rendered against the + auto-resolved x device. + """ + bridges = _fake_bridge_factory(monkeypatch) + wf = create_widget(qtbot, Waveform, client=mocked_client) + inject_scan_history(wf, scan_history_factory, ("hist1", 1)) + + c = wf.plot(device_y="bpm4i", signal_y="bpm4i", scan_id="hist1") + history_bridge = wf._history_bridges["hist1"] + assert history_bridge in bridges + assert history_bridge.scan == "hist1" + assert ("bpm4i", "bpm4i") in history_bridge.sources + # Auto mode pulls in the scan's first report device as x source. + assert ("samx", "samx") in history_bridge.sources + + update = _make_update( + [ + _monitored_source("bpm4i", values=(5, 6, 7)), + _monitored_source("samx", values=(10, 20, 30)), + ], + scan_id="hist1", + reason="history", + ) + wf._on_data_update(update) + + x_data, y_data = c.get_data() + np.testing.assert_array_equal(x_data, [10, 20, 30]) + np.testing.assert_array_equal(y_data, [5, 6, 7]) + + # Updates for other scans do not touch the pinned curve. + other = _make_update([_monitored_source("bpm4i", values=(1, 1, 1))], scan_id="other-scan") + wf._on_data_update(other) + _, y_data = c.get_data() + np.testing.assert_array_equal(y_data, [5, 6, 7]) + @pytest.mark.parametrize("mode", ["auto", "timestamp", "index", "samx"]) -def test_history_curve_x_modes_pre_plot(qtbot, mocked_client, scan_history_factory, mode): +def test_history_curve_x_modes_pre_plot( + qtbot, mocked_client, scan_history_factory, mode, monkeypatch +): """ Test that history curves respect x_mode when set before plotting. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) hist1, hist2 = inject_scan_history(wf, scan_history_factory, ("hist1", 1), ("hist2", 2)) wf.x_mode = mode @@ -1656,10 +1831,13 @@ def test_history_curve_x_modes_pre_plot(qtbot, mocked_client, scan_history_facto @pytest.mark.parametrize("mode", ["auto", "timestamp", "index", "samx"]) -def test_history_curve_x_modes_post_plot(qtbot, mocked_client, scan_history_factory, mode): +def test_history_curve_x_modes_post_plot( + qtbot, mocked_client, scan_history_factory, mode, monkeypatch +): """ Test that changing x_mode after plotting history curves updates the curve on refresh. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) hist1, hist2 = inject_scan_history(wf, scan_history_factory, ("hist1", 1), ("hist2", 2)) c = wf.plot(device_y="bpm4i", signal_y="bpm4i", scan_id="hist1") @@ -1670,10 +1848,13 @@ def test_history_curve_x_modes_post_plot(qtbot, mocked_client, scan_history_fact assert c.config.current_x_mode == mode -def test_history_curve_incompatible_x_mode_hides_curve(qtbot, mocked_client, scan_history_factory): +def test_history_curve_incompatible_x_mode_hides_curve( + qtbot, mocked_client, scan_history_factory, monkeypatch +): """ Test that setting an x_mode not present in stored data hides the history curve. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.x_mode = "nonexistent_device" # Inject history scan for this test @@ -1684,11 +1865,12 @@ def test_history_curve_incompatible_x_mode_hides_curve(qtbot, mocked_client, sca assert not c.isVisible() -def test_fetch_history_data_no_stored_data_raises( +def test_history_curve_no_stored_data_raises( qtbot, mocked_client, monkeypatch, suppress_message_box ): """ - Test that fetching history data when stored_data_info is missing raises ValueError. + Test that plotting a history curve when stored_data_info is missing + raises ValueError (metadata-level validation kept from the legacy fetch). """ wf = create_widget(qtbot, Waveform, client=mocked_client) # Create a dummy scan_item lacking stored_data_info @@ -1717,11 +1899,12 @@ def test_history_curve_device_missing_returns_none(qtbot, mocked_client, scan_hi def test_history_curve_custom_shape_mismatch_hides_curve( - qtbot, mocked_client, scan_history_factory + qtbot, mocked_client, scan_history_factory, monkeypatch ): """ For custom x-mode, if x and y shapes mismatch, curve should be hidden. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.x_mode = "async_device" [history_msg] = inject_scan_history(wf, scan_history_factory, ("hist_custom_shape", 1)) @@ -1731,10 +1914,13 @@ def test_history_curve_custom_shape_mismatch_hides_curve( assert not c.isVisible() -def test_history_curve_index_mode_plots_curve(qtbot, mocked_client, scan_history_factory): +def test_history_curve_index_mode_plots_curve( + qtbot, mocked_client, scan_history_factory, monkeypatch +): """ Test that setting x_mode to 'index' plots and shows the history curve correctly. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.x_mode = "index" [history_msg] = inject_scan_history(wf, scan_history_factory, ("hist_index", 1)) @@ -1744,10 +1930,13 @@ def test_history_curve_index_mode_plots_curve(qtbot, mocked_client, scan_history assert c.config.current_x_mode == "index" -def test_history_curve_timestamp_mode_plots_curve(qtbot, mocked_client, scan_history_factory): +def test_history_curve_timestamp_mode_plots_curve( + qtbot, mocked_client, scan_history_factory, monkeypatch +): """ Test that setting x_mode to 'timestamp' plots and shows the history curve correctly. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.x_mode = "timestamp" [history_msg] = inject_scan_history(wf, scan_history_factory, ("hist_time", 1)) @@ -1758,11 +1947,12 @@ def test_history_curve_timestamp_mode_plots_curve(qtbot, mocked_client, scan_his def test_history_curve_auto_valid_uses_first_report_device( - qtbot, mocked_client, scan_history_factory + qtbot, mocked_client, scan_history_factory, monkeypatch ): """ Test that 'auto' x_mode uses the first available report device and shows the curve. """ + _fake_bridge_factory(monkeypatch) wf = create_widget(qtbot, Waveform, client=mocked_client) wf.x_mode = "auto" [history_msg] = inject_scan_history(wf, scan_history_factory, ("hist_auto_valid", 1)) @@ -1841,3 +2031,97 @@ def test_categorise_device_curves_falls_back_to_readout_priority(qtbot, mocked_c wf._categorise_device_curves() assert c_async in wf._async_curves + + +def test_x_source_resolution_does_not_open_the_data_file(qtbot, mocked_client, monkeypatch): + """Auto x-mode must resolve the report device from in-memory metadata: + ScanDataContainer.metadata lazily opens the HDF5 file, which would block + the GUI thread (and would even precede the large-dataset gate).""" + wf = create_widget(qtbot, Waveform, client=mocked_client) + + class _ExplodingMetadata: + def __getitem__(self, key): + raise AssertionError("data file was opened on the GUI thread") + + def get(self, *args, **kwargs): + raise AssertionError("data file was opened on the GUI thread") + + # History container: report device comes from the scan history message. + history_item = SimpleNamespace( + metadata=_ExplodingMetadata(), + _msg=SimpleNamespace(request_inputs={"arg_bundle": ["samx", -5, 5, 10], "kwargs": {}}), + status_message=None, + ) + assert wf._report_devices_no_file_io(history_item) == ["samx"] + assert wf._history_x_source_key(history_item) == ("samx", "samx") + + # Live scan item: report devices come from the status message. + live_item = SimpleNamespace( + metadata=_ExplodingMetadata(), + status_message=SimpleNamespace(scan_report_devices=["samy"], info={}), + _msg=None, + ) + assert wf._report_devices_no_file_io(live_item) == ["samy"] + + +def test_x_source_falls_back_to_file_metadata_when_unavailable(qtbot, mocked_client): + """When neither the status message nor the history message names the + report devices, the file metadata is still consulted (last resort).""" + wf = create_widget(qtbot, Waveform, client=mocked_client) + scan_item = SimpleNamespace( + metadata={"bec": {"scan_report_devices": ["samx"]}}, status_message=None, _msg=None + ) + assert wf._report_devices_no_file_io(scan_item) == [] + assert wf._history_x_source_key(scan_item) == ("samx", "samx") + + +def test_detector_shaped_history_curve_is_not_hidden(qtbot, mocked_client, monkeypatch): + """A large detector dataset has more rows than the scan has points; it is + plotted against the sample index, so the x/y row-count difference must not + hide the curve (regression: large history datasets disappeared).""" + _fake_bridge_factory(monkeypatch) + wf = create_widget(qtbot, Waveform, client=mocked_client) + wf.x_mode = "samx" # custom device x-axis: the strictest visibility path + + stored = { + "samx": {"samx": SimpleNamespace(shape=(100,))}, + # Flat async concatenation: 100 acquisitions x 1000 samples appended + # along one axis — the shape a large "add" async signal really has. + "waveform": {"waveform_waveform": SimpleNamespace(shape=(100_000,))}, + # 2-D detector data (one row per acquisition). + "eiger": {"eiger_data": SimpleNamespace(shape=(100, 1000))}, + "bpm4i": {"bpm4i": SimpleNamespace(shape=(100,))}, + } + scan_item = SimpleNamespace( + _msg=SimpleNamespace( + stored_data_info=stored, + num_points=100, + num_monitored_readouts=100, + request_inputs={"arg_bundle": ["samx", -5, 5, 100], "kwargs": {}}, + ), + status_message=None, + ) + monkeypatch.setattr(Waveform, "get_history_scan_item", lambda self, **kwargs: scan_item) + + def curve_for(device, signal): + return SimpleNamespace( + config=SimpleNamespace( + scan_id="scan-1", + scan_number=None, + signal=SimpleNamespace(device=device, signal=signal), + ), + name=lambda: f"{device}-{signal}", + ) + + # Flat async data (100 000 samples vs a 100-point motor): visible. + assert wf._history_curve_compatible(curve_for("waveform", "waveform_waveform")) is True + # 2-D detector data: visible. + assert wf._history_curve_compatible(curve_for("eiger", "eiger_data")) is True + # A monitored signal with the same length as x: visible. + assert wf._history_curve_compatible(curve_for("bpm4i", "bpm4i")) is True + # A monitored signal whose length disagrees with x is still hidden + # (unchanged behaviour for point-per-scan-point data). + stored["bpm4i"]["bpm4i"] = SimpleNamespace(shape=(37,)) + scan_item._msg.num_monitored_readouts = 37 + scan_item._msg.num_points = 37 + assert wf._history_curve_compatible(curve_for("bpm4i", "bpm4i")) is False