1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-04 16:02:51 +01:00

refactor: add additional components for history metadata, device view and popup ui

This commit is contained in:
2025-07-07 08:51:06 +02:00
committed by Jan Wyzula
parent 694a6c4960
commit cf97cc1805
5 changed files with 458 additions and 161 deletions

View File

@@ -0,0 +1,5 @@
from .scan_history_device_viewer import ScanHistoryDeviceViewer
from .scan_history_metadata_viewer import ScanHistoryMetadataViewer
from .scan_history_view import ScanHistoryView
__all__ = ["ScanHistoryDeviceViewer", "ScanHistoryMetadataViewer", "ScanHistoryView"]

View File

@@ -0,0 +1,172 @@
from __future__ import annotations
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.messages import ScanHistoryMessage
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_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
logger = bec_logger.logger
# TODO check cleanup
# Custom model
class DeviceModel(QtCore.QAbstractListModel):
def __init__(self, devices=None):
super().__init__()
if devices is None:
devices = {}
self._devices = sorted(devices.items(), key=lambda x: -x[1])
@property
def devices(self):
"""Return the list of devices."""
return self._devices
@devices.setter
def devices(self, value: dict[str, int]):
self.beginResetModel()
self._devices = sorted(value.items(), key=lambda x: -x[1])
self.endResetModel()
def rowCount(self, parent=QtCore.QModelIndex()):
return len(self.devices)
def data(self, index, role=QtCore.Qt.DisplayRole):
if not index.isValid():
return None
name, num_points = self.devices[index.row()]
if role == QtCore.Qt.DisplayRole:
return f"{name} ({num_points})" # fallback display
elif role == QtCore.Qt.UserRole:
return name
elif role == QtCore.Qt.UserRole + 1:
return num_points
return None
# Custom delegate for better formatting
class DeviceDelegate(QtWidgets.QStyledItemDelegate):
def paint(self, painter, option, index):
name = index.data(QtCore.Qt.UserRole)
points = index.data(QtCore.Qt.UserRole + 1)
painter.save()
painter.drawText(
option.rect.adjusted(5, 0, -5, 0), QtCore.Qt.AlignVCenter | QtCore.Qt.AlignLeft, name
)
painter.drawText(
option.rect.adjusted(5, 0, -5, 0),
QtCore.Qt.AlignVCenter | QtCore.Qt.AlignRight,
str(points),
)
painter.restore()
def sizeHint(self, option, index):
return QtCore.QSize(200, 24)
class ScanHistoryDeviceViewer(BECWidget, QtWidgets.QWidget):
"""ScanHistoryTree is a widget that displays the scan history in a tree format."""
RPC = False
PLUGIN = False
request_history_plot = QtCore.Signal(str, dict) # (str, ScanHistoryMessage.model_dump())
def __init__(
self,
parent: QtWidgets.QWidget = None,
client=None,
config: ConnectionConfig = None,
gui_id: str = None,
theme_update: bool = True,
**kwargs,
):
super().__init__(
parent=parent,
client=client,
config=config,
gui_id=gui_id,
theme_update=theme_update,
**kwargs,
)
# Current scan history message
self.scan_history_msg: ScanHistoryMessage | None = None
self._selected_device: str = ""
# Init layout
layout = QtWidgets.QHBoxLayout(self)
self.setLayout(layout)
# Init ComboBox
self.device_combo = QtWidgets.QComboBox(self)
colors = get_accent_colors()
self.request_plotting_button = QtWidgets.QPushButton(
material_icon("play_arrow", size=(24, 24), color=colors.success),
"Request Plotting",
self,
)
self.device_model = DeviceModel({})
self.device_combo.setModel(self.device_model)
layout.addWidget(self.device_combo)
layout.addWidget(self.request_plotting_button)
self.device_combo.setItemDelegate(DeviceDelegate())
# Connect signals
self.request_plotting_button.clicked.connect(self._on_request_plotting_clicked)
@SafeProperty(str)
def device(self) -> str:
"""Get the currently selected device name."""
return self._selected_device
@device.setter
def device(self, value: str):
"""Set the currently selected device name."""
if not isinstance(value, str):
logger.info(f"Device name must be a string {value}.")
if value not in self.scan_history_msg.device_data_info:
logger.info(f"Device name must in the list of selected devices {value}.")
self._selected_device = value
@SafeSlot()
def update_devices_from_scan_history(self, msg: ScanHistoryMessage) -> None:
"""Update the device combo box with the scan history message."""
if not isinstance(msg, ScanHistoryMessage):
logger.info(f"Received message of type {type(msg)} instead of ScanHistoryMessage.")
return
self.scan_history_msg = msg
self.device_model.devices = msg.device_data_info
@SafeSlot()
def clear_view(self, msg: ScanHistoryMessage | None = None) -> None:
"""Clear the device combo box."""
self.scan_history_msg = None
self.device_model.devices = {}
self.device_combo.clear()
@SafeSlot()
def _on_request_plotting_clicked(self):
"""Handle the request plotting button click."""
if self.scan_history_msg is None:
logger.info("No scan history message available for plotting.")
return
current_index = self.device_combo.currentIndex()
device_name = self.device_combo.model().data(
self.device_combo.model().index(current_index, 0), QtCore.Qt.UserRole
)
logger.info(
f"Requesting plotting for device: {device_name} with {self.scan_history_msg} points."
)
self.request_history_plot.emit((device_name, self.scan_history_msg))
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
viewer = ScanHistoryDeviceViewer()
viewer.show()
sys.exit(app.exec_())

View File

