Compare commits

...

2 Commits

Author SHA1 Message Date
appel_c 797a5046d1 refactor: improve device-manager-view 2025-10-09 15:13:58 +02:00
appel_c 9ff0db4831 feat(help-inspector): add help inspector widget 2025-10-09 15:13:35 +02:00
9 changed files with 797 additions and 99 deletions
+1
View File
@@ -195,6 +195,7 @@ if __name__ == "__main__": # pragma: no cover
app = QApplication([sys.argv[0], *qt_args])
apply_theme("dark")
w = BECMainApp(show_examples=args.examples)
w.resize(1920, 1200)
w.show()
sys.exit(app.exec())
@@ -14,10 +14,23 @@ from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path
from bec_qthemes import apply_theme
from PySide6QtAds import CDockManager, CDockWidget
from qtpy.QtCore import Qt, QThreadPool, QTimer
from qtpy.QtWidgets import QFileDialog, QMessageBox, QSplitter, QVBoxLayout, QWidget
from qtpy.QtWidgets import (
QDialog,
QFileDialog,
QHBoxLayout,
QLabel,
QMessageBox,
QPushButton,
QSizePolicy,
QSplitter,
QTextEdit,
QVBoxLayout,
QWidget,
)
from bec_widgets import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.help_inspector.help_inspector import HelpInspector
from bec_widgets.utils.toolbars.actions import MaterialIconAction
from bec_widgets.utils.toolbars.bundles import ToolbarBundle
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
@@ -83,6 +96,53 @@ def set_splitter_weights(splitter: QSplitter, weights: List[float]) -> None:
QTimer.singleShot(0, apply)
class ConfigChoiceDialog(QDialog):
REPLACE = 1
ADD = 2
CANCEL = 0
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("Load Config")
layout = QVBoxLayout(self)
label = QLabel("Do you want to replace the current config or add to it?")
label.setWordWrap(True)
layout.addWidget(label)
# Buttons: equal size, stacked vertically
self.replace_btn = QPushButton("Replace")
self.add_btn = QPushButton("Add")
self.cancel_btn = QPushButton("Cancel")
btn_layout = QHBoxLayout()
for btn in (self.replace_btn, self.add_btn, self.cancel_btn):
btn.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred)
btn_layout.addWidget(btn)
layout.addLayout(btn_layout)
# Connect signals to explicit slots
self.replace_btn.clicked.connect(self.accept_replace)
self.add_btn.clicked.connect(self.accept_add)
self.cancel_btn.clicked.connect(self.reject_cancel)
self._result = self.CANCEL
def accept_replace(self):
self._result = self.REPLACE
self.accept()
def accept_add(self):
self._result = self.ADD
self.accept()
def reject_cancel(self):
self._result = self.CANCEL
self.reject()
def result(self):
return self._result
class DeviceManagerView(BECWidget, QWidget):
def __init__(self, parent=None, *args, **kwargs):
@@ -99,14 +159,14 @@ class DeviceManagerView(BECWidget, QWidget):
self.dock_manager.setStyleSheet("")
self._root_layout.addWidget(self.dock_manager)
# Available Resources Widget
self.available_devices = AvailableDeviceResources(
self, shared_selection_signal=self._shared_selection
)
self.available_devices_dock = QtAds.CDockWidget(
self.dock_manager, "Available Devices", self
)
self.available_devices_dock.setWidget(self.available_devices)
# # Available Resources Widget
# self.available_devices = AvailableDeviceResources(
# self, shared_selection_signal=self._shared_selection
# )
# self.available_devices_dock = QtAds.CDockWidget(
# self.dock_manager, "Available Devices", self
# )
# self.available_devices_dock.setWidget(self.available_devices)
# Device Table View widget
self.device_table_view = DeviceTableView(
@@ -130,29 +190,62 @@ class DeviceManagerView(BECWidget, QWidget):
self.ophyd_test_dock_view = QtAds.CDockWidget(self.dock_manager, "Ophyd Test View", self)
self.ophyd_test_dock_view.setWidget(self.ophyd_test_view)
# Arrange widgets within the QtAds dock manager
# Help Inspector
widget = QWidget(self)
layout = QVBoxLayout(widget)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
self.help_inspector = HelpInspector(self)
layout.addWidget(self.help_inspector)
text_box = QTextEdit(self)
text_box.setReadOnly(False)
text_box.setPlaceholderText("Help text will appear here...")
layout.addWidget(text_box)
self.help_inspector_dock = QtAds.CDockWidget(self.dock_manager, "Help Inspector", self)
self.help_inspector_dock.setWidget(widget)
# # Hook inspector signals
# def _class_cb(text: str):
# print(text)
# # Register callback
self.help_inspector.bec_widget_help.connect(text_box.setMarkdown)
# Error Logs View
self.error_logs_view = QTextEdit(self)
self.error_logs_view.setReadOnly(True)
self.error_logs_view.setPlaceholderText("Error logs will appear here...")
self.error_logs_dock = QtAds.CDockWidget(self.dock_manager, "Error Logs", self)
self.error_logs_dock.setWidget(self.error_logs_view)
self.ophyd_test_view.validation_msg_md.connect(self.error_logs_view.setMarkdown)
# Arrange widgets within the QtAds dock manager
# Central widget area
self.central_dock_area = self.dock_manager.setCentralWidget(self.device_table_view_dock)
# Right area - should be pushed into view if something is active
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea,
self.dm_docs_view_dock,
QtAds.DockWidgetArea.RightDockWidgetArea,
self.ophyd_test_dock_view,
self.central_dock_area,
)
# Left Area
self.left_dock_area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.LeftDockWidgetArea, self.available_devices_dock
)
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea, self.dm_config_view_dock, self.left_dock_area
# create bottom area (2-arg -> area)
self.bottom_dock_area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea, self.dm_docs_view_dock
)
# Right area
# YAML view left of docstrings (docks relative to bottom area)
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.RightDockWidgetArea, self.ophyd_test_dock_view
QtAds.DockWidgetArea.LeftDockWidgetArea, self.dm_config_view_dock, self.bottom_dock_area
)
# Error/help area right of docstrings (dock relative to bottom area)
area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.RightDockWidgetArea,
self.help_inspector_dock,
self.bottom_dock_area,
)
self.dock_manager.addDockWidgetTabToArea(self.error_logs_dock, area)
for dock in self.dock_manager.dockWidgets():
# dock.setFeature(CDockWidget.DockWidgetDeleteOnClose, True)#TODO implement according to MonacoDock or AdvancedDockArea
# dock.setFeature(CDockWidget.CustomCloseHandling, True) #TODO same
@@ -160,13 +253,16 @@ class DeviceManagerView(BECWidget, QWidget):
dock.setFeature(CDockWidget.DockWidgetFloatable, False)
dock.setFeature(CDockWidget.DockWidgetMovable, False)
# TODO decide if we like to hide the title bars..
# Fetch all dock areas of the dock widgets (on our case always one dock area)
for dock in self.dock_manager.dockWidgets():
area = dock.dockAreaWidget()
area.titleBar().setVisible(False)
# for dock in self.dock_manager.dockWidgets():
# if dock.objectName() in ["Help Inspector", "Error Logs"]:
# continue
# area = dock.dockAreaWidget()
# area.titleBar().setVisible(False)
# Apply stretch after the layout is done
self.set_default_view([2, 8, 2], [3, 1])
self.set_default_view([2, 8, 2], [7, 3])
# self.set_default_view([2, 8, 2], [2, 2, 4])
# Connect slots
@@ -175,29 +271,29 @@ class DeviceManagerView(BECWidget, QWidget):
self.device_table_view.selected_devices,
(self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
),
(
self.available_devices.selected_devices,
(self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
),
# (
# self.available_devices.selected_devices,
# (self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
# ),
(
self.ophyd_test_view.device_validated,
(self.device_table_view.update_device_validation,),
),
(
self.device_table_view.device_configs_changed,
(
self.ophyd_test_view.change_device_configs,
self.available_devices.mark_devices_used,
),
),
(
self.available_devices.add_selected_devices,
(self.device_table_view.add_device_configs,),
),
(
self.available_devices.del_selected_devices,
(self.device_table_view.remove_device_configs,),
),
# (
# self.device_table_view.device_configs_changed,
# (
# self.ophyd_test_view.change_device_configs,
# self.available_devices.mark_devices_used,
# ),
# ),
# (
# self.available_devices.add_selected_devices,
# (self.device_table_view.add_device_configs,),
# ),
# (
# self.available_devices.del_selected_devices,
# (self.device_table_view.remove_device_configs,),
# ),
]:
for slot in slots:
signal.connect(slot)
@@ -217,7 +313,9 @@ class DeviceManagerView(BECWidget, QWidget):
# Create IO bundle
io_bundle = ToolbarBundle("IO", self.toolbar.components)
# Load from disk
load = MaterialIconAction(
text_position="under",
icon_name="file_open",
parent=self,
tooltip="Load configuration file from disk",
@@ -229,6 +327,7 @@ class DeviceManagerView(BECWidget, QWidget):
# Add safe to disk
safe_to_disk = MaterialIconAction(
text_position="under",
icon_name="file_save",
parent=self,
tooltip="Save config to disk",
@@ -240,10 +339,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Add load config from redis
load_redis = MaterialIconAction(
text_position="under",
icon_name="cached",
parent=self,
tooltip="Load current config from Redis",
label_text="Reload Config",
label_text="Get Current Config",
)
load_redis.action.triggered.connect(self._load_redis_action)
self.toolbar.components.add_safe("load_redis", load_redis)
@@ -251,11 +351,13 @@ class DeviceManagerView(BECWidget, QWidget):
# Update config action
update_config_redis = MaterialIconAction(
text_position="under",
icon_name="cloud_upload",
parent=self,
tooltip="Update current config in Redis",
label_text="Update Config",
)
update_config_redis.action.setEnabled(False)
update_config_redis.action.triggered.connect(self._update_redis_action)
self.toolbar.components.add_safe("update_config_redis", update_config_redis)
io_bundle.add_action("update_config_redis")
@@ -270,6 +372,7 @@ class DeviceManagerView(BECWidget, QWidget):
# Reset composed view
reset_composed = MaterialIconAction(
text_position="under",
icon_name="delete_sweep",
parent=self,
tooltip="Reset current composed config view",
@@ -281,7 +384,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Add device
add_device = MaterialIconAction(
icon_name="add", parent=self, tooltip="Add new device", label_text="Add Device"
text_position="under",
icon_name="add",
parent=self,
tooltip="Add new device",
label_text="Add Device",
)
add_device.action.triggered.connect(self._add_device_action)
self.toolbar.components.add_safe("add_device", add_device)
@@ -289,7 +396,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Remove device
remove_device = MaterialIconAction(
icon_name="remove", parent=self, tooltip="Remove device", label_text="Remove Device"
text_position="under",
icon_name="remove",
parent=self,
tooltip="Remove device",
label_text="Remove Device",
)
remove_device.action.triggered.connect(self._remove_device_action)
self.toolbar.components.add_safe("remove_device", remove_device)
@@ -297,10 +408,11 @@ class DeviceManagerView(BECWidget, QWidget):
# Rerun validation
rerun_validation = MaterialIconAction(
text_position="under",
icon_name="checklist",
parent=self,
tooltip="Run device validation with 'connect' on selected devices",
label_text="Rerun Validation",
label_text="Validate Connection",
)
rerun_validation.action.triggered.connect(self._rerun_validation_action)
self.toolbar.components.add_safe("rerun_validation", rerun_validation)
@@ -346,15 +458,26 @@ class DeviceManagerView(BECWidget, QWidget):
file_path, _ = QFileDialog.getOpenFileName(
self, caption="Select Config File", dir=start_dir
)
if file_path:
try:
config = [{"name": k, **v} for k, v in yaml_load(file_path).items()]
except Exception as e:
logger.error(f"Failed to load config from file {file_path}. Error: {e}")
return
self.device_table_view.set_device_config(
config
) # TODO ADD QDialog with 'replace', 'add' & 'cancel'
self._load_config_from_file(file_path)
def _load_config_from_file(self, file_path: str):
"""
Load device config from a given file path and update the device table view.
Args:
file_path (str): Path to the configuration file.
"""
try:
config = [{"name": k, **v} for k, v in yaml_load(file_path).items()]
except Exception as e:
logger.error(f"Failed to load config from file {file_path}. Error: {e}")
return
dialog = ConfigChoiceDialog(self)
if dialog.exec():
if dialog.result() == ConfigChoiceDialog.REPLACE:
self.device_table_view.set_device_config(config)
elif dialog.result() == ConfigChoiceDialog.ADD:
self.device_table_view.add_device_configs(config)
# TODO would we ever like to add the current config to an existing composition
@SafeSlot()
+8
View File
@@ -185,6 +185,14 @@ class BECWidget(BECConnector):
except Exception:
logger.warning(f"Failed to apply theme {theme} to {self}")
def get_help_md(self) -> str:
"""
Method to override in subclasses to provide help text in markdown format.
Returns:
str: The help text in markdown format.
"""
@SafeSlot()
@SafeSlot(str)
@rpc_timeout(None)
@@ -0,0 +1,246 @@
"""Module providing a simple help inspector tool for QtWidgets."""
from functools import partial
from typing import Callable
from uuid import uuid4
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from qtpy import QtCore, QtWidgets
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import AccentColors, get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.widget_io import WidgetHierarchy
logger = bec_logger.logger
class HelpInspector(BECWidget, QtWidgets.QWidget):
"""
A help inspector widget that allows to inspect other widgets in the application.
Per default, it emits signals with the docstring, tooltip and bec help text of the inspected widget.
The method "get_help_md" is called on the widget which is added to the BECWidget base class.
It should return a string with a help text, ideally in proper format to be displayed (i.e. markdown).
The inspector also allows to register custom callback that are called with the inspected widget
as argument. This may be useful in the future to hook up more callbacks with custom signals.
Args:
parent (QWidget | None): The parent widget of the help inspector.
client: Optional client for BECWidget functionality.
size (tuple[int, int]): Optional size of the icon for the help inspector.
"""
widget_docstring = QtCore.Signal(str) # Emits docstring from QWidget
widget_tooltip = QtCore.Signal(str) # Emits tooltip string from QWidget
bec_widget_help = QtCore.Signal(str) # Emits md formatted help string from BECWidget class
def __init__(self, parent=None, client=None):
super().__init__(client=client, parent=parent, theme_update=True)
self._app = QtWidgets.QApplication.instance()
layout = QtWidgets.QHBoxLayout(self) # type: ignore
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
self._active = False
self._init_ui()
self._callbacks = {}
# Register the default callbacks
self._register_default_callbacks()
# Connect the button toggle signal
self._button.toggled.connect(self._toggle_mode)
def _init_ui(self):
"""Init the UI components."""
colors: AccentColors = get_accent_colors()
self._button = QtWidgets.QToolButton(self.parent())
self._button.setCheckable(True)
self._icon_checked = partial(
material_icon, "help", size=(32, 32), color=colors.highlight, filled=True
)
self._icon_unchecked = partial(
material_icon, "help", size=(32, 32), color=colors.highlight, filled=False
)
self._button.setText("Help Inspect Tool")
self._button.setIcon(self._icon_unchecked())
self._button.setToolTip("Click to enter Help Mode")
self.layout().addWidget(self._button)
def apply_theme(self, theme: str) -> None:
colors = get_accent_colors()
self._icon_checked = partial(
material_icon, "help", size=(32, 32), color=colors.highlight, filled=True
)
self._icon_unchecked = partial(
material_icon, "help", size=(32, 32), color=colors.highlight, filled=False
)
if self._active:
self._button.setIcon(self._icon_checked())
else:
self._button.setIcon(self._icon_unchecked())
@SafeSlot(bool)
def _toggle_mode(self, enabled: bool):
"""
Toggle the help inspection mode.
Args:
enabled (bool): Whether to enable or disable the help inspection mode.
"""
if self._app is None:
self._app = QtWidgets.QApplication.instance()
self._active = enabled
if enabled:
self._app.installEventFilter(self)
self._button.setIcon(self._icon_checked())
QtWidgets.QApplication.setOverrideCursor(QtCore.Qt.CursorShape.WhatsThisCursor)
else:
self._app.removeEventFilter(self)
self._button.setIcon(self._icon_unchecked())
self._button.setChecked(False)
QtWidgets.QApplication.restoreOverrideCursor()
def eventFilter(self, obj: QtWidgets.QWidget, event: QtCore.QEvent) -> bool:
"""
Filter events to capture Key_Escape event, and mouse clicks
if event filter is active. Any click event on a widget is suppressed, if
the Inspector is active, and the registered callbacks are called with
the clicked widget as argument.
Args:
obj (QObject): The object that received the event.
event (QEvent): The event to filter.
"""
# If not active, return immediately
if not self._active:
return super().eventFilter(obj, event)
# If active, handle escape key
if event.type() == QtCore.QEvent.KeyPress and event.key() == QtCore.Qt.Key_Escape:
self._toggle_mode(False)
return super().eventFilter(obj, event)
# If active, and left mouse button pressed, handle click
if event.type() == QtCore.QEvent.MouseButtonPress:
if event.button() == QtCore.Qt.LeftButton:
widget = self._app.widgetAt(event.globalPos())
if widget is None:
return super().eventFilter(obj, event)
# Get BECWidget ancestor
# TODO check what happens if the HELP Inspector itself is embedded in another BECWidget
# I suppose we would like to get the first ancestor that is a BECWidget, not the topmost one
widget = WidgetHierarchy._get_becwidget_ancestor(widget)
if widget:
if widget is self:
self._toggle_mode(False)
return True
for cb in self._callbacks.values():
try:
cb(widget)
except Exception as e:
logger.error(f"Error occurred in callback {cb}: {e}")
return True
return super().eventFilter(obj, event)
def register_callback(self, callback: Callable[[QtWidgets.QWidget], None]) -> str:
"""
Register a callback to be called when a widget is inspected.
The callback should be callable with the following signature:
callback(widget: QWidget) -> None
Args:
callback (Callable[[QWidget], None]): The callback function to register.
Returns:
str: A unique ID for the registered callback.
"""
cb_id = str(uuid4())
self._callbacks[cb_id] = callback
return cb_id
def unregister_callback(self, cb_id: str):
"""Unregister a previously registered callback."""
self._callbacks.pop(cb_id, None)
def _register_default_callbacks(self):
"""Default behavior: publish tooltip, docstring, bec_help"""
def cb_doc(widget: QtWidgets.QWidget):
docstring = widget.__doc__ or "No documentation available."
self.widget_docstring.emit(docstring)
def cb_help(widget: QtWidgets.QWidget):
tooltip = widget.toolTip() or "No tooltip available."
self.widget_tooltip.emit(tooltip)
def cb_bec_help(widget: QtWidgets.QWidget):
help_text = None
if hasattr(widget, "get_help_md") and callable(widget.get_help_md):
try:
help_text = widget.get_help_md()
except Exception as e:
logger.debug(f"Error retrieving help text from {widget}: {e}")
if help_text is None:
help_text = widget.toolTip() or "No help available."
if not isinstance(help_text, str):
logger.error(
f"Help text from {widget.__class__} is not a string: {type(help_text)}"
)
help_text = str(help_text)
self.bec_widget_help.emit(help_text)
self.register_callback(cb_doc)
self.register_callback(cb_help)
self.register_callback(cb_bec_help)
if __name__ == "__main__":
import sys
from bec_qthemes import apply_theme
from bec_widgets.widgets.control.device_control.positioner_box import PositionerBox
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
app = QtWidgets.QApplication(sys.argv)
main_window = QtWidgets.QMainWindow()
apply_theme("dark")
main_window.setWindowTitle("Help Inspector Test")
central_widget = QtWidgets.QWidget()
main_layout = QtWidgets.QVBoxLayout(central_widget)
dark_mode_button = DarkModeButton(parent=main_window)
main_layout.addWidget(dark_mode_button)
help_inspector = HelpInspector()
main_layout.addWidget(help_inspector)
test_button = QtWidgets.QPushButton("Test Button")
test_button.setToolTip("This is a test button.")
test_line_edit = QtWidgets.QLineEdit()
test_line_edit.setToolTip("This is a test line edit.")
test_label = QtWidgets.QLabel("Test Label")
test_label.setToolTip("")
box = PositionerBox()
layout_1 = QtWidgets.QHBoxLayout()
layout_1.addWidget(test_button)
layout_1.addWidget(test_line_edit)
layout_1.addWidget(test_label)
layout_1.addWidget(box)
main_layout.addLayout(layout_1)
doc_label = QtWidgets.QLabel("Docstring will appear here.")
tool_tip_label = QtWidgets.QLabel("Tooltip will appear here.")
bec_help_label = QtWidgets.QLabel("BEC Help text will appear here.")
main_layout.addWidget(doc_label)
main_layout.addWidget(tool_tip_label)
main_layout.addWidget(bec_help_label)
help_inspector.widget_tooltip.connect(tool_tip_label.setText)
help_inspector.widget_docstring.connect(doc_label.setText)
help_inspector.bec_widget_help.connect(bec_help_label.setText)
main_window.setCentralWidget(central_widget)
main_window.resize(400, 200)
main_window.show()
sys.exit(app.exec())
@@ -6,3 +6,58 @@ MIME_DEVICE_CONFIG: Final[str] = "application/x-bec_device_config"
# Custom user roles
SORT_KEY_ROLE: Final[int] = 117
CONFIG_DATA_ROLE: Final[int] = 118
# TODO keep in sync with header...
HEADERS_HELP_MD: dict[str, str] = {
"status": (
"## Status"
"\n"
"The current status of the device. Can be one of the following values: \n"
"### **LOADED** \n The device with the specified configuration is loaded in the current config.\n"
"### **CONNECT_READY** \n The device config is valid and the connection has been validated. It has not yet been loaded to the current config.\n"
"### **CONNECT_FAILED** \n The device config is valid, but the connection could not be established.\n"
"### **VALID** \n The device config is valid, but the connection has not yet been validated.\n"
"### **INVALID** \n The device config is invalid and can not be loaded to the current config.\n"
),
"name": ("## Name " "\n" "The name of the device."),
"deviceClass": (
"## Device Class"
"\n"
"The device class specifies the type of the device. It will be used to create the instance."
),
"readoutPriority": (
"## Readout Priority"
"\n"
"The readout priority of the device. Can be one of the following values: \n"
"### **monitored** \n The monitored priority is used for devices that are read out during the scan (i.e. at every step) and whose value may change during the scan.\n"
"### **baseline** \n The baseline priority is used for devices that are read out at the beginning of the scan and whose value does not change during the scan.\n"
"### **async** \n The async priority is used for devices that are asynchronous to the monitored devices, and send their data independently.\n"
"### **continuous** \n The continuous priority is used for devices that are read out continuously during the scan.\n"
"### **on_request** \n The on_request priority is used for devices that should not be read out during the scan, yet are configured to be read out manually.\n"
),
"deviceTags": (
"## Device Tags"
"\n"
"A list of tags associated with the device. Tags can be used to group devices and filter them in the device manager."
),
"enabled": (
"## Enabled"
"\n"
"Indicator whether the device is enabled or disabled. Disabled devices can not be used."
),
"readOnly": ("## Read Only" "\n" "Indicator that a device is read-only or can be modified."),
"onFailure": (
"## On Failure"
"\n"
"Specifies the behavior of the device in case of a failure. Can be one of the following values: \n"
"### **buffer** \n The device readback will fall back to the last known value.\n"
"### **retry** \n The device readback will be retried once, and raises an error if it fails again.\n"
"### **raise** \n The device readback will raise immediately.\n"
),
"softwareTrigger": (
"## Software Trigger"
"\n"
"Indicator whether the device receives a software trigger from BEC during a scan."
),
"description": ("## Description" "\n" "A short description of the device."),
}
@@ -4,11 +4,13 @@ from __future__ import annotations
import copy
import json
import textwrap
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, Iterable, List
from uuid import uuid4
from bec_lib.atlas_models import Device
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from qtpy import QtCore, QtGui, QtWidgets
@@ -21,7 +23,10 @@ from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
from bec_widgets.widgets.control.device_manager.components.constants import MIME_DEVICE_CONFIG
from bec_widgets.widgets.control.device_manager.components.constants import (
HEADERS_HELP_MD,
MIME_DEVICE_CONFIG,
)
from bec_widgets.widgets.control.device_manager.components.dm_ophyd_test import ValidationStatus
if TYPE_CHECKING: # pragma: no cover
@@ -75,6 +80,111 @@ class CustomDisplayDelegate(DictToolTipDelegate):
painter.restore()
class WrappingTextDelegate(CustomDisplayDelegate):
"""A lightweight delegate that wraps text without expensive size recalculation."""
def __init__(self, parent=None, max_width=300, margin=6):
super().__init__(parent)
self._parent = parent
self.max_width = max_width
self.margin = margin
self._cache = {} # cache text metrics for performance
def _do_custom_paint(self, painter, option, index, value: str):
text = str(value)
if not text:
return
painter.save()
painter.setClipRect(option.rect)
# Use cached layout if available
cache_key = (text, option.rect.width())
layout = self._cache.get(cache_key)
if layout is None:
layout = QtGui.QTextLayout(text, option.font)
layout.beginLayout()
height = 0
while True:
line = layout.createLine()
if not line.isValid():
break
line.setLineWidth(option.rect.width() - self.margin)
line.setPosition(QtCore.QPointF(self.margin / 2, height))
height += line.height()
layout.endLayout()
self._cache[cache_key] = layout
# # Draw background if selected
# if option.state & QtWidgets.QStyle.State_Selected:
# painter.fillRect(option.rect, option.palette.highlight())
# Draw text
painter.setPen(option.palette.text().color())
layout.draw(painter, option.rect.topLeft())
painter.restore()
def sizeHint(self, option, index):
"""Return a cached or approximate height; avoids costly recomputation."""
text = str(index.data(QtCore.Qt.DisplayRole) or "")
view = self._parent
view.initViewItemOption(option)
if view.isColumnHidden(index.column()) or not view.isVisible() or not text:
return QtCore.QSize(0, option.fontMetrics.height() + 2 * self.margin)
# Use cache for consistent size computation
cache_key = (text, self.max_width)
if cache_key in self._cache:
layout = self._cache[cache_key]
height = 0
for i in range(layout.lineCount()):
height += layout.lineAt(i).height()
return QtCore.QSize(self.max_width, int(height + self.margin))
# Approximate without layout (fast path)
metrics = option.fontMetrics
pixel_width = max(self._parent.columnWidth(index.column()), 100)
if pixel_width > 2000: # safeguard against uninitialized columns, may return large values
pixel_width = 100
char_per_line = self.estimate_chars_per_line(text, option, pixel_width - 2 * self.margin)
wrapped_lines = textwrap.wrap(text, width=char_per_line)
lines = len(wrapped_lines)
return QtCore.QSize(pixel_width, lines * (metrics.height()) + 2 * self.margin)
def estimate_chars_per_line(self, text: str, option, column_width: int) -> int:
"""Estimate number of characters that fit in a line for given width."""
metrics = option.fontMetrics
elided = metrics.elidedText(text, Qt.ElideRight, column_width)
return len(elided.rstrip(""))
@SafeSlot(int, int, int)
def _on_section_resized(self, logical_index, old_size=None, new_size=None):
"""Only update rows if a wrapped column was resized."""
self._cache.clear()
self._update_row_heights()
def _update_row_heights(self):
"""Efficiently adjust row heights based on wrapped columns."""
view = self._parent
proxy = view.model()
model = proxy.sourceModel()
option = QtWidgets.QStyleOptionViewItem()
view.initViewItemOption(option)
# wrapping delegates
wrap_delegate_columns = []
for row in range(proxy.rowCount()):
max_height = 18
for column in [5, 6]: # TODO don't hardcode columns.. to be improved
index = proxy.index(row, column)
# model_index = proxy.mapToSource(index)
# delegate = view.itemDelegateForColumn(model_index) or view.itemDelegate()
delegate = view.itemDelegateForColumn(column)
hint = delegate.sizeHint(option, index)
max_height = max(max_height, hint.height())
if view.rowHeight(row) != max_height:
view.setRowHeight(row, max_height)
class CenterCheckBoxDelegate(CustomDisplayDelegate):
"""Custom checkbox delegate to center checkboxes in table cells."""
@@ -89,8 +199,9 @@ class CenterCheckBoxDelegate(CustomDisplayDelegate):
def apply_theme(self, theme: str | None = None):
colors = get_accent_colors()
self._icon_checked.setColor(colors.default)
self._icon_unchecked.setColor(colors.default)
_icon = partial(material_icon, size=(16, 16), color=colors.default, filled=True)
self._icon_checked = _icon("check_box")
self._icon_unchecked = _icon("check_box_outline_blank")
def _do_custom_paint(self, painter, option, index, value):
pixmap = self._icon_checked if value == Qt.CheckState.Checked else self._icon_unchecked
@@ -123,8 +234,12 @@ class DeviceValidatedDelegate(CustomDisplayDelegate):
def apply_theme(self, theme: str | None = None):
colors = get_accent_colors()
for status, icon in self._icons.items():
icon.setColor(colors[status])
_icon = partial(material_icon, icon_name="circle", size=(12, 12), filled=True)
self._icons = {
ValidationStatus.PENDING: _icon(color=colors.default),
ValidationStatus.VALID: _icon(color=colors.success),
ValidationStatus.FAILED: _icon(color=colors.emergency),
}
def _do_custom_paint(self, painter, option, index, value):
if pixmap := self._icons.get(value):
@@ -148,15 +263,19 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
self._device_config: list[dict[str, Any]] = []
self._validation_status: dict[str, ValidationStatus] = {}
self.headers = [
"",
"status",
"name",
"deviceClass",
"readoutPriority",
"onFailure",
"deviceTags",
"description",
"enabled",
"readOnly",
"softwareTrigger",
]
self._checkable_columns_enabled = {"enabled": True, "readOnly": True}
self._device_model_schema = Device.model_json_schema()
###############################################
########## Override custom Qt methods #########
@@ -172,6 +291,8 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
def headerData(self, section, orientation, role=int(Qt.ItemDataRole.DisplayRole)):
if role == Qt.ItemDataRole.DisplayRole and orientation == Qt.Orientation.Horizontal:
if section == 9: # softwareTrigger
return "softTrig"
return self.headers[section]
return None
@@ -192,20 +313,24 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
return self._validation_status.get(dev_name, ValidationStatus.PENDING)
key = self.headers[col]
value = self._device_config[row].get(key)
value = self._device_config[row].get(key, None)
if value is None:
value = (
self._device_model_schema.get("properties", {}).get(key, {}).get("default", None)
)
if role == Qt.ItemDataRole.DisplayRole:
if key in ("enabled", "readOnly"):
if key in ("enabled", "readOnly", "softwareTrigger"):
return bool(value)
if key == "deviceTags":
return ", ".join(str(tag) for tag in value) if value else ""
if key == "deviceClass":
return str(value).split(".")[-1]
return str(value) if value is not None else ""
if role == USER_CHECK_DATA_ROLE and key in ("enabled", "readOnly"):
if role == USER_CHECK_DATA_ROLE and key in ("enabled", "readOnly", "softwareTrigger"):
return Qt.CheckState.Checked if value else Qt.CheckState.Unchecked
if role == Qt.ItemDataRole.TextAlignmentRole:
if key in ("enabled", "readOnly"):
if key in ("enabled", "readOnly", "softwareTrigger"):
return Qt.AlignmentFlag.AlignCenter
return Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter
if role == Qt.ItemDataRole.FontRole:
@@ -223,7 +348,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable | Qt.ItemFlag.ItemIsDropEnabled
)
if key in ("enabled", "readOnly"):
if key in ("enabled", "readOnly", "softwareTrigger"):
if self._checkable_columns_enabled.get(key, True):
return base_flags | Qt.ItemFlag.ItemIsUserCheckable
else:
@@ -245,7 +370,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
if not index.isValid():
return False
key = self.headers[index.column()]
if key in ("enabled", "readOnly") and role == USER_CHECK_DATA_ROLE:
if key in ("enabled", "readOnly", "softwareTrigger") and role == USER_CHECK_DATA_ROLE:
if not self._checkable_columns_enabled.get(key, True):
return False # ignore changes if column is disabled
self._device_config[index.row()][key] = value == Qt.CheckState.Checked
@@ -301,6 +426,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
if self._name_exists_in_config(name := cfg.get("name", "<not found>"), True):
logger.warning(f"Device {name} already exists in the model.")
already_in_list.append(name)
# TODO add a warning that some devices were already in the list, how is this handled...
continue
row = len(self._device_config)
self.beginInsertRows(QtCore.QModelIndex(), row, row)
@@ -489,6 +615,12 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
self._filter_text = ""
self._enable_fuzzy = True
self._filter_columns = [1, 2] # name and deviceClass for search
# TODO refactor if enums are changed!!
self._status_order = {
ValidationStatus.VALID: 0,
ValidationStatus.PENDING: 1,
ValidationStatus.FAILED: 2,
}
def get_row_data(self, rows: Iterable[QModelIndex]) -> Iterable[dict[str, Any]]:
return (self.sourceModel().get_row_data(self.mapToSource(idx)) for idx in rows)
@@ -506,6 +638,14 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
self._hidden_rows.update(row_indices)
self.invalidateFilter()
def lessThan(self, left, right):
"""Add custom sorting for the status column"""
if left.column() != 0 or right.column() != 0:
return super().lessThan(left, right)
left_data = self.sourceModel().data(left, Qt.ItemDataRole.DisplayRole)
right_data = self.sourceModel().data(right, Qt.ItemDataRole.DisplayRole)
return self._status_order.get(left_data, 99) < self._status_order.get(right_data, 99)
def show_rows(self, row_indices: list[int]):
"""
Show specific rows in the model.
@@ -602,6 +742,21 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
# Connect signals
self._model.configs_changed.connect(self.device_configs_changed.emit)
def get_help_md(self) -> str:
"""
Generate Markdown help for a cell or header.
"""
pos = self.table.mapFromGlobal(QtGui.QCursor.pos())
model: DeviceTableModel = self._model # access underlying model
index = self.table.indexAt(pos)
if index.isValid():
column = index.column()
label = model.headerData(column, QtCore.Qt.Horizontal, QtCore.Qt.DisplayRole)
if label == "softTrig":
label = "softwareTrigger"
return HEADERS_HELP_MD.get(label, "")
return ""
def _setup_search(self):
"""Create components related to the search functionality"""
@@ -653,15 +808,20 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
self.checkbox_delegate = CenterCheckBoxDelegate(self.table, colors=colors)
self.tool_tip_delegate = DictToolTipDelegate(self.table)
self.validated_delegate = DeviceValidatedDelegate(self.table, colors=colors)
self.table.setItemDelegateForColumn(0, self.validated_delegate) # ValidationStatus
self.wrapped_delegate = WrappingTextDelegate(self.table, max_width=300)
# Add resize handling for wrapped delegate
header = self.table.horizontalHeader()
self.table.setItemDelegateForColumn(0, self.validated_delegate) # status
self.table.setItemDelegateForColumn(1, self.tool_tip_delegate) # name
self.table.setItemDelegateForColumn(2, self.tool_tip_delegate) # deviceClass
self.table.setItemDelegateForColumn(3, self.tool_tip_delegate) # readoutPriority
self.table.setItemDelegateForColumn(
4, self.tool_tip_delegate
) # deviceTags (was wrap_delegate)
self.table.setItemDelegateForColumn(5, self.checkbox_delegate) # enabled
self.table.setItemDelegateForColumn(6, self.checkbox_delegate) # readOnly
self.table.setItemDelegateForColumn(4, self.tool_tip_delegate) # onFailure
self.table.setItemDelegateForColumn(5, self.wrapped_delegate) # deviceTags
self.table.setItemDelegateForColumn(6, self.wrapped_delegate) # description
self.table.setItemDelegateForColumn(7, self.checkbox_delegate) # enabled
self.table.setItemDelegateForColumn(8, self.checkbox_delegate) # readOnly
self.table.setItemDelegateForColumn(9, self.checkbox_delegate) # softwareTrigger
# Disable wrapping, use eliding, and smooth scrolling
self.table.setWordWrap(False)
@@ -675,19 +835,35 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
header.setSectionResizeMode(1, QHeaderView.ResizeMode.Interactive) # name
header.setSectionResizeMode(2, QHeaderView.ResizeMode.Interactive) # deviceClass
header.setSectionResizeMode(3, QHeaderView.ResizeMode.Interactive) # readoutPriority
header.setSectionResizeMode(4, QHeaderView.ResizeMode.Stretch) # deviceTags: expand to fill
header.setSectionResizeMode(5, QHeaderView.ResizeMode.Fixed) # enabled
header.setSectionResizeMode(6, QHeaderView.ResizeMode.Fixed) # readOnly
header.setSectionResizeMode(4, QHeaderView.ResizeMode.Interactive) # onFailure
header.setSectionResizeMode(
5, QHeaderView.ResizeMode.Interactive
) # deviceTags: expand to fill
header.setSectionResizeMode(6, QHeaderView.ResizeMode.Stretch) # descript: expand to fill
header.setSectionResizeMode(7, QHeaderView.ResizeMode.Fixed) # enabled
header.setSectionResizeMode(8, QHeaderView.ResizeMode.Fixed) # readOnly
header.setSectionResizeMode(9, QHeaderView.ResizeMode.Fixed) # softwareTrigger
self.table.setColumnWidth(0, 25)
self.table.setColumnWidth(5, 70)
self.table.setColumnWidth(6, 70)
self.table.setColumnWidth(0, 70)
self.table.setColumnWidth(5, 200)
self.table.setColumnWidth(6, 200)
self.table.setColumnWidth(7, 70)
self.table.setColumnWidth(8, 70)
self.table.setColumnWidth(9, 70)
# Ensure column widths stay fixed
header.setMinimumSectionSize(25)
header.setDefaultSectionSize(90)
header.setStretchLastSection(False)
# Resize policy for wrapped text delegate
self._resize_proxy = BECSignalProxy(
header.sectionResized,
rateLimit=25,
slot=self.wrapped_delegate._on_section_resized,
timeout=1.0,
)
# Selection behavior
self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
@@ -322,28 +322,36 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
self.validation_msg_md.emit("")
def _format_markdown_text(self, device_name: str, raw_msg: str) -> str:
"""Simple HTML formatting for validation messages, wrapping text naturally."""
if not raw_msg.strip():
return f"### Validation in progress for {device_name}... \n\n"
if raw_msg == "Validation in progress...":
"""
Simple HTML formatting for validation messages, wrapping text naturally.
Args:
device_name (str): The name of the device.
raw_msg (str): The raw validation message.
"""
if not raw_msg.strip() or raw_msg.strip() == "Validation in progress...":
return f"### Validation in progress for {device_name}... \n\n"
m = re.search(r"ERROR:\s*([^\s]+)\s+is not valid:\s*(.+?errors?)", raw_msg)
device, summary = m.group(1), m.group(2)
lines = [f"## Error for '{device}'", f"'{device}' is not valid: {summary}"]
# Find each field block: \n<field>\n Field required ...
field_pat = re.compile(
r"\n(?P<field>\w+)\n\s+(?P<rest>Field required.*?(?=\n\w+\n|$))", re.DOTALL
# Regex to capture repeated ERROR patterns
pat = re.compile(
r"ERROR:\s*(?P<device>[^\s]+)\s+"
r"(?P<status>is not valid|is not connectable|failed):\s*"
r"(?P<detail>.*?)(?=ERROR:|$)",
re.DOTALL,
)
blocks = []
for m in pat.finditer(raw_msg):
dev = m.group("device")
status = m.group("status")
detail = m.group("detail").strip()
lines = [f"## Error for {dev}", f"**{dev} {status}**", f"```\n{detail}\n```"]
blocks.append("\n\n".join(lines))
for m in field_pat.finditer(raw_msg):
field = m.group("field")
rest = m.group("rest").rstrip()
lines.append(f"### {field}")
lines.append(rest)
# Fallback: If no patterns matched, return the raw message
if not blocks:
return f"## Error for {device_name}\n```\n{raw_msg.strip()}\n```"
return "\n".join(lines)
return "\n\n---\n\n".join(blocks)
def validation_running(self):
return self._device_list_items != {}
@@ -386,7 +394,7 @@ if __name__ == "__main__":
layout.setSpacing(0)
device_manager_ophyd_test = DMOphydTest()
try:
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/endstation.yaml"
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/first_light.yaml"
config = [{"name": k, **v} for k, v in yaml_load(config_path).items()]
except Exception as e:
logger.error(f"Error loading config: {e}")
+81
View File
@@ -0,0 +1,81 @@
# pylint: disable=missing-function-docstring, missing-module-docstring, unused-import
import pytest
from qtpy import QtCore, QtWidgets
from bec_widgets.utils.help_inspector.help_inspector import HelpInspector
from bec_widgets.widgets.control.buttons.button_abort.button_abort import AbortButton
from .client_mocks import mocked_client
@pytest.fixture
def help_inspector(qtbot, mocked_client):
widget = HelpInspector(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
@pytest.fixture
def abort_button(qtbot):
widget = AbortButton()
widget.setToolTip("This is an abort button.")
def get_help_md():
return "This is **markdown** help text for the abort button."
widget.get_help_md = get_help_md # type: ignore
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_help_inspector_button(help_inspector):
"""Test the HelpInspector widget."""
assert not help_inspector._active
help_inspector._button.click()
assert help_inspector._active
assert help_inspector._button.isChecked()
cursor = QtWidgets.QApplication.overrideCursor()
assert cursor is not None
assert cursor.shape() == QtCore.Qt.CursorShape.WhatsThisCursor
help_inspector._button.click()
assert not help_inspector._active
assert not help_inspector._button.isChecked()
assert QtWidgets.QApplication.overrideCursor() is None
def test_help_inspector_register_callback(help_inspector):
"""Test registering a callback in the HelpInspector widget."""
assert len(help_inspector._callbacks) == 3 # default callbacks
def my_callback(widget):
pass
cb_id = help_inspector.register_callback(my_callback)
assert len(help_inspector._callbacks) == 4
assert help_inspector._callbacks[cb_id] == my_callback
cb_id2 = help_inspector.register_callback(my_callback)
assert len(help_inspector._callbacks) == 5
assert help_inspector._callbacks[cb_id2] == my_callback
help_inspector.unregister_callback(cb_id)
assert len(help_inspector._callbacks) == 4
help_inspector.unregister_callback(cb_id2)
assert len(help_inspector._callbacks) == 3
def test_help_inspector_escape_key(qtbot, help_inspector):
"""Test that pressing the Escape key deactivates the HelpInspector."""
help_inspector._button.click()
assert help_inspector._active
qtbot.keyClick(help_inspector, QtCore.Qt.Key.Key_Escape)
assert not help_inspector._active
assert not help_inspector._button.isChecked()
assert QtWidgets.QApplication.overrideCursor() is None