From c5303dc8ab7f138924da3d2b643fa37aa35c8b7c Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Mon, 30 Jun 2025 10:28:03 +0200 Subject: [PATCH] feat(scan-history-browser): Add history browser and history metadata viewer --- .../services/scan_history_browser/__init__.py | 0 .../scan_history_browser.py | 198 ++++++++++++++++++ .../scan_history_metadata_viewer.py | 161 ++++++++++++++ 3 files changed, 359 insertions(+) create mode 100644 bec_widgets/widgets/services/scan_history_browser/__init__.py create mode 100644 bec_widgets/widgets/services/scan_history_browser/scan_history_browser.py create mode 100644 bec_widgets/widgets/services/scan_history_browser/scan_history_metadata_viewer.py diff --git a/bec_widgets/widgets/services/scan_history_browser/__init__.py b/bec_widgets/widgets/services/scan_history_browser/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/services/scan_history_browser/scan_history_browser.py b/bec_widgets/widgets/services/scan_history_browser/scan_history_browser.py new file mode 100644 index 00000000..f4317a2d --- /dev/null +++ b/bec_widgets/widgets/services/scan_history_browser/scan_history_browser.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import random +import sys +from datetime import datetime, timedelta + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.messages import ScanHistoryMessage +from qtpy import QtCore, QtGui, QtWidgets + +from bec_widgets.utils.bec_widget import BECWidget, ConnectionConfig +from bec_widgets.utils.colors import get_accent_colors +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + +logger = bec_logger.logger + + +class ScanHistoryBrowser(BECWidget, QtWidgets.QTreeWidget): + """ScanHistoryTree is a widget that displays the scan history in a tree format.""" + + scan_selected = QtCore.Signal(object) + scan_removed = QtCore.Signal(object) + + def __init__( + self, + parent: QtWidgets.QWidget = None, + client=None, + config: ConnectionConfig = None, + gui_id: str = None, + max_length: int = 100, + **kwargs, + ): + super().__init__( + parent=parent, client=client, config=config, gui_id=gui_id, theme_update=True, **kwargs + ) + colors = get_accent_colors() + self.status_colors = { + "closed": colors.success, + "halted": colors.warning, + "aborted": colors.emergency, + } + # self.status_colors = {"closed": "#00e676", "halted": "#ffca28", "aborted": "#ff5252"} + self.column_header = ["Scan Nr", "Scan Name", "Status"] + self.scan_history: list[ScanHistoryMessage] = [] # newest at index 0 + self.max_length = max_length # Maximum number of scan history entries to keep + self._set_policies() + self.apply_theme() + self._start_subscription() + self.itemActivated.connect(self._on_item_clicked) + self.itemClicked.connect(self._on_item_clicked) + + def keyPressEvent(self, event): + """Handle Enter key press to activate emitting the selected item.""" + if event.key() in (QtCore.Qt.Key_Enter, QtCore.Qt.Key_Return): + item = self.currentItem() + if item: + self.itemActivated.emit(item, self.currentColumn()) + else: + super().keyPressEvent(event) + + def _set_policies(self): + self.setColumnCount(len(self.column_header)) + self.setHeaderLabels(self.column_header) + self.setRootIsDecorated(False) # allow expand arrow for per‑scan details + self.setUniformRowHeights(True) + self.setFocusPolicy(QtCore.Qt.StrongFocus) + self.setAlternatingRowColors(True) + self.setSelectionMode(QtWidgets.QAbstractItemView.SingleSelection) + self.setIndentation(12) + self.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff) + self.setAnimated(True) + + header = self.header() + header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeMode.ResizeToContents) + for column in range(1, self.columnCount()): + header.setSectionResizeMode(column, QtWidgets.QHeaderView.ResizeMode.Stretch) + + def apply_theme(self, theme: str | None = None): + colors = get_accent_colors() + self.status_colors = { + "closed": colors.success, + "halted": colors.warning, + "aborted": colors.emergency, + } + self.repaint() + + def _on_item_clicked(self, item: QtWidgets.QTreeWidgetItem, column: int): + """ + Handle item click events in the tree widget. + Emits a signal with the selected scan message when an item is clicked. + """ + if not item: + return + index = self.indexOfTopLevelItem(item) + self.scan_selected.emit(self.scan_history[index]) + + def _start_subscription(self): + """ + Subscribe to scan history updates. + """ + self.bec_dispatcher.connect_slot( + slot=self.update_history, topics=MessageEndpoints.scan_history(), from_start=True + ) + + @SafeSlot() + def update_history(self, msg_content: dict, metdata: dict): + """ + This method is called whenever a new scan history is available. + """ + # TODO directly receive ScanHistoryMessage through dispatcher + msg = ScanHistoryMessage(**msg_content) + msg.metadata = metdata + self.add_scan(msg) + self.ensure_history_max_length() + + def ensure_history_max_length(self) -> None: + """ + Clean up the scan history by clearing the list. + This method can be called when the widget is closed or no longer needed. + """ + while len(self.scan_history) > self.max_length: + logger.warning( + f"Removing oldest scan history entry to maintain max length of {self.max_length}." + ) + self.remove_scan(index=-1) + + def add_scan(self, msg: ScanHistoryMessage): + """ + Add a scan entry to the tree widget. + Args: + msg (ScanHistoryMessage): The scan history message containing scan details. + """ + self.scan_history.insert(0, msg) + tree_item = QtWidgets.QTreeWidgetItem([str(msg.scan_number), msg.scan_name, ""]) + color = QtGui.QColor(self.status_colors.get(msg.exit_status, "#b0bec5")) + pix = QtGui.QPixmap(10, 10) + pix.fill(QtCore.Qt.transparent) + with QtGui.QPainter(pix) as p: + p.setRenderHint(QtGui.QPainter.Antialiasing) + p.setPen(QtCore.Qt.NoPen) + p.setBrush(color) + p.drawEllipse(0, 0, 10, 10) + tree_item.setIcon(2, QtGui.QIcon(pix)) + tree_item.setForeground(2, QtGui.QBrush(color)) + for col in range(tree_item.columnCount()): + tree_item.setToolTip(col, f"Status: {msg.exit_status}") + self.insertTopLevelItem(0, tree_item) + tree_item.setExpanded(False) + + def remove_scan(self, index: int): + """ + Remove a scan entry from the tree widget. We supoprt negative indexing where -1, -2, etc. refer to the last, second last, etc. entry. + Args: + index (int): The index of the scan entry to remove. + """ + if index < 0: + index = len(self.scan_history) + index + try: + msg = self.scan_history.pop(index) + self.scan_removed.emit(msg) + except IndexError: + logger.warning(f"Invalid index {index} for removing scan entry from history.") + return + self.takeTopLevelItem(index) + + +if __name__ == "__main__": # pragma: no cover + # pylint: disable=import-outside-toplevel + + from bec_widgets.widgets.services.scan_history_browser.scan_history_metadata_viewer import ( + ScanHistoryMetadataViewer, + ) + from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + + app = QtWidgets.QApplication([]) + + main_window = QtWidgets.QMainWindow() + central_widget = QtWidgets.QWidget() + button = DarkModeButton() + layout = QtWidgets.QVBoxLayout(central_widget) + main_window.setCentralWidget(central_widget) + + # Create a ScanHistoryBrowser instance + browser = ScanHistoryBrowser() + + # Create a ScanHistoryView instance + view = ScanHistoryMetadataViewer() + + layout.addWidget(button) + layout.addWidget(browser) + layout.addWidget(view) + browser.scan_selected.connect(view.update_view) + browser.scan_removed.connect(view.clear_view) + + main_window.show() + app.exec_() diff --git a/bec_widgets/widgets/services/scan_history_browser/scan_history_metadata_viewer.py b/bec_widgets/widgets/services/scan_history_browser/scan_history_metadata_viewer.py new file mode 100644 index 00000000..e58053a4 --- /dev/null +++ b/bec_widgets/widgets/services/scan_history_browser/scan_history_metadata_viewer.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from bec_lib.logger import bec_logger +from bec_qthemes import material_icon +from qtpy import QtCore, QtGui, QtWidgets + +from bec_widgets.utils.bec_widget import BECWidget, ConnectionConfig +from bec_widgets.utils.colors import get_theme_palette +from bec_widgets.utils.error_popups import SafeSlot + +if TYPE_CHECKING: # pragma: no cover + from bec_lib.messages import ScanHistoryMessage + +logger = bec_logger.logger + + +class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox): + """ScanHistoryView is a widget to display the metadata of a ScanHistoryMessage in a structured format.""" + + def __init__( + self, + client=None, + config: ConnectionConfig | None = None, + gui_id: str | None = None, + theme_update: bool = True, + scan_history_msg: ScanHistoryMessage | None = None, + parent=None, + ): + super().__init__( + parent=parent, client=client, config=config, gui_id=gui_id, theme_update=theme_update + ) + self._scan_history_msg_labels = { + "scan_id": "Scan ID", + "dataset_number": "Dataset Nr", + "file_path": "File Path", + "start_time": "Start Time", + "end_time": "End Time", + "elapsed_time": "Elapsed Time", + "exit_status": "Status", + "scan_name": "Scan Name", + "num_points": "Nr of Points", + } + self.setTitle("No Scan Selected") + # self.setMinimumHeight(100) + # self.setSizePolicy(QtWidgets.QSizePolicy.Preferred, QtWidgets.QSizePolicy.Expanding) + layout = QtWidgets.QGridLayout() + self.setLayout(layout) + self._init_grid_layout() + self.scan_history_msg = scan_history_msg + if scan_history_msg is not None: + self.update_view(self.scan_history_msg) + self.apply_theme() + + def apply_theme(self, theme: str | None = None): + """Apply the theme to the widget.""" + colors = get_theme_palette() + palette = QtGui.QPalette() + palette.setColor(self.backgroundRole(), colors.midlight().color()) + self.setPalette(palette) + + def _init_grid_layout(self): + """Initialize the layout of the widget.""" + layout: QtWidgets.QGridLayout = self.layout() + layout.setContentsMargins(10, 10, 10, 10) + layout.setHorizontalSpacing(0) + layout.setVerticalSpacing(0) + layout.setColumnStretch(0, 0) + layout.setColumnStretch(1, 1) + layout.setColumnStretch(2, 0) + + def setup_content_widget_label(self) -> None: + """Setup the labels for the content widget for the scan history view.""" + layout = self.layout() + for row, k in enumerate(self._scan_history_msg_labels.keys()): + v = self._scan_history_msg_labels[k] + # Label for the key + label = QtWidgets.QLabel(f"{v}:") + # label.setSizePolicy(QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Preferred) + layout.addWidget(label, row, 0) + # Value field + value_field = QtWidgets.QLabel("") + value_field.setSizePolicy( + QtWidgets.QSizePolicy.Ignored, QtWidgets.QSizePolicy.Preferred + ) + layout.addWidget(value_field, row, 1) # Placeholder for value + # Copy button + copy_button = QtWidgets.QToolButton() + copy_button.setSizePolicy(QtWidgets.QSizePolicy.Fixed, QtWidgets.QSizePolicy.Minimum) + copy_button.setContentsMargins(0, 0, 0, 0) + copy_button.setStyleSheet("padding: 0px; margin: 0px; border: none;") + copy_button.setIcon(material_icon(icon_name="content_copy", size=(16, 16))) + copy_button.setToolTip("Copy to clipboard") + copy_button.setVisible(False) + copy_button.setEnabled(False) # Initially disabled + copy_button.clicked.connect( + lambda _, field=value_field: QtWidgets.QApplication.clipboard().setText( + field.text() + ) + ) + layout.addWidget(copy_button, row, 2) # Placeholder for copy button + + @SafeSlot() + def update_view(self, msg: ScanHistoryMessage): + """ + Update the view with the given ScanHistoryMessage. + + Args: + msg (ScanHistoryMessage): The message containing scan metadata. + """ + if msg == self.scan_history_msg: + return # Same message, no update needed + self.scan_history_msg = msg + layout = self.layout() + if layout.count() == 0: + self.setup_content_widget_label() + self.setTitle(f"Metadata - Scan {msg.scan_number}") + for row, k in enumerate(self._scan_history_msg_labels.keys()): + if k == "elapsed_time": + value = ( + f"{(msg.end_time - msg.start_time):.3f}s" + if msg.start_time and msg.end_time + else None + ) + else: + value = getattr(msg, k, None) + if k in ["start_time", "end_time"]: + value = ( + datetime.fromtimestamp(value).strftime("%a %b %d %H:%M:%S %Y") + if value + else None + ) + if value is None: + logger.warning(f"ScanHistoryMessage missing value for {k} and msg {msg}.") + continue + layout.itemAtPosition(row, 1).widget().setText(str(value)) + if k in ["file_path", "scan_id"]: + layout.itemAtPosition(row, 2).widget().setVisible(True) + layout.itemAtPosition(row, 2).widget().setEnabled(True) + else: + layout.itemAtPosition(row, 2).widget().setText("") + layout.itemAtPosition(row, 2).widget().setToolTip("") + + @SafeSlot() + def clear_view(self, msg: ScanHistoryMessage | None = None): + """ + Clear the view by resetting the labels and values. + """ + if self.scan_history_msg is not None and msg != self.scan_history_msg: + return + layout = self.layout() + lauout_counts = layout.count() + for i in range(lauout_counts): + item = layout.itemAt(i) + if item.widget(): + item.widget().close() + item.widget().deleteLater() + self.scan_history_msg = None + self.setTitle("No Scan Selected")