Compare commits

...

6 Commits

Author SHA1 Message Date
perl_d fb452fe9b0 wip sorting list 2025-09-01 16:41:00 +02:00
perl_d 495405240e wip 2025-09-01 15:30:10 +02:00
perl_d 42d93f9446 refactor: make sorting more generic 2025-09-01 15:25:15 +02:00
perl_d c06eb36701 feat: add available devices widget 2025-09-01 15:25:15 +02:00
perl_d 91988c4fef feat: add ListOfExpandableFrames util 2025-09-01 15:24:51 +02:00
appel_c 787bfe09ef feat(dm-view): initial commit for config_view, ophyd_test and dm_widget 2025-09-01 15:24:51 +02:00
20 changed files with 1653 additions and 178 deletions
@@ -0,0 +1,214 @@
from typing import List
import PySide6QtAds as QtAds
import yaml
from bec_qthemes import material_icon
from PySide6QtAds import CDockManager, CDockWidget
from qtpy.QtCore import Qt, QTimer
from qtpy.QtWidgets import (
QPushButton,
QSizePolicy,
QSplitter,
QStackedLayout,
QTreeWidget,
QVBoxLayout,
QWidget,
)
from bec_widgets import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.containers.advanced_dock_area.advanced_dock_area import AdvancedDockArea
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources import (
AvailableDeviceResources,
)
from bec_widgets.widgets.control.device_manager.components.device_table_view import DeviceTableView
from bec_widgets.widgets.control.device_manager.components.dm_config_view import DMConfigView
from bec_widgets.widgets.control.device_manager.components.dm_ophyd_test import (
DeviceManagerOphydTest,
)
from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget
from bec_widgets.widgets.editors.web_console.web_console import WebConsole
from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatusBox
from bec_widgets.widgets.utility.ide_explorer.ide_explorer import IDEExplorer
def set_splitter_weights(splitter: QSplitter, weights: List[float]) -> None:
"""
Apply initial sizes to a splitter using weight ratios, e.g. [1,3,2,1].
Works for horizontal or vertical splitters and sets matching stretch factors.
"""
def apply():
n = splitter.count()
if n == 0:
return
w = list(weights[:n]) + [1] * max(0, n - len(weights))
w = [max(0.0, float(x)) for x in w]
tot_w = sum(w)
if tot_w <= 0:
w = [1.0] * n
tot_w = float(n)
total_px = (
splitter.width() if splitter.orientation() == Qt.Horizontal else splitter.height()
)
if total_px < 2:
QTimer.singleShot(0, apply)
return
sizes = [max(1, int(total_px * (wi / tot_w))) for wi in w]
diff = total_px - sum(sizes)
if diff != 0:
idx = max(range(n), key=lambda i: w[i])
sizes[idx] = max(1, sizes[idx] + diff)
splitter.setSizes(sizes)
for i, wi in enumerate(w):
splitter.setStretchFactor(i, max(1, int(round(wi * 100))))
QTimer.singleShot(0, apply)
class DeviceManagerView(BECWidget, QWidget):
def __init__(self, parent=None, *args, **kwargs):
super().__init__(parent, *args, **kwargs)
# Top-level layout hosting a toolbar and the dock manager
self._root_layout = QVBoxLayout(self)
self._root_layout.setContentsMargins(0, 0, 0, 0)
self._root_layout.setSpacing(0)
self.dock_manager = CDockManager(self)
self._root_layout.addWidget(self.dock_manager)
# Initialize the widgets
self.available_devices = AvailableDeviceResources(self)
self.device_table_view = DeviceTableView(self)
# Placeholder
self.dm_config_view = DMConfigView(self)
# Placeholder for ophyd test
WebConsole.startup_cmd = "ipython"
self.ophyd_test = DeviceManagerOphydTest(self)
self.ophyd_test_dock = QtAds.CDockWidget("Ophyd Test", self)
self.ophyd_test_dock.setWidget(self.ophyd_test)
# Create the dock widgets
self.available_devices_dock = QtAds.CDockWidget("Explorer", self)
self.available_devices_dock.setWidget(self.available_devices)
self.device_table_view_dock = QtAds.CDockWidget("Device Table", self)
self.device_table_view_dock.setWidget(self.device_table_view)
# Device Table will be central widget
self.dock_manager.setCentralWidget(self.device_table_view_dock)
self.dm_config_view_dock = QtAds.CDockWidget("YAML Editor", self)
self.dm_config_view_dock.setWidget(self.dm_config_view)
# Add the dock widgets to the dock manager
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.LeftDockWidgetArea, self.available_devices_dock
)
monaco_yaml_area = self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.RightDockWidgetArea, self.dm_config_view_dock
)
self.dock_manager.addDockWidget(
QtAds.DockWidgetArea.BottomDockWidgetArea, self.ophyd_test_dock, monaco_yaml_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)
# 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)
# Apply stretch after the layout is done
self.set_default_view([2, 5, 3], [5, 5])
# Connect slots
self.device_table_view.selected_device.connect(self.dm_config_view.on_select_config)
self.device_table_view.model.devices_reset.connect(
self.available_devices.update_devices_state
)
####### Default view has to be done with setting up splitters ########
def set_default_view(self, horizontal_weights: list, vertical_weights: list):
"""Apply initial weights to every horizontal and vertical splitter.
Examples:
horizontal_weights = [1, 3, 2, 1]
vertical_weights = [3, 7] # top:bottom = 30:70
"""
splitters_h = []
splitters_v = []
for splitter in self.findChildren(QSplitter):
if splitter.orientation() == Qt.Horizontal:
splitters_h.append(splitter)
elif splitter.orientation() == Qt.Vertical:
splitters_v.append(splitter)
def apply_all():
for s in splitters_h:
set_splitter_weights(s, horizontal_weights)
for s in splitters_v:
set_splitter_weights(s, vertical_weights)
QTimer.singleShot(0, apply_all)
def set_stretch(self, *, horizontal=None, vertical=None):
"""Update splitter weights and re-apply to all splitters.
Accepts either a list/tuple of weights (e.g., [1,3,2,1]) or a role dict
for convenience: horizontal roles = {"left","center","right"},
vertical roles = {"top","bottom"}.
"""
def _coerce_h(x):
if x is None:
return None
if isinstance(x, (list, tuple)):
return list(map(float, x))
if isinstance(x, dict):
return [
float(x.get("left", 1)),
float(x.get("center", x.get("middle", 1))),
float(x.get("right", 1)),
]
return None
def _coerce_v(x):
if x is None:
return None
if isinstance(x, (list, tuple)):
return list(map(float, x))
if isinstance(x, dict):
return [float(x.get("top", 1)), float(x.get("bottom", 1))]
return None
h = _coerce_h(horizontal)
v = _coerce_v(vertical)
if h is None:
h = [1, 1, 1]
if v is None:
v = [1, 1]
self.set_default_view(h, v)
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
device_manager_view = DeviceManagerView()
config = device_manager_view.client.device_manager._get_redis_device_config()
device_manager_view.device_table_view.set_device_config(config)
device_manager_view.show()
device_manager_view.setWindowTitle("Device Manager View")
device_manager_view.resize(1200, 800)
# developer_view.set_stretch(horizontal=[1, 3, 2], vertical=[5, 5]) #can be set during runtime
sys.exit(app.exec_())
@@ -0,0 +1,73 @@
"""Top Level wrapper for device_manager widget"""
from __future__ import annotations
from bec_qthemes import material_icon
from qtpy import QtCore, QtWidgets
from bec_widgets.examples.device_manager_view.device_manager_view import DeviceManagerView
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
class DeviceManagerWidget(BECWidget, QtWidgets.QWidget):
def __init__(self, parent=None, client=None):
super().__init__(client=client, parent=parent)
self.stacked_layout = QtWidgets.QStackedLayout()
self.stacked_layout.setContentsMargins(0, 0, 0, 0)
self.stacked_layout.setSpacing(0)
self.stacked_layout.setStackingMode(QtWidgets.QStackedLayout.StackAll)
self.setLayout(self.stacked_layout)
# Add device manager view
self.device_manager_view = DeviceManagerView()
self.stacked_layout.addWidget(self.device_manager_view)
# Add overlay widget
self._overlay_widget = QtWidgets.QWidget(self)
self._customize_overlay()
self.stacked_layout.addWidget(self._overlay_widget)
self.stacked_layout.setCurrentWidget(self._overlay_widget)
def _customize_overlay(self):
self._overlay_widget.setStyleSheet(
"background: qlineargradient(x1:0, y1:0, x2:0, y2:1,stop:0 #ffffff, stop:1 #e0e0e0);"
)
self._overlay_widget.setAutoFillBackground(True)
self._overlay_layout = QtWidgets.QVBoxLayout()
self._overlay_layout.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
self._overlay_widget.setLayout(self._overlay_layout)
self._overlay_widget.setSizePolicy(
QtWidgets.QSizePolicy.Policy.Expanding, QtWidgets.QSizePolicy.Policy.Expanding
)
self.button_load_current_config = QtWidgets.QPushButton("Load Current Config")
icon = material_icon(icon_name="database", size=(24, 24), convert_to_pixmap=False)
self.button_load_current_config.setIcon(icon)
self._overlay_layout.addWidget(self.button_load_current_config)
self.button_load_current_config.clicked.connect(self._load_config_clicked)
self._overlay_widget.setVisible(True)
@SafeSlot()
def _load_config_clicked(self):
"""Handle click on 'Load Current Config' button."""
config = self.client.device_manager._get_redis_device_config()
self.device_manager_view.device_table_view.set_device_config(config)
self.device_manager_view.ophyd_test.on_device_config_update(config)
self.stacked_layout.setCurrentWidget(self.device_manager_view)
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
device_manager = DeviceManagerWidget()
# config = device_manager.client.device_manager._get_redis_device_config()
# device_manager.device_table_view.set_device_config(config)
device_manager.show()
device_manager.setWindowTitle("Device Manager View")
device_manager.resize(1600, 1200)
# developer_view.set_stretch(horizontal=[1, 3, 2], vertical=[5, 5]) #can be set during runtime
sys.exit(app.exec_())
+26 -6
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
from bec_qthemes import material_icon
from qtpy.QtCore import Signal
from qtpy.QtCore import QSize, Signal
from qtpy.QtWidgets import (
QApplication,
QFrame,
@@ -19,7 +19,8 @@ from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
class ExpandableGroupFrame(QFrame):
broadcast_size_hint = Signal(QSize)
imminent_deletion = Signal()
expansion_state_changed = Signal()
EXPANDED_ICON_NAME: str = "collapse_all"
@@ -31,6 +32,7 @@ class ExpandableGroupFrame(QFrame):
super().__init__(parent=parent)
self._expanded = expanded
self._title_text = f"<b>{title}</b>"
self.setFrameStyle(QFrame.Shape.StyledPanel | QFrame.Shadow.Plain)
self.setSizePolicy(QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Minimum)
self._layout = QVBoxLayout()
@@ -49,21 +51,27 @@ class ExpandableGroupFrame(QFrame):
def _create_title_layout(self, title: str, icon: str):
self._title_layout = QHBoxLayout()
self._layout.addLayout(self._title_layout)
self._internal_title_layout = QHBoxLayout()
self._title_layout.addLayout(self._internal_title_layout)
self._title = ClickableLabel(f"<b>{title}</b>")
self._title = ClickableLabel()
self._set_title_text(self._title_text)
self._title_icon = ClickableLabel()
self._title_layout.addWidget(self._title_icon)
self._title_layout.addWidget(self._title)
self._internal_title_layout.addWidget(self._title_icon)
self._internal_title_layout.addWidget(self._title)
self.icon_name = icon
self._title.clicked.connect(self.switch_expanded_state)
self._title_icon.clicked.connect(self.switch_expanded_state)
self._title_layout.addStretch(1)
self._internal_title_layout.addStretch(1)
self._expansion_button = QToolButton()
self._update_expansion_icon()
self._title_layout.addWidget(self._expansion_button, stretch=1)
def get_title_layout(self) -> QHBoxLayout:
return self._internal_title_layout
def set_layout(self, layout: QLayout) -> None:
self._contents.setLayout(layout)
self._contents.layout().setContentsMargins(0, 0, 0, 0) # type: ignore
@@ -112,6 +120,18 @@ class ExpandableGroupFrame(QFrame):
else:
self._title_icon.setVisible(False)
@SafeProperty(str)
def title_text(self): # type: ignore
return self._title_text
@title_text.setter
def title_text(self, title_text: str):
self._title_text = title_text
self._set_title_text(self._title_text)
def _set_title_text(self, title_text: str):
self._title.setText(title_text)
# Application example
if __name__ == "__main__": # pragma: no cover
@@ -0,0 +1,133 @@
import re
from functools import partial
from re import Pattern
from typing import Generic, Iterable, NamedTuple, TypeVar
from bec_lib.logger import bec_logger
from more_itertools import consume
from PySide6.QtGui import QFont
from PySide6.QtWidgets import QListWidgetItem, QWidget
from qtpy.QtCore import QSize, Qt
from qtpy.QtWidgets import QListWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
logger = bec_logger.logger
_SORT_KEY_ROLE = 117
_EF = TypeVar("_EF", bound=ExpandableGroupFrame)
class ListOfExpandableFrames(QListWidget, Generic[_EF]):
def __init__(
self, /, parent: QWidget | None = None, item_class: type[_EF] = ExpandableGroupFrame
) -> None:
super().__init__(parent)
_Items = NamedTuple("_Items", (("item", QListWidgetItem), ("widget", _EF)))
self.item_tuple = _Items
self._item_class = item_class
self._item_dict: dict[str, _Items] = {}
def __contains__(self, id: str):
return id in self._item_dict
def clear(self) -> None:
self._item_dict = {}
return super().clear()
def add_item(self, id: str, *args, **kwargs) -> _EF:
"""Adds the specified type of widget as an item. args and kwargs are passed to the constructor.
Args:
id (str): the key under which to store the list item in the internal dict
Returns:
The widget created in the addition process
"""
def _remove_item(item: QListWidgetItem):
self.takeItem(self.row(item))
del self._item_dict[id]
self.sortItems()
def _updatesize(item: QListWidgetItem, item_widget: _EF):
item_widget.adjustSize()
item.setSizeHint(QSize(item_widget.width(), item_widget.height()))
item = QListWidgetItem(self)
item.setData(_SORT_KEY_ROLE, id) # used for sorting
item_widget = self._item_class(*args, **kwargs)
item_widget.expansion_state_changed.connect(partial(_updatesize, item, item_widget))
item_widget.imminent_deletion.connect(partial(_remove_item, item))
item_widget.broadcast_size_hint.connect(item.setSizeHint)
self.addItem(item)
self.setItemWidget(item, item_widget)
self._item_dict[id] = self.item_tuple(item, item_widget)
item.setSizeHint(item_widget.sizeHint())
return item_widget
def sort_by_key(self, role=_SORT_KEY_ROLE, order=Qt.SortOrder.AscendingOrder):
items = [self.takeItem(0) for i in range(self.count())]
items.sort(key=lambda it: it.data(role), reverse=(order == Qt.SortOrder.DescendingOrder))
for it in items:
self.addItem(it)
# reattach its custom widget
widget = self.itemWidget(it)
if widget:
self.setItemWidget(it, widget)
def item_widget_pairs(self):
return self._item_dict.values()
def widgets(self):
return (i.widget for i in self._item_dict.values())
def get_item_widget(self, id: str):
if (item := self._item_dict.get(id)) is None:
return None
return item
def set_hidden_pattern(self, pattern: Pattern):
self.hide_all()
self._set_hidden(filter(pattern.search, self._item_dict.keys()), False)
def set_hidden(self, ids: Iterable[str]):
self._set_hidden(ids, True)
def _set_hidden(self, ids: Iterable[str], hidden: bool):
for id in ids:
if (_item := self._item_dict.get(id)) is not None:
_item.item.setHidden(hidden)
_item.widget.setHidden(hidden)
else:
logger.warning(
f"List {self.__qualname__} does not have an item with ID {id} to hide!"
)
self.sortItems()
def hide_all(self):
self.set_hidden_state_on_all(True)
def unhide_all(self):
self.set_hidden_state_on_all(False)
def set_hidden_state_on_all(self, hidden: bool):
for _item in self._item_dict.values():
_item.item.setHidden(hidden)
_item.widget.setHidden(hidden)
self.sortItems()
@SafeSlot(str)
def update_filter(self, value: str):
if value == "":
return self.unhide_all()
try:
self.set_hidden_pattern(re.compile(value, re.IGNORECASE))
except Exception:
self.unhide_all()
@@ -0,0 +1,3 @@
from .available_device_resources import AvailableDeviceResources
__all__ = ["AvailableDeviceResources"]
@@ -0,0 +1,12 @@
from typing import Any, Callable, Generator, Iterable, TypeVar
_T = TypeVar("_T")
_RT = TypeVar("_RT")
def yield_only_passing(fn: Callable[[_T], _RT], vals: Iterable[_T]) -> Generator[_RT, Any, None]:
for v in vals:
try:
yield fn(v)
except BaseException:
pass
@@ -0,0 +1,200 @@
from textwrap import dedent
from typing import NamedTuple
from bec_qthemes import material_icon
from qtpy.QtCore import QSize
from qtpy.QtWidgets import QFrame, QHBoxLayout, QLabel, QListWidgetItem, QVBoxLayout, QWidget
from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group_ui import (
Ui_AvailableDeviceGroup,
)
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
HashableDevice,
)
def _warning_string(spec: HashableDevice):
name_warning = (
f"Device defined with multiple names! Please check:\n {'\n '.join(spec.names)}\n"
if len(spec.names) > 1
else ""
)
source_warning = (
f"Device found in multiple source files! Please check:\n {'\n '.join(spec._source_files)}"
if len(spec._source_files) > 1
else ""
)
return f"{name_warning}{source_warning}"
class _DeviceEntryWidget(QFrame):
def __init__(self, device_spec: HashableDevice, parent=None, **kwargs):
super().__init__(parent, **kwargs)
self._device_spec = device_spec
self.included: bool = False
self.setFrameShape(QFrame.Shape.StyledPanel)
self.setFrameShadow(QFrame.Shadow.Raised)
self._layout = QVBoxLayout()
self._layout.setContentsMargins(0, 0, 0, 0)
self.setLayout(self._layout)
self.setup_title_layout(device_spec)
self.check_and_display_warning()
self.setToolTip(self._rich_text())
def _rich_text(self):
return dedent(
f"""
<b><u><h2> {self._device_spec.name}: </h2></u></b>
<table>
<tr><td> description: </td><td><i> {self._device_spec.description} </i></td></tr>
<tr><td> config: </td><td><i> {self._device_spec.deviceConfig} </i></td></tr>
<tr><td> enabled: </td><td><i> {self._device_spec.enabled} </i></td></tr>
<tr><td> read only: </td><td><i> {self._device_spec.readOnly} </i></td></tr>
</table>
"""
)
def setup_title_layout(self, device_spec: HashableDevice):
self._title_layout = QHBoxLayout()
self._title_layout.setContentsMargins(0, 0, 0, 0)
self._title_container = QWidget(parent=self)
self._title_container.setLayout(self._title_layout)
self._warning_label = QLabel()
self._title_layout.addWidget(self._warning_label)
self.title = QLabel(device_spec.name)
self.title.setToolTip(device_spec.name)
self.title.setStyleSheet(self.title_style("#FF0000"))
self._title_layout.addWidget(self.title)
self._title_layout.addStretch(1)
self._layout.addWidget(self._title_container)
def check_and_display_warning(self):
if len(self._device_spec.names) == 1 and len(self._device_spec._source_files) == 1:
self._warning_label.setText("")
self._warning_label.setToolTip("")
else:
self._warning_label.setPixmap(material_icon("warning", size=(12, 12), color="#FFAA00"))
self._warning_label.setToolTip(_warning_string(self._device_spec))
@property
def device_hash(self):
return hash(self._device_spec)
def title_style(self, color: str) -> str:
return f"QLabel {{ color: {color}; font-weight: bold; font-size: 10pt; }}"
def setTitle(self, text: str):
self.title.setText(text)
def set_included(self, included: bool):
self.included = included
self.title.setStyleSheet(self.title_style("#00FF00" if included else "#FF0000"))
class _DeviceEntry(NamedTuple):
list_item: QListWidgetItem
widget: _DeviceEntryWidget
class AvailableDeviceGroup(ExpandableGroupFrame, Ui_AvailableDeviceGroup):
def __init__(
self, parent=None, name: str = "TagGroupTitle", data: set[HashableDevice] = set(), **kwargs
):
super().__init__(parent=parent, **kwargs)
self.setupUi(self)
self.title_text = name # type: ignore
self._devices: dict[str, _DeviceEntry] = {}
for device in data:
self._add_item(device)
self.device_list.sortItems()
self.setMinimumSize(self.device_list.sizeHint())
self._update_num_included()
self.add_to_composition_button.clicked.connect(self.test)
def _add_item(self, device: HashableDevice):
item = QListWidgetItem(self.device_list)
widget = _DeviceEntryWidget(device, self)
item.setSizeHint(QSize(widget.width(), widget.height()))
self.device_list.setItemWidget(item, widget)
self.device_list.addItem(item)
self._devices[device.name] = _DeviceEntry(item, widget)
def reset_devices_state(self):
for dev in self._devices.values():
dev.widget.set_included(False)
self._update_num_included()
def set_item_state(self, /, device_hash: int, included: bool):
for dev in self._devices.values():
if dev.widget.device_hash == device_hash:
dev.widget.set_included(included)
self._update_num_included()
def _update_num_included(self):
n_included = sum(int(dev.widget.included) for dev in self._devices.values())
if n_included == 0:
color = "#FF0000"
elif n_included == len(self._devices):
color = "#00FF00"
else:
color = "#FFAA00"
self.n_included.setText(f"{n_included} / {len(self._devices)}")
self.n_included.setStyleSheet(f"QLabel {{ color: {color}; }}")
def sizeHint(self) -> QSize:
if not getattr(self, "device_list", None) or not self.expanded:
return super().sizeHint()
return QSize(
max(150, self.device_list.viewport().width()),
self.device_list.sizeHintForRow(0) * self.device_list.count() + 50,
)
def resizeEvent(self, event):
super().resizeEvent(event)
self.setMinimumHeight(self.sizeHint().height())
self.setMaximumHeight(self.sizeHint().height())
def get_selection(self) -> set[HashableDevice]:
selection = self.device_list.selectedItems()
widgets = (w.widget for _, w in self._devices.items() if w.list_item in selection)
return set(w._device_spec for w in widgets)
def test(self, *args):
print(self.get_selection())
def __repr__(self) -> str:
return f"{self.__class__.__name__}: {self.title_text}"
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = AvailableDeviceGroup(name="Tag group 1")
for item in [
HashableDevice(
**{
"name": f"test_device_{i}",
"deviceClass": "TestDeviceClass",
"readoutPriority": "baseline",
"enabled": True,
}
)
for i in range(5)
]:
widget._add_item(item)
widget._update_num_included()
widget.show()
sys.exit(app.exec())
@@ -0,0 +1,66 @@
from functools import partial
from bec_qthemes import material_icon
from PySide6.QtWidgets import QFrame
from qtpy.QtCore import QMetaObject
from qtpy.QtWidgets import QLabel, QListWidget, QToolButton, QVBoxLayout
class Ui_AvailableDeviceGroup(object):
def setupUi(self, AvailableDeviceGroup):
if not AvailableDeviceGroup.objectName():
AvailableDeviceGroup.setObjectName("AvailableDeviceGroup")
AvailableDeviceGroup.setMinimumWidth(150)
self.verticalLayout = QVBoxLayout()
self.verticalLayout.setObjectName("verticalLayout")
AvailableDeviceGroup.set_layout(self.verticalLayout)
title_layout = AvailableDeviceGroup.get_title_layout()
self.n_included = QLabel(AvailableDeviceGroup, text="...")
self.n_included.setObjectName("n_included")
title_layout.addWidget(self.n_included)
self.delete_tag_button = QToolButton(AvailableDeviceGroup)
self.delete_tag_button.setObjectName("delete_tag_button")
title_layout.addWidget(self.delete_tag_button)
self.remove_from_composition_button = QToolButton(AvailableDeviceGroup)
self.remove_from_composition_button.setObjectName("remove_from_composition_button")
title_layout.addWidget(self.remove_from_composition_button)
self.add_to_composition_button = QToolButton(AvailableDeviceGroup)
self.add_to_composition_button.setObjectName("add_to_composition_button")
title_layout.addWidget(self.add_to_composition_button)
self.remove_all_button = QToolButton(AvailableDeviceGroup)
self.remove_all_button.setObjectName("remove_all_from_composition_button")
title_layout.addWidget(self.remove_all_button)
self.add_all_button = QToolButton(AvailableDeviceGroup)
self.add_all_button.setObjectName("add_all_to_composition_button")
title_layout.addWidget(self.add_all_button)
self.device_list = QListWidget(AvailableDeviceGroup)
self.device_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
self.device_list.setObjectName("device_list")
self.device_list.setFrameStyle(0)
self.verticalLayout.addWidget(self.device_list)
self.set_icons()
QMetaObject.connectSlotsByName(AvailableDeviceGroup)
def set_icons(self):
icon = partial(material_icon, size=(15, 15), convert_to_pixmap=False)
self.delete_tag_button.setIcon(icon("delete"))
self.delete_tag_button.setToolTip("Delete tag group")
self.remove_from_composition_button.setIcon(icon("remove"))
self.remove_from_composition_button.setToolTip("Remove selected from composition")
self.add_to_composition_button.setIcon(icon("add"))
self.add_to_composition_button.setToolTip("Add selected to composition")
self.remove_all_button.setIcon(icon("chips"))
self.remove_all_button.setToolTip("Remove all with this tag from composition")
self.add_all_button.setIcon(icon("add_box"))
self.add_all_button.setToolTip("Add all with this tag to composition")
@@ -0,0 +1,83 @@
from random import randint
from typing import Any, Iterable
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.control.device_manager.components.available_device_resources._util import (
yield_only_passing,
)
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources_ui import (
Ui_availableDeviceResources,
)
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
HashableDevice,
get_backend,
)
class AvailableDeviceResources(BECWidget, QWidget, Ui_availableDeviceResources):
def __init__(self, parent=None, **kwargs):
super().__init__(parent=parent, **kwargs)
self.setupUi(self)
self._backend = get_backend()
self.grouping_selector.addItem("deviceTags")
self.grouping_selector.addItems(self._backend.allowed_sort_keys)
self._grouping_selection_changed("deviceTags")
self.grouping_selector.currentTextChanged.connect(self._grouping_selection_changed)
self.search_box.textChanged.connect(self.device_groups_list.update_filter)
def refresh_full_list(self, device_groups: dict[str, set[HashableDevice]]):
self.device_groups_list.clear()
for device_group, devices in device_groups.items():
self._add_device_group(device_group, devices)
if self.grouping_selector.currentText == "deviceTags":
self._add_device_group("Untagged devices", self._backend.untagged_devices)
self.device_groups_list.sort_by_key()
def _add_device_group(self, device_group: str, devices: set[HashableDevice]):
self.device_groups_list.add_item(
device_group, self.device_groups_list, device_group, devices, expanded=False
)
def _reset_devices_state(self):
for device_group in self.device_groups_list.widgets():
device_group.reset_devices_state()
def set_devices_state(self, devices: Iterable[HashableDevice], included: bool):
for device in devices:
for device_group in self.device_groups_list.widgets():
device_group.set_item_state(hash(device), included)
def resizeEvent(self, event):
super().resizeEvent(event)
for list_item, device_group_widget in self.device_groups_list.item_widget_pairs():
list_item.setSizeHint(device_group_widget.sizeHint())
@SafeSlot(list)
def update_devices_state(self, config_list: list[dict[str, Any]]):
self.set_devices_state(yield_only_passing(HashableDevice.model_validate, config_list), True)
@SafeSlot(str)
def _grouping_selection_changed(self, sort_key: str):
self.search_box.setText("")
if sort_key == "deviceTags":
device_groups = self._backend.tag_groups
else:
device_groups = self._backend.group_by_key(sort_key)
self.refresh_full_list(device_groups)
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = AvailableDeviceResources()
widget.set_devices_state(
list(filter(lambda _: randint(0, 1) == 1, widget._backend.all_devices)), True
)
widget.show()
sys.exit(app.exec())
@@ -0,0 +1,55 @@
from qtpy.QtCore import QMetaObject, Qt
from qtpy.QtWidgets import (
QAbstractItemView,
QComboBox,
QHBoxLayout,
QLabel,
QLineEdit,
QListView,
QListWidget,
QVBoxLayout,
)
from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group import (
AvailableDeviceGroup,
)
class Ui_availableDeviceResources(object):
def setupUi(self, availableDeviceResources):
if not availableDeviceResources.objectName():
availableDeviceResources.setObjectName("availableDeviceResources")
self.verticalLayout = QVBoxLayout(availableDeviceResources)
self.verticalLayout.setObjectName("verticalLayout")
self.search_layout = QHBoxLayout()
self.verticalLayout.addLayout(self.search_layout)
self.search_layout.addWidget(QLabel("Filter groups: "))
self.search_box = QLineEdit()
self.search_layout.addWidget(self.search_box)
self.search_layout.addWidget(QLabel("Group by: "))
self.grouping_selector = QComboBox()
self.search_layout.addWidget(self.grouping_selector)
self.device_groups_list = ListOfExpandableFrames(
availableDeviceResources, AvailableDeviceGroup
)
self.device_groups_list.setObjectName("device_groups_list")
self.device_groups_list.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
self.device_groups_list.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
self.device_groups_list.setHorizontalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
self.device_groups_list.setMovement(QListView.Movement.Static)
self.device_groups_list.setSpacing(2)
self.device_groups_list.setDragDropMode(QListWidget.DragDropMode.DragOnly)
self.device_groups_list.setSelectionBehavior(QListWidget.SelectionBehavior.SelectItems)
self.device_groups_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
self.device_groups_list.setDragEnabled(True)
self.device_groups_list.setAcceptDrops(False)
self.device_groups_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
availableDeviceResources.setMinimumWidth(250)
availableDeviceResources.resize(250, availableDeviceResources.height())
self.verticalLayout.addWidget(self.device_groups_list)
QMetaObject.connectSlotsByName(availableDeviceResources)
@@ -0,0 +1,131 @@
from __future__ import annotations
import operator
import os
from enum import Enum, auto
from functools import partial, reduce
from glob import glob
from pathlib import Path
from typing import Protocol
import bec_lib
from bec_lib.atlas_models import HashableDevice, HashableDeviceSet
from bec_lib.bec_yaml_loader import yaml_load
from bec_lib.logger import bec_logger
from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path
logger = bec_logger.logger
_BASE_REPO_PATH = Path(os.path.dirname(bec_lib.__file__)) / "../.."
class HashModel(str, Enum):
DEFAULT = auto()
DEFAULT_DEVICECONFIG = auto()
DEFAULT_EPICS = auto()
class DeviceResourceBackend(Protocol):
@property
def tag_groups(self) -> dict[str, set[HashableDevice]]:
"""A dictionary of all availble devices separated by tag groups. The same device may
appear more than once (in different groups)."""
...
@property
def all_devices(self) -> set[HashableDevice]:
"""A set of all availble devices. The same device may not appear more than once."""
...
@property
def untagged_devices(self) -> set[HashableDevice]:
"""A set of all untagged devices. The same device may not appear more than once."""
...
@property
def allowed_sort_keys(self) -> set[str]:
"""A set of all fields which you may group devices by"""
...
def tags(self) -> set[str]:
"""Returns a set of all the tags in all available devices."""
...
def tag_group(self, tag: str) -> set[HashableDevice]:
"""Returns a set of the devices in the tag group with the given key."""
...
def group_by_key(self, key: str) -> dict[str, set[HashableDevice]]:
"""Return a dict of all devices, organised by the specified key, which must be one of
the string keys in the Device model."""
...
def _devices_from_file(file: str, include_source: bool = True):
data = yaml_load(file, process_includes=False)
return HashableDeviceSet(
HashableDevice.model_validate(
dev | {"name": name, "source_files": {file} if include_source else set()}
)
for name, dev in data.items()
)
class _ConfigFileBackend(DeviceResourceBackend):
def __init__(self) -> None:
self._raw_device_set: set[
HashableDevice
] = self._get_config_from_backup_files() | self._get_configs_from_plugin_files(
Path(plugin_repo_path()) / plugin_package_name() / "device_configs/"
)
self._device_groups = self._get_tag_groups()
def _get_config_from_backup_files(self):
dir = _BASE_REPO_PATH / "logs/device_configs/recovery_configs"
files = glob("*.yaml", root_dir=dir)
return reduce(
operator.or_,
map(partial(_devices_from_file, include_source=False), (str(dir / f) for f in files)),
)
def _get_configs_from_plugin_files(self, dir: Path):
files = glob("*.yaml", root_dir=dir, recursive=True)
return reduce(operator.or_, map(_devices_from_file, (str(dir / f) for f in files)))
def _get_tag_groups(self) -> dict[str, set[HashableDevice]]:
return {
tag: set(filter(lambda dev: tag in dev.deviceTags, self._raw_device_set))
for tag in self.tags()
}
@property
def tag_groups(self):
return self._device_groups
@property
def all_devices(self):
return self._raw_device_set
@property
def untagged_devices(self):
return {d for d in self._raw_device_set if d.deviceTags == set()}
@property
def allowed_sort_keys(self) -> set[str]:
return {n for n, info in HashableDevice.model_fields.items() if info.annotation is str}
def tags(self) -> set[str]:
return reduce(operator.or_, (dev.deviceTags for dev in self._raw_device_set))
def tag_group(self, tag: str) -> set[HashableDevice]:
return self.tag_groups[tag]
def group_by_key(self, key: str) -> dict[str, set[HashableDevice]]:
if key not in self.allowed_sort_keys:
raise ValueError(f"Cannot group available devices by model key {key}")
group_names: set[str] = {getattr(item, key) for item in self._raw_device_set}
return {g: {d for d in self._raw_device_set if getattr(d, key) == g} for g in group_names}
def get_backend() -> DeviceResourceBackend:
return _ConfigFileBackend()
@@ -23,20 +23,14 @@ FUZZY_SEARCH_THRESHOLD = 80
class DictToolTipDelegate(QtWidgets.QStyledItemDelegate):
"""Delegate that shows all key-value pairs of a rows's data as a YAML-like tooltip."""
@staticmethod
def dict_to_str(d: dict) -> str:
"""Convert a dictionary to a formatted string."""
return json.dumps(d, indent=4)
def helpEvent(self, event, view, option, index):
"""Override to show tooltip when hovering."""
if event.type() != QtCore.QEvent.ToolTip:
return super().helpEvent(event, view, option, index)
model: DeviceFilterProxyModel = index.model()
model_index = model.mapToSource(index)
row_dict = model.sourceModel().row_data(model_index)
row_dict.pop("description", None)
QtWidgets.QToolTip.showText(event.globalPos(), self.dict_to_str(row_dict), view)
row_dict = model.sourceModel().get_row_data(model_index)
QtWidgets.QToolTip.showText(event.globalPos(), row_dict["description"], view)
return True
@@ -47,10 +41,10 @@ class CenterCheckBoxDelegate(DictToolTipDelegate):
super().__init__(parent)
colors = get_accent_colors()
self._icon_checked = material_icon(
"check_box", size=QtCore.QSize(16, 16), color=colors.default
"check_box", size=QtCore.QSize(16, 16), color=colors.default, filled=True
)
self._icon_unchecked = material_icon(
"check_box_outline_blank", size=QtCore.QSize(16, 16), color=colors.default
"check_box_outline_blank", size=QtCore.QSize(16, 16), color=colors.default, filled=True
)
def apply_theme(self, theme: str | None = None):
@@ -121,6 +115,9 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
Sort logic is implemented directly on the data of the table view.
"""
device_added = QtCore.Signal(dict)
devices_reset = QtCore.Signal(list)
def __init__(self, device_config: list[dict] | None = None, parent=None):
super().__init__(parent)
self._device_config = device_config or []
@@ -128,10 +125,9 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
"name",
"deviceClass",
"readoutPriority",
"deviceTags",
"enabled",
"readOnly",
"deviceTags",
"description",
]
self._checkable_columns_enabled = {"enabled": True, "readOnly": True}
@@ -150,7 +146,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
return self.headers[section]
return None
def row_data(self, index: QtCore.QModelIndex) -> dict:
def get_row_data(self, index: QtCore.QModelIndex) -> dict:
"""Return the row data for the given index."""
if not index.isValid():
return {}
@@ -169,6 +165,8 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
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 == QtCore.Qt.CheckStateRole and key in ("enabled", "readOnly"):
return QtCore.Qt.Checked if value else QtCore.Qt.Unchecked
@@ -255,6 +253,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
self.beginResetModel()
self._device_config = list(device_config)
self.endResetModel()
self.devices_reset.emit(self._device_config)
@SafeSlot(dict)
def add_device(self, device: dict):
@@ -436,6 +435,8 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
class DeviceTableView(BECWidget, QtWidgets.QWidget):
"""Device Table View for the device manager."""
selected_device = QtCore.Signal(dict)
RPC = False
PLUGIN = False
devices_removed = QtCore.Signal(list)
@@ -508,10 +509,9 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
self.table.setItemDelegateForColumn(0, self.tool_tip_delegate) # name
self.table.setItemDelegateForColumn(1, self.tool_tip_delegate) # deviceClass
self.table.setItemDelegateForColumn(2, self.tool_tip_delegate) # readoutPriority
self.table.setItemDelegateForColumn(3, self.checkbox_delegate) # enabled
self.table.setItemDelegateForColumn(4, self.checkbox_delegate) # readOnly
self.table.setItemDelegateForColumn(5, self.wrap_delegate) # deviceTags
self.table.setItemDelegateForColumn(6, self.wrap_delegate) # description
self.table.setItemDelegateForColumn(3, self.wrap_delegate) # deviceTags
self.table.setItemDelegateForColumn(4, self.checkbox_delegate) # enabled
self.table.setItemDelegateForColumn(5, self.checkbox_delegate) # readOnly
# Column resize policies
# TODO maybe we need here a flexible header options as deviceClass
@@ -520,13 +520,12 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents) # name
header.setSectionResizeMode(1, QtWidgets.QHeaderView.ResizeToContents) # deviceClass
header.setSectionResizeMode(2, QtWidgets.QHeaderView.ResizeToContents) # readoutPriority
header.setSectionResizeMode(3, QtWidgets.QHeaderView.Fixed) # enabled
header.setSectionResizeMode(4, QtWidgets.QHeaderView.Fixed) # readOnly
# TODO maybe better stretch...
header.setSectionResizeMode(5, QtWidgets.QHeaderView.ResizeToContents) # deviceTags
header.setSectionResizeMode(6, QtWidgets.QHeaderView.Stretch) # description
self.table.setColumnWidth(3, 82)
self.table.setColumnWidth(4, 82)
header.setSectionResizeMode(3, QtWidgets.QHeaderView.Stretch) # deviceTags
header.setSectionResizeMode(4, QtWidgets.QHeaderView.Fixed) # enabled
header.setSectionResizeMode(5, QtWidgets.QHeaderView.Fixed) # readOnly
self.table.setColumnWidth(3, 70)
self.table.setColumnWidth(4, 70)
# Ensure column widths stay fixed
header.setMinimumSectionSize(70)
@@ -538,6 +537,8 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
# Selection behavior
self.table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectRows)
self.table.setSelectionMode(QtWidgets.QAbstractItemView.ExtendedSelection)
# Connect to selection model to get selection changes
self.table.selectionModel().selectionChanged.connect(self._on_selection_changed)
self.table.horizontalHeader().setHighlightSections(False)
# QtCore.QTimer.singleShot(0, lambda: header.sectionResized.emit(0, 0, 0))
@@ -566,6 +567,45 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
height = delegate.sizeHint(option, index).height()
self.table.setRowHeight(row, height)
@SafeSlot(QtCore.QItemSelection, QtCore.QItemSelection)
def _on_selection_changed(
self, selected: QtCore.QItemSelection, deselected: QtCore.QItemSelection
) -> None:
"""
Handle selection changes in the device table.
Args:
selected (QtCore.QItemSelection): The selected items.
deselected (QtCore.QItemSelection): The deselected items.
"""
# TODO also hook up logic if a config update is propagated from somewhere!
# selected_indexes = selected.indexes()
selected_indexes = self.table.selectionModel().selectedIndexes()
if not selected_indexes:
return
source_indexes = [self.proxy.mapToSource(idx) for idx in selected_indexes]
source_rows = {idx.row() for idx in source_indexes}
# Ignore if multiple are selected
if len(source_rows) > 1:
self.selected_device.emit({})
return
# Get the single row
(row,) = source_rows
source_index = self.model.index(row, 0) # pick column 0 or whichever
device = self.model.get_row_data(source_index)
self.selected_device.emit(device)
@SafeSlot(QtCore.QModelIndex)
def _on_row_selected(self, index: QtCore.QModelIndex):
"""Handle row selection in the device table."""
if not index.isValid():
return
source_index = self.proxy.mapToSource(index)
device = self.model.get_device_at_index(source_index)
self.selected_device.emit(device)
######################################
##### Ext. Slot API #################
######################################
@@ -0,0 +1,71 @@
"""Module with a config view for the device manager."""
from __future__ import annotations
import yaml
from qtpy import QtCore, QtWidgets
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget
class DMConfigView(BECWidget, QtWidgets.QWidget):
def __init__(self, parent=None, client=None):
super().__init__(client=client, parent=parent)
self.stacked_layout = QtWidgets.QStackedLayout()
self.stacked_layout.setContentsMargins(0, 0, 0, 0)
self.stacked_layout.setSpacing(0)
self.setLayout(self.stacked_layout)
# Monaco widget
self.monaco_editor = MonacoWidget()
self._customize_monaco()
self.stacked_layout.addWidget(self.monaco_editor)
self._overlay_widget = QtWidgets.QLabel(text="Select single device to show config")
self._customize_overlay()
self.stacked_layout.addWidget(self._overlay_widget)
self.stacked_layout.setCurrentWidget(self._overlay_widget)
def _customize_monaco(self):
self.monaco_editor.set_language("yaml")
self.monaco_editor.set_vim_mode_enabled(False)
self.monaco_editor.set_minimap_enabled(False)
# self.monaco_editor.setFixedHeight(600)
self.monaco_editor.set_readonly(True)
def _customize_overlay(self):
self._overlay_widget.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
self._overlay_widget.setStyleSheet(
"background: qlineargradient(x1:0, y1:0, x2:0, y2:1,stop:0 #ffffff, stop:1 #e0e0e0);"
)
self._overlay_widget.setAutoFillBackground(True)
self._overlay_widget.setSizePolicy(
QtWidgets.QSizePolicy.Policy.Expanding, QtWidgets.QSizePolicy.Policy.Expanding
)
@SafeSlot(dict)
def on_select_config(self, device: dict):
"""Handle selection of a device from the device table."""
if not device:
text = ""
self.stacked_layout.setCurrentWidget(self._overlay_widget)
else:
text = yaml.dump(device, default_flow_style=False)
self.stacked_layout.setCurrentWidget(self.monaco_editor)
self.monaco_editor.set_readonly(False) # Enable editing
self.monaco_editor.set_text(text)
self.monaco_editor.set_readonly(True) # Disable editing again
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
config_view = DMConfigView()
config_view.show()
sys.exit(app.exec_())
@@ -0,0 +1,333 @@
"""Module to run a static test for the current config and see if it is valid."""
from __future__ import annotations
import enum
import bec_lib
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
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.widgets.editors.web_console.web_console import WebConsole
READY_TO_TEST = False
logger = bec_logger.logger
try:
import bec_server
import ophyd_devices
READY_TO_TEST = True
except ImportError:
logger.warning(f"Optional dependencies not available: {ImportError}")
ophyd_devices = None
bec_server = None
class ValidationStatus(int, enum.Enum):
"""Validation status for device configurations."""
UNKNOWN = 0 # colors.default
ERROR = 1 # colors.emergency
VALID = 2 # colors.highlight
CANT_CONNECT = 3 # colors.warning
CONNECTED = 4 # colors.success
class DeviceValidationListItem(QtWidgets.QWidget):
"""Custom list item widget showing device name and validation status."""
status_changed = QtCore.Signal(int) # Signal emitted when status changes -> ValidationStatus
# Signal emitted when device was validated with name, success, msg
device_validated = QtCore.Signal(str, str)
def __init__(
self,
device_config: dict[str, dict],
status: ValidationStatus,
status_icons: dict[ValidationStatus, QtGui.QPixmap],
validate_icon: QtGui.QPixmap,
parent=None,
static_device_test=None,
):
super().__init__(parent)
if len(device_config.keys()) > 1:
logger.warning(
f"Multiple devices found for config: {list(device_config.keys())}, using first one"
)
self._static_device_test = static_device_test
self.device_name = list(device_config.keys())[0]
self.device_config = device_config
self.status: ValidationStatus = status
colors = get_accent_colors()
self._status_icon = status_icons
self._validate_icon = validate_icon
self._setup_ui()
self._update_status_indicator()
def _setup_ui(self):
"""Setup the UI for the list item."""
layout = QtWidgets.QHBoxLayout(self)
layout.setContentsMargins(4, 4, 4, 4)
# Device name label
self.name_label = QtWidgets.QLabel(self.device_name)
self.name_label.setStyleSheet("font-weight: bold;")
layout.addWidget(self.name_label)
# Make sure status is on the right
layout.addStretch()
self.request_validation_button = QtWidgets.QPushButton("Validate")
self.request_validation_button.setIcon(self._validate_icon)
if self._static_device_test is None:
self.request_validation_button.setDisabled(True)
else:
self.request_validation_button.clicked.connect(self.on_request_validation)
# self.request_validation_button.setVisible(False) -> Hide it??
layout.addWidget(self.request_validation_button)
# Status indicator
self.status_indicator = QtWidgets.QLabel()
self._update_status_indicator()
layout.addWidget(self.status_indicator)
@SafeSlot()
def on_request_validation(self):
"""Handle validate button click."""
if self._static_device_test is None:
logger.warning("Static device test not available.")
return
self._static_device_test.config = self.device_config
# TODO logic if connect is allowed
ret = self._static_device_test.run_with_list_output(connect=False)[0]
if ret.success:
self.set_status(ValidationStatus.VALID)
else:
self.set_status(ValidationStatus.ERROR)
self.device_validated.emit(ret.name, ret.message)
def _update_status_indicator(self):
"""Update the status indicator color based on validation status."""
self.status_indicator.setPixmap(self._status_icon[self.status])
def set_status(self, status: ValidationStatus):
"""Update the validation status."""
self.status = status
self._update_status_indicator()
self.status_changed.emit(self.status)
def get_status(self) -> ValidationStatus:
"""Get the current validation status."""
return self.status
class DeviceManagerOphydTest(BECWidget, QtWidgets.QWidget):
config_changed = QtCore.Signal(
dict, dict
) # Signal emitted when the device config changed, new_config, old_config
def __init__(self, parent=None, client=None):
super().__init__(parent=parent, client=client)
if not READY_TO_TEST:
self._set_disabled()
static_device_test = None
else:
from ophyd_devices.utils.static_device_test import StaticDeviceTest
static_device_test = StaticDeviceTest(config_dict={})
self._static_device_test = static_device_test
self._device_config: dict[str, dict] = {}
self._main_layout = QtWidgets.QVBoxLayout(self)
self._main_layout.setContentsMargins(0, 0, 0, 0)
self._main_layout.setSpacing(4)
# Setup icons
colors = get_accent_colors()
self._validate_icon = material_icon(
icon_name="play_arrow", color=colors.default, filled=True
)
self._status_icons = {
ValidationStatus.UNKNOWN: material_icon(
icon_name="circle", size=(12, 12), color=colors.default, filled=True
),
ValidationStatus.ERROR: material_icon(
icon_name="circle", size=(12, 12), color=colors.emergency, filled=True
),
ValidationStatus.VALID: material_icon(
icon_name="circle", size=(12, 12), color=colors.highlight, filled=True
),
ValidationStatus.CANT_CONNECT: material_icon(
icon_name="circle", size=(12, 12), color=colors.warning, filled=True
),
ValidationStatus.CONNECTED: material_icon(
icon_name="circle", size=(12, 12), color=colors.success, filled=True
),
}
self.setLayout(self._main_layout)
# splitter
self.splitter = QtWidgets.QSplitter(QtCore.Qt.Orientation.Vertical)
self._main_layout.addWidget(self.splitter)
# Add custom list
self.setup_device_validation_list()
# Setup text box
self.setup_text_box()
# Connect signals
self.config_changed.connect(self.on_config_updated)
@SafeSlot(list)
def on_device_config_update(self, config: list[dict]):
old_cfg = self._device_config
self._device_config = self._compile_device_config_list(config)
self.config_changed.emit(self._device_config, old_cfg)
def _compile_device_config_list(self, config: list[dict]) -> dict[str, dict]:
return {dev["name"]: {k: v for k, v in dev.items() if k != "name"} for dev in config}
@SafeSlot(dict, dict)
def on_config_updated(self, new_config: dict, old_config: dict):
"""Handle config updates and refresh the validation list."""
# Find differences for potential re-validation
diffs = self._find_diffs(new_config, old_config)
# Check diff first
for diff in diffs:
if not diff:
continue
if len(diff) > 1:
logger.warning(f"Multiple devices found in diff: {diff}, using first one")
name = list(diff.keys())[0]
if name in self.client.device_manager.devices:
status = ValidationStatus.CONNECTED
else:
status = ValidationStatus.UNKNOWN
if self.get_device_status(diff) is None:
self.add_device(diff, status)
else:
self.update_device_status(diff, status)
def _find_diffs(self, new_config: dict, old_config: dict) -> list[dict]:
"""
Return list of keys/paths where d1 and d2 differ. This goes recursively through the dictionary.
Args:
new_config: The first dictionary to compare.
old_config: The second dictionary to compare.
"""
diffs = []
keys = set(new_config.keys()) | set(old_config.keys())
for k in keys:
if k not in old_config: # New device
diffs.append({k: new_config[k]})
continue
if k not in new_config: # Removed device
diffs.append({k: old_config[k]})
continue
# Compare device config if exists in both
v1, v2 = old_config[k], new_config[k]
if isinstance(v1, dict) and isinstance(v2, dict):
if self._find_diffs(v2, v1): # recurse: something inside changed
diffs.append({k: new_config[k]})
elif v1 != v2:
diffs.append({k: new_config[k]})
return diffs
def setup_device_validation_list(self):
"""Setup the device validation list."""
# Create the custom validation list widget
self.validation_list = QtWidgets.QListWidget()
self.validation_list.setSelectionMode(QtWidgets.QAbstractItemView.SingleSelection)
self.splitter.addWidget(self.validation_list)
# self._main_layout.addWidget(self.validation_list)
def setup_text_box(self):
"""Setup the text box for device validation messages."""
self.validation_text_box = QtWidgets.QTextEdit()
self.validation_text_box.setReadOnly(True)
self.splitter.addWidget(self.validation_text_box)
# self._main_layout.addWidget(self.validation_text_box)
@SafeSlot(str, str)
def on_device_validated(self, device_name: str, message: str):
"""Handle device validation results."""
text = f"Device {device_name} was validated. Message: {message}"
self.validation_text_box.setText(text)
def _set_disabled(self) -> None:
"""Disable the full view"""
self.setDisabled(True)
def add_device(
self, device_config: dict[str, dict], status: ValidationStatus = ValidationStatus.UNKNOWN
):
"""Add a device to the validation list."""
# Create the custom widget
item_widget = DeviceValidationListItem(
device_config=device_config,
status=status,
status_icons=self._status_icons,
validate_icon=self._validate_icon,
static_device_test=self._static_device_test,
)
# Create a list widget item
list_item = QtWidgets.QListWidgetItem()
list_item.setSizeHint(item_widget.sizeHint())
# Add item to list and set custom widget
self.validation_list.addItem(list_item)
self.validation_list.setItemWidget(list_item, item_widget)
item_widget.device_validated.connect(self.on_device_validated)
def update_device_status(self, device_config: dict[str, dict], status: ValidationStatus):
"""Update the validation status for a specific device."""
for i in range(self.validation_list.count()):
item = self.validation_list.item(i)
widget = self.validation_list.itemWidget(item)
if (
isinstance(widget, DeviceValidationListItem)
and widget.device_config == device_config
):
widget.set_status(status)
break
def clear_devices(self):
"""Clear all devices from the list."""
self.validation_list.clear()
def get_device_status(self, device_config: dict[str, dict]) -> ValidationStatus | None:
"""Get the validation status for a specific device."""
for i in range(self.validation_list.count()):
item = self.validation_list.item(i)
widget = self.validation_list.itemWidget(item)
if (
isinstance(widget, DeviceValidationListItem)
and widget.device_config == device_config
):
return widget.get_status()
return None
if __name__ == "__main__":
import sys
# pylint: disable=ungrouped-imports
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
device_manager_ophyd_test = DeviceManagerOphydTest()
cfg = device_manager_ophyd_test.client.device_manager._get_redis_device_config()
cfg.append({"name": "Wrong_Device", "type": "test"})
device_manager_ophyd_test.on_device_config_update(cfg)
device_manager_ophyd_test.show()
device_manager_ophyd_test.setWindowTitle("Device Manager Ophyd Test")
device_manager_ophyd_test.resize(800, 600)
sys.exit(app.exec_())
@@ -11,19 +11,13 @@ from bec_lib.logger import bec_logger
from bec_lib.messages import ConfigAction, ScanStatusMessage
from bec_qthemes import material_icon
from pyqtgraph import SignalProxy
from qtpy.QtCore import QSize, QThreadPool, Signal
from qtpy.QtWidgets import (
QFileDialog,
QListWidget,
QListWidgetItem,
QToolButton,
QVBoxLayout,
QWidget,
)
from qtpy.QtCore import QThreadPool, Signal
from qtpy.QtWidgets import QFileDialog, QListWidget, QToolButton, QVBoxLayout, QWidget
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.services.device_browser.device_item import DeviceItem
from bec_widgets.widgets.services.device_browser.device_item.device_config_dialog import (
@@ -59,7 +53,8 @@ class DeviceBrowser(BECWidget, QWidget):
self._q_threadpool = QThreadPool()
self.ui = None
self.init_ui()
self.dev_list: QListWidget = self.ui.device_list
self.dev_list = ListOfExpandableFrames(self, DeviceItem)
self.ui.verticalLayout.addWidget(self.dev_list)
self.dev_list.setVerticalScrollMode(QListWidget.ScrollMode.ScrollPerPixel)
self.proxy_device_update = SignalProxy(
self.ui.filter_input.textChanged, rateLimit=500, slot=self.update_device_list
@@ -132,25 +127,15 @@ class DeviceBrowser(BECWidget, QWidget):
def init_device_list(self):
self.dev_list.clear()
self._device_items: dict[str, QListWidgetItem] = {}
with RPCRegister.delayed_broadcast():
for device, device_obj in self.dev.items():
self._add_item_to_list(device, device_obj)
def _add_item_to_list(self, device: str, device_obj):
def _updatesize(item: QListWidgetItem, device_item: DeviceItem):
device_item.adjustSize()
item.setSizeHint(QSize(device_item.width(), device_item.height()))
logger.debug(f"Adjusting {item} size to {device_item.width(), device_item.height()}")
def _remove_item(item: QListWidgetItem):
self.dev_list.takeItem(self.dev_list.row(item))
del self._device_items[device]
self.dev_list.sortItems()
item = QListWidgetItem(self.dev_list)
device_item = DeviceItem(
device_item = self.dev_list.add_item(
id=device,
parent=self,
device=device,
devices=self.dev,
@@ -158,18 +143,11 @@ class DeviceBrowser(BECWidget, QWidget):
config_helper=self._config_helper,
q_threadpool=self._q_threadpool,
)
device_item.expansion_state_changed.connect(partial(_updatesize, item, device_item))
device_item.imminent_deletion.connect(partial(_remove_item, item))
self.editing_enabled.connect(device_item.set_editable)
self.device_update.connect(device_item.config_update)
tooltip = self.dev[device]._config.get("description", "")
device_item.setToolTip(tooltip)
device_item.broadcast_size_hint.connect(item.setSizeHint)
item.setSizeHint(device_item.sizeHint())
self.dev_list.setItemWidget(item, device_item)
self.dev_list.addItem(item)
self._device_items[device] = item
@SafeSlot(dict, dict)
def scan_status_changed(self, scan_info: dict, _: dict):
@@ -198,20 +176,11 @@ class DeviceBrowser(BECWidget, QWidget):
Either way, the function will filter the devices based on the filter input text and update the device list.
"""
filter_text = self.ui.filter_input.text()
for device in self.dev:
if device not in self._device_items:
if device not in self.dev_list:
# it is possible the device has just been added to the config
self._add_item_to_list(device, self.dev[device])
try:
self.regex = re.compile(filter_text, re.IGNORECASE)
except re.error:
self.regex = None # Invalid regex, disable filtering
for device in self.dev:
self._device_items[device].setHidden(False)
return
for device in self.dev:
self._device_items[device].setHidden(not self.regex.search(device))
self.dev_list.update_filter(self.ui.filter_input.text())
@SafeSlot()
def _load_from_file(self):
@@ -1,93 +1,90 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>406</width>
<height>500</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="QGroupBox" name="browser_group_box">
<property name="title">
<string>Device Browser</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<layout class="QHBoxLayout" name="filter_layout">
<item>
<widget class="QLineEdit" name="filter_input">
<property name="placeholderText">
<string>Filter</string>
</property>
</widget>
</item>
<item>
<widget class="QGroupBox" name="button_box">
<layout class="QHBoxLayout" name="horizontalLayout">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
<item>
<widget class="QToolButton" name="add_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="save_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="import_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="scan_running_warning">
<property name="styleSheet">
<string notr="true"/>
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>406</width>
<height>500</height>
</rect>
</property>
<property name="text">
<string>warning</string>
<property name="windowTitle">
<string>Form</string>
</property>
</widget>
</item>
<item>
<widget class="QListWidget" name="device_list"/>
</item>
</layout>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="QGroupBox" name="browser_group_box">
<property name="title">
<string>Device Browser</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<layout class="QHBoxLayout" name="filter_layout">
<item>
<widget class="QLineEdit" name="filter_input">
<property name="placeholderText">
<string>Filter</string>
</property>
</widget>
</item>
<item>
<widget class="QGroupBox" name="button_box">
<layout class="QHBoxLayout" name="horizontalLayout">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
<item>
<widget class="QToolButton" name="add_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="save_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="import_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="scan_running_warning">
<property name="styleSheet">
<string notr="true" />
</property>
<property name="text">
<string>warning</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</widget>
<resources/>
<connections/>
</ui>
<resources />
<connections />
</ui>
@@ -35,9 +35,6 @@ logger = bec_logger.logger
class DeviceItem(ExpandableGroupFrame):
broadcast_size_hint = Signal(QSize)
imminent_deletion = Signal()
RPC = False
def __init__(
@@ -0,0 +1,78 @@
from copy import copy
import pytest
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
HashableDevice,
_HashableDeviceSet,
)
TEST_DEVICE_DICT = {
"name": "test_device",
"deviceClass": "TestDeviceClass",
"readoutPriority": "baseline",
"enabled": True,
}
def _test_device_dict(**kwargs):
new = copy(TEST_DEVICE_DICT)
new.update(kwargs)
return new
@pytest.mark.parametrize(
"kwargs_1, kwargs_2, kwargs_3, kwargs_4, n",
[
({}, {}, {}, {}, 1),
({}, {}, {}, {"deviceConfig": {"a": 1}}, 1),
({}, {}, {}, {"name": "test_device_2"}, 2),
({}, {}, {"name": "test_device_2"}, {"deviceClass": "OtherDeviceClass"}, 3),
],
)
def test_hashable_device_set_merges_equal(kwargs_1, kwargs_2, kwargs_3, kwargs_4, n):
item_1 = HashableDevice(**_test_device_dict(**kwargs_1))
item_2 = HashableDevice(**_test_device_dict(**kwargs_2))
item_3 = HashableDevice(**_test_device_dict(**kwargs_3))
item_4 = HashableDevice(**_test_device_dict(**kwargs_4))
test_set = _HashableDeviceSet((item_1, item_2, item_3, item_4))
assert len(test_set) == n
def test_hashable_device_set_or_adds_sources():
item_1 = HashableDevice(**_test_device_dict(), source_files={"a", "b"})
item_2 = HashableDevice(**_test_device_dict(), source_files={"c", "d"})
set_1 = _HashableDeviceSet((item_1,))
set_2 = _HashableDeviceSet((item_2,))
combined = set_1 | set_2
assert len(combined) == 1
assert combined.pop().source_files == {"a", "b", "c", "d"}
def test_hashable_device_set_or_adds_tags():
item_1 = HashableDevice(
**_test_device_dict(deviceTags={"tag1"}, deviceConfig={"param": "value"}),
source_files={"a", "b"},
)
item_2 = HashableDevice(
**_test_device_dict(deviceTags={"tag2"}, deviceConfig={"param": "value"}),
source_files={"c", "d"},
)
item_3 = HashableDevice(
**_test_device_dict(deviceTags={"tag3"}, deviceConfig={"param": "other_value"}),
source_files={"q"},
)
set_1 = _HashableDeviceSet((item_1,))
set_2 = _HashableDeviceSet((item_2,))
set_3 = _HashableDeviceSet((item_3,))
combined = sorted(set_1 | set_2 | set_3, key=lambda hd: hd.deviceConfig["param"])
assert len(combined) == 2
assert combined[0].source_files == {"q"}
assert combined[0].deviceTags == {"tag3"}
assert combined[1].source_files == {"a", "b", "c", "d"}
assert combined[1].deviceTags == {"tag1", "tag2"}
+16 -16
View File
@@ -37,11 +37,11 @@ def device_browser(qtbot, mocked_client):
yield dev_browser
def test_device_browser_init_with_devices(device_browser):
def test_device_browser_init_with_devices(device_browser: DeviceBrowser):
"""
Test that the device browser is initialized with the correct number of devices.
"""
device_list = device_browser.ui.device_list
device_list = device_browser.dev_list
assert device_list.count() == len(device_browser.dev)
@@ -58,11 +58,11 @@ def test_device_browser_filtering(
expected = expected_num_visible if expected_num_visible >= 0 else len(device_browser.dev)
def num_visible(item_dict):
return len(list(filter(lambda i: not i.isHidden(), item_dict.values())))
return len(list(filter(lambda i: not i.widget.isHidden(), item_dict.values())))
device_browser.ui.filter_input.setText(search_term)
qtbot.wait(100)
assert num_visible(device_browser._device_items) == expected
assert num_visible(device_browser.dev_list._item_dict) == expected
def test_device_item_mouse_press_event(device_browser, qtbot):
@@ -70,8 +70,8 @@ def test_device_item_mouse_press_event(device_browser, qtbot):
Test that the mousePressEvent is triggered correctly.
"""
# Simulate a left mouse press event on the device item
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
qtbot.mouseClick(widget._title, Qt.MouseButton.LeftButton)
@@ -88,8 +88,8 @@ def test_device_item_expansion(device_browser, qtbot):
Test that the form is displayed when the item is expanded, and that the expansion is triggered
by clicking on the expansion button, the title, or the device icon
"""
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
qtbot.mouseClick(widget._expansion_button, Qt.MouseButton.LeftButton)
tab_widget: QTabWidget = widget._contents.layout().itemAt(0).widget()
qtbot.waitUntil(lambda: tab_widget.widget(0) is not None, timeout=100)
@@ -115,8 +115,8 @@ def test_device_item_mouse_press_and_move_events_creates_drag(device_browser, qt
"""
Test that the mousePressEvent is triggered correctly and initiates a drag.
"""
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
device_name = widget.device
with mock.patch("qtpy.QtGui.QDrag.exec_") as mock_exec:
with mock.patch("qtpy.QtGui.QDrag.setMimeData") as mock_set_mimedata:
@@ -133,19 +133,19 @@ def test_device_item_double_click_event(device_browser, qtbot):
Test that the mouseDoubleClickEvent is triggered correctly.
"""
# Simulate a left mouse press event on the device item
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
qtbot.mouseDClick(widget, Qt.LeftButton)
def test_device_deletion(device_browser, qtbot):
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
widget._config_helper = mock.MagicMock()
assert widget.device in device_browser._device_items
assert widget.device in device_browser.dev_list._item_dict
qtbot.mouseClick(widget.delete_button, Qt.LeftButton)
qtbot.waitUntil(lambda: widget.device not in device_browser._device_items, timeout=10000)
qtbot.waitUntil(lambda: widget.device not in device_browser.dev_list._item_dict, timeout=10000)
def test_signal_display(mocked_client, qtbot):