diff --git a/bec_widgets/widgets/plots/heatmap/heatmap.py b/bec_widgets/widgets/plots/heatmap/heatmap.py index 70b8a422..7af9cc77 100644 --- a/bec_widgets/widgets/plots/heatmap/heatmap.py +++ b/bec_widgets/widgets/plots/heatmap/heatmap.py @@ -8,10 +8,10 @@ import numpy as np import pyqtgraph as pg from bec_lib import bec_logger, messages from bec_lib.endpoints import MessageEndpoints -from bec_lib.utils.import_utils import lazy_import, lazy_import_from +from bec_lib.utils.import_utils import lazy_import_from from bec_qthemes import material_icon from pydantic import BaseModel, Field, field_validator -from qtpy.QtCore import QObject, QRectF, Qt, QThread, QTimer, Signal +from qtpy.QtCore import QObject, Qt, QThread, QTimer, Signal from qtpy.QtGui import QTransform from qtpy.QtWidgets import QDialog, QPushButton, QVBoxLayout from toolz import partition @@ -93,30 +93,6 @@ class HeatmapConfig(ConnectionConfig): _validate_color_palette = field_validator("color_map")(Colors.validate_color_map) -class _TextOnlyLegendSample(pg.graphicsItems.LegendItem.ItemSample): - """Zero-size legend sample for text-only rows in the config label. - - The stock ItemSample expects a plottable item with an ``opts`` dict; since - PySide 6.10 an exception raised inside its paint() override propagates out - of the C++ paint loop and crashes the application, so the config label rows - must not carry a real sample item. - """ - - def __init__(self): - super().__init__(item=None) - self.setFixedWidth(0) - self.setFixedHeight(0) - - def boundingRect(self): - return QRectF(0, 0, 0, 0) - - def paint(self, p, *args): - pass - - def mouseClickEvent(self, event): - event.ignore() - - @dataclass class _InterpolationRequest: """Immutable payload describing an interpolation request for the worker thread. @@ -300,12 +276,8 @@ 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 - bg_color = pg.mkColor((240, 240, 240, 150)) - self.config_label = pg.LegendItem( - labelTextColor=(0, 0, 0), offset=(-30, 1), brush=pg.mkBrush(bg_color), horSpacing=0 - ) - self.config_label.setParentItem(self.plot_item.vb) - self.config_label.setVisible(False) + 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()) self.bec_dispatcher.connect_slot(self.on_scan_progress, MessageEndpoints.scan_progress()) @@ -325,7 +297,6 @@ class Heatmap(ImageBase): "image_colorbar", "image_processing", "axis_popup", - "interpolation_info", ] ) @@ -344,15 +315,7 @@ class Heatmap(ImageBase): Apply the current theme to the heatmap widget. """ super().apply_theme(theme) - if theme == "dark": - brush = pg.mkBrush(pg.mkColor(50, 50, 50, 150)) - color = pg.mkColor(255, 255, 255) - else: - brush = pg.mkBrush(pg.mkColor(240, 240, 240, 150)) - color = pg.mkColor(0, 0, 0) - if hasattr(self, "config_label"): - self.config_label.setBrush(brush) - self.config_label.setLabelTextColor(color) + if hasattr(self, "info_label"): self.redraw_config_label() @SafeSlot(popup_error=True) @@ -588,19 +551,6 @@ class Heatmap(ImageBase): if name not in ["image_processing_fft", "image_processing_log"]: action().action.setVisible(False) - self.toolbar.add_action( - "interpolation_info", - MaterialIconAction( - icon_name="info", tooltip="Show Interpolation Info", checkable=True, parent=self - ), - ) - self.toolbar.components.get_action("interpolation_info").action.triggered.connect( - self.toggle_interpolation_info - ) - self.toolbar.components.get_action("interpolation_info").action.setChecked( - self._image_config.show_config_label - ) - def show_heatmap_settings(self): """ Show the heatmap settings dialog. @@ -687,15 +637,27 @@ 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 interpolation info label. + Toggle the visibility of the heatmap information label. """ - self._image_config.show_config_label = not self._image_config.show_config_label - self.toolbar.components.get_action("interpolation_info").action.setChecked( - self._image_config.show_config_label - ) - self.redraw_config_label() + self.toggle_info_label() def _heatmap_dialog_closed(self): """ @@ -989,27 +951,29 @@ class Heatmap(ImageBase): self._latest_interpolation_version = -1 def redraw_config_label(self): - scan_msg = self.status_message + scan_msg = getattr(self, "status_message", None) if scan_msg is None: return if not self._image_config.show_config_label: - self.config_label.setVisible(False) + self.show_info_label = False return - self.config_label.setOffset((-30, 1)) - self.config_label.setVisible(True) - self.config_label.clear() # 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" - self.config_label.addItem(_TextOnlyLegendSample(), f"Scan: {scan_msg.scan_number} ({mode})") - self.config_label.addItem(_TextOnlyLegendSample(), f"Scan Name: {scan_msg.scan_name}") + extra_rows = [] if scan_msg.scan_name != "grid_scan" or self._image_config.enforce_interpolation: - self.config_label.addItem( - _TextOnlyLegendSample(), f"Interpolation: {self._image_config.interpolation}" - ) - self.config_label.addItem( - _TextOnlyLegendSample(), f"Oversampling: {self._image_config.oversampling_factor}x" + extra_rows.extend( + [ + ("Interpolation", self._image_config.interpolation), + ("Oversampling", f"{self._image_config.oversampling_factor}x"), + ] ) + self.set_scan_info( + scan_number=scan_msg.scan_number, + scan_name=scan_msg.scan_name, + mode=mode, + extra_rows=extra_rows, + ) def get_image_data( self, diff --git a/bec_widgets/widgets/plots/motor_map/motor_map.py b/bec_widgets/widgets/plots/motor_map/motor_map.py index 32c3d151..0121054e 100644 --- a/bec_widgets/widgets/plots/motor_map/motor_map.py +++ b/bec_widgets/widgets/plots/motor_map/motor_map.py @@ -167,6 +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") 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 291a15cc..3875173d 100644 --- a/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py +++ b/bec_widgets/widgets/plots/multi_waveform/multi_waveform.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections import deque -from typing import TYPE_CHECKING, cast +from typing import cast import pyqtgraph as pg from bec_lib.endpoints import MessageEndpoints @@ -361,7 +361,8 @@ class MultiWaveform(PlotBase): current_scan_id = metadata.get("scan_id", None) if current_scan_id != self.scan_id: - self.scan_id = current_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.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 b2c494b1..f35c8aae 100644 --- a/bec_widgets/widgets/plots/plot_base.py +++ b/bec_widgets/widgets/plots/plot_base.py @@ -1,10 +1,13 @@ from __future__ import annotations from enum import Enum +from typing import TypeAlias import numpy as np import pyqtgraph as pg -from bec_lib import bec_logger +from bec_lib import bec_logger, messages +from bec_lib.scan_data_container import ScanDataContainer +from bec_lib.scan_items import ScanItem from qtpy.QtCore import QPoint, QPointF, Qt, Signal from qtpy.QtWidgets import QHBoxLayout, QLabel, QMainWindow, QVBoxLayout, QWidget @@ -17,10 +20,12 @@ from bec_widgets.utils.fps_counter import FPSCounter from bec_widgets.utils.plot_indicator_items import BECArrowItem, BECTickItem from bec_widgets.utils.round_frame import RoundedFrame from bec_widgets.utils.side_panel import SidePanel +from bec_widgets.utils.toolbars.actions import MaterialIconAction from bec_widgets.utils.toolbars.performance import PerformanceConnection, performance_bundle from bec_widgets.utils.toolbars.toolbar import ModularToolBar from bec_widgets.utils.widget_state_manager import WidgetStateManager from bec_widgets.widgets.containers.layout_manager.layout_manager import LayoutManagerWidget +from bec_widgets.widgets.plots.plot_info_label import PlotInfoLabel from bec_widgets.widgets.plots.setting_menus.axis_settings import AxisSettings from bec_widgets.widgets.plots.toolbar_components.axis_settings_popup import ( AxisSettingsPopupConnection, @@ -38,6 +43,8 @@ from bec_widgets.widgets.plots.toolbar_components.roi import RoiConnection, roi_ logger = bec_logger.logger +ScanInfoSource: TypeAlias = ScanItem | ScanDataContainer | None + class BECViewBox(pg.ViewBox): sigPaint = Signal() @@ -162,6 +169,8 @@ class PlotBase(BECWidget, QWidget): # PlotItem Addons self.plot_item.addLegend() + self.info_label = PlotInfoLabel() + self.info_label.setParentItem(self.plot_item.vb) 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. @@ -183,6 +192,7 @@ class PlotBase(BECWidget, QWidget): # Visibility States self._toolbar_visible = True self._enable_fps_monitor = False + self._show_info_label = False self._outer_axes_visible = self.plot_item.getAxis("top").isVisible() self._inner_axes_visible = self.plot_item.getAxis("bottom").isVisible() @@ -195,6 +205,8 @@ class PlotBase(BECWidget, QWidget): def apply_theme(self, theme: str): self.round_plot_widget.apply_theme(theme) + if hasattr(self, "info_label"): + self.info_label.set_theme(theme) def _init_ui(self): self.layout.addWidget(self.layout_manager) @@ -219,6 +231,16 @@ class PlotBase(BECWidget, QWidget): self.toolbar.add_bundle(mouse_interaction_bundle(self.toolbar.components)) self.toolbar.add_bundle(roi_bundle(self.toolbar.components)) self.toolbar.add_bundle(axis_popup_bundle(self.toolbar.components)) + self.toolbar.components.add_safe( + "plot_info_label", + MaterialIconAction( + icon_name="info", tooltip="Show Plot Info", checkable=True, parent=self + ), + ) + self.toolbar.get_bundle("axis_popup").add_action("plot_info_label") + self.toolbar.components.get_action("plot_info_label").action.triggered.connect( + self.toggle_info_label + ) self.toolbar.connect_bundle( "plot_base", PlotExportConnection(self.toolbar.components, self) @@ -236,6 +258,7 @@ class PlotBase(BECWidget, QWidget): # hide some options by default self.toolbar.toggle_action_visibility("fps_monitor", False) + self._sync_info_label_action() # Get default viewbox state self.toolbar.show_bundles( @@ -261,6 +284,137 @@ class PlotBase(BECWidget, QWidget): """In the case that the legend is not visible, reset it to be visible to top left corner""" self.plot_item.legend.autoAnchor(50) + ################################################################################ + # Plot Info Label + ################################################################################ + + @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.""" + return self._show_info_label + + @show_info_label.setter + def show_info_label(self, value: bool) -> None: + """Show or hide the plot info label.""" + visible = bool(value) + if visible and not self._show_info_label: + self.info_label.reset_position() + self._show_info_label = visible + self._sync_info_label_visibility() + self._sync_info_label_action() + + @SafeSlot() + def toggle_info_label(self) -> None: + """Toggle the plot info label visibility.""" + 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]] + ) -> 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() + + def clear_info_label(self) -> None: + """Clear all rows from the plot info label.""" + self.info_label.clear_rows() + 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, + ) -> None: + """Submit common scan metadata rows to the plot 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)) + 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) + + 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_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 + + 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 _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) + ################################################################################ # Toggle UI Elements ################################################################################ diff --git a/bec_widgets/widgets/plots/plot_info_label.py b/bec_widgets/widgets/plots/plot_info_label.py new file mode 100644 index 00000000..0e859bd1 --- /dev/null +++ b/bec_widgets/widgets/plots/plot_info_label.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +from collections.abc import Iterable, Mapping + +import pyqtgraph as pg +from qtpy.QtCore import QRectF + + +class TextOnlyLegendSample(pg.graphicsItems.LegendItem.ItemSample): + """Zero-size legend sample for text-only rows. + + The stock ItemSample expects a plottable item with an ``opts`` dict; since + PySide 6.10 an exception raised inside its paint() override propagates out + of the C++ paint loop and crashes the application. + """ + + def __init__(self): + super().__init__(item=None) + self.setFixedWidth(0) + self.setFixedHeight(0) + + def boundingRect(self): + return QRectF(0, 0, 0, 0) + + def paint(self, p, *args): + pass + + def mouseClickEvent(self, event): + event.ignore() + + +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. + """ + + 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._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.""" + 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 add_row(self, label: str, value: object | None = None) -> None: + """Append one displayed row.""" + self._rows.append((str(label), value)) + 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) + + def set_theme(self, theme: str) -> None: + """Update the label colors for the active application theme.""" + if theme == "dark": + brush = pg.mkBrush(pg.mkColor(50, 50, 50, 150)) + color = pg.mkColor(255, 255, 255) + else: + brush = pg.mkBrush(pg.mkColor(240, 240, 240, 150)) + color = pg.mkColor(0, 0, 0) + self.setBrush(brush) + self.setLabelTextColor(color) + 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) + + @staticmethod + def _format_row(label: str, value: object | 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 54640730..7b462283 100644 --- a/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py +++ b/bec_widgets/widgets/plots/scatter_waveform/scatter_waveform.py @@ -318,6 +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") # First trigger to update the scan curves self.sync_signal_update.emit() @@ -703,6 +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.sync_signal_update.emit() return @@ -714,6 +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.sync_signal_update.emit() return @@ -724,6 +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.sync_signal_update.emit() diff --git a/bec_widgets/widgets/plots/waveform/waveform.py b/bec_widgets/widgets/plots/waveform/waveform.py index 123c5295..4a3c6e5f 100644 --- a/bec_widgets/widgets/plots/waveform/waveform.py +++ b/bec_widgets/widgets/plots/waveform/waveform.py @@ -1594,6 +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._slice_index = None # Reset the slice index self._update_curve_visibility() self._mode = self._categorise_device_curves() @@ -2495,6 +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._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 dc8f6000..83c3dad3 100644 --- a/tests/unit_tests/client_mocks.py +++ b/tests/unit_tests/client_mocks.py @@ -7,6 +7,7 @@ import pytest from bec_lib.bec_service import messages from bec_lib.endpoints import MessageEndpoints from bec_lib.scan_history import ScanHistory +from bec_lib.scan_items import ScanItem from bec_widgets.tests.utils import FakePositioner, Positioner @@ -197,7 +198,7 @@ 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])}, } - dummy_scan = MagicMock() + dummy_scan = MagicMock(spec=ScanItem) dummy_scan.live_data = dummy_live_data dummy_scan.metadata = { "bec": { @@ -206,10 +207,14 @@ def create_dummy_scan_item(): "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, } } - dummy_scan.status_message.info = { - "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, - "scan_report_devices": ["samx"], - } + dummy_scan.status_message = messages.ScanStatusMessage( + scan_id="dummy", + status="open", + info={ + "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, + "scan_report_devices": ["samx"], + }, + ) return dummy_scan diff --git a/tests/unit_tests/test_heatmap_widget.py b/tests/unit_tests/test_heatmap_widget.py index 4d0092fb..2ff51fd4 100644 --- a/tests/unit_tests/test_heatmap_widget.py +++ b/tests/unit_tests/test_heatmap_widget.py @@ -13,8 +13,8 @@ from bec_widgets.widgets.plots.heatmap.heatmap import ( HeatmapDeviceSignal, _InterpolationRequest, _StepInterpolationWorker, - _TextOnlyLegendSample, ) +from bec_widgets.widgets.plots.plot_info_label import TextOnlyLegendSample # pytest: disable=unused-import from tests.unit_tests.client_mocks import mocked_client @@ -1048,7 +1048,9 @@ def test_heatmap_config_label_shows_live_or_history(heatmap_widget): 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 assert "Scan: 5 (live)" in labels + assert "Interpolation: linear" in labels heatmap_widget._history_scan_id = "scan-1" heatmap_widget.redraw_config_label() @@ -1071,7 +1073,7 @@ def test_heatmap_config_label_paints_without_error(heatmap_widget): samples = [sample for sample, _ in heatmap_widget.config_label.items] assert samples - assert all(isinstance(sample, _TextOnlyLegendSample) for sample in samples) + assert all(isinstance(sample, TextOnlyLegendSample) for sample in samples) pixmap = heatmap_widget.grab() assert not pixmap.isNull() diff --git a/tests/unit_tests/test_image_view_next_gen.py b/tests/unit_tests/test_image_view_next_gen.py index 5e82c47d..269480b9 100644 --- a/tests/unit_tests/test_image_view_next_gen.py +++ b/tests/unit_tests/test_image_view_next_gen.py @@ -674,6 +674,7 @@ def test_handle_scan_change_clears_buffers_and_resets_crosshair(qtbot, mocked_cl assert view.old_scan_id == "scan_1" assert view.scan_id == "scan_2" + assert view.info_label.rows == [] assert clear_called == [True] assert view.main_image.buffer == [] assert view.main_image.max_len == 0 diff --git a/tests/unit_tests/test_multi_waveform_next_gen.py b/tests/unit_tests/test_multi_waveform_next_gen.py index 0b7ae96f..825570e0 100644 --- a/tests/unit_tests/test_multi_waveform_next_gen.py +++ b/tests/unit_tests/test_multi_waveform_next_gen.py @@ -1,4 +1,6 @@ import numpy as np +from bec_lib import messages +from bec_lib.scan_items import ScanItem from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform from tests.unit_tests.client_mocks import mocked_client @@ -88,6 +90,22 @@ def test_multiwaveform_curve_limit_no_flush(qtbot, mocked_client): assert len(visible_curves) == 3 +def test_multiwaveform_scan_update_populates_info_label(qtbot, mocked_client, monkeypatch): + mw = create_widget(qtbot, MultiWaveform, client=mocked_client) + 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={} + ) + monkeypatch.setattr(mw.queue.scan_storage, "find_scan_by_ID", lambda scan_id: scan_item) + + mw.on_monitor_1d_update({"data": np.array([1, 2, 3])}, metadata={"scan_id": "scan_1"}) + + assert [label.text for _, label in mw.info_label.items] == [ + "Scan: 1 (live)", + "Scan Name: line_scan", + ] + + 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 cadd5083..6db2c9f0 100644 --- a/tests/unit_tests/test_plot_base_next_gen.py +++ b/tests/unit_tests/test_plot_base_next_gen.py @@ -1,6 +1,17 @@ import numpy as np +import pytest +from bec_lib import messages +from bec_lib.scan_data_container import ScanDataContainer +from bec_lib.scan_items import ScanItem +from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap +from bec_widgets.widgets.plots.image.image import Image +from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap +from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform from bec_widgets.widgets.plots.plot_base import PlotBase, UIMode +from bec_widgets.widgets.plots.plot_info_label import TextOnlyLegendSample +from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform +from bec_widgets.widgets.plots.waveform.waveform import Waveform from .client_mocks import mocked_client from .conftest import create_widget @@ -73,6 +84,185 @@ def test_set_y_label_emits_signal(qtbot, mocked_client): assert pb.plot_item.getAxis("left").labelText == "Current [A]" +def test_info_label_rows_are_submitted_without_forcing_visibility(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + action = pb.toolbar.components.get_action("plot_info_label").action + + assert action.isVisible() + assert action.isEnabled() + assert not action.isChecked() + + pb.set_info_label_rows([("Scan", "5 (live)"), ("Mode", "alignment")]) + + labels = [label.text for _, label in pb.info_label.items] + assert labels == ["Scan: 5 (live)", "Mode: alignment"] + assert pb.info_label.rows == [("Scan", "5 (live)"), ("Mode", "alignment")] + assert not pb.info_label.isVisible() + assert action.isVisible() + assert action.isEnabled() + assert not action.isChecked() + + +def test_info_label_can_be_enabled_before_rows_are_available(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + action = pb.toolbar.components.get_action("plot_info_label").action + + action.trigger() + + assert pb.show_info_label + assert action.isChecked() + assert not pb.info_label.isVisible() + + pb.set_info_label_rows({"Scan": "5 (live)"}) + + assert pb.info_label.isVisible() + assert action.isChecked() + + +def test_info_label_visibility_property_and_toolbar_action(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + pb.set_info_label_rows({"Scan": "5 (live)"}) + + with qtbot.waitSignal(pb.property_changed, timeout=500) as signal: + pb.show_info_label = True + assert signal.args == ["show_info_label", True] + assert pb.info_label.isVisible() + assert pb.toolbar.components.get_action("plot_info_label").action.isChecked() + + pb.toolbar.components.get_action("plot_info_label").action.trigger() + assert not pb.show_info_label + assert not pb.info_label.isVisible() + + +def test_set_scan_info_formats_common_scan_rows(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + + pb.show_info_label = True + pb.set_scan_info( + scan_number=7, scan_name="grid_scan", mode="history", extra_rows=[("Detector", "eiger")] + ) + + labels = [label.text for _, label in pb.info_label.items] + assert labels == ["Scan: 7 (history)", "Scan Name: grid_scan", "Detector: eiger"] + + +def test_update_scan_info_from_scan_item_status_message(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + scan_item = ScanItem(queue_id="queue-1", scan_number=7, scan_id="scan-7", status="open") + status_message = messages.ScanStatusMessage( + scan_id="scan-7", scan_number=7, scan_name="line_scan", status="open", info={} + ) + scan_item.status_message = status_message + + pb.update_scan_info_from_source(scan_item, mode="live") + + labels = [label.text for _, label in pb.info_label.items] + assert labels == ["Scan: 7 (live)", "Scan Name: line_scan"] + + +def test_update_scan_info_from_history_scan_item_uses_scan_number( + qtbot, mocked_client, scan_history_factory +): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + history_message = scan_history_factory( + scan_id="history-id", scan_number=42, scan_name="line_scan" + ) + scan_item = ScanDataContainer(file_path=history_message.file_path, msg=history_message) + + pb.update_scan_info_from_source(scan_item, mode="history") + + 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): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + + pb.set_scan_info(scan_id="history-id", mode="history") + + assert pb.info_label.rows == [] + + +def test_update_scan_info_noops_when_source_has_no_scan_info(qtbot, mocked_client): + 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") + + assert [label.text for _, label in pb.info_label.items] == ["Scan ID: scan-7 (live)"] + + +def test_info_label_paints_without_error(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + + pb.show_info_label = True + pb.set_info_label_rows({"Scan": "5 (live)"}) + + samples = [sample for sample, _ in pb.info_label.items] + assert samples + assert all(isinstance(sample, TextOnlyLegendSample) for sample in samples) + + pixmap = pb.grab() + assert not pixmap.isNull() + + +def test_info_label_redraw_preserves_user_position(qtbot, mocked_client): + pb = create_widget(qtbot, PlotBase, client=mocked_client) + pb.show_info_label = True + pb.set_info_label_rows({"Scan": "5 (live)"}) + + pb.info_label.autoAnchor((42, 24), relative=False) + position = pb.info_label.pos() + pb.set_info_label_rows({"Scan": "6 (live)"}) + + assert pb.info_label.pos() == position + + +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)"}) + pb.show_info_label = True + default_position = pb.info_label.pos() + + pb.info_label.autoAnchor((42, 24), relative=False) + assert pb.info_label.pos() != default_position + + pb.show_info_label = False + pb.show_info_label = True + + assert pb.info_label.pos() == default_position + + +@pytest.mark.parametrize( + "widget_cls", [PlotBase, Waveform, ScatterWaveform, MultiWaveform, MotorMap, Image, Heatmap] +) +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.set_info_label_rows([("Widget", widget_cls.__name__), ("Mode", "custom")]) + assert [label.text for _, label in widget.info_label.items] == [ + f"Widget: {widget_cls.__name__}", + "Mode: custom", + ] + assert not widget.info_label.isVisible() + + if isinstance(widget, MotorMap): + assert "plot_info_label" not in widget.toolbar.get_bundle("axis_popup").bundle_actions + return + + action = widget.toolbar.components.get_action("plot_info_label").action + action.trigger() + assert widget.show_info_label + assert widget.info_label.isVisible() + assert action.isChecked() + + def test_set_x_min_max(qtbot, mocked_client): """ Test setting x_min, x_max changes the actual X-range of the plot diff --git a/tests/unit_tests/test_scatter_waveform.py b/tests/unit_tests/test_scatter_waveform.py index 04852368..694f416d 100644 --- a/tests/unit_tests/test_scatter_waveform.py +++ b/tests/unit_tests/test_scatter_waveform.py @@ -1,6 +1,8 @@ from unittest.mock import MagicMock, 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 ( ScatterCurveConfig, @@ -53,26 +55,43 @@ def test_scatter_waveform_update_with_scan_history(qtbot, mocked_client, monkeyp 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() - # .get_by_scan_id() typically returns historical data, but we abuse it here - # to return mock live data - mocked_client.history.get_by_scan_id.return_value = dummy_scan - mocked_client.history.__getitem__.return_value = dummy_scan + mocked_client.history.get_by_scan_id.return_value = history_scan + mocked_client.history.__getitem__.return_value = history_scan swf.plot("samx", "samy", "bpm4i", label="test_curve") swf.update_with_scan_history(scan_id="dummy") - qtbot.waitUntil(lambda: swf.scan_item == dummy_scan, timeout=500) + qtbot.waitUntil(lambda: swf.scan_item == history_scan, timeout=500) qtbot.wait(200) 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 [label.text for _, label in swf.info_label.items] == [ + "Scan: 24 (history)", + "Scan Name: line_scan", + ] 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"], + }, + ) monkeypatch.setattr(swf.queue.scan_storage, "find_scan_by_ID", lambda scan_id: dummy_scan) swf.plot("samx", "samy", "bpm4i", label="live_curve") @@ -84,6 +103,10 @@ def test_scatter_waveform_live_update(qtbot, mocked_client, monkeypatch): assert swf.scan_id == "dummy" assert swf.scan_item == dummy_scan + assert [label.text for _, label in swf.info_label.items] == [ + "Scan: 13 (live)", + "Scan Name: line_scan", + ] qtbot.wait(500) diff --git a/tests/unit_tests/test_waveform.py b/tests/unit_tests/test_waveform.py index bf50a022..756346a3 100644 --- a/tests/unit_tests/test_waveform.py +++ b/tests/unit_tests/test_waveform.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock import numpy as np import pyqtgraph as pg import pytest +from bec_lib import messages from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem from qtpy.QtCore import QTimer from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox @@ -459,6 +460,16 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls): # 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"], + }, + ) 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 @@ -475,6 +486,10 @@ def test_on_scan_status(qtbot, mocked_client, monkeypatch, mode, calls): assert wf.scan_id == "1234" assert wf.scan_item == dummy_scan assert wf._mode == mode + assert [label.text for _, label in wf.info_label.items] == [ + "Scan: 12 (live)", + "Scan Name: line_scan", + ] 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" @@ -1642,6 +1657,13 @@ 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" + wf.update_with_scan_history(scan_id="hist1") + + assert [label.text for _, label in wf.info_label.items] == [ + "Scan: 1 (history)", + "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):