+ | 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 000000000..bea0a1c34
--- /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 000000000..93e810156
--- /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 000000000..05701864a
--- /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 000000000..145d21109
--- /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 000000000..8ac82f0b4
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/constants.py
@@ -0,0 +1,113 @@
+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] = {
+ "valid": {
+ "long": "\n".join(
+ [
+ "## Valid",
+ "The current configuration status of the device. Can be one of the following values: ",
+ "### **VALID** \n The device configuration is valid and can be used.",
+ "### **INVALID** \n The device configuration is invalid.",
+ "### **UNKNOWN** \n The device configuration has not been validated yet.",
+ ]
+ ),
+ "short": "Validation status of the device configuration.",
+ },
+ "connect": {
+ "long": "\n".join(
+ [
+ "## Connect",
+ "The current connection status of the device. Can be one of the following values: ",
+ "### **CONNECTED** \n The device is connected and in current session.",
+ "### **CAN_CONNECT** \n The connection to the device has been validated. It's not yet loaded in the current session.",
+ "### **CANNOT_CONNECT** \n The connection to the device could not be established.",
+ "### **UNKNOWN** \n The connection status of the device is unknown.",
+ ]
+ ),
+ "short": "Connection status of the device.",
+ },
+ "name": {
+ "long": "\n".join(["## Name ", "The name of the device."]),
+ "short": "Name of the device.",
+ },
+ "deviceClass": {
+ "long": "\n".join(
+ [
+ "## Device Class",
+ "The device class specifies the type of the device. It will be used to create the instance.",
+ ]
+ ),
+ "short": "Python class for the device.",
+ },
+ "readoutPriority": {
+ "long": "\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.",
+ ]
+ ),
+ "short": "Readout priority of the device for scans in BEC.",
+ },
+ "deviceTags": {
+ "long": "\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.",
+ ]
+ ),
+ "short": "Tags associated with the device.",
+ },
+ "enabled": {
+ "long": "\n".join(
+ [
+ "## Enabled",
+ "Indicator whether the device is enabled or disabled. Disabled devices can not be used.",
+ ]
+ ),
+ "short": "Enabled status of the device.",
+ },
+ "readOnly": {
+ "long": "\n".join(
+ ["## Read Only", "Indicator that a device is read-only or can be modified."]
+ ),
+ "short": "Read-only status of the device.",
+ },
+ "onFailure": {
+ "long": "\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.",
+ ]
+ ),
+ "short": "On failure behavior of the device.",
+ },
+ "softwareTrigger": {
+ "long": "\n".join(
+ [
+ "## Software Trigger",
+ "Indicator whether the device receives a software trigger from BEC during a scan.",
+ ]
+ ),
+ "short": "Software trigger status of the device.",
+ },
+ "description": {
+ "long": "\n".join(["## Description", "A short description of the device."]),
+ "short": "Description of the device.",
+ },
+}
diff --git a/bec_widgets/widgets/control/device_manager/components/device_config_template/__init__.py b/bec_widgets/widgets/control/device_manager/components/device_config_template/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/bec_widgets/widgets/control/device_manager/components/device_config_template/device_config_template.py b/bec_widgets/widgets/control/device_manager/components/device_config_template/device_config_template.py
new file mode 100644
index 000000000..508cef68b
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/device_config_template/device_config_template.py
@@ -0,0 +1,519 @@
+"""Module for the device configuration form widget for EpicsMotor, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV"""
+
+from copy import deepcopy
+from typing import Any, Type
+
+from bec_lib.atlas_models import Device as DeviceModel
+from bec_lib.logger import bec_logger
+from ophyd_devices.interfaces.device_config_templates.ophyd_templates import OPHYD_DEVICE_TEMPLATES
+from pydantic import BaseModel
+from pydantic_core import PydanticUndefinedType
+from qtpy import QtWidgets
+
+from bec_widgets.widgets.control.device_manager.components.device_config_template.template_items import (
+ DEVICE_CONFIG_FIELDS,
+ DEVICE_FIELDS,
+ DeviceConfigField,
+ DeviceTagsWidget,
+ InputLineEdit,
+ LimitInputWidget,
+ OnFailureComboBox,
+ ParameterValueWidget,
+ ReadoutPriorityComboBox,
+)
+from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
+
+logger = bec_logger.logger
+
+
+class DeviceConfigTemplate(QtWidgets.QWidget):
+ """
+ Device Configuration Template Widget.
+ Current supported templates follow the structure in
+ ophyd_devices.interfaces.device_config_templates.ophyd_templates.OPHYD_DEVICE_TEMPLATES.
+
+ Args:
+ parent (QtWidgets.QWidget, optional) : Parent widget. Defaults to None.
+ client (BECClient, optional) : BECClient instance. Defaults to None.
+ template (dict[str, any], optional) : Device configuration template. If None,
+ the "CustomDevice" template will be used. Defaults to None.
+ """
+
+ RPC = False
+
+ def __init__(self, parent=None, template: dict[str, any] = None):
+ super().__init__(parent=parent)
+ if template is None:
+ template = OPHYD_DEVICE_TEMPLATES["CustomDevice"]["CustomDevice"]
+ self.template = template
+ self._device_fields = deepcopy(DEVICE_FIELDS)
+ self._device_config_fields = deepcopy(DEVICE_CONFIG_FIELDS)
+ self._unknown_device_config_entry: dict[str, any] = {}
+
+ # Dict to store references to input widgets
+ self._widgets: dict[str, QtWidgets.QWidget] = {}
+
+ # Two column layout
+ layout = QtWidgets.QHBoxLayout(self)
+ layout.setContentsMargins(2, 0, 2, 0)
+ layout.setSpacing(2)
+ self.setLayout(layout)
+
+ # Left hand side, settings, connection and advanced settings
+ self._left_layout = QtWidgets.QVBoxLayout()
+ self._left_layout.setContentsMargins(2, 2, 2, 2)
+ self._left_layout.setSpacing(4)
+ # Settings box, name | deviceClass | description
+ self.settings_box = self._create_settings_box()
+ # Device Config settings box | dynamic fields from deviceConfig
+ self.connection_settings_box = self._create_connection_settings_box()
+ # Advanced Control box | readoutPriority | onFailure | softwareTrigger | enabled | readOnly
+ self.advanced_control_box = self._create_advanced_control_box()
+ # Add boxes to left layout
+ self._left_layout.addWidget(self.settings_box)
+ self._left_layout.addWidget(self.connection_settings_box)
+ self._left_layout.addWidget(self.advanced_control_box)
+ layout.addLayout(self._left_layout)
+
+ # Right hand side, advanced settings
+ self._right_layout = QtWidgets.QVBoxLayout()
+ self._right_layout.setContentsMargins(2, 2, 2, 2)
+ self._right_layout.setSpacing(4)
+ layout.addLayout(self._right_layout)
+ # Create Additional Settings box
+ self.additional_settings_box = self.create_additional_settings()
+ self._right_layout.addWidget(self.additional_settings_box)
+
+ # Set default values
+ self.reset_to_defaults()
+
+ def _clear_layout(self, layout: QtWidgets.QLayout) -> None:
+ """Clear a layout recursively. If the layout contains sub-layouts, they will also be cleared."""
+ while layout.count():
+ item = layout.takeAt(0)
+ if item.widget():
+ item.widget().close()
+ item.widget().deleteLater()
+ if item.layout():
+ self._clear_layout(item.layout())
+
+ def reset_to_defaults(self) -> None:
+ """Reset all fields to default values."""
+ self._widgets.pop("deviceConfig", None)
+ self._clear_layout(self.connection_settings_box.layout())
+
+ # Recreate Connection Settings box
+ layout: QtWidgets.QGridLayout = self.connection_settings_box.layout()
+ self._fill_connection_settings_box(self.connection_settings_box, layout)
+
+ # Reset Settings and Advanced Control boxes
+ for field_name, widget in self._widgets.items():
+ if field_name in self.template:
+ self._set_value_for_widget(widget, self.template[field_name])
+ else:
+ self._set_default_entry(field_name, widget)
+
+ def change_template(self, template: dict[str, any]) -> None:
+ """
+ Change the template and update the form fields accordingly.
+
+ Args:
+ template (dict[str, any]): New device configuration template.
+ """
+ self.template = template
+ self.reset_to_defaults()
+
+ def get_config_fields(self) -> dict:
+ """Retrieve the current configuration from the input fields."""
+ config: dict[str, any] = {}
+ for device_entry, widget in self._widgets.items():
+ config[device_entry] = self._get_entry_for_widget(widget)
+ if self._unknown_device_config_entry:
+ if "deviceConfig" not in config:
+ config["deviceConfig"] = {}
+ config["deviceConfig"].update(self._unknown_device_config_entry)
+ return config
+
+ def set_config_fields(self, config: dict) -> None:
+ """
+ Set the configuration fields based on the provided config dictionary.
+
+ Args:
+ config (dict): Configuration dictionary to set the fields.
+ """
+ # Clear storage for unknown entries
+ self._unknown_device_config_entry.clear()
+ if self.template.get("deviceClass", "") != config.get("deviceClass", ""):
+ logger.warning(
+ f"Device class {config.get('deviceClass', '')} does not match template device class {self.template.get('deviceClass', '')}. Using custom device template."
+ )
+ self.change_template(OPHYD_DEVICE_TEMPLATES["CustomDevice"]["CustomDevice"])
+ else:
+ self.reset_to_defaults()
+ self._fill_fields_from_config(config)
+
+ def _fill_fields_from_config(self, model: dict) -> None:
+ """
+ Fill the form fields base on the provided configuration dictionary.
+ Please note, deviceConfig is handled separately through _fill_connection_settings_box
+ as this depends on the template used.
+
+ Args:
+ model (dict): Configuration dictionary to fill the fields.
+ """
+ for key, value in model.items():
+ if key == "name":
+ wid = self._widgets["name"]
+ wid.setText(value or "")
+ elif key == "deviceClass":
+ wid = self._widgets["deviceClass"]
+ wid.setText(value or "")
+ if "deviceClass" in self.template:
+ wid.setEnabled(False)
+ else:
+ wid.setEnabled(True)
+ elif key == "deviceConfig" and isinstance(
+ self._widgets.get("deviceConfig", None), dict
+ ):
+ # If _widgets["deviceConfig"] is a dict, we have individual widgets for each field
+ for sub_key, sub_value in value.items():
+ widget = self._widgets["deviceConfig"].get(sub_key, None)
+ if widget is None:
+ logger.warning(
+ f"Widget for key {sub_key} not found in deviceConfig widgets."
+ )
+ # Store any unknown entry fields
+ self._unknown_device_config_entry[sub_key] = sub_value
+ continue
+ self._set_value_for_widget(widget, sub_value)
+ else:
+ widget = self._widgets.get(key, None)
+ if widget is not None:
+ self._set_value_for_widget(widget, value)
+
+ def _set_value_for_widget(self, widget: QtWidgets.QWidget, value: Any) -> None:
+ """
+ Set the value for a widget based on its type.
+
+ Args:
+ widget (QtWidgets.QWidget): The widget to set the value for.
+ value (any): The value to set.
+ """
+ if isinstance(widget, (ParameterValueWidget)) and isinstance(value, dict):
+ for param, val in value.items():
+ widget.add_parameter_line(param, val)
+ elif isinstance(widget, DeviceTagsWidget) and isinstance(value, (list, tuple, set)):
+ for tag in value:
+ widget.add_parameter_line(tag or "")
+ elif isinstance(widget, InputLineEdit):
+ widget.setText(str(value or ""))
+ elif isinstance(widget, ToggleSwitch):
+ widget.setChecked(bool(value))
+ elif isinstance(widget, LimitInputWidget):
+ widget.set_limits(value)
+ elif isinstance(widget, QtWidgets.QComboBox):
+ index = widget.findText(value)
+ if index != -1:
+ widget.setCurrentIndex(index)
+ elif isinstance(widget, QtWidgets.QTextEdit):
+ widget.setPlainText(str(value or ""))
+ else:
+ logger.warning(f"Unsupported widget type for setting value: {type(widget)}")
+
+ def _get_entry_for_widget(self, widget: QtWidgets.QWidget) -> any:
+ """
+ Get the value from a widget based on its type.
+
+ Args:
+ widget (QtWidgets.QWidget): The widget to get the value from.
+ Returns:
+ any: The value retrieved from the widget.
+ """
+ if isinstance(widget, (ParameterValueWidget, DeviceTagsWidget)):
+ return widget.parameters()
+ elif isinstance(widget, InputLineEdit):
+ return widget.text().strip()
+ elif isinstance(widget, ToggleSwitch):
+ return widget.isChecked()
+ elif isinstance(widget, LimitInputWidget):
+ return widget.get_limits()
+ elif isinstance(widget, QtWidgets.QComboBox):
+ return widget.currentText()
+ elif isinstance(widget, QtWidgets.QTextEdit):
+ return widget.toPlainText()
+ elif isinstance(widget, dict):
+ result = {}
+ for sub_entry, sub_widget in widget.items():
+ result[sub_entry] = self._get_entry_for_widget(sub_widget)
+ return result
+ else:
+ logger.warning(f"Unsupported widget type for getting entry: {type(widget)}")
+ return None
+
+ def _create_device_field(
+ self, field_name: str, field_info: DeviceConfigField | None = None
+ ) -> tuple[QtWidgets.QLabel, QtWidgets.QWidget]:
+ """
+ Create a device field based on the field name. If field_info is not provided,
+ a default label and input widget will be created.
+
+ Args:
+ field_name (str): Name of the field.
+ field_info (DeviceConfigField | None, optional): Information about the field. Defaults to None.
+ """
+ if field_info is None:
+ label = QtWidgets.QLabel(field_name, parent=self)
+ input_widget = QtWidgets.QLineEdit(parent=self)
+ return label, input_widget
+
+ label_text = field_info.label
+ label = QtWidgets.QLabel(label_text, parent=self)
+ if field_info.required:
+ label_text = label.text()
+ label_text += " *"
+ label.setText(label_text)
+ label.setStyleSheet("font-weight: bold;")
+ input_widget = field_info.widget_cls(parent=self)
+ if field_info.placeholder_text:
+ if hasattr(input_widget, "setPlaceholderText"):
+ input_widget.setPlaceholderText(field_info.placeholder_text)
+ if field_info.static:
+ input_widget.setEnabled(False)
+ if field_info.validation_callback:
+ # Attach validation callback if provided
+ if isinstance(input_widget, InputLineEdit):
+ input_widget: InputLineEdit
+ for callback in field_info.validation_callback:
+ input_widget.register_validation_callback(callback)
+ if field_info.default is not None:
+ # Set default value
+ if isinstance(input_widget, QtWidgets.QLineEdit):
+ input_widget.setText(str(field_info.default))
+ elif isinstance(input_widget, QtWidgets.QTextEdit):
+ input_widget.setPlainText(str(field_info.default))
+ elif isinstance(input_widget, ToggleSwitch):
+ input_widget.setChecked(bool(field_info.default))
+ elif isinstance(input_widget, (ReadoutPriorityComboBox, OnFailureComboBox)):
+ index = input_widget.findText(field_info.default)
+ if index != -1:
+ input_widget.setCurrentIndex(index)
+ return label, input_widget
+
+ def _create_group_box_with_grid_layout(
+ self, title: str
+ ) -> tuple[QtWidgets.QGroupBox, QtWidgets.QGridLayout]:
+ """Create a group box with a grid layout."""
+ box = QtWidgets.QGroupBox(title)
+ layout = QtWidgets.QGridLayout(box)
+ layout.setContentsMargins(4, 8, 4, 8)
+ layout.setSpacing(4)
+ box.setLayout(layout)
+ return box, layout
+
+ def _set_default_entry(self, field_name: str, widget: QtWidgets.QWidget) -> None:
+ """
+ Set the default value for a given field in the form based on the Pydantic model.
+
+ Args:
+ field_name (str): Name of the field.
+ widget (QtWidgets.QWidget): The widget to set the default value for.
+ """
+ if field_name == "enabled":
+ widget.setChecked(True)
+ return
+ if field_name == "readOnly":
+ widget.setChecked(False)
+ return
+ default = self._get_default_for_device_config_field(field_name) or ""
+ widget.setEnabled(True)
+ if isinstance(widget, QtWidgets.QComboBox):
+ index = widget.findText(default)
+ if index != -1:
+ widget.setCurrentIndex(index)
+ elif isinstance(widget, (QtWidgets.QTextEdit, QtWidgets.QLineEdit)):
+ widget.setText(str(default))
+ elif isinstance(widget, ToggleSwitch):
+ widget.setChecked(bool(default))
+ elif isinstance(widget, (ParameterValueWidget, DeviceTagsWidget)):
+ widget.clear_widget()
+
+ def _get_default_for_device_config_field(self, field_name: str) -> any:
+ """
+ Get the default value for a given deviceConfig field based on the Pydantic model.
+
+ Args:
+ field_name (str): Name of the deviceConfig field.
+ Returns:
+ any: The default value for the field, or None if not found.
+ """
+ model_properties: dict = DeviceModel.model_json_schema()["properties"]
+ if field_name in model_properties:
+ field_info = model_properties[field_name]
+ default = field_info.get("default", None)
+ if default:
+ return default
+ return None
+
+ ### Box creation methods ###
+
+ def _create_box(self, box_title: str, field_names: list[str]) -> QtWidgets.QGroupBox:
+ """
+ Create a box layout with specific fields. If field_names are in _device_fields,
+ their corresponding widgets will be used.
+ """
+ # Create box
+ box, layout = self._create_group_box_with_grid_layout(box_title)
+ box.setLayout(layout)
+
+ for ii, field_name in enumerate(field_names):
+ label, input_widget = self._create_device_field(
+ field_name, self._device_fields.get(field_name, None)
+ )
+ layout.addWidget(label, ii, 0)
+ layout.addWidget(input_widget, ii, 1)
+ self._widgets[field_name] = input_widget
+ return box
+
+ def _create_settings_box(self) -> QtWidgets.QGroupBox:
+ """Create the settings box widget."""
+ box = self._create_box("Settings", ["name", "deviceClass", "description"])
+ layout = box.layout()
+ # Set column stretch
+ layout.setColumnStretch(0, 0)
+ layout.setColumnStretch(1, 1)
+ return box
+
+ def _create_advanced_control_box(self) -> QtWidgets.QGroupBox:
+ """Create the advanced control box widget."""
+ # Set up advanced control box
+ box = self._create_box("Advanced Control", ["readoutPriority", "onFailure"])
+ layout = box.layout()
+ for ii, field_name in enumerate(["enabled", "readOnly", "softwareTrigger"]):
+ label, input_widget = self._create_device_field(
+ field_name, self._device_fields.get(field_name, None)
+ )
+ layout.addWidget(label, ii, 2)
+ layout.addWidget(input_widget, ii, 3)
+ self._widgets[field_name] = input_widget
+ return box
+
+ def _create_connection_settings_box(self) -> QtWidgets.QGroupBox:
+ """Create the connection settings box widget. These are all entries in the deviceConfig field."""
+ box, layout = self._create_group_box_with_grid_layout("Connection Settings")
+ box = self._fill_connection_settings_box(box, layout)
+ return box
+
+ def _fill_connection_settings_box(
+ self, box: QtWidgets.QGroupBox, layout: QtWidgets.QGridLayout
+ ) -> QtWidgets.QGroupBox:
+ """Fill the connection settings box based on the deviceConfig template."""
+ if not self.template.get("deviceConfig", {}):
+ widget = ParameterValueWidget(parent=self)
+ widget.setToolTip(
+ "Add custom deviceConfig entries as key-value pairs in the tree view."
+ )
+ layout.addWidget(widget, 0, 0)
+ self._widgets["deviceConfig"] = widget
+ return box
+ # If template specifies deviceConfig fields, create them
+ self._widgets["deviceConfig"] = {}
+ model: Type[BaseModel] = self.template["deviceConfig"]
+ for field_name, field in model.model_fields.items():
+ field_info = self._device_config_fields.get(field_name, None)
+ default = field.get_default()
+ if isinstance(default, PydanticUndefinedType):
+ default = None
+ if field_info:
+ if field.is_required():
+ field_info.required = True
+ if field.description:
+ field_info.placeholder_text = field.description
+ if default is not None:
+ field_info.default = default
+ label, input_widget = self._create_device_field(field_name, field_info)
+ row = layout.rowCount()
+ layout.addWidget(label, row, 0)
+ layout.addWidget(input_widget, row, 1)
+ self._widgets["deviceConfig"][field_name] = input_widget
+ return box
+
+ def create_additional_settings(self) -> QtWidgets.QGroupBox:
+ """Create the additional settings box widget."""
+ box, layout = self._create_group_box_with_grid_layout("Additional Settings")
+ toolbox = QtWidgets.QToolBox(parent=self)
+ layout.addWidget(toolbox, 0, 0)
+ user_parameters_widget = ParameterValueWidget(parent=self)
+ self._widgets["userParameter"] = user_parameters_widget
+ toolbox.addItem(user_parameters_widget, "User Parameter")
+ device_tags_widget = DeviceTagsWidget(parent=self)
+ toolbox.addItem(device_tags_widget, "Device Tags")
+ toolbox.setCurrentIndex(1)
+ self._widgets["deviceTags"] = device_tags_widget
+ return box
+
+
+if __name__ == """__main__""": # pragma: no cover
+ import sys
+
+ app = QtWidgets.QApplication(sys.argv)
+ import yaml
+ from bec_qthemes import apply_theme
+
+ from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
+
+ apply_theme("light")
+
+ class TestWidget(QtWidgets.QWidget):
+ pass
+
+ w = TestWidget()
+ w_layout = QtWidgets.QVBoxLayout(w)
+ w_layout.setContentsMargins(0, 0, 0, 0)
+ w_layout.setSpacing(20)
+ dark_mode_button = DarkModeButton()
+ w_layout.addWidget(dark_mode_button)
+ test_motor = "EpicsMotor"
+ config_form = DeviceConfigTemplate(template=OPHYD_DEVICE_TEMPLATES[test_motor][test_motor])
+ w_layout.addWidget(config_form)
+ button_layout = QtWidgets.QHBoxLayout()
+ button = QtWidgets.QPushButton("Get Config")
+ button.clicked.connect(
+ lambda: print("Device Config:", yaml.dump(config_form.get_config_fields(), indent=4))
+ )
+ button_layout.addWidget(button)
+ button2 = QtWidgets.QPushButton("Reset")
+ button2.clicked.connect(config_form.reset_to_defaults)
+ button_layout.addWidget(button2)
+ combo = QtWidgets.QComboBox()
+ combo_keys = [
+ "EpicsMotor",
+ "EpicsSignal",
+ "EpicsSignalRO",
+ "EpicsSignalWithRBV",
+ "CustomDevice",
+ ]
+ combo.addItems(combo_keys)
+ combo.setCurrentText(test_motor)
+
+ def text_changed(text: str) -> None:
+ if text.startswith("EpicsMotor"):
+ if text == "EpicsMotor":
+ template = OPHYD_DEVICE_TEMPLATES[text][text]
+ else:
+ template = OPHYD_DEVICE_TEMPLATES["EpicsMotor"][text]
+ elif text.startswith("EpicsSignal"):
+ if text == "EpicsSignal":
+ template = OPHYD_DEVICE_TEMPLATES[text][text]
+ else:
+ template = OPHYD_DEVICE_TEMPLATES["EpicsSignal"][text]
+ else:
+ template = OPHYD_DEVICE_TEMPLATES["CustomDevice"]["CustomDevice"]
+ config_form.change_template(template)
+
+ combo.currentTextChanged.connect(text_changed)
+ button_layout.addWidget(button)
+ button_layout.addWidget(combo)
+ w_layout.addLayout(button_layout)
+ w.resize(1200, 600)
+ w.show()
+ sys.exit(app.exec_())
diff --git a/bec_widgets/widgets/control/device_manager/components/device_config_template/template_items.py b/bec_widgets/widgets/control/device_manager/components/device_config_template/template_items.py
new file mode 100644
index 000000000..9dc661589
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/device_config_template/template_items.py
@@ -0,0 +1,481 @@
+"""Module for custom input widgets used in device configuration templates."""
+
+from ast import literal_eval
+from typing import Any, Callable
+
+from bec_lib.logger import bec_logger
+from bec_qthemes import material_icon
+from pydantic import BaseModel, ConfigDict
+from qtpy import QtWidgets
+
+from bec_widgets.utils.colors import get_accent_colors
+from bec_widgets.widgets.control.scan_control.scan_group_box import ScanDoubleSpinBox
+from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
+
+logger = bec_logger.logger
+
+
+def _try_literal_eval(value: str) -> Any:
+ """Consolidated function for literal evaluation of a value."""
+ if value in ["true", "True"]:
+ return True
+ if value in ["false", "False"]:
+ return False
+ if value == "":
+ return ""
+ try:
+ return literal_eval(f"{value}")
+ except ValueError:
+ return value
+ except Exception:
+ logger.warning(f"Could not literal_eval value: {value}, returning as string")
+ return value
+
+
+class InputLineEdit(QtWidgets.QLineEdit):
+ """
+ Custom QLineEdit for input fields with validation.
+
+ Args:
+ parent (QtWidgets.QWidget, optional): Parent widget. Defaults to None.
+ config_field (str, optional): Configuration field name. Defaults to "no_field_specified"
+ required (bool, optional): Whether the field is required. Defaults to True.
+ placeholder_text (str, optional): Placeholder text for the input field. Defaults to "".
+ """
+
+ def __init__(
+ self,
+ parent=None,
+ config_field: str = "no_field_specified",
+ required: bool = True,
+ placeholder_text: str = "",
+ ):
+ super().__init__(parent)
+ self._config_field = config_field
+ self._colors = get_accent_colors()
+ self._required = required
+ self.textChanged.connect(self._update_input_field_style)
+ self._validation_callbacks: list[Callable[[bool], str]] = []
+ self.setPlaceholderText(placeholder_text)
+ self._update_input_field_style()
+
+ def register_validation_callback(self, callback: Callable[[str], bool]) -> None:
+ """
+ Register a custom validation callback.
+
+ Args:
+ callback (Callable[[str], bool]): A function that takes the input string
+ and returns True if valid, False otherwise.
+ """
+ self._validation_callbacks.append(callback)
+
+ def apply_theme(self, theme: str) -> None:
+ """Apply the theme to the widget."""
+ self._colors = get_accent_colors()
+ self._update_input_field_style()
+
+ def _update_input_field_style(self) -> None:
+ """Update the input field style based on validation."""
+ name = self.text()
+ if not self.is_valid_input(name) and self._required is True:
+ self.setStyleSheet(f"border: 1px solid {self._colors.emergency.name()};")
+ return
+ self.setStyleSheet("")
+ return
+
+ def is_valid_input(self, name: str) -> bool:
+ """Validate the input string using plugin helper."""
+ name = name.strip() # Remove leading/trailing whitespace
+ # Run registered validation callbacks
+ for callback in self._validation_callbacks:
+ try:
+ valid = callback(name)
+ except Exception as exc:
+ logger.warning(
+ f"Validation callback raised an exception: {exc}. Defaulting to valid"
+ )
+ valid = True
+ if not valid:
+ return False
+ if not self._required:
+ return True
+ if not name:
+ return False
+ return True
+
+
+class OnFailureComboBox(QtWidgets.QComboBox):
+ """Custom QComboBox for the onFailure input field."""
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ self.addItems(["buffer", "retry", "raise"])
+
+
+class ReadoutPriorityComboBox(QtWidgets.QComboBox):
+ """Custom QComboBox for the readoutPriority input field."""
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ self.addItems(["monitored", "baseline", "async", "continuous", "on_request"])
+
+
+class LimitInputWidget(QtWidgets.QWidget):
+ """Custom widget for inputting limits as a tuple (min, max)."""
+
+ def __init__(self, parent=None, **kwargs):
+ super().__init__(parent)
+ self._layout = QtWidgets.QHBoxLayout(self)
+ self._layout.setContentsMargins(0, 0, 0, 0)
+ self._layout.setSpacing(4)
+
+ # Colors
+ self._colors = get_accent_colors()
+
+ self.min_input = ScanDoubleSpinBox(self, arg_name="min_limit", default=0.0)
+ self.min_input.setPrefix("Min: ")
+ self.min_input.setEnabled(False)
+ self.min_input.setRange(-1e12, 1e12)
+ self._layout.addWidget(self.min_input)
+
+ self.max_input = ScanDoubleSpinBox(self, arg_name="max_limit", default=0.0)
+ self.max_input.setPrefix("Max: ")
+ self.max_input.setRange(-1e12, 1e12)
+ self.max_input.setEnabled(False)
+ self._layout.addWidget(self.max_input)
+
+ # Add validity checks
+ self.min_input.valueChanged.connect(self._check_valid_inputs)
+ self.max_input.valueChanged.connect(self._check_valid_inputs)
+
+ # Add checkbox to enable/disable limits
+ self.enable_toggle = ToggleSwitch(self)
+ self.enable_toggle.setToolTip("Enable editing limits")
+ self.enable_toggle.setChecked(False)
+ self.enable_toggle.enabled.connect(self._toggle_limits_enabled)
+ self._layout.addWidget(self.enable_toggle)
+
+ def reset_defaults(self) -> None:
+ """Reset limits to default values."""
+ self.min_input.setValue(0.0)
+ self.max_input.setValue(0.0)
+ self.enable_toggle.setChecked(False)
+
+ def _is_valid_limit(self) -> bool:
+ """Check if the current limits are valid (min < max)."""
+ return self.min_input.value() <= self.max_input.value()
+
+ def _check_valid_inputs(self) -> None:
+ """Check if the current inputs are valid and update styles accordingly."""
+ if not self._is_valid_limit():
+ self.min_input.setStyleSheet(f"border: 1px solid {self._colors.emergency.name()};")
+ self.max_input.setStyleSheet(f"border: 1px solid {self._colors.emergency.name()};")
+ else:
+ self.min_input.setStyleSheet("")
+ self.max_input.setStyleSheet("")
+
+ def _toggle_limits_enabled(self, enable: bool) -> None:
+ """Enable or disable the limit inputs based on the checkbox state."""
+ self.min_input.setEnabled(enable)
+ self.max_input.setEnabled(enable)
+
+ def get_limits(self) -> list[float, float]:
+ """Return the limits as a list [min, max]."""
+ min_val = self.min_input.value()
+ max_val = self.max_input.value()
+ return [min_val, max_val]
+
+ def set_limits(self, limits: tuple) -> None:
+ """Set the limits from a tuple (min, max)."""
+ checked_state = self.enable_toggle.isChecked()
+ if not checked_state:
+ self.enable_toggle.setChecked(True)
+ self.min_input.setValue(limits[0])
+ self.max_input.setValue(limits[1])
+ self.enable_toggle.setChecked(checked_state)
+
+
+class ParameterValueWidget(QtWidgets.QWidget):
+ """Custom QTreeWidget for user parameters input field."""
+
+ def __init__(self, parent=None):
+ super().__init__(parent=parent)
+ self._layout = QtWidgets.QVBoxLayout(self)
+ self._layout.setContentsMargins(0, 0, 0, 0)
+ self._layout.setSpacing(4)
+ self.tree_widget = QtWidgets.QTreeWidget(self)
+ self._layout.addWidget(self.tree_widget)
+ self.tree_widget.setColumnCount(2)
+ self.tree_widget.setHeaderLabels(["Parameter", "Value"])
+ self.tree_widget.setIndentation(0)
+ self.tree_widget.setRootIsDecorated(False)
+ header = self.tree_widget.header()
+ header.setSectionResizeMode(0, QtWidgets.QHeaderView.Stretch)
+ header.setSectionResizeMode(1, QtWidgets.QHeaderView.Stretch)
+ self._add_tool_buttons()
+
+ def clear_widget(self) -> None:
+ """Clear all tags."""
+ for i in reversed(range(self.tree_widget.topLevelItemCount())):
+ item = self.tree_widget.topLevelItem(i)
+ index = self.tree_widget.indexOfTopLevelItem(item)
+ if index != -1:
+ self.tree_widget.takeTopLevelItem(index)
+
+ def _add_tool_buttons(self) -> None:
+ """Add tool buttons for adding/removing parameter lines."""
+ button_layout = QtWidgets.QHBoxLayout()
+ button_layout.setContentsMargins(0, 0, 0, 0)
+ button_layout.setSpacing(4)
+ self._layout.addLayout(button_layout)
+ self._button_add = QtWidgets.QPushButton(self)
+ self._button_add.setIcon(material_icon("add", size=(16, 16), convert_to_pixmap=False))
+ self._button_add.setToolTip("Add parameter")
+ self._button_add.clicked.connect(self._add_button_clicked)
+ button_layout.addWidget(self._button_add)
+
+ self._button_remove = QtWidgets.QPushButton(self)
+ self._button_remove.setIcon(material_icon("remove", size=(16, 16), convert_to_pixmap=False))
+ self._button_remove.setToolTip("Remove selected parameter")
+ self._button_remove.clicked.connect(self.remove_parameter_line)
+ button_layout.addWidget(self._button_remove)
+
+ def _add_button_clicked(self, *args, **kwargs) -> None:
+ """Handle the add button click event."""
+ self.add_parameter_line()
+
+ def add_parameter_line(self, parameter: str | None = None, value: str | None = None) -> None:
+ """Add a new row with editable Parameter/Value QLineEdits."""
+ item = QtWidgets.QTreeWidgetItem(self.tree_widget)
+ self.tree_widget.addTopLevelItem(item)
+
+ # Parameter field
+ param_edit = QtWidgets.QLineEdit(self.tree_widget)
+ param_edit.setPlaceholderText("Parameter")
+ self.tree_widget.setItemWidget(item, 0, param_edit)
+
+ # Value field
+ value_edit = QtWidgets.QLineEdit(self.tree_widget)
+ value_edit.setPlaceholderText("Value")
+ self.tree_widget.setItemWidget(item, 1, value_edit)
+ if parameter is not None:
+ param_edit.setText(str(parameter))
+ if value is not None:
+ value_edit.setText(str(value))
+
+ def remove_parameter_line(self) -> None:
+ """Remove the selected row."""
+ selected_items = self.tree_widget.selectedItems()
+ for item in selected_items:
+ index = self.tree_widget.indexOfTopLevelItem(item)
+ if index != -1:
+ self.tree_widget.takeTopLevelItem(index)
+
+ # ---------------------------------------------------------------------
+
+ def parameters(self) -> dict:
+ """Return all parameters as a dictionary {parameter: value}."""
+ result = {}
+ for i in range(self.tree_widget.topLevelItemCount()):
+ item = self.tree_widget.topLevelItem(i)
+ param_edit = self.tree_widget.itemWidget(item, 0)
+ value_edit = self.tree_widget.itemWidget(item, 1)
+ if param_edit and value_edit:
+ key = param_edit.text().strip()
+ val = value_edit.text().strip()
+ if key and val:
+ result[key] = _try_literal_eval(val)
+ return result
+
+
+class DeviceTagsWidget(QtWidgets.QWidget):
+ """Custom QTreeWidget for deviceTags input field."""
+
+ def __init__(self, parent=None):
+ super().__init__(parent=parent)
+ self._layout = QtWidgets.QVBoxLayout(self)
+ self._layout.setContentsMargins(0, 0, 0, 0)
+ self._layout.setSpacing(4)
+ self.tree_widget = QtWidgets.QTreeWidget(self)
+ self._layout.addWidget(self.tree_widget)
+ self.tree_widget.setColumnCount(1)
+ self.tree_widget.setHeaderLabels(["Tags"])
+ self.tree_widget.setIndentation(0)
+ self.tree_widget.setRootIsDecorated(False)
+ self._add_tool_buttons()
+
+ def clear_widget(self) -> None:
+ """Clear all tags."""
+ for i in reversed(range(self.tree_widget.topLevelItemCount())):
+ item = self.tree_widget.topLevelItem(i)
+ index = self.tree_widget.indexOfTopLevelItem(item)
+ if index != -1:
+ self.tree_widget.takeTopLevelItem(index)
+
+ def _add_tool_buttons(self) -> None:
+ """Add tool buttons for adding/removing parameter lines."""
+ button_layout = QtWidgets.QHBoxLayout()
+ button_layout.setContentsMargins(0, 0, 0, 0)
+ button_layout.setSpacing(4)
+ self._layout.addLayout(button_layout)
+ self._button_add = QtWidgets.QPushButton(self)
+ self._button_add.setIcon(material_icon("add", size=(16, 16), convert_to_pixmap=False))
+ self._button_add.setToolTip("Add parameter")
+ self._button_add.clicked.connect(self._add_button_clicked)
+ button_layout.addWidget(self._button_add)
+
+ self._button_remove = QtWidgets.QPushButton(self)
+ self._button_remove.setIcon(material_icon("remove", size=(16, 16), convert_to_pixmap=False))
+ self._button_remove.setToolTip("Remove selected parameter")
+ self._button_remove.clicked.connect(self.remove_parameter_line)
+ button_layout.addWidget(self._button_remove)
+
+ def _add_button_clicked(self, *args, **kwargs) -> None:
+ """Handle the add button click event."""
+ self.add_parameter_line()
+
+ def add_parameter_line(self, parameter: str | None = None) -> None:
+ """Add a new row with editable Tag QLineEdit."""
+ item = QtWidgets.QTreeWidgetItem(self.tree_widget)
+ self.tree_widget.addTopLevelItem(item)
+
+ # Tag field
+ param_edit = QtWidgets.QLineEdit(self.tree_widget)
+ param_edit.setPlaceholderText("Tag")
+ self.tree_widget.setItemWidget(item, 0, param_edit)
+ if parameter is not None:
+ param_edit.setText(str(parameter))
+
+ def remove_parameter_line(self) -> None:
+ """Remove the selected row."""
+ selected_items = self.tree_widget.selectedItems()
+ for item in selected_items:
+ index = self.tree_widget.indexOfTopLevelItem(item)
+ if index != -1:
+ self.tree_widget.takeTopLevelItem(index)
+
+ # ---------------------------------------------------------------------
+
+ def parameters(self) -> list[str]:
+ """Return all parameters as a list of tags."""
+ result = []
+ for i in range(self.tree_widget.topLevelItemCount()):
+ item = self.tree_widget.topLevelItem(i)
+ param_edit = self.tree_widget.itemWidget(item, 0)
+ if param_edit:
+ tag = param_edit.text().strip()
+ if tag:
+ result.append(tag)
+ return result
+
+
+# Validation callback for name field
+def validate_name(name: str) -> bool:
+ """Check that the name does not contain spaces."""
+ if " " in name:
+ return False
+ if not name.replace("_", "").isalnum():
+ return False
+ return True
+
+
+# Validation callback for deviceClass field
+def validate_device_cls(name: str) -> bool:
+ """Check that the name does not contain spaces."""
+ if " " in name:
+ return False
+ if not name.replace("_", "").replace(".", "").isalnum():
+ return False
+ return True
+
+
+def validate_prefix(value: str) -> bool:
+ """Check that the prefix does not contain spaces."""
+ if " " in value:
+ return False
+ if not value.replace("_", "").replace(".", "").replace("-", "").replace(":", "").isalnum():
+ return False
+ return True
+
+
+class DeviceConfigField(BaseModel):
+ """Pydantic model for device configuration fields."""
+
+ label: str
+ widget_cls: type[QtWidgets.QWidget]
+ required: bool = False
+ static: bool = False
+ placeholder_text: str | None = None
+ validation_callback: list[Callable[[str], bool]] | None = None
+ default: Any = None
+
+ model_config = ConfigDict(arbitrary_types_allowed=True)
+
+
+DEVICE_FIELDS = {
+ "name": DeviceConfigField(
+ label="Name",
+ widget_cls=InputLineEdit,
+ required=True,
+ placeholder_text="Device name (no spaces or special characters)",
+ validation_callback=[validate_name],
+ ),
+ "deviceClass": DeviceConfigField(
+ label="Device Class",
+ widget_cls=InputLineEdit,
+ required=True,
+ placeholder_text="Device class (no spaces or special characters)",
+ validation_callback=[validate_device_cls],
+ ),
+ "description": DeviceConfigField(
+ label="Description",
+ widget_cls=QtWidgets.QTextEdit,
+ required=False,
+ placeholder_text="Short device description",
+ ),
+ "enabled": DeviceConfigField(
+ label="Enabled", widget_cls=ToggleSwitch, required=False, default=True
+ ),
+ "readOnly": DeviceConfigField(
+ label="Read Only", widget_cls=ToggleSwitch, required=False, default=False
+ ),
+ "softwareTrigger": DeviceConfigField(
+ label="Software Trigger", widget_cls=ToggleSwitch, required=False, default=False
+ ),
+ "readoutPriority": DeviceConfigField(
+ label="Readout Priority", widget_cls=ReadoutPriorityComboBox, default="baseline"
+ ),
+ "onFailure": DeviceConfigField(
+ label="On Failure", widget_cls=OnFailureComboBox, default="retry"
+ ),
+ "userParameter": DeviceConfigField(
+ label="User Parameters", widget_cls=ParameterValueWidget, static=False
+ ),
+ "deviceTags": DeviceConfigField(label="Device Tags", widget_cls=DeviceTagsWidget, static=False),
+}
+
+DEVICE_CONFIG_FIELDS = {
+ "prefix": DeviceConfigField(
+ label="Prefix",
+ widget_cls=InputLineEdit,
+ static=False,
+ placeholder_text="EPICS IOC prefix, e.g. X25DA-ES1-MOT:",
+ validation_callback=[validate_prefix],
+ ),
+ "read_pv": DeviceConfigField(
+ label="Read PV",
+ widget_cls=InputLineEdit,
+ static=False,
+ placeholder_text="EPICS read PV: e.g. X25DA-ES1-MOT:GET",
+ validation_callback=[validate_prefix],
+ ),
+ "write_pv": DeviceConfigField(
+ label="Write PV",
+ widget_cls=InputLineEdit,
+ static=False,
+ placeholder_text="EPICS write PV (if different from read_pv): e.g. X25DA-ES1-MOT:SET",
+ validation_callback=[validate_prefix],
+ ),
+ "limits": DeviceConfigField(label="Limits", widget_cls=LimitInputWidget, static=False),
+ "DEFAULT": DeviceConfigField(label="DEFAULT FIELD", widget_cls=InputLineEdit, static=False),
+}
diff --git a/bec_widgets/widgets/control/device_manager/components/device_table/__init__.py b/bec_widgets/widgets/control/device_manager/components/device_table/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py b/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py
new file mode 100644
index 000000000..a7b716cfa
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py
@@ -0,0 +1,1128 @@
+"""
+Module for a TableWidget for the device manager view. Row data is encapsulated
+in DeviceTableRow entries.
+"""
+
+from __future__ import annotations
+
+import traceback
+from copy import deepcopy
+from typing import TYPE_CHECKING, Any, Callable, Iterable, Literal, Tuple
+
+from bec_lib.atlas_models import Device as DeviceModel
+from bec_lib.callback_handler import EventType
+from bec_lib.logger import bec_logger
+from bec_qthemes import material_icon
+from qtpy import QtCore, QtGui, QtWidgets
+from thefuzz import fuzz
+
+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.device_table.device_table_row import (
+ DeviceTableRow,
+)
+from bec_widgets.widgets.control.device_manager.components.ophyd_validation import (
+ ConfigStatus,
+ ConnectionStatus,
+ get_validation_icons,
+)
+
+if TYPE_CHECKING: # pragma: no cover
+ from bec_lib.messages import ConfigAction
+
+logger = bec_logger.logger
+
+_DeviceCfgIter = Iterable[dict[str, Any]]
+# DeviceValidationResult: device_config, config_status, connection_status, error_message
+_ValidationResultIter = Iterable[Tuple[dict[str, Any], ConfigStatus, ConnectionStatus, str]]
+
+FUZZY_SEARCH_THRESHOLD = 80
+
+
+def is_match(
+ text: str, row_data: dict[str, Any], relevant_keys: list[str], enable_fuzzy: bool
+) -> bool:
+ """
+ Check if the text matches any of the relevant keys in the row data.
+
+ Args:
+ text (str): The text to search for.
+ row_data (dict[str, Any]): The row data to search in.
+ relevant_keys (list[str]): The keys to consider for searching.
+ enable_fuzzy (bool): Whether to use fuzzy matching.
+ Returns:
+ bool: True if a match is found, False otherwise.
+ """
+ for key in relevant_keys:
+ data = str(row_data.get(key, "") or "")
+ if enable_fuzzy:
+ match_ratio = fuzz.partial_ratio(text.lower(), data.lower())
+ if match_ratio >= FUZZY_SEARCH_THRESHOLD:
+ return True
+ else:
+ if text.lower() in data.lower():
+ return True
+ return False
+
+
+class TableSortOnHold:
+ """Context manager for putting table sorting on hold. Works with nested calls."""
+
+ def __init__(self, table: QtWidgets.QTableWidget) -> None:
+ self.table = table
+ self._call_depth = 0
+ self._registered_methods = []
+
+ def register_on_hold_method(
+ self, method: Callable[[QtWidgets.QTableWidget, bool], None]
+ ) -> None:
+ """
+ Register a method to be called when sorting is put on hold.
+
+ Args:
+ method (Callable[[QtWidgets.QTableWidget, bool], None]): The method to register.
+ The method should accept the QTableWidget and a bool indicating
+ whether sorting is being enabled (True) or disabled (False).
+ """
+ self._registered_methods.append(method)
+
+ def __enter__(self):
+ """Enter the context manager"""
+ self._call_depth += 1 # Needed for nested calls
+ self.table.setSortingEnabled(False)
+ for method in self._registered_methods:
+ method(self.table, False)
+
+ def __exit__(self, *exc):
+ """Exit the context manager"""
+ self._call_depth -= 1 # Remove nested calls
+ if self._call_depth == 0: # Only re-enable sorting on outermost exit
+ self.table.setSortingEnabled(True)
+ for method in self._registered_methods:
+ method(self.table, True)
+
+
+class CenterIconDelegate(QtWidgets.QStyledItemDelegate):
+ """Custom delegate to center icons in table cells."""
+
+ def paint(
+ self,
+ painter: QtGui.QPainter,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QtCore.QModelIndex,
+ ):
+ # First draw the default cell (without icon)
+ opt = QtWidgets.QStyleOptionViewItem(option)
+ self.initStyleOption(opt, index)
+ opt.icon = QtGui.QIcon() # Create empty icon to avoid default to be drawn at given position
+ option.widget.style().drawControl(
+ QtWidgets.QStyle.ControlElement.CE_ItemViewItem, opt, painter, option.widget
+ )
+ # Check if there is an icon to draw
+ icon = index.data(QtCore.Qt.ItemDataRole.DecorationRole)
+ if not icon:
+ return
+ # Draw the icon centered in the cell
+ icon_size = option.decorationSize
+ if icon_size.isValid():
+ size = icon_size
+ else:
+ size = icon.actualSize(option.rect.size())
+
+ x = option.rect.x() + (option.rect.width() - size.width()) // 2
+ y = option.rect.y() + (option.rect.height() - size.height()) // 2
+
+ icon.paint(painter, QtCore.QRect(QtCore.QPoint(x, y), size))
+
+
+class CheckBoxDelegate(QtWidgets.QStyledItemDelegate):
+ """Custom delegate to handle checkbox interactions in the table."""
+
+ # Signal to indicate a checkbox was clicked
+ checkbox_clicked = QtCore.Signal(int, int, bool) # row, column, checked
+
+ def editorEvent(
+ self,
+ event: QtCore.QEvent,
+ model: QtCore.QAbstractItemModel,
+ option: QtWidgets.QStyleOptionViewItem,
+ index: QtCore.QModelIndex,
+ ):
+ if event.type() == QtCore.QEvent.Type.MouseButtonRelease:
+ if model and (model.flags(index) & QtCore.Qt.ItemFlag.ItemIsUserCheckable):
+ old_state = QtCore.Qt.CheckState(
+ model.data(index, QtCore.Qt.ItemDataRole.CheckStateRole)
+ )
+ new_state = (
+ QtCore.Qt.CheckState.Unchecked
+ if old_state == QtCore.Qt.CheckState.Checked
+ else QtCore.Qt.CheckState.Checked
+ )
+ model.setData(index, new_state, QtCore.Qt.ItemDataRole.CheckStateRole)
+ model.setData(
+ index,
+ new_state == QtCore.Qt.CheckState.Checked,
+ QtCore.Qt.ItemDataRole.UserRole,
+ )
+ self.checkbox_clicked.emit(
+ index.row(), index.column(), new_state == QtCore.Qt.CheckState.Checked
+ )
+ return True
+ return super().editorEvent(event, model, option, index)
+
+
+class SortTableItem(QtWidgets.QTableWidgetItem):
+ """Custom TableWidgetItem with hidden __column_data attribute for sorting."""
+
+ def __lt__(self, other: QtWidgets.QTableWidgetItem) -> bool:
+ """Override less-than operator for sorting."""
+ if not isinstance(other, QtWidgets.QTableWidgetItem):
+ return NotImplemented
+ self_data = self.data(QtCore.Qt.ItemDataRole.UserRole)
+ other_data = other.data(QtCore.Qt.ItemDataRole.UserRole)
+ if self_data is not None and other_data is not None:
+ return self_data < other_data
+ return super().__lt__(other)
+
+ def __gt__(self, other: QtWidgets.QTableWidgetItem) -> bool:
+ """Override less-than operator for sorting."""
+ if not isinstance(other, QtWidgets.QTableWidgetItem):
+ return NotImplemented
+ self_data = self.data(QtCore.Qt.ItemDataRole.UserRole)
+ other_data = other.data(QtCore.Qt.ItemDataRole.UserRole)
+ if self_data is not None and other_data is not None:
+ return self_data > other_data
+ return super().__gt__(other)
+
+
+class DeviceTable(BECWidget, QtWidgets.QWidget):
+ """Custom table to display device configurations."""
+
+ RPC = False # TODO discuss if this should be available for RPC
+
+ # Signal emitted if devices are added (updated) or removed
+ # - device_configs: List of device configurations.
+ # - added: True if devices were added/updated, False if removed.
+ # - skip validation: True if validation should be skipped for added/updated devices.
+ device_configs_changed = QtCore.Signal(list, bool, bool)
+ # Signal emitted when device selection changes, emits list of selected device configs
+ selected_devices = QtCore.Signal(list)
+ # Signal emitted when a device row is double-clicked, emits the device config
+ device_row_dbl_clicked = QtCore.Signal(dict)
+ # Signal emitted when the device config is in sync with Redis
+ device_config_in_sync_with_redis = QtCore.Signal(bool)
+
+ # Request multiple validation updates for devices
+ request_update_multiple_device_validations = QtCore.Signal(list)
+ # Request update after client DEVICE_UPDATE event
+ request_update_after_client_device_update = QtCore.Signal()
+
+ _auto_size_request = QtCore.Signal()
+
+ def __init__(self, parent: QtWidgets.QWidget | None = None, client=None):
+ super().__init__(parent=parent, client=client)
+ self.headers_key_map: dict[str, str] = {
+ "Valid": "valid",
+ "Connect": "connect",
+ "Name": "name",
+ "Device Class": "deviceClass",
+ "Readout Priority": "readoutPriority",
+ "On Failure": "onFailure",
+ "Device Tags": "deviceTags",
+ "Description": "description",
+ "Enabled": "enabled",
+ "Read Only": "readOnly",
+ "Software Trigger": "softwareTrigger",
+ }
+
+ # General attributes
+ self._icon_size = (18, 18)
+ self._colors = get_accent_colors()
+ self._icons = get_validation_icons(self._colors, self._icon_size)
+ self._check_box_icons = {
+ "checked": material_icon(
+ "check_box", size=(24, 24), color=self._colors.default, convert_to_pixmap=False
+ ),
+ "unchecked": material_icon(
+ "check_box_outline_blank",
+ size=(24, 24),
+ color=self._colors.default,
+ convert_to_pixmap=False,
+ ),
+ }
+ self._layout = QtWidgets.QVBoxLayout(self)
+ self._layout.setContentsMargins(0, 0, 0, 0)
+ self._layout.setSpacing(4)
+ self.setLayout(self._layout)
+
+ # Table related attributes
+ self.row_data: dict[str, DeviceTableRow] = {}
+ self.table = QtWidgets.QTableWidget(self)
+ self.table_sort_on_hold = TableSortOnHold(self.table)
+ self._setup_table()
+ self.table_sort_on_hold.register_on_hold_method(self._resize_table_policy)
+ self.table_sort_on_hold.register_on_hold_method(self._set_table_signals_on_hold)
+
+ # Search related attributes
+ self._searchable_keys: list[str] = ["name", "deviceClass", "deviceTags", "description"]
+ self._hidden_rows: set[int] = set()
+ self._enable_fuzzy_search: bool = True
+ self._setup_search()
+
+ # Add components to layout
+ self._layout.addLayout(self.search_controls)
+ self._layout.addWidget(self.table)
+
+ # Connect slots
+ self.table.selectionModel().selectionChanged.connect(self._on_selection_changed)
+ self.table.cellDoubleClicked.connect(self._on_cell_double_clicked)
+ self.request_update_multiple_device_validations.connect(
+ self.update_multiple_device_validations
+ )
+ self.request_update_after_client_device_update.connect(self._on_device_config_update)
+ # Install event filter
+ self.table.installEventFilter(self)
+
+ # Add hook to BECClient for DeviceUpdates
+ self.client_callback_id = self.client.callbacks.register(
+ event_type=EventType.DEVICE_UPDATE, callback=self.__on_client_device_update_event
+ )
+
+ def cleanup(self):
+ """Cleanup resources."""
+ self.row_data.clear() # Drop references to row data..
+ self.client.callbacks.remove(self.client_callback_id) # Unregister callback
+ super().cleanup()
+
+ def __on_client_device_update_event(
+ self, action: "ConfigAction", config: dict[str, dict[str, Any]]
+ ) -> None:
+ """Handle DEVICE_UPDATE events from the BECClient."""
+ self.request_update_after_client_device_update.emit()
+
+ @SafeSlot()
+ def _on_device_config_update(self) -> None:
+ """Handle device configuration updates from the BECClient."""
+ # Determine the overlapping device configs between Redis and the table
+ device_config_overlap_with_bec = self._get_overlapping_configs()
+ if len(device_config_overlap_with_bec) > 0:
+ # Notify any listeners about the update, the device manager devices will now be up to date
+ self.device_configs_changed.emit(device_config_overlap_with_bec, True, True)
+
+ # Correct all connection statuses in the table which are ConnectionStatus.CONNECTED
+ # to ConnectionStatus.CAN_CONNECT
+ device_status_updates = []
+ validation_results = self.get_validation_results()
+ for device_name, (cfg, config_status, connection_status) in validation_results.items():
+ if device_name is None:
+ continue
+ # Check if config is not in the overlap, but connection status is CONNECTED
+ # Update to CAN_CONNECT
+ if cfg not in device_config_overlap_with_bec:
+ if connection_status == ConnectionStatus.CONNECTED.value:
+ device_status_updates.append(
+ (cfg, config_status, ConnectionStatus.CAN_CONNECT.value, "")
+ )
+ # Update only if there are any updates
+ if len(device_status_updates) > 0:
+ # NOTE We need to emit here a signal to call update_multiple_device_validations
+ # as this otherwise can cause problems with being executed from a python callback
+ # thread which are not properly scheduled in the Qt event loop. We see that this
+ # has caused issues in form of segfaults under certain usage of the UI. Please
+ # do not remove this signal & slot mechanism!
+ self.request_update_multiple_device_validations.emit(device_status_updates)
+
+ # Check if in sync with BEC server session
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+
+ # -------------------------------------------------------------------------
+ # Custom hooks for table events
+ # -------------------------------------------------------------------------
+
+ def _on_selection_changed(
+ self, selected: QtCore.QItemSelection, deselected: QtCore.QItemSelection
+ ):
+ """Handle selection changes in the table."""
+ rows = set()
+ for index in selected.indexes():
+ row = index.row()
+ rows.add(row)
+ selected_configs = []
+ for row in rows:
+ device_name = self._get_cell_data(row, 2) # Name column
+ if device_name:
+ row_data = self.row_data.get(device_name)
+ if row_data:
+ cfg = deepcopy(row_data.data)
+ cfg.pop("name")
+ selected_configs.append({device_name: cfg})
+ self.selected_devices.emit(selected_configs)
+
+ def _on_cell_double_clicked(self, row: int, column: int):
+ """Handle double-click events on table cells."""
+ device_name = self._get_cell_data(row, 2) # Name column
+ if device_name:
+ row_data = self.row_data.get(device_name)
+ if row_data:
+ self.device_row_dbl_clicked.emit(row_data.data)
+
+ def eventFilter(self, source: QtCore.QObject, event: QtCore.QEvent) -> bool:
+ """Customize event filtering for table interactions."""
+ if source is self.table:
+ if event.type() == QtCore.QEvent.Type.KeyPress:
+ if event.key() in (QtCore.Qt.Key.Key_Backspace, QtCore.Qt.Key.Key_Delete):
+ configs = self.get_selected_device_configs()
+ if configs:
+ if self._remove_configs_dialog([cfg["name"] for cfg in configs]):
+ self.remove_device_configs(configs)
+ return True # Event handled
+ if event.key() == QtCore.Qt.Key.Key_Escape:
+ self.table.clearSelection()
+ return True # handled
+ return super().eventFilter(source, event)
+
+ def _on_table_checkbox_clicked(self, row: int, column: int, checked: bool):
+ """Handle checkbox clicks in the table."""
+ name_index = list(self.headers_key_map.values()).index("name")
+ device_name = self._get_cell_data(row, name_index)
+ row_data = self.row_data.get(device_name)
+ if not row_data:
+ return
+ row_data.data[self.headers_key_map[list(self.headers_key_map.keys())[column]]] = checked
+ self._on_device_row_data_changed(row_data.data)
+
+ def _on_device_row_data_changed(self, data: dict):
+ """Handle data change events from device rows."""
+ device_name = data.get("name", None)
+ cfg = deepcopy(data)
+ cfg.pop("name")
+ self.selected_devices.emit([{device_name: cfg}])
+ self.device_config_in_sync_with_redis.emit(self._is_config_in_sync_with_redis())
+
+ def _apply_row_filter(self, text_input: str):
+ """Apply a filter to the table rows based on the filter text."""
+ for row in range(self.table.rowCount()):
+ device_name = self._get_cell_data(row, 2) # Name column
+ if not device_name:
+ continue
+ row_data = self.row_data.get(device_name)
+ if not row_data:
+ continue
+ if is_match(
+ text_input, row_data.data, self._searchable_keys, self._enable_fuzzy_search
+ ):
+ self.table.setRowHidden(row, False)
+ self._hidden_rows.discard(row)
+ else:
+ self.table.setRowHidden(row, True)
+ self._hidden_rows.add(row)
+
+ def _state_change_fuzzy_search(self, enabled: int):
+ """Handle state changes for the fuzzy search toggle."""
+ self._enable_fuzzy_search = not bool(enabled)
+ # Re-apply filter with updated fuzzy search setting
+ current_text = self.search_input.text()
+ self._apply_row_filter(current_text)
+
+ # -------------------------------------------------------------------------
+ # Custom Dialog
+ # -------------------------------------------------------------------------
+
+ def _remove_configs_dialog(self, device_names: list[str]) -> bool:
+ """
+ Prompt the user to confirm removal of rows and remove them from the model if accepted.
+
+ Args:
+ device_names (list[str]): List of device names to be removed.
+
+ Returns:
+ bool: True if the user confirmed removal, False otherwise.
+ """
+ msg = QtWidgets.QMessageBox(self)
+ msg.setIcon(QtWidgets.QMessageBox.Icon.Warning)
+ msg.setWindowTitle("Confirm device removal")
+ msg.setText(
+ f"Remove device '{device_names[0]}'?"
+ if len(device_names) == 1
+ else f"Remove {len(device_names)} devices?"
+ )
+ separator = "\n" if len(device_names) < 12 else ", "
+ msg.setInformativeText("Selected devices: \n" + separator.join(device_names))
+ msg.setStandardButtons(
+ QtWidgets.QMessageBox.StandardButton.Ok | QtWidgets.QMessageBox.StandardButton.Cancel
+ )
+ msg.setDefaultButton(QtWidgets.QMessageBox.StandardButton.Cancel)
+
+ res = msg.exec_()
+ if res == QtWidgets.QMessageBox.StandardButton.Ok:
+ return True
+ return False
+
+ # -------------------------------------------------------------------------
+ # Setup table
+ # -------------------------------------------------------------------------
+ def _setup_table(self):
+ """Initializes the table configuration and headers."""
+ # Temporary instance to get headers dynamically
+ headers = list(self.headers_key_map.keys())
+ self.table.setColumnCount(len(headers))
+ self.table.setHorizontalHeaderLabels(headers)
+ # Smooth scrolling
+ self.table.setVerticalScrollMode(QtWidgets.QAbstractItemView.ScrollPerPixel)
+ self.table.setHorizontalScrollMode(QtWidgets.QAbstractItemView.ScrollPerPixel)
+ self.table.setEditTriggers(QtWidgets.QAbstractItemView.EditTrigger.NoEditTriggers)
+
+ # Hide vertical header
+ self.table.verticalHeader().setVisible(False)
+
+ # Column resize policies
+ header = self.table.horizontalHeader()
+ header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeMode.Fixed)
+ header.setSectionResizeMode(1, QtWidgets.QHeaderView.ResizeMode.Fixed)
+ header.setSectionResizeMode(2, QtWidgets.QHeaderView.ResizeMode.Interactive)
+ header.setSectionResizeMode(3, QtWidgets.QHeaderView.ResizeMode.Interactive)
+ header.setSectionResizeMode(4, QtWidgets.QHeaderView.ResizeMode.Fixed)
+ header.setSectionResizeMode(5, QtWidgets.QHeaderView.ResizeMode.Fixed)
+ header.setSectionResizeMode(6, QtWidgets.QHeaderView.ResizeMode.Interactive)
+ header.setSectionResizeMode(7, QtWidgets.QHeaderView.ResizeMode.Stretch)
+ header.setSectionResizeMode(8, QtWidgets.QHeaderView.ResizeMode.Fixed)
+ header.setSectionResizeMode(9, QtWidgets.QHeaderView.ResizeMode.Fixed)
+ header.setSectionResizeMode(10, QtWidgets.QHeaderView.ResizeMode.Fixed)
+
+ for sizes, col in [
+ (0, 85),
+ (1, 85),
+ (2, 200),
+ (3, 200),
+ (6, 200),
+ (7, 200),
+ (8, 90),
+ (9, 90),
+ (10, 120),
+ ]:
+ self.table.setColumnWidth(sizes, col)
+
+ # Ensure column widths stay fixed
+ header.setStretchLastSection(False)
+
+ # Sorting
+ self.table.setSortingEnabled(True)
+ header.setSortIndicatorShown(True)
+ header.setSortIndicator(2, QtCore.Qt.SortOrder.AscendingOrder) # Default sort by name
+
+ # Selection behavior
+ self.table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectionBehavior.SelectRows)
+ self.table.setSelectionMode(QtWidgets.QAbstractItemView.SelectionMode.ExtendedSelection)
+ # Connect to selection model to get selection changes
+ self.table.selectionModel().selectionChanged.connect(self._on_selection_changed)
+ self.table.horizontalHeader().setHighlightSections(False)
+
+ # Set delegate for checkboxes
+ checkbox_delegate = CheckBoxDelegate(self.table)
+ icon_delegate = CenterIconDelegate(self.table)
+ self.table.setItemDelegateForColumn(0, icon_delegate) # Config status
+ self.table.setItemDelegateForColumn(1, icon_delegate) # Connection status
+ self.table.setWordWrap(True)
+ for col in (8, 9, 10): # enabled, readOnly, softwareTrigger
+ self.table.setItemDelegateForColumn(col, checkbox_delegate)
+ checkbox_delegate.checkbox_clicked.connect(self._on_table_checkbox_clicked)
+
+ def _set_table_signals_on_hold(self, table: QtWidgets.QTableWidget, enable: bool):
+ """Enable or disable table signals."""
+ if enable:
+ table.blockSignals(False)
+ else:
+ table.blockSignals(True)
+
+ def _resize_table_policy(self, table: QtWidgets.QTableWidget, enable: bool):
+ """Enable or disable column resizing."""
+ if enable:
+ table.resizeColumnToContents(2) # Name
+ table.resizeColumnToContents(3) # Device Class
+ # table.resizeRowsToContents()
+
+ def _setup_search(self):
+ """Create components related to the search functionality"""
+
+ # Create search bar
+ self.search_layout = QtWidgets.QHBoxLayout()
+ self.search_label = QtWidgets.QLabel("Search:")
+ self.search_input = QtWidgets.QLineEdit()
+ self.search_input.setPlaceholderText("Filter devices (approximate matching)...")
+ self.search_input.setClearButtonEnabled(True)
+ self.search_input.textChanged.connect(self._apply_row_filter)
+ self.search_layout.addWidget(self.search_label)
+ self.search_layout.addWidget(self.search_input)
+
+ # Add exact match toggle
+ self.fuzzy_layout = QtWidgets.QHBoxLayout()
+ self.fuzzy_label = QtWidgets.QLabel("Exact Match:")
+ self.fuzzy_is_disabled = QtWidgets.QCheckBox()
+
+ self.fuzzy_is_disabled.stateChanged.connect(self._state_change_fuzzy_search)
+ self.fuzzy_is_disabled.setToolTip(
+ "Enable approximate matching (OFF) and exact matching (ON)"
+ )
+ self.fuzzy_label.setToolTip("Enable approximate matching (OFF) and exact matching (ON)")
+ self.fuzzy_layout.addWidget(self.fuzzy_label)
+ self.fuzzy_layout.addWidget(self.fuzzy_is_disabled)
+ self.fuzzy_layout.addStretch()
+
+ # Add both search components to the layout
+ self.search_controls = QtWidgets.QHBoxLayout()
+ self.search_controls.addLayout(self.search_layout)
+ self.search_controls.addSpacing(20) # Add some space between the search box and toggle
+ self.search_controls.addLayout(self.fuzzy_layout)
+ QtCore.QTimer.singleShot(0, lambda: self.fuzzy_is_disabled.stateChanged.emit(0))
+
+ # -------------------------------------------------------------------------
+ # Row Management, internal methods.
+ # -------------------------------------------------------------------------
+
+ def _add_row(
+ self,
+ data: dict,
+ config_status: ConfigStatus | int,
+ connection_status: ConnectionStatus | int,
+ ):
+ """
+ Adds a new row at the bottom and populates it with data. The row widgets
+ are stored in self.row_widgets for easy access. Consider to disable sorting
+ when adding rows as this method is not responsible for maintaining sort order.
+
+ Args:
+ data (dict): The device data to populate the row.
+ config_status (ConfigStatus | int): The configuration validation status.
+ connection_status (ConnectionStatus | int): The connection status.
+ """
+ with self.table_sort_on_hold:
+ if data["name"] in self.row_data:
+ logger.warning(f"Overwriting existing device row for {data['name']}")
+ self._remove_rows_by_name([data["name"]])
+ row_index = self.table.rowCount()
+ self.table.insertRow(row_index)
+
+ # Create row for the table
+ device_row = DeviceTableRow(data=data)
+ device_row.set_validation_status(config_status, connection_status)
+
+ # Populate cells
+ self._populate_device_row_cells(row_index, device_row)
+
+ def _populate_device_row_cells(self, row: int, device_row: DeviceTableRow):
+ """Populate the cells of a given row with the widgets from the DeviceTableRow."""
+ with self.table_sort_on_hold:
+ config_status, connect_status = device_row.validation_status
+ column_keys = list(self.headers_key_map.values())
+ for ii, key in enumerate(column_keys):
+ if key in ("enabled", "readOnly", "softwareTrigger"): # flags for checkboxes
+ item = SortTableItem()
+ item.setFlags(
+ item.flags()
+ | QtCore.Qt.ItemFlag.ItemIsUserCheckable
+ | QtCore.Qt.ItemFlag.ItemIsEnabled
+ )
+ item.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
+ elif key in ("valid", "connect"): # status columns
+ item = SortTableItem()
+ item.setTextAlignment(
+ QtCore.Qt.AlignmentFlag.AlignCenter | QtCore.Qt.AlignmentFlag.AlignVCenter
+ )
+ item.setIcon(
+ self._icons["connection_status"][connect_status]
+ if key == "connect"
+ else self._icons["config_status"][config_status]
+ )
+ else:
+ item = QtWidgets.QTableWidgetItem()
+ item.setTextAlignment(
+ QtCore.Qt.AlignmentFlag.AlignLeft | QtCore.Qt.AlignmentFlag.AlignVCenter
+ )
+ self.table.setItem(row, ii, item) # +2 offset for status columns
+ self.__update_device_row_data(row, device_row.data)
+
+ def __update_device_row_data(self, row: int, data: dict):
+ """
+ Update an existing device row with new data.
+
+ Args:
+ row (int): The row index to update.
+ data (dict): The device data to populate the row.
+ """
+ # Update stored row data
+ if data["name"] in self.row_data:
+ self.row_data[data["name"]].set_data(data)
+ else:
+ self.row_data[data["name"]] = DeviceTableRow(data)
+ # Update table cells
+ with self.table_sort_on_hold:
+ column_keys = list(self.headers_key_map.values()) # map columns
+ for key, value in data.items():
+ if key not in column_keys:
+ continue # Skip userParameters and deviceConfig
+ column = column_keys.index(key)
+ item = self.table.item(row, column)
+ if not item:
+ continue
+ if key in ("enabled", "readOnly", "softwareTrigger"):
+ item.setCheckState(
+ QtCore.Qt.CheckState.Checked if value else QtCore.Qt.CheckState.Unchecked
+ )
+ item.setData(QtCore.Qt.ItemDataRole.UserRole, value)
+ item.setText("") # No text for checkboxes
+ elif key == "deviceTags":
+ item.setText(
+ ", ".join(value) if isinstance(value, (list, set, tuple)) else str(value)
+ )
+ elif key == "deviceClass":
+ item.setText(
+ value.split(".")[-1]
+ ) # Only show the DeviceClass, not the full module
+ else:
+ if value is None:
+ value = ""
+ item.setText(str(value))
+ self._update_device_row_status(
+ row,
+ self.row_data[data["name"]].validation_status[0],
+ self.row_data[data["name"]].validation_status[1],
+ )
+ self.table.resizeRowToContents(row)
+ self._on_device_row_data_changed(self.row_data[data["name"]].data)
+ return True
+
+ def _update_device_row_status(
+ self, row: int, config_status: int, connection_status: int
+ ) -> bool:
+ """
+ Update an existing device row's validation status.
+
+ Args:
+ device_name (str): The name of the device.
+ config_status (int): The configuration validation status.
+ connection_status (int): The connection status.
+ """
+ with self.table_sort_on_hold:
+ item = self.table.item(row, 0) # Config status column
+ if item:
+ item.setData(QtCore.Qt.ItemDataRole.UserRole, config_status)
+ item.setIcon(self._icons["config_status"][config_status])
+ item = self.table.item(row, 1) # Connect status column
+ if item:
+ item.setData(QtCore.Qt.ItemDataRole.UserRole, connection_status)
+ item.setIcon(self._icons["connection_status"][connection_status])
+
+ # Update the stored row data as well
+ device_name = self._get_cell_data(row, 2) # Name column
+ device_row = self.row_data.get(device_name, None)
+ if not device_row:
+ return False
+ device_row: DeviceTableRow
+ device_row.set_validation_status(config_status, connection_status)
+ return True
+
+ def _get_cell_data(self, row: int, column: int) -> str | bool | None:
+ """
+ Get the data from a specific cell.
+
+ Args:
+ row (int): The row index.
+ column (int): The column index.
+ """
+ item = self.table.item(row, column)
+ if item is None:
+ return None
+ if column in (8, 9, 10): # Checkboxes
+ return item.checkState() == QtCore.Qt.CheckState.Checked
+ return item.text()
+
+ def _update_row(self, data: dict) -> int | None:
+ """
+ Update an existing row with new data.
+
+ Args:
+ data (dict): The device data to populate the row.
+ Returns:
+ int | None: The row index if updated, else None.
+ """
+ device_row = self.row_data.get(data.get("name"), {})
+ if self._compare_configs(device_row.data, data):
+ return None # No update needed
+ row = self._find_row_by_name(data.get("name", ""))
+ if row is not None:
+ self.__update_device_row_data(row, data)
+ return row
+
+ def _compare_configs(self, cfg1: dict, cfg2: dict) -> bool:
+ """Compare two device configurations for equality."""
+ try:
+ cfg1_model = DeviceModel.model_validate(cfg1)
+ cfg2_model = DeviceModel.model_validate(cfg2)
+ return cfg1_model == cfg2_model
+ except Exception as e:
+ logger.error(f"Error comparing device configs: {e}")
+ return False
+
+ def _clear_table(self):
+ """Remove all rows."""
+ with self.table_sort_on_hold:
+ n_rows = self.table.rowCount()
+ for _ in range(n_rows):
+ self.table.removeRow(0)
+ self.row_data.clear()
+
+ def _find_row_by_name(self, name: str) -> int | None:
+ """
+ Find a row by device name.
+
+ Args:
+ name (str): The name of the device to find.
+ Returns:
+ int | None: The row index if found, else None.
+ """
+ for row in range(self.table.rowCount()):
+ data = self._get_cell_data(row, 2)
+ if data and data == name:
+ return row
+ return None
+
+ def _remove_rows_by_name(self, device_names: list[str]):
+ """
+ Remove a row by device name.
+
+ Args:
+ device_name (str): The name of the device to remove.
+ """
+ if not device_names:
+ return
+ with self.table_sort_on_hold:
+ for device_name in device_names:
+ row = self._find_row_by_name(device_name)
+ if row is None:
+ logger.warning(f"Device {device_name} not found in table for removal.")
+ return
+ self.table.removeRow(row)
+ self.row_data.pop(device_name, None)
+
+ def _is_config_in_sync_with_redis(self):
+ """Check if the current config is in sync with Redis."""
+ if (
+ not self.client
+ or not self.client.device_manager
+ or not self.client.device_manager.devices
+ ):
+ return False # No proper client connection
+ redis_config = [
+ DeviceModel.model_validate(device._config)
+ for device in self.client.device_manager.devices.values()
+ ]
+ try:
+ current_config = [
+ DeviceModel.model_validate(row_data.data) for row_data in self.row_data.values()
+ ]
+ if redis_config == current_config:
+ return True
+ else:
+ return False
+ except Exception as e:
+ logger.error(f"Error comparing device configs: {e}")
+ return False
+
+ def _get_overlapping_configs(self) -> list[dict[str, Any]]:
+ """
+ Get the device configs that overlap between the table and the config in the current running BEC session.
+ A device will be ignored if it is disabled in the BEC session.
+
+ Args:
+ device_configs (Iterable[dict[str, Any]]): The device configs to check.
+
+ Returns:
+ list[dict[str, Any]]: The list of overlapping device configs.
+ """
+ overlapping_configs = []
+ for cfg in self.get_device_config():
+ device_name = cfg.get("name", None)
+ if device_name is None:
+ continue
+ if self._is_device_in_redis_session(device_name, cfg):
+ overlapping_configs.append(cfg)
+
+ return overlapping_configs
+
+ def _is_device_in_redis_session(self, device_name: str, device_config: dict) -> bool:
+ """Check if a device is in the running section."""
+ dev_obj = self.client.device_manager.devices.get(device_name, None)
+ if dev_obj is None or dev_obj.enabled is False:
+ return False
+ return self._compare_device_configs(dev_obj._config, device_config)
+
+ def _compare_device_configs(self, config1: dict, config2: dict) -> bool:
+ """Compare two device configurations through the Device model in bec_lib.atlas_models.
+
+ Args:
+ config1 (dict): The first device configuration.
+ config2 (dict): The second device configuration.
+
+ Returns:
+ bool: True if the configurations are equivalent, False otherwise.
+ """
+ try:
+ model1 = DeviceModel.model_validate(config1)
+ model2 = DeviceModel.model_validate(config2)
+ return model1 == model2
+ except Exception:
+ return False
+
+ # -------------------------------------------------------------------------
+ # Public API to manage device configs in the table
+ # -------------------------------------------------------------------------
+
+ def get_device_config(self) -> list[dict]:
+ """
+ Get the current device configurations in the table.
+
+ Returns:
+ list[dict]: The list of device configurations.
+ """
+ cfgs = [
+ {"name": device_name, **row_data.data}
+ for device_name, row_data in self.row_data.items()
+ ]
+ return cfgs
+
+ def get_validation_results(self) -> dict[str, Tuple[dict, int, int]]:
+ """
+ Get the current device validation results in the table.
+
+ Returns:
+ dict[str, Tuple[dict, int, int]]: Dictionary mapping of device name to
+ (device config, config status, connection status).
+ """
+ return {
+ row_data.data.get("name"): (row_data.data, *row_data.validation_status)
+ for row_data in self.row_data.values()
+ if row_data.data.get("name") is not None
+ }
+
+ def get_selected_device_configs(self) -> list[dict]:
+ """
+ Get the currently selected device configurations in the table.
+
+ Returns:
+ list[dict]: The list of selected device configurations.
+ """
+ selected_configs = []
+ selected_rows = set()
+ for index in self.table.selectionModel().selectedIndexes():
+ selected_rows.add(index.row())
+ for row in selected_rows:
+ device_name = self._get_cell_data(row, 2) # Name column
+ if device_name:
+ row_data = self.row_data.get(device_name)
+ if row_data:
+ selected_configs.append(row_data.data)
+ return selected_configs
+
+ # -------------------------------------------------------------------------
+ # Public API to be called via signals/slots
+ # -------------------------------------------------------------------------
+
+ @SafeSlot(list, bool)
+ def set_device_config(self, device_configs: _DeviceCfgIter, skip_validation: bool = False):
+ """
+ Set the device config. This will clear any existing configs.
+
+ Args:
+ device_configs (Iterable[dict[str, Any]]): The device configs to set.
+ skip_validation (bool): Whether to skip validation for the set devices.
+ """
+ self.set_busy(True)
+ with self.table_sort_on_hold:
+ self.clear_device_configs()
+ cfgs_added = []
+ for cfg in device_configs:
+ self._add_row(cfg, ConfigStatus.UNKNOWN, ConnectionStatus.UNKNOWN)
+ cfgs_added.append(cfg)
+ self.device_configs_changed.emit(cfgs_added, True, skip_validation)
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
+
+ @SafeSlot()
+ def clear_device_configs(self):
+ """Clear the device configs. Skips validation by default."""
+ self.set_busy(True)
+ device_configs = self.get_device_config()
+ with self.table_sort_on_hold:
+ self._clear_table()
+ self.device_configs_changed.emit(
+ device_configs, False, True
+ ) # Skip validation for removals
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
+
+ @SafeSlot(list, bool)
+ def add_device_configs(self, device_configs: _DeviceCfgIter, skip_validation: bool = False):
+ """
+ Add devices to the config. If a device already exists, it will be replaced. If the validation is
+ skipped, the device will be added with UNKNOWN state to the table and has to be manually adjusted
+ by the user later on.
+
+ Args:
+ device_configs (Iterable[dict[str, Any]]): The device configs to add.
+ skip_validation (bool): Whether to skip validation for the added devices.
+ """
+ self.set_busy(True)
+ already_in_table = []
+ not_in_table = []
+ with self.table_sort_on_hold:
+ for cfg in device_configs:
+ if cfg["name"] in self.row_data:
+ already_in_table.append(cfg)
+ else:
+ not_in_table.append(cfg)
+ with self.table_sort_on_hold:
+ # Remove existing rows first
+ if len(already_in_table) > 0:
+ self._remove_rows_by_name([cfg["name"] for cfg in already_in_table])
+ self.device_configs_changed.emit(
+ already_in_table, False, True
+ ) # Skip validation for removals
+
+ all_configs = already_in_table + not_in_table
+ if len(all_configs) > 0:
+ for cfg in already_in_table + not_in_table:
+ self._add_row(cfg, ConfigStatus.UNKNOWN, ConnectionStatus.UNKNOWN)
+
+ self.device_configs_changed.emit(already_in_table + not_in_table, True, skip_validation)
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
+
+ @SafeSlot(list, bool)
+ def update_device_configs(self, device_configs: _DeviceCfgIter, skip_validation: bool = False):
+ """
+ Update devices in the config. If a device does not exist, it will be added.
+
+ Args:
+ device_configs (Iterable[dict[str, Any]]): The device configs to update.
+ skip_validation (bool): Whether to skip validation for the updated devices.
+ """
+ self.set_busy(True)
+ cfgs_updated = []
+ with self.table_sort_on_hold:
+ for cfg in device_configs:
+ if cfg["name"] not in self.row_data:
+ self._add_row(cfg, ConfigStatus.UNKNOWN, ConnectionStatus.UNKNOWN)
+ cfgs_updated.append(cfg)
+ continue
+ # Update existing row if device config has changed
+ row = self._update_row(cfg)
+ if row is not None:
+ cfgs_updated.append(cfg)
+ self.device_configs_changed.emit(cfgs_updated, True, skip_validation)
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
+
+ @SafeSlot(list)
+ def remove_device_configs(self, device_configs: _DeviceCfgIter):
+ """
+ Remove devices from the config.
+
+ Args:
+ device_configs (dict[str, dict]): The device configs to remove.
+ """
+ self.set_busy(True)
+ cfgs_to_be_removed = list(device_configs)
+ with self.table_sort_on_hold:
+ self._remove_rows_by_name([cfg["name"] for cfg in cfgs_to_be_removed])
+ self.device_configs_changed.emit(
+ cfgs_to_be_removed, False, True
+ ) # Skip validation for removals
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
+
+ @SafeSlot(str)
+ def remove_device(self, device_name: str):
+ """
+ Remove a device from the config.
+
+ Args:
+ device_name (str): The name of the device to remove.
+ """
+ self.set_busy(True)
+ row_data = self.row_data.get(device_name)
+ if not row_data:
+ logger.warning(f"Device {device_name} not found in table for removal.")
+ self.set_busy(False)
+ return
+ with self.table_sort_on_hold:
+ self._remove_rows_by_name([row_data.data["name"]])
+ cfgs = [{"name": device_name, **row_data.data}]
+ self.device_configs_changed.emit(cfgs, False, True) # Skip validation for removals
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
+
+ @SafeSlot(list)
+ def update_multiple_device_validations(self, validation_results: _ValidationResultIter):
+ """
+ Slot to update multiple device validation statuses. This is recommended and more
+ efficient than updating individual device validation statuses which may affect
+ the performance of the UI when many devices are being updated in quick succession.
+
+ Args:
+ device_configs (Iterable[dict[str, Any]]): The device configs to update.
+ """
+ self.set_busy(True)
+ self.table.setSortingEnabled(False)
+ logger.info(
+ f"Updating multiple device validation statuses with names {[cfg.get('name', '') for cfg, _, _, _ in validation_results]}..."
+ )
+ for cfg, config_status, connection_status, _ in validation_results:
+ logger.info(
+ f"Updating device {cfg.get('name', '')} with config status {config_status} and connection status {connection_status}..."
+ )
+ row = self._find_row_by_name(cfg.get("name", ""))
+ if row is None:
+ logger.warning(f"Device {cfg.get('name')} not found in table for session update.")
+ continue
+ self._update_device_row_status(row, config_status, connection_status)
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.table.setSortingEnabled(True)
+ self.set_busy(False)
+
+ @SafeSlot(dict, int, int, str)
+ def update_device_validation(
+ self, device_config: dict, config_status: int, connection_status: int, validation_msg: str
+ ) -> None:
+ """
+ Update the validation status of a device. If multiple devices are being updated in a batch,
+ consider using the `update_multiple_device_validations` method instead for better performance.
+
+ Args:
+
+ """
+ self.set_busy(True)
+ row = self._find_row_by_name(device_config.get("name", ""))
+ if row is None:
+ logger.warning(
+ f"Device {device_config.get('name')} not found in table for validation update."
+ )
+ self.set_busy(False)
+ return
+ # Disable here sorting without context manager to avoid triggering of registered
+ # resizing methods. Those can be quite heavy, thus, should not run on every
+ # update of a validation status.
+ self.table.setSortingEnabled(False)
+ self._update_device_row_status(row, config_status, connection_status)
+ self.table.setSortingEnabled(True)
+ in_sync_with_redis = self._is_config_in_sync_with_redis()
+ self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
+ self.set_busy(False)
diff --git a/bec_widgets/widgets/control/device_manager/components/device_table/device_table_row.py b/bec_widgets/widgets/control/device_manager/components/device_table/device_table_row.py
new file mode 100644
index 000000000..4a777e08a
--- /dev/null
+++ b/bec_widgets/widgets/control/device_manager/components/device_table/device_table_row.py
@@ -0,0 +1,56 @@
+"""Module with custom table row for the device manager device table view."""
+
+from bec_lib.atlas_models import Device as DeviceModel
+
+from bec_widgets.widgets.control.device_manager.components.ophyd_validation import (
+ ConfigStatus,
+ ConnectionStatus,
+)
+
+
+class DeviceTableRow:
+ """
+ Custom class to hold data and validation status for a device table row.
+
+ Args:
+ data (list[str, dict] | None): Initial data for the row.
+ """
+
+ def __init__(self, data: list[str, dict] | None = None):
+ """Initialize the DeviceTableRow with optional data.
+
+ Args:
+ data (list[str, dict] | None): Initial data for the row.
+ """
+ self._data = {}
+ self.validation_status: tuple[int, int] = (ConfigStatus.UNKNOWN, ConnectionStatus.UNKNOWN)
+ self.set_data(data or {})
+
+ @property
+ def data(self) -> dict:
+ """Get the current data from the row widgets as a dictionary."""
+ return self._data
+
+ def set_data(self, data: DeviceModel | dict) -> None:
+ """Set the data for the row widgets."""
+ if isinstance(data, dict):
+ data = DeviceModel.model_validate(data)
+ old_data = DeviceModel.model_validate(self._data) if self._data else None
+ if old_data is not None and old_data == data:
+ return # No change needed
+ self._data = data.model_dump()
+ self.set_validation_status(ConfigStatus.UNKNOWN, ConnectionStatus.UNKNOWN)
+
+ def set_validation_status(
+ self, valid: ConfigStatus | int, connect_status: ConnectionStatus | int
+ ) -> None:
+ """
+ Set the validation and connection status icons.
+
+ Args:
+ valid (ConfigStatus | int): The configuration validation status.
+ connect_status (ConnectionStatus | int): The connection status.
+ """
+ valid = int(valid)
+ connect_status = int(connect_status)
+ self.validation_status = valid, connect_status
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
deleted file mode 100644
index b541916b6..000000000
--- a/bec_widgets/widgets/control/device_manager/components/device_table_view.py
+++ /dev/null
@@ -1,631 +0,0 @@
-"""Module with the device table view implementation."""
-
-from __future__ import annotations
-
-import copy
-import json
-
-from bec_lib.logger import bec_logger
-from bec_qthemes import material_icon
-from qtpy import QtCore, QtGui, QtWidgets
-from thefuzz import fuzz
-
-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
-
-logger = bec_logger.logger
-
-# Threshold for fuzzy matching, careful with adjusting this. 80 seems good
-FUZZY_SEARCH_THRESHOLD = 80
-
-
-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):
- """Override to show tooltip when hovering."""
- if event.type() != QtCore.QEvent.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)
- return True
-
-
-class CenterCheckBoxDelegate(DictToolTipDelegate):
- """Custom checkbox delegate to center checkboxes in table cells."""
-
- def __init__(self, parent=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
- )
-
- def apply_theme(self, theme: str | None = None):
- colors = get_accent_colors()
- self._icon_checked.setColor(colors.default)
- self._icon_unchecked.setColor(colors.default)
-
- 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
- pix_rect = pixmap.rect()
- pix_rect.moveCenter(rect.center())
- painter.drawPixmap(pix_rect.topLeft(), pixmap)
-
- def editorEvent(self, event, model, option, index):
- if event.type() != QtCore.QEvent.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)
-
-
-class WrappingTextDelegate(DictToolTipDelegate):
- """Custom delegate for wrapping text in table cells."""
-
- def paint(self, painter, option, index):
- text = index.model().data(index, QtCore.Qt.DisplayRole)
- if not text:
- return super().paint(painter, option, index)
-
- 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 sizeHint(self, option, index):
- text = str(index.model().data(index, QtCore.Qt.DisplayRole) or "")
- # if not text:
- # return super().sizeHint(option, index)
-
- # 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)
-
-
-class DeviceTableModel(QtCore.QAbstractTableModel):
- """
- Custom Device Table Model for managing device configurations.
-
- Sort logic is implemented directly on the data of the table view.
- """
-
- def __init__(self, device_config: list[dict] | None = None, parent=None):
- super().__init__(parent)
- self._device_config = device_config or []
- self.headers = [
- "name",
- "deviceClass",
- "readoutPriority",
- "enabled",
- "readOnly",
- "deviceTags",
- "description",
- ]
- self._checkable_columns_enabled = {"enabled": True, "readOnly": True}
-
- ###############################################
- ########## Overwrite custom Qt methods ########
- ###############################################
-
- def rowCount(self, parent=QtCore.QModelIndex()) -> int:
- return len(self._device_config)
-
- def columnCount(self, parent=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:
- return self.headers[section]
- return None
-
- def 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):
- """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"):
- return bool(value)
- if key == "deviceTags":
- return ", ".join(str(tag) for tag in value) if value else ""
- 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:
- font = QtGui.QFont()
- return font
- return None
-
- def flags(self, index):
- """Flags for the table model."""
- if not index.isValid():
- return QtCore.Qt.NoItemFlags
- key = self.headers[index.column()]
-
- if key in ("enabled", "readOnly"):
- base_flags = QtCore.Qt.ItemIsEnabled | QtCore.Qt.ItemIsSelectable
- if self._checkable_columns_enabled.get(key, True):
- return base_flags | QtCore.Qt.ItemIsUserCheckable
- else:
- return base_flags # disable editing but still visible
- return QtCore.Qt.ItemIsEnabled | QtCore.Qt.ItemIsSelectable
-
- def setData(self, index, value, role=QtCore.Qt.EditRole) -> bool:
- """
- Method to set the data of the table.
-
- Args:
- index (QModelIndex): The index of the item to modify.
- value (Any): The new value to set.
- role (Qt.ItemDataRole): The role of the data being set.
-
- Returns:
- bool: True if the data was set successfully, False otherwise.
- """
- 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 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])
- return True
- return False
-
- ####################################
- ############ Public methods ########
- ####################################
-
- def get_device_config(self) -> list[dict]:
- """Return the current device config (with checkbox updates applied)."""
- return self._device_config
-
- def set_checkbox_enabled(self, column_name: str, enabled: bool):
- """
- Enable/Disable the checkbox column.
-
- Args:
- column_name (str): The name of the column to modify.
- enabled (bool): Whether the checkbox should be enabled or disabled.
- """
- 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]
- )
-
- def set_device_config(self, device_config: list[dict]):
- """
- Replace the device config.
-
- Args:
- device_config (list[dict]): The new device config to set.
- """
- self.beginResetModel()
- self._device_config = list(device_config)
- self.endResetModel()
-
- @SafeSlot(dict)
- def add_device(self, device: dict):
- """
- Add an extra device to the device config at the bottom.
-
- Args:
- device (dict): The device configuration to add.
- """
- row = len(self._device_config)
- self.beginInsertRows(QtCore.QModelIndex(), row, row)
- self._device_config.append(device)
- self.endInsertRows()
-
- @SafeSlot(int)
- def remove_device_by_row(self, row: int):
- """
- Remove one device row by index. This maps to the row to the source of the data model
-
- Args:
- row (int): The index of the device row to remove.
- """
- if 0 <= row < len(self._device_config):
- self.beginRemoveRows(QtCore.QModelIndex(), row, row)
- self._device_config.pop(row)
- self.endRemoveRows()
-
- @SafeSlot(list)
- def remove_devices_by_rows(self, rows: list[int]):
- """
- Remove multiple device rows by their indices.
-
- Args:
- rows (list[int]): The indices of the device rows to remove.
- """
- for row in sorted(rows, reverse=True):
- self.remove_device_by_row(row)
-
- @SafeSlot(str)
- def remove_device_by_name(self, name: str):
- """
- Remove one device row by name.
-
- Args:
- name (str): The name of the device to remove.
- """
- for row, device in enumerate(self._device_config):
- if device.get("name") == name:
- self.remove_device_by_row(row)
- break
-
-
-class BECTableView(QtWidgets.QTableView):
- """Table View with custom keyPressEvent to delete rows with backspace or delete key"""
-
- def keyPressEvent(self, event) -> None:
- """
- Delete selected rows with backspace or delete key
-
- Args:
- event: keyPressEvent
- """
- if event.key() not in (QtCore.Qt.Key_Backspace, QtCore.Qt.Key_Delete):
- return super().keyPressEvent(event)
-
- proxy_indexes = self.selectedIndexes()
- if not proxy_indexes:
- return
-
- # Get unique rows (proxy indices) in reverse order so removal indexes stay valid
- proxy_rows = sorted({idx.row() for idx in proxy_indexes}, reverse=True)
- # Map to source model rows
- source_rows = [
- self.model().mapToSource(self.model().index(row, 0)).row() for row in proxy_rows
- ]
-
- model: DeviceTableModel = self.model().sourceModel() # access underlying model
- # Delegate confirmation and removal to helper
- removed = self._confirm_and_remove_rows(model, source_rows)
- if not removed:
- return
-
- def _confirm_and_remove_rows(self, model: DeviceTableModel, source_rows: list[int]) -> bool:
- """
- Prompt the user to confirm removal of rows and remove them from the model if accepted.
-
- Returns True if rows were removed, False otherwise.
- """
- cfg = model.get_device_config()
- names = [str(cfg[r].get("name", "