+ | 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}