@@ -20,6 +20,9 @@ logger = bec_logger.logger
class ScanHistoryMetadataViewer(BECWidget, QtWidgets.QGroupBox):
"""ScanHistoryView is a widget to display the metadata of a ScanHistoryMessage in a structured format."""
RPC = False
PLUGIN = False
def __init__(
self,
client=None,

View File

@@ -0,0 +1,210 @@
from __future__ import annotations
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 ScanHistoryView(BECWidget, QtWidgets.QTreeWidget):
"""ScanHistoryTree is a widget that displays the scan history in a tree format."""
RPC = False
PLUGIN = False
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,
theme_update: bool = True,
**kwargs,
):
super().__init__(
parent=parent,
client=client,
config=config,
gui_id=gui_id,
theme_update=theme_update,
**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.itemClicked.connect(self._on_item_clicked)
self.currentItemChanged.connect(self._current_item_changed)
def _set_policies(self):
self.setColumnCount(len(self.column_header))
self.setHeaderLabels(self.column_header)
self.setRootIsDecorated(False) # allow expand arrow for perscan 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 _current_item_changed(
self, current: QtWidgets.QTreeWidgetItem, previous: QtWidgets.QTreeWidgetItem
):
"""
Handle current item change events in the tree widget.
Emits a signal with the selected scan message when the current item changes.
"""
if not current:
return
self._on_item_clicked(current, self.currentColumn())
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.components import (
ScanHistoryDeviceViewer,
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 = ScanHistoryView()
# Create a ScanHistoryView instance
view = ScanHistoryMetadataViewer()
device_viewer = ScanHistoryDeviceViewer()
layout.addWidget(button)
layout.addWidget(browser)
layout.addWidget(view)
layout.addWidget(device_viewer)
browser.scan_selected.connect(view.update_view)
browser.scan_selected.connect(device_viewer.update_devices_from_scan_history)
browser.scan_removed.connect(view.clear_view)
browser.scan_removed.connect(device_viewer.clear_view)
main_window.show()
app.exec_()

View File

@@ -1,198 +1,105 @@
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 qtpy import QtCore, 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
from bec_widgets.widgets.services.scan_history_browser.components import (
ScanHistoryDeviceViewer,
ScanHistoryMetadataViewer,
ScanHistoryView,
)
class ScanHistoryBrowser(BECWidget, QtWidgets.QTreeWidget):
"""ScanHistoryTree is a widget that displays the scan history in a tree format."""
class ScanHistoryBrowser(BECWidget, QtWidgets.QWidget):
scan_selected = QtCore.Signal(object)
scan_removed = QtCore.Signal(object)
RPC = False
PLUGIN = False
def __init__(
self,
parent: QtWidgets.QWidget = None,
parent: QtWidgets.QWidget | None = None,
client=None,
config: ConnectionConfig = None,
gui_id: str = None,
max_length: int = 100,
gui_id: str | None = None,
theme_update: bool = False,
**kwargs,
):
super().__init__(
parent=parent, client=client, config=config, gui_id=gui_id, theme_update=True, **kwargs
parent=parent,
client=client,
config=config,
gui_id=gui_id,
theme_update=theme_update,
**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)
layout = QtWidgets.QHBoxLayout()
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 perscan 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
self.setLayout(layout)
self.scan_history_view = ScanHistoryView(
parent=self, client=client, config=config, gui_id=gui_id, theme_update=theme_update
)
self.scan_history_metadata_viewer = ScanHistoryMetadataViewer(
parent=self, client=client, config=config, gui_id=gui_id, theme_update=theme_update
)
self.scan_history_device_viewer = ScanHistoryDeviceViewer(
parent=self, client=client, config=config, gui_id=gui_id, theme_update=theme_update
)
@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()
self.init_layout()
self.connect_signals()
QtCore.QTimer.singleShot(0, self.select_first_history_entry)
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 init_layout(self):
"""Initialize the layout of the widget."""
# Add Scan history view
layout: QtWidgets.QHBoxLayout = self.layout()
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
layout.addWidget(self.scan_history_view)
# Add metadata and device viewers in a vertical layout
widget = QtWidgets.QWidget(self)
vertical_layout = QtWidgets.QVBoxLayout()
vertical_layout.setContentsMargins(0, 0, 0, 0)
vertical_layout.setSpacing(0)
vertical_layout.addWidget(self.scan_history_metadata_viewer)
vertical_layout.addWidget(self.scan_history_device_viewer)
widget.setLayout(vertical_layout)
# Add the vertical layout widget to the main layout
layout.addWidget(widget)
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 connect_signals(self):
"""Connect signals to the appropriate slots."""
self.scan_history_view.scan_selected.connect(self.scan_history_metadata_viewer.update_view)
self.scan_history_view.scan_selected.connect(
self.scan_history_device_viewer.update_devices_from_scan_history
)
self.scan_history_view.scan_removed.connect(self.scan_history_metadata_viewer.clear_view)
self.scan_history_view.scan_removed.connect(self.scan_history_device_viewer.clear_view)
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)
def select_first_history_entry(self):
"""Select the first entry in the scan history view."""
if self.scan_history_view.topLevelItemCount() > 0:
self.scan_history_view.setCurrentItem(self.scan_history_view.topLevelItem(0))
self.scan_history_view.itemActivated.emit(self.scan_history_view.topLevelItem(0), 0)
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 qtpy.QtWidgets import QApplication
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
app = QtWidgets.QApplication([])
app = 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()
browser = ScanHistoryBrowser() # type: ignore
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.setWindowTitle("Scan History Browser")
main_window.resize(800, 600)
main_window.show()
app.exec_()