From 48b854a9ab59aa58206290e36ce1a29db47da0da Mon Sep 17 00:00:00 2001 From: David Perl Date: Tue, 2 Sep 2025 11:25:02 +0200 Subject: [PATCH] feat: add available device resource browser --- .../available_device_resources/__init__.py | 3 + .../available_device_resources/_util.py | 36 ++++ .../available_device_group.py | 200 ++++++++++++++++++ .../available_device_group_ui.py | 66 ++++++ .../available_device_resources.py | 83 ++++++++ .../available_device_resources_ui.py | 55 +++++ .../device_resource_backend.py | 139 ++++++++++++ 7 files changed, 582 insertions(+) create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/__init__.py create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/_util.py create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group.py create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py create mode 100644 bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/__init__.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/__init__.py new file mode 100644 index 00000000..83d4d4d0 --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/__init__.py @@ -0,0 +1,3 @@ +from .available_device_resources import AvailableDeviceResources + +__all__ = ["AvailableDeviceResources"] diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/_util.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/_util.py new file mode 100644 index 00000000..c848da92 --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/_util.py @@ -0,0 +1,36 @@ +from typing import Any, Callable, Generator, Iterable, TypeVar + +from qtpy.QtWidgets import QListWidgetItem + +_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 + + +SORT_KEY_ROLE = 117 + + +class SortableQListWidgetItem(QListWidgetItem): + """Store a sorting string key with .setData(SORT_KEY_ROLE, key) to be able to sort a list with \ + custom widgets and this item.""" + + def __gt__(self, other): + if (self_key := self.data(SORT_KEY_ROLE)) is None or ( + other_key := other.data(SORT_KEY_ROLE) + ) is None: + return False + return self_key.lower() > other_key.lower() + + def __lt__(self, other): + if (self_key := self.data(SORT_KEY_ROLE)) is None or ( + other_key := other.data(SORT_KEY_ROLE) + ) is None: + return False + return self_key.lower() < other_key.lower() diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group.py new file mode 100644 index 00000000..1fc2810c --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group.py @@ -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""" +

{self._device_spec.name}:

+ + + + + +
description: {self._device_spec.description}
config: {self._device_spec.deviceConfig}
enabled: {self._device_spec.enabled}
read only: {self._device_spec.readOnly}
+ """ + ) + + 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()) diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py new file mode 100644 index 00000000..d2565b4a --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py @@ -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") diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py new file mode 100644 index 00000000..a4b8fc70 --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py @@ -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.sortItems() + + 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()) diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py new file mode 100644 index 00000000..4cfac859 --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py @@ -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) diff --git a/bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py b/bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py new file mode 100644 index 00000000..46cc2aaf --- /dev/null +++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py @@ -0,0 +1,139 @@ +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__)) / "../.." + + +def get_backend() -> DeviceResourceBackend: + return _ConfigFileBackend() + + +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() + ) + + +# use the last n recovery files +_N_RECOVERY_FILES = 3 + + +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 = sorted(glob("*.yaml", root_dir=dir)) + last_n_files = files[-_N_RECOVERY_FILES:] + return reduce( + operator.or_, + map( + partial(_devices_from_file, include_source=False), + (str(dir / f) for f in last_n_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}