+ | 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
new file mode 100644
index 00000000..bea0a1c3
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_group_ui.py
@@ -0,0 +1,56 @@
+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 _DeviceListWiget(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 = _DeviceListWiget(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
new file mode 100644
index 00000000..93e81015
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources.py
@@ -0,0 +1,128 @@
+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
new file mode 100644
index 00000000..05701864
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/available_device_resources_ui.py
@@ -0,0 +1,135 @@
+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
new file mode 100644
index 00000000..145d2110
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py
@@ -0,0 +1,140 @@
+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 availble devices separated by tag groups. The same device may
+ appear more than once (in different groups)."""
+ ...
+
+ @property
+ def all_devices(self) -> set[HashableDevice]:
+ """A set of all availble devices. The same device may not appear more than once."""
+ ...
+
+ @property
+ def untagged_devices(self) -> set[HashableDevice]:
+ """A set of all untagged devices. The same device may not appear more than once."""
+ ...
+
+ @property
+ def allowed_sort_keys(self) -> set[str]:
+ """A set of all fields which you may group devices by"""
+ ...
+
+ def tags(self) -> set[str]:
+ """Returns a set of all the tags in all available devices."""
+ ...
+
+ def tag_group(self, tag: str) -> set[HashableDevice]:
+ """Returns a set of the devices in the tag group with the given key."""
+ ...
+
+ def group_by_key(self, key: str) -> dict[str, set[HashableDevice]]:
+ """Return a dict of all devices, organised by the specified key, which must be one of
+ the string keys in the Device model."""
+ ...
+
+
+def _devices_from_file(file: str, include_source: bool = True):
+ data = yaml_load(file, process_includes=False)
+ return HashableDeviceSet(
+ HashableDevice.model_validate(
+ dev | {"name": name, "source_files": {file} if include_source else set()}
+ )
+ for name, dev in data.items()
+ )
+
+
+class _ConfigFileBackend(DeviceResourceBackend):
+ def __init__(self) -> None:
+ self._raw_device_set: set[HashableDevice] = self._get_config_from_backup_files()
+ 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
new file mode 100644
index 00000000..b3f72051
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/constants.py
@@ -0,0 +1,72 @@
+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] = {
+ "status": "\n".join(
+ [
+ "## Status",
+ "The current status of the device. Can be one of the following values: ",
+ "### **LOADED** \n The device with the specified configuration is loaded in the current config.",
+ "### **CONNECT_READY** \n The device config is valid and the connection has been validated. It has not yet been loaded to the current config.",
+ "### **CONNECT_FAILED** \n The device config is valid, but the connection could not be established.",
+ "### **VALID** \n The device config is valid, but the connection has not yet been validated.",
+ "### **INVALID** \n The device config is invalid and can not be loaded to the current config.",
+ ]
+ ),
+ "name": "\n".join(["## Name ", "The name of the device."]),
+ "deviceClass": "\n".join(
+ [
+ "## Device Class",
+ "The device class specifies the type of the device. It will be used to create the instance.",
+ ]
+ ),
+ "readoutPriority": "\n".join(
+ [
+ "## Readout Priority",
+ "The readout priority of the device. Can be one of the following values: ",
+ "### **monitored** \n The monitored priority is used for devices that are read out during the scan (i.e. at every step) and whose value may change during the scan.",
+ "### **baseline** \n The baseline priority is used for devices that are read out at the beginning of the scan and whose value does not change during the scan.",
+ "### **async** \n The async priority is used for devices that are asynchronous to the monitored devices, and send their data independently.",
+ "### **continuous** \n The continuous priority is used for devices that are read out continuously during the scan.",
+ "### **on_request** \n The on_request priority is used for devices that should not be read out during the scan, yet are configured to be read out manually.",
+ ]
+ ),
+ "deviceTags": "\n".join(
+ [
+ "## Device Tags",
+ "A list of tags associated with the device. Tags can be used to group devices and filter them in the device manager.",
+ ]
+ ),
+ "enabled": "\n".join(
+ [
+ "## Enabled",
+ "Indicator whether the device is enabled or disabled. Disabled devices can not be used.",
+ ]
+ ),
+ "readOnly": "\n".join(
+ ["## Read Only", "Indicator that a device is read-only or can be modified."]
+ ),
+ "onFailure": "\n".join(
+ [
+ "## On Failure",
+ "Specifies the behavior of the device in case of a failure. Can be one of the following values: ",
+ "### **buffer** \n The device readback will fall back to the last known value.",
+ "### **retry** \n The device readback will be retried once, and raises an error if it fails again.",
+ "### **raise** \n The device readback will raise immediately.",
+ ]
+ ),
+ "softwareTrigger": "\n".join(
+ [
+ "## Software Trigger",
+ "Indicator whether the device receives a software trigger from BEC during a scan.",
+ ]
+ ),
+ "description": "\n".join(["## Description", "A short description of the device."]),
+}
diff --git a/bec_widgets/widgets/control/device_manager/components/device_table_view.py b/bec_widgets/widgets/control/device_manager/components/device_table_view.py
index b541916b..886b02c7 100644
--- a/bec_widgets/widgets/control/device_manager/components/device_table_view.py
+++ b/bec_widgets/widgets/control/device_manager/components/device_table_view.py
@@ -4,114 +4,327 @@ from __future__ import annotations
import copy
import json
+import textwrap
+from contextlib import contextmanager
+from functools import partial
+from typing import TYPE_CHECKING, Any, Iterable, List, Literal
+from uuid import uuid4
+from bec_lib.atlas_models import Device
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from qtpy import QtCore, QtGui, QtWidgets
+from qtpy.QtCore import QModelIndex, QPersistentModelIndex, Qt, QTimer
+from qtpy.QtWidgets import QAbstractItemView, QHeaderView, QMessageBox
from thefuzz import fuzz
+from bec_widgets.utils.bec_signal_proxy import BECSignalProxy
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
+from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
+from bec_widgets.widgets.control.device_manager.components.constants import (
+ HEADERS_HELP_MD,
+ MIME_DEVICE_CONFIG,
+)
+from bec_widgets.widgets.control.device_manager.components.dm_ophyd_test import ValidationStatus
+
+if TYPE_CHECKING: # pragma: no cover
+ from bec_qthemes._theme import AccentColors
logger = bec_logger.logger
+_DeviceCfgIter = Iterable[dict[str, Any]]
+
# Threshold for fuzzy matching, careful with adjusting this. 80 seems good
FUZZY_SEARCH_THRESHOLD = 80
+#
+USER_CHECK_DATA_ROLE = 101
+
class DictToolTipDelegate(QtWidgets.QStyledItemDelegate):
"""Delegate that shows all key-value pairs of a rows's data as a YAML-like tooltip."""
- @staticmethod
- def dict_to_str(d: dict) -> str:
- """Convert a dictionary to a formatted string."""
- return json.dumps(d, indent=4)
-
- def helpEvent(self, event, view, option, index):
+ def helpEvent(
+ self,
+ event: QtCore.QEvent,
+ view: QtWidgets.QAbstractItemView,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QModelIndex,
+ ):
"""Override to show tooltip when hovering."""
- if event.type() != QtCore.QEvent.ToolTip:
+ if event.type() != QtCore.QEvent.Type.ToolTip:
return super().helpEvent(event, view, option, index)
model: DeviceFilterProxyModel = index.model()
model_index = model.mapToSource(index)
- row_dict = model.sourceModel().row_data(model_index)
- row_dict.pop("description", None)
- QtWidgets.QToolTip.showText(event.globalPos(), self.dict_to_str(row_dict), view)
+ row_dict = model.sourceModel().get_row_data(model_index)
+ description = row_dict.get("description", "")
+ QtWidgets.QToolTip.showText(event.globalPos(), description, view)
return True
-class CenterCheckBoxDelegate(DictToolTipDelegate):
+class CustomDisplayDelegate(DictToolTipDelegate):
+ _paint_test_role = Qt.ItemDataRole.DisplayRole
+
+ def displayText(self, value: Any, locale: QtCore.QLocale | QtCore.QLocale.Language) -> str:
+ return ""
+
+ def _test_custom_paint(
+ self, painter: QtGui.QPainter, option: QtWidgets.QStyleOptionViewItem, index: QModelIndex
+ ):
+ v = index.model().data(index, self._paint_test_role)
+ return (v is not None), v
+
+ def _do_custom_paint(
+ self,
+ painter: QtGui.QPainter,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QModelIndex,
+ value: Any,
+ ): ...
+
+ def paint(
+ self, painter: QtGui.QPainter, option: QtWidgets.QStyleOptionViewItem, index: QModelIndex
+ ) -> None:
+ (check, value) = self._test_custom_paint(painter, option, index)
+ if not check:
+ return super().paint(painter, option, index)
+ super().paint(painter, option, index)
+ painter.save()
+ self._do_custom_paint(painter, option, index, value)
+ painter.restore()
+
+
+class WrappingTextDelegate(CustomDisplayDelegate):
+ """A lightweight delegate that wraps text without expensive size recalculation."""
+
+ def __init__(self, parent: BECTableView | None = None, max_width: int = 300, margin: int = 6):
+ super().__init__(parent)
+ self._parent = parent
+ self.max_width = max_width
+ self.margin = margin
+ self._cache = {} # cache text metrics for performance
+ self._wrapping_text_columns = None
+
+ @property
+ def wrapping_text_columns(self) -> List[int]:
+ # Compute once, cache for later
+ if self._wrapping_text_columns is None:
+ self._wrapping_text_columns = []
+ view = self._parent
+ proxy: DeviceFilterProxyModel = self._parent.model()
+ for col in range(proxy.columnCount()):
+ delegate = view.itemDelegateForColumn(col)
+ if isinstance(delegate, WrappingTextDelegate):
+ self._wrapping_text_columns.append(col)
+ return self._wrapping_text_columns
+
+ def _do_custom_paint(
+ self,
+ painter: QtGui.QPainter,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QModelIndex,
+ value: str,
+ ):
+ text = str(value)
+ if not text:
+ return
+ painter.save()
+ painter.setClipRect(option.rect)
+
+ # Use cached layout if available
+ cache_key = (text, option.rect.width())
+ layout = self._cache.get(cache_key)
+ if layout is None:
+ layout = self._compute_layout(text, option)
+ self._cache[cache_key] = layout
+
+ # Draw text
+ painter.setPen(option.palette.text().color())
+ layout.draw(painter, option.rect.topLeft())
+ painter.restore()
+
+ def _compute_layout(
+ self, text: str, option: QtWidgets.QStyleOptionViewItem
+ ) -> QtGui.QTextLayout:
+ """Compute and return the text layout for given text and option."""
+ layout = self._get_layout(text, option.font)
+ text_option = QtGui.QTextOption()
+ text_option.setWrapMode(QtGui.QTextOption.WrapAnywhere)
+ layout.setTextOption(text_option)
+ layout.beginLayout()
+ height = 0
+ max_lines = 100 # safety cap, should never be more than 100 lines..
+ for _ in range(max_lines):
+ line = layout.createLine()
+ if not line.isValid():
+ break
+ line.setLineWidth(option.rect.width() - self.margin)
+ line.setPosition(QtCore.QPointF(self.margin / 2, height))
+ line_height = line.height()
+ if line_height <= 0:
+ break # avoid negative or zero height lines to be added
+ height += line_height
+ layout.endLayout()
+ return layout
+
+ def _get_layout(self, text: str, font_option: QtGui.QFont) -> QtGui.QTextLayout:
+ return QtGui.QTextLayout(text, font_option)
+
+ def sizeHint(self, option: QtWidgets.QStyleOptionViewItem, index: QModelIndex) -> QtCore.QSize:
+ """Return a cached or approximate height; avoids costly recomputation."""
+ text = str(index.data(QtCore.Qt.DisplayRole) or "")
+ view = self._parent
+ view.initViewItemOption(option)
+ if view.isColumnHidden(index.column()) or not view.isVisible() or not text:
+ return QtCore.QSize(0, option.fontMetrics.height() + 2 * self.margin)
+
+ # Use cache for consistent size computation
+ cache_key = (text, self.max_width)
+ if cache_key in self._cache:
+ layout = self._cache[cache_key]
+ height = 0
+ for i in range(layout.lineCount()):
+ height += layout.lineAt(i).height()
+ return QtCore.QSize(self.max_width, int(height + self.margin))
+
+ # Approximate without layout (fast path)
+ metrics = option.fontMetrics
+ pixel_width = max(self._parent.columnWidth(index.column()), 100)
+ if pixel_width > 2000: # safeguard against uninitialized columns, may return large values
+ pixel_width = 100
+ char_per_line = self.estimate_chars_per_line(text, option, pixel_width - 2 * self.margin)
+ wrapped_lines = textwrap.wrap(text, width=char_per_line)
+ lines = len(wrapped_lines)
+ return QtCore.QSize(pixel_width, lines * (metrics.height()) + 2 * self.margin)
+
+ def estimate_chars_per_line(
+ self, text: str, option: QtWidgets.QStyleOptionViewItem, column_width: int
+ ) -> int:
+ """Estimate number of characters that fit in a line for given width."""
+ metrics = option.fontMetrics
+ elided = metrics.elidedText(text, Qt.ElideRight, column_width)
+ return len(elided.rstrip("…"))
+
+ @SafeSlot(int, int, int)
+ @SafeSlot(int)
+ def _on_section_resized(
+ self, logical_index: int, old_size: int | None = None, new_size: int | None = None
+ ):
+ """Only update rows if a wrapped column was resized."""
+ self._cache.clear()
+ # Make sure layout is computed first
+ QtCore.QTimer.singleShot(0, self._update_row_heights)
+
+ def _update_row_heights(self):
+ """Efficiently adjust row heights based on wrapped columns."""
+ view = self._parent
+ proxy = view.model()
+ option = QtWidgets.QStyleOptionViewItem()
+ view.initViewItemOption(option)
+ for row in range(proxy.rowCount()):
+ max_height = 18
+ for column in self.wrapping_text_columns:
+ index = proxy.index(row, column)
+ delegate = view.itemDelegateForColumn(column)
+ hint = delegate.sizeHint(option, index)
+ max_height = max(max_height, hint.height())
+ if view.rowHeight(row) != max_height:
+ view.setRowHeight(row, max_height)
+
+
+class CenterCheckBoxDelegate(CustomDisplayDelegate):
"""Custom checkbox delegate to center checkboxes in table cells."""
- def __init__(self, parent=None):
+ _paint_test_role = USER_CHECK_DATA_ROLE
+
+ def __init__(self, parent: BECTableView | None = None, colors: AccentColors | None = None):
super().__init__(parent)
- colors = get_accent_colors()
- self._icon_checked = material_icon(
- "check_box", size=QtCore.QSize(16, 16), color=colors.default
- )
- self._icon_unchecked = material_icon(
- "check_box_outline_blank", size=QtCore.QSize(16, 16), color=colors.default
- )
+ colors: AccentColors = colors if colors else get_accent_colors() # type: ignore
+ _icon = partial(material_icon, size=(16, 16), color=colors.default, filled=True)
+ self._icon_checked = _icon("check_box")
+ self._icon_unchecked = _icon("check_box_outline_blank")
def apply_theme(self, theme: str | None = None):
colors = get_accent_colors()
- self._icon_checked.setColor(colors.default)
- self._icon_unchecked.setColor(colors.default)
+ _icon = partial(material_icon, size=(16, 16), color=colors.default, filled=True)
+ self._icon_checked = _icon("check_box")
+ self._icon_unchecked = _icon("check_box_outline_blank")
- def paint(self, painter, option, index):
- value = index.model().data(index, QtCore.Qt.CheckStateRole)
- if value is None:
- super().paint(painter, option, index)
- return
-
- # Choose icon based on state
- pixmap = self._icon_checked if value == QtCore.Qt.Checked else self._icon_unchecked
-
- # Draw icon centered
- rect = option.rect
+ def _do_custom_paint(
+ self,
+ painter: QtGui.QPainter,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QModelIndex,
+ value: Literal[
+ Qt.CheckState.Checked | Qt.CheckState.Unchecked | Qt.CheckState.PartiallyChecked
+ ],
+ ):
+ pixmap = self._icon_checked if value == Qt.CheckState.Checked else self._icon_unchecked
pix_rect = pixmap.rect()
- pix_rect.moveCenter(rect.center())
+ pix_rect.moveCenter(option.rect.center())
painter.drawPixmap(pix_rect.topLeft(), pixmap)
- def editorEvent(self, event, model, option, index):
- if event.type() != QtCore.QEvent.MouseButtonRelease:
+ def editorEvent(
+ self,
+ event: QtCore.QEvent,
+ model: QtCore.QSortFilterProxyModel,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QModelIndex,
+ ):
+ if event.type() != QtCore.QEvent.Type.MouseButtonRelease:
return False
- current = model.data(index, QtCore.Qt.CheckStateRole)
- new_state = QtCore.Qt.Unchecked if current == QtCore.Qt.Checked else QtCore.Qt.Checked
- return model.setData(index, new_state, QtCore.Qt.CheckStateRole)
+ current = model.data(index, USER_CHECK_DATA_ROLE)
+ new_state = (
+ Qt.CheckState.Unchecked if current == Qt.CheckState.Checked else Qt.CheckState.Checked
+ )
+ return model.setData(index, new_state, USER_CHECK_DATA_ROLE)
-class WrappingTextDelegate(DictToolTipDelegate):
- """Custom delegate for wrapping text in table cells."""
+class DeviceValidatedDelegate(CustomDisplayDelegate):
+ """Custom delegate for displaying validated device configurations."""
- def paint(self, painter, option, index):
- text = index.model().data(index, QtCore.Qt.DisplayRole)
- if not text:
- return super().paint(painter, option, index)
+ def __init__(self, parent: BECTableView | None = None, colors: AccentColors | None = None):
+ super().__init__(parent)
+ colors = colors if colors else get_accent_colors()
+ _icon = partial(material_icon, icon_name="circle", size=(12, 12), filled=True)
+ self._icons = {
+ ValidationStatus.PENDING: _icon(color=colors.default),
+ ValidationStatus.VALID: _icon(color=colors.success),
+ ValidationStatus.FAILED: _icon(color=colors.emergency),
+ }
- painter.save()
- painter.setClipRect(option.rect)
- text_option = QtCore.Qt.TextWordWrap | QtCore.Qt.AlignLeft | QtCore.Qt.AlignTop
- painter.drawText(option.rect.adjusted(4, 2, -4, -2), text_option, text)
- painter.restore()
+ def apply_theme(self, theme: str | None = None):
+ colors = get_accent_colors()
+ _icon = partial(material_icon, icon_name="circle", size=(12, 12), filled=True)
+ self._icons = {
+ ValidationStatus.PENDING: _icon(color=colors.default),
+ ValidationStatus.VALID: _icon(color=colors.success),
+ ValidationStatus.FAILED: _icon(color=colors.emergency),
+ }
- def sizeHint(self, option, index):
- text = str(index.model().data(index, QtCore.Qt.DisplayRole) or "")
- # if not text:
- # return super().sizeHint(option, index)
+ def _do_custom_paint(
+ self,
+ painter: QtGui.QPainter,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QModelIndex,
+ value: Literal[0, 1, 2],
+ ):
+ """
+ Paint the validation status icon centered in the cell.
- # Use the actual column width
- table = index.model().parent() # or store reference to QTableView
- column_width = table.columnWidth(index.column()) # - 8
-
- doc = QtGui.QTextDocument()
- doc.setDefaultFont(option.font)
- doc.setTextWidth(column_width)
- doc.setPlainText(text)
-
- layout_height = doc.documentLayout().documentSize().height()
- height = int(layout_height) + 4 # Needs some extra padding, otherwise it gets cut off
- return QtCore.QSize(column_width, height)
+ Args:
+ painter (QtGui.QPainter): The painter object.
+ option (QtWidgets.QStyleOptionViewItem): The style options for the item.
+ index (QModelIndex): The model index of the item.
+ value (Literal[0,1,2]): The validation status value, where 0=Pending, 1=Valid, 2=Failed.
+ Relates to ValidationStatus enum.
+ """
+ if pixmap := self._icons.get(value):
+ pix_rect = pixmap.rect()
+ pix_rect.moveCenter(option.rect.center())
+ painter.drawPixmap(pix_rect.topLeft(), pixmap)
class DeviceTableModel(QtCore.QAbstractTableModel):
@@ -121,62 +334,86 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
Sort logic is implemented directly on the data of the table view.
"""
- def __init__(self, device_config: list[dict] | None = None, parent=None):
+ # tuple of list[dict[str, Any]] of configs which were added and bool True if added or False if removed
+ configs_changed = QtCore.Signal(list, bool)
+
+ def __init__(self, parent: DeviceTableModel | None = None):
super().__init__(parent)
- self._device_config = device_config or []
+ self._device_config: list[dict[str, Any]] = []
+ self._validation_status: dict[str, ValidationStatus] = {}
+ # TODO 882 keep in sync with HEADERS_HELP_MD
self.headers = [
+ "status",
"name",
"deviceClass",
"readoutPriority",
- "enabled",
- "readOnly",
+ "onFailure",
"deviceTags",
"description",
+ "enabled",
+ "readOnly",
+ "softwareTrigger",
]
self._checkable_columns_enabled = {"enabled": True, "readOnly": True}
+ self._device_model_schema = Device.model_json_schema()
###############################################
- ########## Overwrite custom Qt methods ########
+ ########## Override custom Qt methods #########
###############################################
- def rowCount(self, parent=QtCore.QModelIndex()) -> int:
+ def rowCount(self, parent: QModelIndex | QPersistentModelIndex = QtCore.QModelIndex()) -> int:
return len(self._device_config)
- def columnCount(self, parent=QtCore.QModelIndex()) -> int:
+ def columnCount(
+ self, parent: QModelIndex | QPersistentModelIndex = QtCore.QModelIndex()
+ ) -> int:
return len(self.headers)
- def headerData(self, section, orientation, role=QtCore.Qt.DisplayRole):
- if role == QtCore.Qt.DisplayRole and orientation == QtCore.Qt.Horizontal:
+ def headerData(self, section, orientation, role=int(Qt.ItemDataRole.DisplayRole)):
+ if role == Qt.ItemDataRole.DisplayRole and orientation == Qt.Orientation.Horizontal:
+ if section == 9: # softwareTrigger
+ return "softTrig"
return self.headers[section]
return None
- def row_data(self, index: QtCore.QModelIndex) -> dict:
+ def get_row_data(self, index: QtCore.QModelIndex) -> dict:
"""Return the row data for the given index."""
if not index.isValid():
return {}
return copy.deepcopy(self._device_config[index.row()])
- def data(self, index, role=QtCore.Qt.DisplayRole):
+ def data(self, index, role=int(Qt.ItemDataRole.DisplayRole)):
"""Return data for the given index and role."""
if not index.isValid():
return None
row, col = index.row(), index.column()
- key = self.headers[col]
- value = self._device_config[row].get(key)
- if role == QtCore.Qt.DisplayRole:
- if key in ("enabled", "readOnly"):
+ if col == 0 and role == Qt.ItemDataRole.DisplayRole:
+ dev_name = self._device_config[row].get("name", "")
+ return self._validation_status.get(dev_name, ValidationStatus.PENDING)
+
+ key = self.headers[col]
+ value = self._device_config[row].get(key, None)
+ if value is None:
+ value = (
+ self._device_model_schema.get("properties", {}).get(key, {}).get("default", None)
+ )
+
+ if role == Qt.ItemDataRole.DisplayRole:
+ if key in ("enabled", "readOnly", "softwareTrigger"):
return bool(value)
if key == "deviceTags":
return ", ".join(str(tag) for tag in value) if value else ""
+ if key == "deviceClass":
+ return str(value).split(".")[-1]
return str(value) if value is not None else ""
- if role == QtCore.Qt.CheckStateRole and key in ("enabled", "readOnly"):
- return QtCore.Qt.Checked if value else QtCore.Qt.Unchecked
- if role == QtCore.Qt.TextAlignmentRole:
- if key in ("enabled", "readOnly"):
- return QtCore.Qt.AlignCenter
- return QtCore.Qt.AlignLeft | QtCore.Qt.AlignVCenter
- if role == QtCore.Qt.FontRole:
+ if role == USER_CHECK_DATA_ROLE and key in ("enabled", "readOnly", "softwareTrigger"):
+ return Qt.CheckState.Checked if value else Qt.CheckState.Unchecked
+ if role == Qt.ItemDataRole.TextAlignmentRole:
+ if key in ("enabled", "readOnly", "softwareTrigger"):
+ return Qt.AlignmentFlag.AlignCenter
+ return Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter
+ if role == Qt.ItemDataRole.FontRole:
font = QtGui.QFont()
return font
return None
@@ -184,18 +421,21 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
def flags(self, index):
"""Flags for the table model."""
if not index.isValid():
- return QtCore.Qt.NoItemFlags
+ return Qt.ItemFlag.NoItemFlags
key = self.headers[index.column()]
- if key in ("enabled", "readOnly"):
- base_flags = QtCore.Qt.ItemIsEnabled | QtCore.Qt.ItemIsSelectable
+ base_flags = super().flags(index) | (
+ Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable | Qt.ItemFlag.ItemIsDropEnabled
+ )
+
+ if key in ("enabled", "readOnly", "softwareTrigger"):
if self._checkable_columns_enabled.get(key, True):
- return base_flags | QtCore.Qt.ItemIsUserCheckable
+ return base_flags | Qt.ItemFlag.ItemIsUserCheckable
else:
return base_flags # disable editing but still visible
- return QtCore.Qt.ItemIsEnabled | QtCore.Qt.ItemIsSelectable
+ return base_flags
- def setData(self, index, value, role=QtCore.Qt.EditRole) -> bool:
+ def setData(self, index, value, role=int(Qt.ItemDataRole.EditRole)) -> bool:
"""
Method to set the data of the table.
@@ -210,106 +450,172 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
if not index.isValid():
return False
key = self.headers[index.column()]
- row = index.row()
-
- if key in ("enabled", "readOnly") and role == QtCore.Qt.CheckStateRole:
+ if key in ("enabled", "readOnly", "softwareTrigger") and role == USER_CHECK_DATA_ROLE:
if not self._checkable_columns_enabled.get(key, True):
return False # ignore changes if column is disabled
- self._device_config[row][key] = value == QtCore.Qt.Checked
- self.dataChanged.emit(index, index, [QtCore.Qt.CheckStateRole])
+ self._device_config[index.row()][key] = value == Qt.CheckState.Checked
+ self.dataChanged.emit(index, index, [Qt.ItemDataRole.DisplayRole, USER_CHECK_DATA_ROLE])
return True
return False
+ ####################################
+ ############ Drag and Drop #########
+ ####################################
+
+ def mimeTypes(self) -> List[str]:
+ return [*super().mimeTypes(), MIME_DEVICE_CONFIG]
+
+ def supportedDropActions(self):
+ return Qt.DropAction.CopyAction | Qt.DropAction.MoveAction
+
+ def dropMimeData(self, data, action, row, column, parent):
+ if action not in [Qt.DropAction.CopyAction, Qt.DropAction.MoveAction]:
+ return False
+ if (raw_data := data.data(MIME_DEVICE_CONFIG)) is None:
+ return False
+ self.add_device_configs(json.loads(raw_data.toStdString()))
+ return True
+
####################################
############ Public methods ########
####################################
- def get_device_config(self) -> list[dict]:
- """Return the current device config (with checkbox updates applied)."""
- return self._device_config
+ def get_device_config(self) -> list[dict[str, Any]]:
+ """Method to get the device configuration."""
+ return copy.deepcopy(self._device_config)
- def set_checkbox_enabled(self, column_name: str, enabled: bool):
+ def device_names(self, configs: _DeviceCfgIter | None = None) -> set[str]:
+ _configs = self._device_config if configs is None else configs
+ return set(cfg.get("name") for cfg in _configs if cfg.get("name") is not None) # type: ignore
+
+ def _name_exists_in_config(self, name: str, exists: bool):
+ if (name in self.device_names()) == exists:
+ return True
+ return not exists
+
+ def add_device_configs(self, device_configs: _DeviceCfgIter):
"""
- Enable/Disable the checkbox column.
+ Add devices to the model.
Args:
- column_name (str): The name of the column to modify.
- enabled (bool): Whether the checkbox should be enabled or disabled.
+ device_configs (_DeviceCfgList): An iterable of device configurations to add.
"""
- if column_name in self._checkable_columns_enabled:
- self._checkable_columns_enabled[column_name] = enabled
- col = self.headers.index(column_name)
- top_left = self.index(0, col)
- bottom_right = self.index(self.rowCount() - 1, col)
- self.dataChanged.emit(
- top_left, bottom_right, [QtCore.Qt.CheckStateRole, QtCore.Qt.DisplayRole]
- )
+ already_in_list = []
+ added_configs = []
+ for cfg in device_configs:
+ if self._name_exists_in_config(name := cfg.get("name", "