bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-12-31 11:11:17 +01:00
Code Activity

refactor(device-manager-view): cleanup, minor bugfixes

Browse Source
This commit is contained in:
appel_c
2025-10-13 07:47:03 +02:00
committed by Christian Appel
parent 3522caba67
commit bc77d3199e
4 changed files with 212 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
bec_widgets/applications/views/device_manager_view/device_manager_view.py
View File

@@ -143,6 +143,9 @@ class ConfigChoiceDialog(QDialog):
return self._result
AVAILABLE_RESOURCE_IS_READY = False
class DeviceManagerView(BECWidget, QWidget):
def __init__(self, parent=None, *args, **kwargs):
@@ -159,15 +162,6 @@ 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)
# Device Table View widget
self.device_table_view = DeviceTableView(
self, shared_selection_signal=self._shared_selection
@@ -204,11 +198,7 @@ class DeviceManagerView(BECWidget, QWidget):
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
# Register callback
self.help_inspector.bec_widget_help.connect(text_box.setMarkdown)
# Error Logs View
@@ -247,57 +237,63 @@ class DeviceManagerView(BECWidget, QWidget):
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
dock.setFeature(CDockWidget.DockWidgetClosable, False)
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():
# 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], [7, 3])
# self.set_default_view([2, 8, 2], [2, 2, 4])
# Connect slots
for signal, slots in [
(
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.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,),
),
]:
for slot in slots:
signal.connect(slot)
# Once available resource is ready, add it to the view again
if AVAILABLE_RESOURCE_IS_READY:
# 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)
# Connect slots for available reosource
for signal, slots in [
(
self.available_devices.selected_devices,
(self.dm_config_view.on_select_config, self.dm_docs_view.on_select_config),
),
(
self.device_table_view.device_configs_changed,
(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)
# Add toolbar
self._add_toolbar()
def _add_toolbar(self):
@@ -421,13 +417,6 @@ class DeviceManagerView(BECWidget, QWidget):
# Add load config from plugin dir
self.toolbar.add_bundle(table_bundle)
# Most likly, no actions on available devices
# Actions (vielleicht bundle fuer available devices )
# - reset composed view
# - add new device (EpicsMotor, EpicsMotorECMC, EpicsSignal, CustomDevice)
# - remove device
# - rerun validation (with/without connect)
# IO actions
def _coming_soon(self):
return QMessageBox.question(
@@ -441,7 +430,6 @@ class DeviceManagerView(BECWidget, QWidget):
@SafeSlot()
def _load_file_action(self):
"""Action for the 'load' action to load a config from disk for the io_bundle of the toolbar."""
# Check if plugin repo is installed...
try:
plugin_path = plugin_repo_path()
plugin_name = plugin_package_name()
@@ -542,7 +530,6 @@ class DeviceManagerView(BECWidget, QWidget):
file.write(yaml.dump(config))
# Table actions
@SafeSlot()
def _reset_composed_view(self):
"""Action for the 'reset_composed_view' action to reset the composed view."""
@@ -554,13 +541,10 @@ class DeviceManagerView(BECWidget, QWidget):
if reply == QMessageBox.StandardButton.Yes:
self.device_table_view.clear_device_configs()
# TODO We want to have a combobox to choose from EpicsMotor, EpicsMotorECMC, EpicsSignal, EpicsSignalRO, and maybe EpicsSignalWithRBV and custom Device
# For all default Epics devices, we would like to preselect relevant fields, and prompt them with the proper deviceConfig args already, i.e. 'prefix', 'read_pv', 'write_pv' etc..
# For custom Device, they should receive all options. It might be cool to get a side panel with docstring view of the class upon inspecting it to make it easier in case deviceConfig entries are required..
# TODO Bespoke Form to add a new device
@SafeSlot()
def _add_device_action(self):
"""Action for the 'add_device' action to add a new device."""
# Implement the logic to add a new device
dialog = PresetClassDeviceConfigDialog(parent=self)
dialog.accepted_data.connect(self._add_to_table_from_dialog)
dialog.open()
@@ -574,13 +558,12 @@ class DeviceManagerView(BECWidget, QWidget):
"""Action for the 'remove_device' action to remove a device."""
self.device_table_view.remove_selected_rows()
# TODO implement proper logic for validation. We should also carefully review how these jobs update the table, and how we can cancel pending validations
# in case they are no longer relevant. We might want to 'block' the interactivity on the items for which validation runs with 'connect'!
@SafeSlot()
def _rerun_validation_action(self):
@SafeSlot(bool)
def _rerun_validation_action(self, connect: bool = True):
"""Action for the 'rerun_validation' action to rerun validation on selected devices."""
configs = self.device_table_view.table.selected_configs()
self.ophyd_test_view.change_device_configs(configs, True, True)
self.ophyd_test_view.change_device_configs(configs, True, connect)
####### Default view has to be done with setting up splitters ########
def set_default_view(

97
bec_widgets/widgets/control/device_manager/components/constants.py
View File

@@ -9,55 +9,64 @@ CONFIG_DATA_ROLE: Final[int] = 118
# TODO 882 keep in sync with headers in device_table_view.py
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"
"status": "\n".join(
[
"## Status",
"The current status of the device. Can be one of the following values: ",
"### **LOADED** \n The device with the specified configuration is loaded in the current config.",
"### **CONNECT_READY** \n The device config is valid and the connection has been validated. It has not yet been loaded to the current config.",
"### **CONNECT_FAILED** \n The device config is valid, but the connection could not be established.",
"### **VALID** \n The device config is valid, but the connection has not yet been validated.",
"### **INVALID** \n The device config is invalid and can not be loaded to the current config.",
]
),
"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."
"name": "\n".join(["## Name ", "The name of the device."]),
"deviceClass": "\n".join(
[
"## Device Class",
"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"
"readoutPriority": "\n".join(
[
"## Readout Priority",
"The readout priority of the device. Can be one of the following values: ",
"### **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.",
"### **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.",
"### **async** \n The async priority is used for devices that are asynchronous to the monitored devices, and send their data independently.",
"### **continuous** \n The continuous priority is used for devices that are read out continuously during the scan.",
"### **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.",
]
),
"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."
"deviceTags": "\n".join(
[
"## Device Tags",
"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."
"enabled": "\n".join(
[
"## Enabled",
"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"
"readOnly": "\n".join(
["## Read Only", "Indicator that a device is read-only or can be modified."]
),
"softwareTrigger": (
"## Software Trigger"
"\n"
"Indicator whether the device receives a software trigger from BEC during a scan."
"onFailure": "\n".join(
[
"## On Failure",
"Specifies the behavior of the device in case of a failure. Can be one of the following values: ",
"### **buffer** \n The device readback will fall back to the last known value.",
"### **retry** \n The device readback will be retried once, and raises an error if it fails again.",
"### **raise** \n The device readback will raise immediately.",
]
),
"description": ("## Description" "\n" "A short description of the device."),
"softwareTrigger": "\n".join(
[
"## Software Trigger",
"Indicator whether the device receives a software trigger from BEC during a scan.",
]
),
"description": "\n".join(["## Description", "A short description of the device."]),
}

151
bec_widgets/widgets/control/device_manager/components/device_table_view.py
View File

@@ -7,7 +7,7 @@ import json
import textwrap
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, Iterable, List
from typing import TYPE_CHECKING, Any, Iterable, List, Literal
from uuid import uuid4
from bec_lib.atlas_models import Device
@@ -46,7 +46,13 @@ USER_CHECK_DATA_ROLE = 101
class DictToolTipDelegate(QtWidgets.QStyledItemDelegate):
"""Delegate that shows all key-value pairs of a rows's data as a YAML-like tooltip."""
def helpEvent(self, event, view, option, index):
def helpEvent(
self,
event: QtCore.QEvent,
view: QtWidgets.QAbstractItemView,
option: QtWidgets.QStyleOptionViewItem,
index: QModelIndex,
):
"""Override to show tooltip when hovering."""
if event.type() != QtCore.QEvent.Type.ToolTip:
return super().helpEvent(event, view, option, index)
@@ -64,13 +70,23 @@ class CustomDisplayDelegate(DictToolTipDelegate):
def displayText(self, value: Any, locale: QtCore.QLocale | QtCore.QLocale.Language) -> str:
return ""
def _test_custom_paint(self, painter, option, index):
def _test_custom_paint(
self, painter: QtGui.QPainter, option: QtWidgets.QStyleOptionViewItem, index: QModelIndex
):
v = index.model().data(index, self._paint_test_role)
return (v is not None), v
def _do_custom_paint(self, painter, option, index, value): ...
def _do_custom_paint(
self,
painter: QtGui.QPainter,
option: QtWidgets.QStyleOptionViewItem,
index: QModelIndex,
value: Any,
): ...
def paint(self, painter, option, index) -> None:
def paint(
self, painter: QtGui.QPainter, option: QtWidgets.QStyleOptionViewItem, index: QModelIndex
) -> None:
(check, value) = self._test_custom_paint(painter, option, index)
if not check:
return super().paint(painter, option, index)
@@ -83,14 +99,34 @@ class CustomDisplayDelegate(DictToolTipDelegate):
class WrappingTextDelegate(CustomDisplayDelegate):
"""A lightweight delegate that wraps text without expensive size recalculation."""
def __init__(self, parent=None, max_width=300, margin=6):
def __init__(self, parent: BECTableView | None = None, max_width: int = 300, margin: int = 6):
super().__init__(parent)
self._parent = parent
self.max_width = max_width
self.margin = margin
self._cache = {} # cache text metrics for performance
self._wrapping_text_columns = None
def _do_custom_paint(self, painter, option, index, value: str):
@property
def wrapping_text_columns(self) -> List[int]:
# Compute once, cache for later
if self._wrapping_text_columns is None:
self._wrapping_text_columns = []
view = self._parent
proxy: DeviceFilterProxyModel = self._parent.model()
for col in range(proxy.columnCount()):
delegate = view.itemDelegateForColumn(col)
if isinstance(delegate, WrappingTextDelegate):
self._wrapping_text_columns.append(col)
return self._wrapping_text_columns
def _do_custom_paint(
self,
painter: QtGui.QPainter,
option: QtWidgets.QStyleOptionViewItem,
index: QModelIndex,
value: str,
):
text = str(value)
if not text:
return
@@ -101,30 +137,39 @@ class WrappingTextDelegate(CustomDisplayDelegate):
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()
layout = self._compute_layout(text, option)
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):
def _compute_layout(
self, text: str, option: QtWidgets.QStyleOptionViewItem
) -> QtGui.QTextLayout:
"""Compute and return the text layout for given text and option."""
layout = self._get_layout(text, option.font)
layout.beginLayout()
height = 0
max_lines = 100 # safety cap, should never be more than 100 lines..
for _ in range(max_lines):
line = layout.createLine()
if not line.isValid():
break
line.setLineWidth(option.rect.width() - self.margin)
line.setPosition(QtCore.QPointF(self.margin / 2, height))
line_height = line.height()
if line_height <= 0:
break # avoid negative or zero height lines to be added
height += line_height
layout.endLayout()
return layout
def _get_layout(self, text: str, font_option: QtGui.QFont) -> QtGui.QTextLayout:
return QtGui.QTextLayout(text, font_option)
def sizeHint(self, option: QtWidgets.QStyleOptionViewItem, index: QModelIndex) -> QtCore.QSize:
"""Return a cached or approximate height; avoids costly recomputation."""
text = str(index.data(QtCore.Qt.DisplayRole) or "")
view = self._parent
@@ -151,14 +196,19 @@ class WrappingTextDelegate(CustomDisplayDelegate):
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:
def estimate_chars_per_line(
self, text: str, option: QtWidgets.QStyleOptionViewItem, 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):
@SafeSlot(int)
def _on_section_resized(
self, logical_index: int, old_size: int | None = None, new_size: int | None = None
):
"""Only update rows if a wrapped column was resized."""
self._cache.clear()
self._update_row_heights()
@@ -167,17 +217,12 @@ class WrappingTextDelegate(CustomDisplayDelegate):
"""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 884 don't hardcode columns.. to be improved
for column in self.wrapping_text_columns:
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())
@@ -190,7 +235,7 @@ class CenterCheckBoxDelegate(CustomDisplayDelegate):
_paint_test_role = USER_CHECK_DATA_ROLE
def __init__(self, parent=None, colors=None):
def __init__(self, parent: BECTableView | None = None, colors: AccentColors | None = None):
super().__init__(parent)
colors: AccentColors = colors if colors else get_accent_colors() # type: ignore
_icon = partial(material_icon, size=(16, 16), color=colors.default, filled=True)
@@ -203,13 +248,27 @@ class CenterCheckBoxDelegate(CustomDisplayDelegate):
self._icon_checked = _icon("check_box")
self._icon_unchecked = _icon("check_box_outline_blank")
def _do_custom_paint(self, painter, option, index, value):
def _do_custom_paint(
self,
painter: QtGui.QPainter,
option: QtWidgets.QStyleOptionViewItem,
index: QModelIndex,
value: Literal[
Qt.CheckState.Checked | Qt.CheckState.Unchecked | Qt.CheckState.PartiallyChecked
],
):
pixmap = self._icon_checked if value == Qt.CheckState.Checked else self._icon_unchecked
pix_rect = pixmap.rect()
pix_rect.moveCenter(option.rect.center())
painter.drawPixmap(pix_rect.topLeft(), pixmap)
def editorEvent(self, event, model, option, index):
def editorEvent(
self,
event: QtCore.QEvent,
model: QtCore.QSortFilterProxyModel,
option: QtWidgets.QStyleOptionViewItem,
index: QModelIndex,
):
if event.type() != QtCore.QEvent.Type.MouseButtonRelease:
return False
current = model.data(index, USER_CHECK_DATA_ROLE)
@@ -222,7 +281,7 @@ class CenterCheckBoxDelegate(CustomDisplayDelegate):
class DeviceValidatedDelegate(CustomDisplayDelegate):
"""Custom delegate for displaying validated device configurations."""
def __init__(self, parent=None, colors=None):
def __init__(self, parent: BECTableView | None = None, colors: AccentColors | None = None):
super().__init__(parent)
colors = colors if colors else get_accent_colors()
_icon = partial(material_icon, icon_name="circle", size=(12, 12), filled=True)
@@ -241,7 +300,23 @@ class DeviceValidatedDelegate(CustomDisplayDelegate):
ValidationStatus.FAILED: _icon(color=colors.emergency),
}
def _do_custom_paint(self, painter, option, index, value):
def _do_custom_paint(
self,
painter: QtGui.QPainter,
option: QtWidgets.QStyleOptionViewItem,
index: QModelIndex,
value: Literal[0, 1, 2],
):
"""
Paint the validation status icon centered in the cell.
Args:
painter (QtGui.QPainter): The painter object.
option (QtWidgets.QStyleOptionViewItem): The style options for the item.
index (QModelIndex): The model index of the item.
value (Literal[0,1,2]): The validation status value, where 0=Pending, 1=Valid, 2=Failed.
Relates to ValidationStatus enum.
"""
if pixmap := self._icons.get(value):
pix_rect = pixmap.rect()
pix_rect.moveCenter(option.rect.center())
@@ -258,7 +333,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
# tuple of list[dict[str, Any]] of configs which were added and bool True if added or False if removed
configs_changed = QtCore.Signal(list, bool)
def __init__(self, parent=None):
def __init__(self, parent: DeviceTableModel | None = None):
super().__init__(parent)
self._device_config: list[dict[str, Any]] = []
self._validation_status: dict[str, ValidationStatus] = {}

2
bec_widgets/widgets/control/device_manager/components/dm_docstring_view.py
View File

@@ -43,7 +43,7 @@ def docstring_to_markdown(obj) -> str:
# Highlight section headers for Markdown
headers = ["Parameters", "Args", "Returns", "Raises", "Attributes", "Examples", "Notes"]
for h in headers:
doc = re.sub(rf"(?m)^({h})\s*:?\s*$", rf"### \1", text)
text = re.sub(rf"(?m)^({h})\s*:?\s*$", rf"### \1", text)
# Preserve code blocks (4+ space indented lines)
def fence_code(match: re.Match) -> str: