diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index aecf751f..56b0f853 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -1832,6 +1832,20 @@ class Heatmap(RPCBase): Enable the FPS monitor. """ + @property + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + + @show_info_label.setter + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + @rpc_call def set(self, **kwargs): """ @@ -2540,6 +2554,20 @@ class Image(RPCBase): Enable the FPS monitor. """ + @property + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + + @show_info_label.setter + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + @rpc_call def set(self, **kwargs): """ @@ -3713,6 +3741,20 @@ class MotorMap(RPCBase): Enable the FPS monitor. """ + @property + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + + @show_info_label.setter + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + @rpc_call def set(self, **kwargs): """ @@ -4185,6 +4227,20 @@ class MultiWaveform(RPCBase): Enable the FPS monitor. """ + @property + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + + @show_info_label.setter + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + @rpc_call def set(self, **kwargs): """ @@ -5446,6 +5502,20 @@ class ScatterWaveform(RPCBase): Enable the FPS monitor. """ + @property + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + + @show_info_label.setter + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + @rpc_call def set(self, **kwargs): """ @@ -6097,6 +6167,20 @@ class Waveform(RPCBase): Enable the FPS monitor. """ + @property + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + + @show_info_label.setter + @rpc_call + def show_info_label(self) -> "bool": + """ + Whether the plot info label is shown. + """ + @rpc_call def set(self, **kwargs): """ diff --git a/bec_widgets/widgets/plots/heatmap/heatmap.py b/bec_widgets/widgets/plots/heatmap/heatmap.py index 7af9cc77..7466a562 100644 --- a/bec_widgets/widgets/plots/heatmap/heatmap.py +++ b/bec_widgets/widgets/plots/heatmap/heatmap.py @@ -276,7 +276,6 @@ class Heatmap(ImageBase): # Scan ID the widget is pinned to when plotting from history; None means live mode. self._history_scan_id: str | None = None self._selected_history_scan_id: str | None = None - self.config_label = self.info_label self.show_info_label = self._image_config.show_config_label self.reload = False self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status()) @@ -309,15 +308,6 @@ class Heatmap(ImageBase): # Widget Specific GUI interactions ################################################################################ - @SafeSlot(str) - def apply_theme(self, theme: str): - """ - Apply the current theme to the heatmap widget. - """ - super().apply_theme(theme) - if hasattr(self, "info_label"): - self.redraw_config_label() - @SafeSlot(popup_error=True) def plot( self, @@ -380,7 +370,7 @@ class Heatmap(ImageBase): lock_aspect_ratio = self._image_config.lock_aspect_ratio if show_config_label is None: - show_config_label = self._image_config.show_config_label + show_config_label = self.show_info_label def _device_key(device: HeatmapDeviceSignal | None) -> tuple[str | None, str | None]: return (device.device if device else None, device.signal if device else None) @@ -409,6 +399,7 @@ class Heatmap(ImageBase): lock_aspect_ratio=lock_aspect_ratio, show_config_label=show_config_label, ) + self.show_info_label = show_config_label self.color_map = color_map self.reload = reload or config_changed if config_changed: @@ -637,28 +628,6 @@ class Heatmap(ImageBase): self._selected_history_scan_id = None self.toolbar.components.get_action("scan_history").action.setChecked(False) - @SafeProperty(bool, auto_emit=True) - def show_config_label(self) -> bool: - """Whether to show heatmap configuration rows in the plot info label.""" - return self._image_config.show_config_label - - @show_config_label.setter - def show_config_label(self, value: bool) -> None: - self._image_config.show_config_label = bool(value) - self.show_info_label = bool(value) - self.redraw_config_label() - - @SafeSlot() - def toggle_info_label(self): - """Toggle the heatmap information label and keep heatmap config in sync.""" - self.show_config_label = not self.show_config_label - - def toggle_interpolation_info(self): - """ - Toggle the visibility of the heatmap information label. - """ - self.toggle_info_label() - def _heatmap_dialog_closed(self): """ Slot for when the heatmap settings dialog is closed. @@ -785,9 +754,7 @@ class Heatmap(ImageBase): logger.warning("Scan message is None; skipping update.") return self.status_message = scan_msg - - if self._image_config.show_config_label: - self.redraw_config_label() + self.redraw_config_label() if self._is_grid_scan_supported(scan_msg): img, transform = self.get_grid_scan_image(z_data, scan_msg) @@ -951,12 +918,9 @@ class Heatmap(ImageBase): self._latest_interpolation_version = -1 def redraw_config_label(self): - scan_msg = getattr(self, "status_message", None) + scan_msg = self.status_message if scan_msg is None: return - if not self._image_config.show_config_label: - self.show_info_label = False - return # Indicate whether the widget follows the live acquisition or is pinned to a history scan mode = "history" if self._history_scan_id is not None else "live" diff --git a/bec_widgets/widgets/plots/image/image.py b/bec_widgets/widgets/plots/image/image.py index 0b5dd483..2b5ad820 100644 --- a/bec_widgets/widgets/plots/image/image.py +++ b/bec_widgets/widgets/plots/image/image.py @@ -159,6 +159,7 @@ class Image(ImageBase): crosshair_bundle = self.toolbar.get_bundle("image_crosshair") crosshair_bundle.add_action("image_autorange") crosshair_bundle.add_action("image_colorbar_switch") + self.remove_info_label_action() self.toolbar.show_bundles( [ diff --git a/bec_widgets/widgets/plots/motor_map/motor_map.py b/bec_widgets/widgets/plots/motor_map/motor_map.py index 0121054e..300b2e54 100644 --- a/bec_widgets/widgets/plots/motor_map/motor_map.py +++ b/bec_widgets/widgets/plots/motor_map/motor_map.py @@ -167,7 +167,7 @@ class MotorMap(PlotBase): ) self.toolbar.components.get_action("reset_legend").action.setVisible(False) - self.toolbar.get_bundle("axis_popup").remove_action("plot_info_label") + self.remove_info_label_action() reset_legend = MaterialIconAction( icon_name="history", diff --git a/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py b/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py index 3875173d..7e81d6a0 100644 --- a/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py +++ b/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py @@ -361,8 +361,10 @@ class MultiWaveform(PlotBase): current_scan_id = metadata.get("scan_id", None) if current_scan_id != self.scan_id: - scan_item = self.update_scan_info_from_scan_id(current_scan_id, mode="live") - self.scan_id = scan_item.scan_id if scan_item is not None else current_scan_id + self.scan_id = current_scan_id + self.update_scan_info_from_source( + self.queue.scan_storage.find_scan_by_ID(current_scan_id) + ) self.clear_curves() self.curves.clear() if self.crosshair: diff --git a/bec_widgets/widgets/plots/plot_base.py b/bec_widgets/widgets/plots/plot_base.py index f35c8aae..6c026073 100644 --- a/bec_widgets/widgets/plots/plot_base.py +++ b/bec_widgets/widgets/plots/plot_base.py @@ -1,11 +1,12 @@ from __future__ import annotations +from collections.abc import Iterable, Mapping from enum import Enum -from typing import TypeAlias +from typing import Literal, TypeAlias import numpy as np import pyqtgraph as pg -from bec_lib import bec_logger, messages +from bec_lib import bec_logger from bec_lib.scan_data_container import ScanDataContainer from bec_lib.scan_items import ScanItem from qtpy.QtCore import QPoint, QPointF, Qt, Signal @@ -44,6 +45,7 @@ from bec_widgets.widgets.plots.toolbar_components.roi import RoiConnection, roi_ logger = bec_logger.logger ScanInfoSource: TypeAlias = ScanItem | ScanDataContainer | None +ScanInfoMode: TypeAlias = Literal["live", "history"] class BECViewBox(pg.ViewBox): @@ -79,6 +81,8 @@ class PlotBase(BECWidget, QWidget): "enable_side_panel.setter", "enable_fps_monitor", "enable_fps_monitor.setter", + "show_info_label", + "show_info_label.setter", "set", "title", "title.setter", @@ -171,6 +175,8 @@ class PlotBase(BECWidget, QWidget): self.plot_item.addLegend() self.info_label = PlotInfoLabel() self.info_label.setParentItem(self.plot_item.vb) + self._scan_info_rows: list[tuple[str, object | None]] = [] + self._custom_info_rows: list[tuple[str, object | None]] = [] self.crosshair = None # Holds a pin that outlived its crosshair (e.g. crosshair toggled off) so it # can be re-adopted when the crosshair is hooked again. @@ -290,7 +296,7 @@ class PlotBase(BECWidget, QWidget): @SafeProperty(bool, auto_emit=True, doc="Show the plot info label.") def show_info_label(self) -> bool: - """Return whether the plot info label is enabled.""" + """Whether the plot info label is shown.""" return self._show_info_label @show_info_label.setter @@ -309,111 +315,85 @@ class PlotBase(BECWidget, QWidget): self.show_info_label = not self.show_info_label def set_info_label_rows( - self, rows: dict[str, object | None] | list[tuple[str, object | None]] + self, rows: Mapping[str, object | None] | Iterable[tuple[str, object | None]] ) -> None: - """Submit arbitrary rows for the plot info label.""" - self.info_label.set_rows(rows) - self._sync_info_label_visibility() - self._sync_info_label_action() + """Show caller-defined rows in the info label, below the scan rows. + + The rows replace previously submitted custom rows and survive scan updates. + """ + self._custom_info_rows = list(rows.items() if isinstance(rows, Mapping) else rows) + self._render_info_label() def clear_info_label(self) -> None: - """Clear all rows from the plot info label.""" - self.info_label.clear_rows() + """Clear the scan rows and the custom rows of the info label.""" + self._scan_info_rows = [] + self._custom_info_rows = [] + self._render_info_label() + + def _render_info_label(self) -> None: + self.info_label.set_rows([*self._scan_info_rows, *self._custom_info_rows]) + self._sync_info_label_visibility() self._sync_info_label_action() def set_scan_info( self, *, - scan_id: str | None = None, scan_number: int | str | None = None, scan_name: str | None = None, - mode: str | None = None, - extra_rows: dict[str, object | None] | list[tuple[str, object | None]] | None = None, + mode: ScanInfoMode | None = None, + extra_rows: Iterable[tuple[str, object | None]] | None = None, ) -> None: - """Submit common scan metadata rows to the plot info label.""" + """Submit the common scan rows (scan number with mode, scan name) to the info label.""" rows: list[tuple[str, object | None]] = [] if scan_number is not None: - scan_value = f"{scan_number} ({mode})" if mode else scan_number - rows.append(("Scan", scan_value)) - elif scan_id is not None and mode != "history": - scan_value = f"{scan_id} ({mode})" if mode else scan_id - rows.append(("Scan ID", scan_value)) + rows.append(("Scan", f"{scan_number} ({mode})" if mode else scan_number)) if scan_name is not None: rows.append(("Scan Name", scan_name)) if extra_rows: - rows.extend(extra_rows.items() if isinstance(extra_rows, dict) else extra_rows) - self.set_info_label_rows(rows) + rows.extend(extra_rows) + self._scan_info_rows = rows + self._render_info_label() - def update_scan_info_from_source( - self, scan_source: ScanInfoSource, *, mode: str | None = None - ) -> None: - """Extract common scan metadata from a scan item and submit it.""" - scan_info = self._extract_scan_info(scan_source) - if scan_info is None: - return - scan_id, scan_number, scan_name = scan_info - self.set_scan_info(scan_id=scan_id, scan_number=scan_number, scan_name=scan_name, mode=mode) + def update_scan_info_from_source(self, scan_source: ScanInfoSource) -> None: + """ + Show the scan rows of a live ScanItem or a history ScanDataContainer. - def update_scan_info_from_scan_id( - self, scan_id: str | None, *, mode: str | None = None - ) -> ScanItem | None: - """Fetch a live scan item by ID and submit common scan metadata.""" - if scan_id is None: - return None - scan_item = self.queue.scan_storage.find_scan_by_ID(scan_id) - if not isinstance(scan_item, ScanItem): - return None - self.update_scan_info_from_source(scan_item, mode=mode) - return scan_item + Args: + scan_source(ScanInfoSource): The source of the scan information, which can be a ScanItem, ScanDataContainer, or None. + """ + if isinstance(scan_source, ScanItem) and scan_source.status_message is not None: + status_message = scan_source.status_message + self.set_scan_info( + scan_number=status_message.scan_number, + scan_name=status_message.scan_name, + mode="live", + ) + elif isinstance(scan_source, ScanDataContainer): + metadata = scan_source.metadata.get("bec", {}) + self.set_scan_info( + scan_number=metadata.get("scan_number"), + scan_name=metadata.get("scan_name"), + mode="history", + ) + else: + self.set_scan_info() - def _extract_scan_info( - self, scan_source: ScanInfoSource - ) -> tuple[str | None, int | str | None, str | None] | None: - if scan_source is None: - return None - - if isinstance(scan_source, ScanItem): - return self._scan_info_from_status_message(scan_source.status_message) - - if isinstance(scan_source, ScanDataContainer): - metadata = scan_source.metadata - scan_info = self._scan_info_from_metadata(metadata.get("bec", metadata)) - if scan_info is not None: - return scan_info - - return None - - @staticmethod - def _scan_info_from_status_message( - status_message: messages.ScanStatusMessage | None, - ) -> tuple[str | None, int | str | None, str | None] | None: - if status_message is None: - return None - scan_info = (status_message.scan_id, status_message.scan_number, status_message.scan_name) - return scan_info if any(value is not None for value in scan_info) else None - - @staticmethod - def _scan_info_from_metadata( - metadata: dict | None, - ) -> tuple[str | None, int | str | None, str | None] | None: - if not isinstance(metadata, dict): - return None - scan_id = metadata.get("scan_id") - scan_number = metadata.get("scan_number") - scan_name = metadata.get("scan_name") - scan_info = (scan_id, scan_number, scan_name) - return scan_info if any(value is not None for value in scan_info) else None + def remove_info_label_action(self) -> None: + """Remove the info label toolbar button, for widgets that never populate the label.""" + component = self.toolbar.components.get_action("plot_info_label") + component.action.triggered.disconnect(self.toggle_info_label) + self.toolbar.components.remove_action("plot_info_label") + component.action.deleteLater() def _sync_info_label_visibility(self) -> None: self.info_label.setVisible(self._show_info_label and bool(self.info_label.rows)) def _sync_info_label_action(self) -> None: - try: - action = self.toolbar.components.get_action("plot_info_label").action - except KeyError: - return - action.setEnabled(True) - action.setChecked(self._show_info_label) + if not self.toolbar.components.exists("plot_info_label"): + return # the widget removed the button with remove_info_label_action() + self.toolbar.components.get_action("plot_info_label").action.setChecked( + self._show_info_label + ) ################################################################################ # Toggle UI Elements diff --git a/bec_widgets/widgets/plots/plot_info_label.py b/bec_widgets/widgets/plots/plot_info_label.py index 0e859bd1..fdbd1b8e 100644 --- a/bec_widgets/widgets/plots/plot_info_label.py +++ b/bec_widgets/widgets/plots/plot_info_label.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Iterable, Mapping +from collections.abc import Iterable import pyqtgraph as pg from qtpy.QtCore import QRectF @@ -32,45 +32,36 @@ class TextOnlyLegendSample(pg.graphicsItems.LegendItem.ItemSample): class PlotInfoLabel(pg.LegendItem): """Paint-safe text overlay for plot metadata. - Plot widgets submit arbitrary rows; this class owns the pyqtgraph legend - mechanics and keeps the display independent from any specific metadata - source such as scans. + Plot widgets submit (label, value) rows; this class owns the pyqtgraph + legend mechanics and keeps the display independent from any specific + metadata source such as scans. """ def __init__(self, offset: tuple[int, int] = (-30, 1), theme: str = "light"): super().__init__(offset=offset, horSpacing=0) - self._rows: list[tuple[str, object | None]] = [] + self._rows: list[tuple[str, str | None]] = [] self._offset = offset - self._has_drawn_rows = False self.set_theme(theme) self.setVisible(False) @property - def rows(self) -> list[tuple[str, object | None]]: - """Return a copy of the currently submitted rows.""" + def rows(self) -> list[tuple[str, str | None]]: + """Return a copy of the currently displayed rows.""" return list(self._rows) - def set_rows( - self, rows: Mapping[str, object | None] | Iterable[tuple[str, object | None]] - ) -> None: - """Replace all displayed rows.""" - if isinstance(rows, Mapping): - rows = rows.items() - self._rows = [(str(label), value) for label, value in rows] - self.redraw() + def set_rows(self, rows: Iterable[tuple[str, object | None]]) -> None: + """ + Replace all displayed rows. Unchanged rows are not re-rendered. - def add_row(self, label: str, value: object | None = None) -> None: - """Append one displayed row.""" - self._rows.append((str(label), value)) + Args: + rows(Iterable[tuple[str, object | None]]): An iterable of (label, value) pairs to display. The label is a string and the value can be any object or None. + """ + rows = [(str(label), None if value is None else str(value)) for label, value in rows] + if rows == self._rows: + return + self._rows = rows self.redraw() - def clear_rows(self) -> None: - """Remove all displayed rows and hide the label.""" - self._rows = [] - self._has_drawn_rows = False - self.clear() - self.setVisible(False) - def reset_position(self) -> None: """Reset the label to its default anchored position.""" self.setOffset(self._offset) @@ -88,24 +79,13 @@ class PlotInfoLabel(pg.LegendItem): self.redraw() def redraw(self) -> None: - """Refresh the pyqtgraph legend rows.""" - previous_pos = self.pos() - preserve_pos = self._has_drawn_rows and self.isVisible() - row_texts = [self._format_row(label, value) for label, value in self._rows] - if len(row_texts) == len(self.items): - for row_text, (_, label_item) in zip(row_texts, self.items, strict=True): - label_item.setText(row_text) - self.updateSize() - else: - self.clear() - for row_text in row_texts: - self.addItem(TextOnlyLegendSample(), row_text) - if preserve_pos: - self.setPos(previous_pos) - self._has_drawn_rows = bool(self._rows) + """Rebuild the pyqtgraph legend rows from scratch.""" + self.clear() + for label, value in self._rows: + self.addItem(TextOnlyLegendSample(), self._format_row(label, value)) @staticmethod - def _format_row(label: str, value: object | None) -> str: + def _format_row(label: str, value: str | None) -> str: if value is None: return label return f"{label}: {value}" diff --git a/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py b/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py index 7b462283..9b815d8e 100644 --- a/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py +++ b/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py @@ -318,7 +318,7 @@ class ScatterWaveform(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) - self.update_scan_info_from_source(self.scan_item, mode="live") + self.update_scan_info_from_source(self.scan_item) # First trigger to update the scan curves self.sync_signal_update.emit() @@ -704,7 +704,7 @@ class ScatterWaveform(PlotBase): if scan_index is None: self.scan_id = scan_id self.scan_item = self.client.history.get_by_scan_id(scan_id) - self.update_scan_info_from_source(self.scan_item, mode="history") + self.update_scan_info_from_source(self.scan_item) self.sync_signal_update.emit() return @@ -716,7 +716,7 @@ class ScatterWaveform(PlotBase): return self.scan_item = scan_item self.scan_id = scan_item.scan_id - self.update_scan_info_from_source(scan_item, mode="live") + self.update_scan_info_from_source(scan_item) self.sync_signal_update.emit() return @@ -727,7 +727,7 @@ class ScatterWaveform(PlotBase): self.scan_item = self.client.history[scan_index] metadata = self.scan_item.metadata self.scan_id = metadata["bec"]["scan_id"] - self.update_scan_info_from_source(self.scan_item, mode="history") + self.update_scan_info_from_source(self.scan_item) self.sync_signal_update.emit() diff --git a/bec_widgets/widgets/plots/waveform/waveform.py b/bec_widgets/widgets/plots/waveform/waveform.py index 4a3c6e5f..9e8f986c 100644 --- a/bec_widgets/widgets/plots/waveform/waveform.py +++ b/bec_widgets/widgets/plots/waveform/waveform.py @@ -1594,7 +1594,7 @@ 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.update_scan_info_from_source(self.scan_item, mode="live") + self.update_scan_info_from_source(self.scan_item) self._slice_index = None # Reset the slice index self._update_curve_visibility() self._mode = self._categorise_device_curves() @@ -2496,7 +2496,7 @@ class Waveform(PlotBase): else: self.scan_id = self.scan_item.scan_id - self.update_scan_info_from_source(self.scan_item, mode="history") + self.update_scan_info_from_source(self.scan_item) self._emit_signal_update() def _emit_signal_update(self): diff --git a/tests/unit_tests/client_mocks.py b/tests/unit_tests/client_mocks.py index 83c3dad3..a3749a99 100644 --- a/tests/unit_tests/client_mocks.py +++ b/tests/unit_tests/client_mocks.py @@ -188,9 +188,16 @@ class DummyData: return default -def create_dummy_scan_item(): +def create_dummy_scan_item( + scan_id: str = "dummy", scan_number: int | None = None, scan_name: str | None = None +): """ Helper to create a dummy scan item with both live_data and metadata/status_message info. + + Args: + scan_id (str): Scan ID used in the metadata and the status message. + scan_number (int | None): Scan number of the status message and the metadata. + scan_name (str | None): Scan name of the status message and the metadata. """ dummy_live_data = { "samx": {"samx": DummyData(val=[10, 20, 30], timestamps=[100, 200, 300])}, @@ -198,17 +205,26 @@ def create_dummy_scan_item(): "bpm4i": {"bpm4i": DummyData(val=[5, 6, 7], timestamps=[101, 201, 301])}, "async_device": {"async_device": DummyData(val=[1, 2, 3], timestamps=[11, 21, 31])}, } + # spec=ScanItem makes isinstance checks pass but hides the instance attributes that + # ScanItem.__init__ assigns, so the ones the widgets read are set explicitly. dummy_scan = MagicMock(spec=ScanItem) + dummy_scan.scan_id = scan_id + dummy_scan.scan_number = scan_number + dummy_scan.status = "open" dummy_scan.live_data = dummy_live_data dummy_scan.metadata = { "bec": { - "scan_id": "dummy", + "scan_id": scan_id, + "scan_number": scan_number, + "scan_name": scan_name, "scan_report_devices": ["samx"], "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, } } dummy_scan.status_message = messages.ScanStatusMessage( - scan_id="dummy", + scan_id=scan_id, + scan_number=scan_number, + scan_name=scan_name, status="open", info={ "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, diff --git a/tests/unit_tests/test_heatmap_widget.py b/tests/unit_tests/test_heatmap_widget.py index 2ff51fd4..871c1977 100644 --- a/tests/unit_tests/test_heatmap_widget.py +++ b/tests/unit_tests/test_heatmap_widget.py @@ -1044,20 +1044,77 @@ def test_heatmap_config_label_shows_live_or_history(heatmap_widget): scan_msg.scan_number = 5 scan_msg.scan_name = "line_scan" heatmap_widget.status_message = scan_msg - heatmap_widget._image_config.show_config_label = True + heatmap_widget.show_info_label = True heatmap_widget.redraw_config_label() - labels = [label.text for _, label in heatmap_widget.config_label.items] - assert heatmap_widget.config_label is heatmap_widget.info_label + labels = [label.text for _, label in heatmap_widget.info_label.items] assert "Scan: 5 (live)" in labels assert "Interpolation: linear" in labels heatmap_widget._history_scan_id = "scan-1" heatmap_widget.redraw_config_label() - labels = [label.text for _, label in heatmap_widget.config_label.items] + labels = [label.text for _, label in heatmap_widget.info_label.items] assert "Scan: 5 (history)" in labels +def test_heatmap_plot_show_config_label_reenables_hidden_label(heatmap_widget): + """Regression: plot(show_config_label=True) after the toolbar hid the label kept it hidden.""" + scan_msg = mock.MagicMock() + scan_msg.scan_number = 5 + scan_msg.scan_name = "grid_scan" + heatmap_widget.status_message = scan_msg + heatmap_widget.redraw_config_label() + assert heatmap_widget.info_label.isVisible() + + action = heatmap_widget.toolbar.components.get_action("plot_info_label").action + action.trigger() + assert not heatmap_widget.show_info_label + assert not heatmap_widget.info_label.isVisible() + + heatmap_widget.plot( + "samx", + "samy", + "bpm4i", + signal_x="samx", + signal_y="samy", + signal_z="bpm4i", + validate_bec=False, + show_config_label=True, + ) + heatmap_widget.redraw_config_label() + + assert heatmap_widget.show_info_label + assert heatmap_widget._image_config.show_config_label + assert heatmap_widget.info_label.isVisible() + assert action.isChecked() + + action.trigger() + assert not heatmap_widget.info_label.isVisible() + + +def test_heatmap_show_info_label_is_the_only_visibility_flag(heatmap_widget): + """Regression: writes to the inherited show_info_label property were reverted on redraw.""" + scan_msg = mock.MagicMock() + scan_msg.scan_number = 5 + scan_msg.scan_name = "grid_scan" + heatmap_widget.status_message = scan_msg + action = heatmap_widget.toolbar.components.get_action("plot_info_label").action + + # what a Designer .ui file / QUiLoader does + heatmap_widget.setProperty("show_info_label", False) + heatmap_widget.redraw_config_label() + assert not heatmap_widget.info_label.isVisible() + assert not action.isChecked() + + action.trigger() + assert heatmap_widget.info_label.isVisible() + + heatmap_widget.apply_theme("dark") + heatmap_widget.redraw_config_label() + assert heatmap_widget.show_info_label + assert heatmap_widget.info_label.isVisible() + + def test_heatmap_config_label_paints_without_error(heatmap_widget): """The config label rows must survive a real paint pass. @@ -1068,10 +1125,10 @@ def test_heatmap_config_label_paints_without_error(heatmap_widget): scan_msg.scan_number = 5 scan_msg.scan_name = "line_scan" heatmap_widget.status_message = scan_msg - heatmap_widget._image_config.show_config_label = True + heatmap_widget.show_info_label = True heatmap_widget.redraw_config_label() - samples = [sample for sample, _ in heatmap_widget.config_label.items] + samples = [sample for sample, _ in heatmap_widget.info_label.items] assert samples assert all(isinstance(sample, TextOnlyLegendSample) for sample in samples) diff --git a/tests/unit_tests/test_multi_waveform_next_gen.py b/tests/unit_tests/test_multi_waveform_next_gen.py index 825570e0..babbc179 100644 --- a/tests/unit_tests/test_multi_waveform_next_gen.py +++ b/tests/unit_tests/test_multi_waveform_next_gen.py @@ -106,6 +106,30 @@ def test_multiwaveform_scan_update_populates_info_label(qtbot, mocked_client, mo ] +def test_multiwaveform_clears_info_label_when_scan_id_is_unknown(qtbot, mocked_client, monkeypatch): + """Regression: monitor data without a known scan_id kept the previous scan's rows.""" + mw = create_widget(qtbot, MultiWaveform, client=mocked_client) + mw.show_info_label = True + scan_item = ScanItem(queue_id="queue-1", scan_number=1, scan_id="scan_1", status="open") + scan_item.status_message = messages.ScanStatusMessage( + scan_id="scan_1", scan_number=1, scan_name="line_scan", status="open", info={} + ) + storage = {"scan_1": scan_item} + monkeypatch.setattr( + mw.queue.scan_storage, "find_scan_by_ID", lambda scan_id: storage.get(scan_id) + ) + + mw.on_monitor_1d_update({"data": np.array([1, 2, 3])}, metadata={"scan_id": "scan_1"}) + assert mw.info_label.isVisible() + + # e.g. the device was triggered outside a scan: no scan_id in the metadata + mw.on_monitor_1d_update({"data": np.array([4, 5, 6])}, metadata={}) + + assert mw.scan_id is None + assert mw.info_label.rows == [] + assert not mw.info_label.isVisible() + + def test_multiwaveform_curve_limit_flush(qtbot, mocked_client): """Check that limiting the number of curves with flush removes older ones.""" mw = create_widget(qtbot, MultiWaveform, client=mocked_client) diff --git a/tests/unit_tests/test_plot_base_next_gen.py b/tests/unit_tests/test_plot_base_next_gen.py index 6db2c9f0..e105b1da 100644 --- a/tests/unit_tests/test_plot_base_next_gen.py +++ b/tests/unit_tests/test_plot_base_next_gen.py @@ -154,7 +154,7 @@ def test_update_scan_info_from_scan_item_status_message(qtbot, mocked_client): ) scan_item.status_message = status_message - pb.update_scan_info_from_source(scan_item, mode="live") + pb.update_scan_info_from_source(scan_item) labels = [label.text for _, label in pb.info_label.items] assert labels == ["Scan: 7 (live)", "Scan Name: line_scan"] @@ -169,28 +169,55 @@ def test_update_scan_info_from_history_scan_item_uses_scan_number( ) scan_item = ScanDataContainer(file_path=history_message.file_path, msg=history_message) - pb.update_scan_info_from_source(scan_item, mode="history") + pb.update_scan_info_from_source(scan_item) labels = [label.text for _, label in pb.info_label.items] assert labels == ["Scan: 42 (history)", "Scan Name: line_scan"] -def test_history_scan_info_does_not_fall_back_to_scan_id(qtbot, mocked_client): +@pytest.mark.parametrize( + "scan_source", + [None, ScanItem(queue_id="queue-1", scan_number=7, scan_id="scan-7", status="open")], + ids=["no_scan_item", "scan_item_without_status_message"], +) +def test_update_scan_info_clears_rows_when_source_has_no_scan_info( + qtbot, mocked_client, scan_source +): + """Regression: a missing scan item left the previous scan's rows on the label.""" pb = create_widget(qtbot, PlotBase, client=mocked_client) + pb.show_info_label = True + pb.set_scan_info(scan_number=6, scan_name="line_scan", mode="live") + assert pb.info_label.isVisible() - pb.set_scan_info(scan_id="history-id", mode="history") + pb.update_scan_info_from_source(scan_source) assert pb.info_label.rows == [] + assert not pb.info_label.isVisible() -def test_update_scan_info_noops_when_source_has_no_scan_info(qtbot, mocked_client): +def test_custom_info_rows_survive_scan_info_updates(qtbot, mocked_client): + """Regression: set_info_label_rows rows were replaced by the next set_scan_info call.""" pb = create_widget(qtbot, PlotBase, client=mocked_client) - scan_item = ScanItem(queue_id="queue-1", scan_number=7, scan_id="scan-7", status="open") - pb.set_scan_info(scan_id="scan-7", mode="live") - pb.update_scan_info_from_source(scan_item, mode="live") + pb.set_info_label_rows({"Sample": "Si"}) + pb.set_scan_info(scan_number=5, scan_name="line_scan", mode="live") + assert [label.text for _, label in pb.info_label.items] == [ + "Scan: 5 (live)", + "Scan Name: line_scan", + "Sample: Si", + ] - assert [label.text for _, label in pb.info_label.items] == ["Scan ID: scan-7 (live)"] + pb.set_scan_info(scan_number=6, scan_name="grid_scan", mode="live") + pb.set_info_label_rows({"Sample": "Ge"}) + assert [label.text for _, label in pb.info_label.items] == [ + "Scan: 6 (live)", + "Scan Name: grid_scan", + "Sample: Ge", + ] + + pb.clear_info_label() + assert pb.info_label.rows == [] + assert not pb.info_label.isVisible() def test_info_label_paints_without_error(qtbot, mocked_client): @@ -219,6 +246,41 @@ def test_info_label_redraw_preserves_user_position(qtbot, mocked_client): assert pb.info_label.pos() == position +def test_info_label_rows_accept_values_without_boolean_equality(qtbot, mocked_client): + """Regression: comparing raw row values raised for numpy arrays on the second submit.""" + pb = create_widget(qtbot, PlotBase, client=mocked_client) + + pb.set_info_label_rows({"Positions": np.array([1, 2]), "Points": 3}) + pb.set_info_label_rows({"Positions": np.array([1, 2]), "Points": 3}) + + assert pb.info_label.rows == [("Positions", "[1 2]"), ("Points", "3")] + assert [label.text for _, label in pb.info_label.items] == ["Positions: [1 2]", "Points: 3"] + + +def test_info_label_box_shrinks_when_row_text_gets_shorter(qtbot, mocked_client): + """Regression: the same-row-count redraw used to keep the widest frame ever drawn.""" + pb = create_widget(qtbot, PlotBase, client=mocked_client) + pb.show_info_label = True + + pb.set_info_label_rows({"Scan": "1 (live)", "Scan Name": "line_scan"}) + short_width = pb.info_label.geometry().width() + pb.set_info_label_rows({"Scan": "2 (live)", "Scan Name": "fermat_scan_with_a_much_longer_name"}) + assert pb.info_label.geometry().width() > short_width + 50 + pb.set_info_label_rows({"Scan": "3 (live)", "Scan Name": "line_scan"}) + + assert pb.info_label.geometry().width() == pytest.approx(short_width, abs=1) + + +def test_info_label_set_rows_skips_redraw_for_unchanged_rows(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + pb.set_info_label_rows({"Scan": "1 (live)"}) + items_before = [label for _, label in pb.info_label.items] + + pb.set_info_label_rows({"Scan": "1 (live)"}) + + assert [label for _, label in pb.info_label.items] == items_before + + def test_info_label_toggle_resets_position(qtbot, mocked_client): pb = create_widget(qtbot, PlotBase, client=mocked_client) pb.set_info_label_rows({"Scan": "5 (live)"}) @@ -240,10 +302,7 @@ def test_info_label_toggle_resets_position(qtbot, mocked_client): def test_info_label_is_available_on_plot_base_subclasses(qtbot, mocked_client, widget_cls): widget = create_widget(qtbot, widget_cls, client=mocked_client) - if hasattr(widget, "show_config_label"): - widget.show_config_label = False - else: - widget.show_info_label = False + widget.show_info_label = False widget.set_info_label_rows([("Widget", widget_cls.__name__), ("Mode", "custom")]) assert [label.text for _, label in widget.info_label.items] == [ @@ -252,8 +311,13 @@ def test_info_label_is_available_on_plot_base_subclasses(qtbot, mocked_client, w ] assert not widget.info_label.isVisible() - if isinstance(widget, MotorMap): + if isinstance(widget, (Image, MotorMap)): + # these widgets never publish scan info and remove the toolbar button + assert not widget.toolbar.components.exists("plot_info_label") assert "plot_info_label" not in widget.toolbar.get_bundle("axis_popup").bundle_actions + # the property itself keeps working without the button + widget.show_info_label = True + assert widget.info_label.isVisible() return action = widget.toolbar.components.get_action("plot_info_label").action diff --git a/tests/unit_tests/test_scatter_waveform.py b/tests/unit_tests/test_scatter_waveform.py index 694f416d..7f50921e 100644 --- a/tests/unit_tests/test_scatter_waveform.py +++ b/tests/unit_tests/test_scatter_waveform.py @@ -1,7 +1,6 @@ -from unittest.mock import MagicMock, patch +from unittest.mock import patch import numpy as np -from bec_lib import messages from bec_lib.scan_data_container import ScanDataContainer from bec_widgets.widgets.plots.scatter_waveform.scatter_curve import ( @@ -12,7 +11,12 @@ from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterW from bec_widgets.widgets.plots.scatter_waveform.settings.scatter_curve_setting import ( ScatterCurveSettings, ) -from tests.unit_tests.client_mocks import DummyData, create_dummy_scan_item, mocked_client +from tests.unit_tests.client_mocks import ( + DummyData, + create_dummy_scan_item, + inject_scan_history, + mocked_client, +) from .conftest import create_widget @@ -51,27 +55,24 @@ def test_scatter_waveform_color_map(qtbot, mocked_client): assert swf.color_map == "plasma" -def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeypatch): +def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, scan_history_factory): + """History scans are read from the HDF5-backed ScanDataContainer, not from live data.""" swf = create_widget(qtbot, ScatterWaveform, client=mocked_client) - - dummy_scan = create_dummy_scan_item() - history_scan = ScanDataContainer() - history_scan.live_data = dummy_scan.live_data - history_scan.metadata = { - "bec": {"scan_id": "dummy", "scan_number": 24, "scan_name": "line_scan"} - } - mocked_client.history = MagicMock() - mocked_client.history.get_by_scan_id.return_value = history_scan - mocked_client.history.__getitem__.return_value = history_scan + inject_scan_history(swf, scan_history_factory, ("dummy", 24)) swf.plot("samx", "samy", "bpm4i", label="test_curve") swf.update_with_scan_history(scan_id="dummy") - qtbot.waitUntil(lambda: swf.scan_item == history_scan, timeout=500) - qtbot.wait(200) + qtbot.waitUntil(lambda: isinstance(swf.scan_item, ScanDataContainer), timeout=500) + + def _has_points(): + x_data = swf.main_curve.getData()[0] + return x_data is not None and len(x_data) > 0 + + qtbot.waitUntil(_has_points, timeout=2000) x_data, y_data = swf.main_curve.getData() - np.testing.assert_array_equal(x_data, [10, 20, 30]) - np.testing.assert_array_equal(y_data, [5, 10, 15]) + assert len(x_data) == len(y_data) > 0 + assert swf.scan_id == "dummy" assert [label.text for _, label in swf.info_label.items] == [ "Scan: 24 (history)", "Scan Name: line_scan", @@ -81,17 +82,7 @@ def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeyp def test_scatter_waveform_live_update(qtbot, mocked_client, monkeypatch): swf = create_widget(qtbot, ScatterWaveform, client=mocked_client) - dummy_scan = create_dummy_scan_item() - dummy_scan.status_message = messages.ScanStatusMessage( - scan_id="dummy", - scan_number=13, - scan_name="line_scan", - status="open", - info={ - "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, - "scan_report_devices": ["samx"], - }, - ) + dummy_scan = create_dummy_scan_item(scan_number=13, scan_name="line_scan") monkeypatch.setattr(swf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan) swf.plot("samx", "samy", "bpm4i", label="live_curve") diff --git a/tests/unit_tests/test_waveform.py b/tests/unit_tests/test_waveform.py index 756346a3..4f08d98c 100644 --- a/tests/unit_tests/test_waveform.py +++ b/tests/unit_tests/test_waveform.py @@ -9,6 +9,7 @@ import numpy as np import pyqtgraph as pg import pytest from bec_lib import messages +from bec_lib.scan_items import ScanItem from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem from qtpy.QtCore import QTimer from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox @@ -458,18 +459,7 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls): wf.plot(arg1="async_device") # We mock out the scan_item, pretending we found a new scan. - dummy_scan = create_dummy_scan_item() - dummy_scan.metadata["bec"]["scan_id"] = "1234" - dummy_scan.status_message = messages.ScanStatusMessage( - scan_id="1234", - scan_number=12, - scan_name="line_scan", - status="open", - info={ - "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, - "scan_report_devices": ["samx"], - }, - ) + dummy_scan = create_dummy_scan_item(scan_id="1234", scan_number=12, scan_name="line_scan") 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 @@ -1665,6 +1655,52 @@ def test_update_with_scan_history_by_index(qtbot, mocked_client, scan_history_fa ] +def test_on_scan_status_clears_info_label_when_scan_item_is_missing( + qtbot, mocked_client, monkeypatch +): + """Regression: a scan whose ScanItem is not in scan storage kept the previous scan's rows.""" + wf = create_widget(qtbot, Waveform, client=mocked_client) + wf.show_info_label = True + scan_item = ScanItem(queue_id="queue-1", scan_number=1, scan_id="scan-1", status="open") + scan_item.status_message = messages.ScanStatusMessage( + scan_id="scan-1", scan_number=1, scan_name="line_scan", status="open", info={} + ) + storage = {"scan-1": scan_item} + monkeypatch.setattr( + wf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: storage.get(scan_id) + ) + + wf.on_scan_status({"scan_id": "scan-1", "status": "open"}, {}) + assert [label.text for _, label in wf.info_label.items] == [ + "Scan: 1 (live)", + "Scan Name: line_scan", + ] + + wf.on_scan_status({"scan_id": "scan-2", "status": "closed"}, {}) + + assert wf.scan_id == "scan-2" + assert wf.info_label.rows == [] + assert not wf.info_label.isVisible() + + +def test_update_with_scan_history_labels_running_scan_as_live(qtbot, mocked_client): + """Regression: a Waveform created during a running scan labelled the live scan '(history)'.""" + wf = create_widget(qtbot, Waveform, client=mocked_client) + scan_item = ScanItem(queue_id="queue-1", scan_number=9, scan_id="scan-9", status="open") + scan_item.status_message = messages.ScanStatusMessage( + scan_id="scan-9", scan_number=9, scan_name="line_scan", status="open", info={} + ) + mocked_client.queue.scan_storage.current_scan = scan_item + + wf.update_with_scan_history(-1) + + assert wf.scan_item is scan_item + assert [label.text for _, label in wf.info_label.items] == [ + "Scan: 9 (live)", + "Scan Name: line_scan", + ] + + @pytest.mark.parametrize("mode", ["auto", "timestamp", "index", "samx"]) def test_history_curve_x_modes_pre_plot(qtbot, mocked_client, scan_history_factory, mode): """