refactor(device_manager): remove unused available device resources package

bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py

@@ -45,7 +45,6 @@ from bec_widgets.widgets.control.device_manager.components import (
     DocstringView,
     OphydValidation,
 )
-from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
 from bec_widgets.widgets.control.device_manager.components.config_communicator import (
     CommunicateConfigAction,
 )
@@ -173,7 +172,6 @@ class DeviceManagerDisplayWidget(DockAreaWidget):
         # other communications to REDIS as uploading a config through a CommunicationConfigAction
         # will block if we use the config_helper from self.client.config._config_helper
         self._config_helper = config_helper.ConfigHelper(self.client.connector)
-        self._shared_selection = SharedSelectionSignal()
 
         # Device Table View widget
         self.device_table_view = DeviceTable(self)

bec_widgets/cli/client.py

@@ -129,28 +129,6 @@ class AutoUpdates(RPCBase):
         """
 
 
-class AvailableDeviceResources(RPCBase):
-    _IMPORT_MODULE = "bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources"
-
-    @rpc_call
-    def remove(self):
-        """
-        Cleanup the BECConnector
-        """
-
-    @rpc_call
-    def attach(self):
-        """
-        None
-        """
-
-    @rpc_call
-    def detach(self):
-        """
-        Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
-        """
-
-
 class BECDockArea(RPCBase):
     _IMPORT_MODULE = "bec_widgets.widgets.containers.dock_area.dock_area"
 

bec_widgets/utils/expandable_frame.py

@@ -1,148 +0,0 @@
-from __future__ import annotations
-
-from bec_qthemes import material_icon
-from qtpy.QtCore import QSize, Signal
-from qtpy.QtWidgets import (
-    QApplication,
-    QFrame,
-    QHBoxLayout,
-    QLabel,
-    QLayout,
-    QSizePolicy,
-    QToolButton,
-    QVBoxLayout,
-    QWidget,
-)
-
-from bec_widgets.utils.clickable_label import ClickableLabel
-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"
-    COLLAPSED_ICON_NAME: str = "expand_all"
-
-    def __init__(
-        self, parent: QWidget | None = None, title: str = "", expanded: bool = True, icon: str = ""
-    ) -> None:
-        super().__init__(parent=parent)
-        self._expanded = expanded
-
-        self._title_text = f"<b>{title}</b>"
-        self.setFrameStyle(QFrame.Shape.StyledPanel | QFrame.Shadow.Raised)
-        self.setSizePolicy(QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Minimum)
-        self._layout = QVBoxLayout()
-        self._layout.setContentsMargins(5, 0, 0, 0)
-        self.setLayout(self._layout)
-
-        self._create_title_layout(title, icon)
-
-        self._contents = QWidget(self)
-        self._layout.addWidget(self._contents)
-
-        self._expansion_button.clicked.connect(self.switch_expanded_state)
-        self.expanded = self._expanded  # type: ignore
-        self.expansion_state_changed.emit()
-
-    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()
-        self._set_title_text(self._title_text)
-        self._title_icon = ClickableLabel()
-        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._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
-
-    @SafeSlot()
-    def switch_expanded_state(self):
-        self.expanded = not self.expanded  # type: ignore
-        self._update_expansion_icon()
-        self.expansion_state_changed.emit()
-
-    @SafeProperty(bool)
-    def expanded(self):  # type: ignore
-        return self._expanded
-
-    @expanded.setter
-    def expanded(self, expanded: bool):
-        self._expanded = expanded
-        self._contents.setVisible(expanded)
-        self.updateGeometry()
-        self.adjustSize()
-
-    def _update_expansion_icon(self):
-        self._expansion_button.setIcon(
-            material_icon(icon_name=self.EXPANDED_ICON_NAME, size=(10, 10), convert_to_pixmap=False)
-            if self.expanded
-            else material_icon(
-                icon_name=self.COLLAPSED_ICON_NAME, size=(10, 10), convert_to_pixmap=False
-            )
-        )
-
-    @SafeProperty(str)
-    def icon_name(self):  # type: ignore
-        return self._title_icon_name
-
-    @icon_name.setter
-    def icon_name(self, icon_name: str):
-        self._title_icon_name = icon_name
-        self._set_title_icon(self._title_icon_name)
-
-    def _set_title_icon(self, icon_name: str):
-        if icon_name:
-            self._title_icon.setVisible(True)
-            self._title_icon.setPixmap(
-                material_icon(icon_name=icon_name, size=(20, 20), convert_to_pixmap=True)
-            )
-        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
-
-    app = QApplication([])
-    frame = ExpandableGroupFrame()
-    layout = QVBoxLayout()
-    frame.set_layout(layout)
-    layout.addWidget(QLabel("test1"))
-    layout.addWidget(QLabel("test2"))
-    layout.addWidget(QLabel("test3"))
-
-    frame.show()
-    app.exec()

bec_widgets/utils/list_of_expandable_frames.py

@@ -1,133 +0,0 @@
-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 qtpy.QtCore import QSize, Qt
-from qtpy.QtWidgets import QListWidget, QListWidgetItem, QWidget
-
-from bec_widgets.utils.error_popups import SafeSlot
-from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
-from bec_widgets.widgets.control.device_manager.components._util import (
-    SORT_KEY_ROLE,
-    SortableQListWidgetItem,
-)
-
-logger = bec_logger.logger
-
-
-_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) -> tuple[QListWidgetItem, _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 = SortableQListWidgetItem(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, 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()

bec_widgets/widgets/control/device_manager/components/_util.py

@@ -1,33 +1,6 @@
-import json
-from typing import Any, Callable, Generator, Iterable, TypeVar
-
-from bec_lib.utils.json_extended import ExtendedEncoder
-from qtpy.QtCore import QByteArray, QMimeData, QObject, Signal  # type: ignore
 from qtpy.QtWidgets import QListWidgetItem
 
-from bec_widgets.widgets.control.device_manager.components.constants import (
-    MIME_DEVICE_CONFIG,
-    SORT_KEY_ROLE,
-)
-
-_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
-
-
-def mimedata_from_configs(configs: Iterable[dict]) -> QMimeData:
-    """Takes an iterable of device configs, gives a QMimeData with the configs json-encoded under the type MIME_DEVICE_CONFIG"""
-    mime_obj = QMimeData()
-    byte_array = QByteArray(json.dumps(list(configs), cls=ExtendedEncoder).encode("utf-8"))
-    mime_obj.setData(MIME_DEVICE_CONFIG, byte_array)
-    return mime_obj
+from bec_widgets.widgets.control.device_manager.components.constants import SORT_KEY_ROLE
 
 
 class SortableQListWidgetItem(QListWidgetItem):
@@ -47,7 +20,3 @@ class SortableQListWidgetItem(QListWidgetItem):
         ) is None:
             return False
         return self_key.lower() < other_key.lower()
-
-
-class SharedSelectionSignal(QObject):
-    proc = Signal(str)

bec_widgets/widgets/control/device_manager/components/available_device_resources/__init__.py

@@ -1,3 +0,0 @@
-from .available_device_resources import AvailableDeviceResources
-
-__all__ = ["AvailableDeviceResources"]

bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group.py

@@ -1,228 +0,0 @@
-from textwrap import dedent
-from typing import NamedTuple
-from uuid import uuid4
-
-from bec_qthemes import material_icon
-from qtpy.QtCore import QItemSelection, QSize, Signal
-from qtpy.QtWidgets import QFrame, QHBoxLayout, QLabel, QListWidgetItem, QVBoxLayout, QWidget
-
-from bec_widgets.utils.error_popups import SafeSlot
-from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
-from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
-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,
-)
-from bec_widgets.widgets.control.device_manager.components.constants import CONFIG_DATA_ROLE
-
-
-def _warning_string(spec: HashableDevice):
-    name_warning = (
-        "Device defined with multiple names! Please check:\n  " + "\n  ".join(spec.names)
-        if len(spec.names) > 1
-        else ""
-    )
-    source_warning = (
-        "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.setFrameStyle(0)
-
-        self._layout = QVBoxLayout()
-        self._layout.setContentsMargins(2, 2, 2, 2)
-        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):
-
-    selected_devices = Signal(list)
-
-    def __init__(
-        self,
-        parent=None,
-        name: str = "TagGroupTitle",
-        data: set[HashableDevice] = set(),
-        shared_selection_signal=SharedSelectionSignal(),
-        **kwargs,
-    ):
-        super().__init__(parent=parent, **kwargs)
-        self.setupUi(self)
-
-        self._shared_selection_signal = shared_selection_signal
-        self._shared_selection_uuid = str(uuid4())
-        self._shared_selection_signal.proc.connect(self._handle_shared_selection_signal)
-        self.device_list.selectionModel().selectionChanged.connect(self._on_selection_changed)
-
-        self.title_text = name  # type: ignore
-        self._mime_data = []
-        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()
-
-    def _add_item(self, device: HashableDevice):
-        item = QListWidgetItem(self.device_list)
-        device_dump = device.model_dump(exclude_defaults=True)
-        item.setData(CONFIG_DATA_ROLE, device_dump)
-        self._mime_data.append(device_dump)
-        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 create_mime_data(self):
-        return self._mime_data
-
-    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,
-        )
-
-    @SafeSlot(QItemSelection, QItemSelection)
-    def _on_selection_changed(self, selected: QItemSelection, deselected: QItemSelection) -> None:
-        self._shared_selection_signal.proc.emit(self._shared_selection_uuid)
-        config = [dev.as_normal_device().model_dump() for dev in self.get_selection()]
-        self.selected_devices.emit(config)
-
-    @SafeSlot(str)
-    def _handle_shared_selection_signal(self, uuid: str):
-        if uuid != self._shared_selection_uuid:
-            self.device_list.clearSelection()
-
-    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 __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())

bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py

@@ -1,56 +0,0 @@
-from typing import TYPE_CHECKING
-
-from qtpy.QtCore import QMetaObject, Qt
-from qtpy.QtWidgets import QFrame, QLabel, QListWidget, QVBoxLayout
-
-from bec_widgets.widgets.control.device_manager.components._util import mimedata_from_configs
-from bec_widgets.widgets.control.device_manager.components.constants import (
-    CONFIG_DATA_ROLE,
-    MIME_DEVICE_CONFIG,
-)
-
-if TYPE_CHECKING:
-    from .available_device_group import AvailableDeviceGroup
-
-
-class _DeviceListWidget(QListWidget):
-
-    def _item_iter(self):
-        return (self.item(i) for i in range(self.count()))
-
-    def all_configs(self):
-        return [item.data(CONFIG_DATA_ROLE) for item in self._item_iter()]
-
-    def mimeTypes(self):
-        return [MIME_DEVICE_CONFIG]
-
-    def mimeData(self, items):
-        return mimedata_from_configs(item.data(CONFIG_DATA_ROLE) for item in items)
-
-
-class Ui_AvailableDeviceGroup(object):
-    def setupUi(self, AvailableDeviceGroup: "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.device_list = _DeviceListWidget(AvailableDeviceGroup)
-        self.device_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
-        self.device_list.setObjectName("device_list")
-        self.device_list.setFrameStyle(0)
-        self.device_list.setDragEnabled(True)
-        self.device_list.setAcceptDrops(False)
-        self.device_list.setDefaultDropAction(Qt.DropAction.CopyAction)
-        self.verticalLayout.addWidget(self.device_list)
-        AvailableDeviceGroup.setFrameStyle(QFrame.Shadow.Plain | QFrame.Shape.Box)
-        QMetaObject.connectSlotsByName(AvailableDeviceGroup)

bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py

@@ -1,128 +0,0 @@
-from random import randint
-from typing import Any, Iterable
-from uuid import uuid4
-
-from qtpy.QtCore import QItemSelection, Signal  # type: ignore
-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._util import (
-    SharedSelectionSignal,
-    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,
-)
-from bec_widgets.widgets.control.device_manager.components.constants import CONFIG_DATA_ROLE
-
-
-class AvailableDeviceResources(BECWidget, QWidget, Ui_availableDeviceResources):
-
-    selected_devices = Signal(list)  # list[dict[str,Any]] of device configs currently selected
-    add_selected_devices = Signal(list)
-    del_selected_devices = Signal(list)
-
-    def __init__(self, parent=None, shared_selection_signal=SharedSelectionSignal(), **kwargs):
-        super().__init__(parent=parent, **kwargs)
-        self.setupUi(self)
-        self._backend = get_backend()
-        self._shared_selection_signal = shared_selection_signal
-        self._shared_selection_uuid = str(uuid4())
-        self._shared_selection_signal.proc.connect(self._handle_shared_selection_signal)
-        self.device_groups_list.selectionModel().selectionChanged.connect(
-            self._on_selection_changed
-        )
-        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)
-
-        self.tb_add_selected.action.triggered.connect(self._add_selected_action)
-        self.tb_del_selected.action.triggered.connect(self._del_selected_action)
-
-    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]):
-        item, widget = self.device_groups_list.add_item(
-            device_group,
-            self.device_groups_list,
-            device_group,
-            devices,
-            shared_selection_signal=self._shared_selection_signal,
-            expanded=False,
-        )
-        item.setData(CONFIG_DATA_ROLE, widget.create_mime_data())
-        # Re-emit the selected items from a subgroup - all other selections should be disabled anyway
-        widget.selected_devices.connect(self.selected_devices)
-
-    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()
-    def _add_selected_action(self):
-        self.add_selected_devices.emit(self.device_groups_list.any_selected_devices())
-
-    @SafeSlot()
-    def _del_selected_action(self):
-        self.del_selected_devices.emit(self.device_groups_list.any_selected_devices())
-
-    @SafeSlot(QItemSelection, QItemSelection)
-    def _on_selection_changed(self, selected: QItemSelection, deselected: QItemSelection) -> None:
-        self.selected_devices.emit(self.device_groups_list.selected_devices_from_groups())
-        self._shared_selection_signal.proc.emit(self._shared_selection_uuid)
-
-    @SafeSlot(str)
-    def _handle_shared_selection_signal(self, uuid: str):
-        if uuid != self._shared_selection_uuid:
-            self.device_groups_list.clearSelection()
-
-    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)
-
-    @SafeSlot(list)
-    def mark_devices_used(self, config_list: list[dict[str, Any]], used: bool):
-        """Set the display color of individual devices and update the group display of numbers
-        included. Accepts a list of dicts with the complete config as used in
-        bec_lib.atlas_models.Device."""
-        self._set_devices_state(
-            yield_only_passing(HashableDevice.model_validate, config_list), used
-        )
-
-    @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())

bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py

@@ -1,135 +0,0 @@
-from __future__ import annotations
-
-import itertools
-
-from qtpy.QtCore import QMetaObject, Qt
-from qtpy.QtWidgets import (
-    QAbstractItemView,
-    QComboBox,
-    QGridLayout,
-    QLabel,
-    QLineEdit,
-    QListView,
-    QListWidget,
-    QListWidgetItem,
-    QSizePolicy,
-    QVBoxLayout,
-)
-
-from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
-from bec_widgets.utils.toolbars.actions import MaterialIconAction
-from bec_widgets.utils.toolbars.bundles import ToolbarBundle
-from bec_widgets.utils.toolbars.toolbar import ModularToolBar
-from bec_widgets.widgets.control.device_manager.components._util import mimedata_from_configs
-from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group import (
-    AvailableDeviceGroup,
-)
-from bec_widgets.widgets.control.device_manager.components.constants import (
-    CONFIG_DATA_ROLE,
-    MIME_DEVICE_CONFIG,
-)
-
-
-class _ListOfDeviceGroups(ListOfExpandableFrames[AvailableDeviceGroup]):
-
-    def itemWidget(self, item: QListWidgetItem) -> AvailableDeviceGroup:
-        return super().itemWidget(item)  # type: ignore
-
-    def any_selected_devices(self):
-        return self.selected_individual_devices() or self.selected_devices_from_groups()
-
-    def selected_individual_devices(self):
-        for widget in (self.itemWidget(self.item(i)) for i in range(self.count())):
-            if (selected := widget.get_selection()) != set():
-                return [dev.as_normal_device().model_dump() for dev in selected]
-        return []
-
-    def selected_devices_from_groups(self):
-        selected_items = (self.item(r.row()) for r in self.selectionModel().selectedRows())
-        widgets = (self.itemWidget(item) for item in selected_items)
-        return list(itertools.chain.from_iterable(w.device_list.all_configs() for w in widgets))
-
-    def mimeTypes(self):
-        return [MIME_DEVICE_CONFIG]
-
-    def mimeData(self, items):
-        return mimedata_from_configs(
-            itertools.chain.from_iterable(item.data(CONFIG_DATA_ROLE) for item in items)
-        )
-
-
-class Ui_availableDeviceResources(object):
-    def setupUi(self, availableDeviceResources):
-        if not availableDeviceResources.objectName():
-            availableDeviceResources.setObjectName("availableDeviceResources")
-        self.verticalLayout = QVBoxLayout(availableDeviceResources)
-        self.verticalLayout.setObjectName("verticalLayout")
-
-        self._add_toolbar()
-
-        # Main area with search and filter using a grid layout
-        self.search_layout = QVBoxLayout()
-        self.grid_layout = QGridLayout()
-
-        self.grouping_selector = QComboBox()
-        self.grouping_selector.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
-        lbl_group = QLabel("Group by:")
-        lbl_group.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
-        self.grid_layout.addWidget(lbl_group, 0, 0)
-        self.grid_layout.addWidget(self.grouping_selector, 0, 1)
-
-        self.search_box = QLineEdit()
-        self.search_box.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
-        lbl_filter = QLabel("Filter:")
-        lbl_filter.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
-        self.grid_layout.addWidget(lbl_filter, 1, 0)
-        self.grid_layout.addWidget(self.search_box, 1, 1)
-
-        self.grid_layout.setColumnStretch(0, 0)
-        self.grid_layout.setColumnStretch(1, 1)
-
-        self.search_layout.addLayout(self.grid_layout)
-        self.verticalLayout.addLayout(self.search_layout)
-
-        self.device_groups_list = _ListOfDeviceGroups(
-            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(4)
-        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.setDefaultDropAction(Qt.DropAction.CopyAction)
-        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)
-
-    def _add_toolbar(self):
-        self.toolbar = ModularToolBar(self)
-        io_bundle = ToolbarBundle("IO", self.toolbar.components)
-
-        self.tb_add_selected = MaterialIconAction(
-            icon_name="add_box", parent=self, tooltip="Add selected devices to composition"
-        )
-        self.toolbar.components.add_safe("add_selected", self.tb_add_selected)
-        io_bundle.add_action("add_selected")
-
-        self.tb_del_selected = MaterialIconAction(
-            icon_name="chips", parent=self, tooltip="Remove selected devices from composition"
-        )
-        self.toolbar.components.add_safe("del_selected", self.tb_del_selected)
-        io_bundle.add_action("del_selected")
-
-        self.verticalLayout.addWidget(self.toolbar)
-        self.toolbar.add_bundle(io_bundle)
-        self.toolbar.show_bundles(["IO"])

bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py

@@ -1,140 +0,0 @@
-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, plugins_installed
-
-logger = bec_logger.logger
-
-# use the last n recovery files
-_N_RECOVERY_FILES = 3
-_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 available 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 available 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()
-        if plugins_installed() == 1:
-            self._raw_device_set.update(
-                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),
-            ),
-            set(),
-        )
-
-    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)), set())
-
-    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), 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}

bec_widgets/widgets/control/device_manager/components/constants.py

@@ -1,11 +1,7 @@
 from typing import Final
 
-# Denotes a MIME type for JSON-encoded list of device config dictionaries
-MIME_DEVICE_CONFIG: Final[str] = "application/x-bec_device_config"
-
 # Custom user roles
 SORT_KEY_ROLE: Final[int] = 117
-CONFIG_DATA_ROLE: Final[int] = 118
 
 # TODO 882 keep in sync with headers in device_table_view.py
 HEADERS_HELP_MD: dict[str, str] = {