diff --git a/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py b/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py index 0efbc50c..c6c15ac5 100644 --- a/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py +++ b/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) diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 214a10ac..f04deac4 100644 --- a/bec_widgets/cli/client.py +++ b/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" diff --git a/bec_widgets/utils/expandable_frame.py b/bec_widgets/utils/expandable_frame.py deleted file mode 100644 index 08a4d95f..00000000 --- a/bec_widgets/utils/expandable_frame.py +++ /dev/null @@ -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"{title}" - 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() diff --git a/bec_widgets/utils/list_of_expandable_frames.py b/bec_widgets/utils/list_of_expandable_frames.py deleted file mode 100644 index 7ad85a71..00000000 --- a/bec_widgets/utils/list_of_expandable_frames.py +++ /dev/null @@ -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() diff --git a/bec_widgets/widgets/control/device_manager/components/_util.py b/bec_widgets/widgets/control/device_manager/components/_util.py index 8c106b7b..9e47930f 100644 --- a/bec_widgets/widgets/control/device_manager/components/_util.py +++ b/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) 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 deleted file mode 100644 index 83d4d4d0..00000000 --- a/bec_widgets/widgets/control/device_manager/components/available_device_resources/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .available_device_resources import AvailableDeviceResources - -__all__ = ["AvailableDeviceResources"] 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 deleted file mode 100644 index bf5746c9..00000000 --- a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group.py +++ /dev/null @@ -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""" -

{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): - - 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()) 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 deleted file mode 100644 index f6fe77ad..00000000 --- a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py +++ /dev/null @@ -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) 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 deleted file mode 100644 index 93e81015..00000000 --- a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py +++ /dev/null @@ -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()) 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 deleted file mode 100644 index 05701864..00000000 --- a/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py +++ /dev/null @@ -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"]) 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 deleted file mode 100644 index 862798ee..00000000 --- a/bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py +++ /dev/null @@ -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} diff --git a/bec_widgets/widgets/control/device_manager/components/constants.py b/bec_widgets/widgets/control/device_manager/components/constants.py index 8ac82f0b..89fa67e9 100644 --- a/bec_widgets/widgets/control/device_manager/components/constants.py +++ b/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] = {