mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-12-30 02:31:20 +01:00
feat: available resources basic structure
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
from .available_device_resources import AvailableDeviceResources
|
||||
|
||||
__all__ = ["AvailableDeviceResources"]
|
||||
@@ -0,0 +1,61 @@
|
||||
from random import randint
|
||||
from typing import Iterable
|
||||
|
||||
from qtpy.QtCore import QSize
|
||||
from qtpy.QtWidgets import QListWidgetItem, QWidget
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
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.available_device_resources.device_tag_group import (
|
||||
DeviceTagGroup,
|
||||
)
|
||||
|
||||
|
||||
class AvailableDeviceResources(BECWidget, QWidget, Ui_availableDeviceResources):
|
||||
def __init__(self, parent=None, **kwargs):
|
||||
super().__init__(parent=parent, **kwargs)
|
||||
self.setupUi(self)
|
||||
self._backend = get_backend()
|
||||
self._items: dict[str, tuple[QListWidgetItem, DeviceTagGroup]] = {}
|
||||
self.refresh_full_list()
|
||||
|
||||
def refresh_full_list(self):
|
||||
self.tag_groups_list.clear()
|
||||
self._items = {}
|
||||
for tag_group, devices in self._backend.tag_groups.items():
|
||||
item = QListWidgetItem(self.tag_groups_list)
|
||||
tag_group_widget = DeviceTagGroup(self.tag_groups_list, tag_group, devices)
|
||||
self.tag_groups_list.setItemWidget(item, tag_group_widget)
|
||||
self.tag_groups_list.addItem(item)
|
||||
self._items[tag_group] = (item, tag_group_widget)
|
||||
item.setSizeHint(QSize(tag_group_widget.width(), tag_group_widget.height()))
|
||||
|
||||
def set_devices_state(self, devices: Iterable[HashableDevice], included: bool):
|
||||
for _, tag_group in self._items.values():
|
||||
for device in devices:
|
||||
tag_group.set_item_state(hash(device), included)
|
||||
|
||||
def resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
for list_item, tag_group_widget in self._items.values():
|
||||
list_item.setSizeHint(tag_group_widget.sizeHint())
|
||||
|
||||
|
||||
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())
|
||||
@@ -0,0 +1,27 @@
|
||||
from qtpy.QtCore import QMetaObject, Qt
|
||||
from qtpy.QtWidgets import QAbstractItemView, QListView, QListWidget, QVBoxLayout
|
||||
|
||||
|
||||
class Ui_availableDeviceResources(object):
|
||||
def setupUi(self, availableDeviceResources):
|
||||
if not availableDeviceResources.objectName():
|
||||
availableDeviceResources.setObjectName("availableDeviceResources")
|
||||
self.verticalLayout = QVBoxLayout(availableDeviceResources)
|
||||
self.verticalLayout.setObjectName("verticalLayout")
|
||||
self.tag_groups_list = QListWidget(availableDeviceResources)
|
||||
self.tag_groups_list.setObjectName("tag_groups_list")
|
||||
self.tag_groups_list.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
|
||||
self.tag_groups_list.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.tag_groups_list.setHorizontalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.tag_groups_list.setMovement(QListView.Movement.Static)
|
||||
self.tag_groups_list.setSpacing(2)
|
||||
self.tag_groups_list.setDragDropMode(QListWidget.DragDropMode.DragOnly)
|
||||
self.tag_groups_list.setDragEnabled(True)
|
||||
self.tag_groups_list.setAcceptDrops(False)
|
||||
self.tag_groups_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
availableDeviceResources.setMinimumWidth(250)
|
||||
availableDeviceResources.resize(250, availableDeviceResources.height())
|
||||
|
||||
self.verticalLayout.addWidget(self.tag_groups_list)
|
||||
|
||||
QMetaObject.connectSlotsByName(availableDeviceResources)
|
||||
@@ -0,0 +1,143 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import operator
|
||||
from functools import reduce
|
||||
from glob import glob
|
||||
from pathlib import Path
|
||||
from textwrap import dedent
|
||||
from typing import AbstractSet, Protocol
|
||||
|
||||
from bec_lib.atlas_models import Device
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path
|
||||
from pydantic import model_validator
|
||||
|
||||
|
||||
class HashableDevice(Device):
|
||||
source_files: set[str] = set()
|
||||
names: set[str] = set()
|
||||
|
||||
@model_validator(mode="after")
|
||||
def add_name(self) -> HashableDevice:
|
||||
self.names.add(self.name)
|
||||
return self
|
||||
|
||||
def as_normal_device(self):
|
||||
return Device.model_validate(self)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
config_values = sorted(
|
||||
(str(kv) for kv in self.deviceConfig.items()) if self.deviceConfig else []
|
||||
)
|
||||
return (reduce(operator.add, (self.name, self.deviceClass, *config_values))).__hash__()
|
||||
|
||||
def __eq__(self, value: object) -> bool:
|
||||
if not isinstance(value, self.__class__):
|
||||
return False
|
||||
if hash(self) == hash(value):
|
||||
return True
|
||||
return False
|
||||
|
||||
def rich_text(self) -> str:
|
||||
return dedent(
|
||||
f"""
|
||||
<b><u><h2> {self.name}: </h2></u></b>
|
||||
<table>
|
||||
<tr><td> description: </td><td><i> {self.description} </i></td></tr>
|
||||
<tr><td> config: </td><td><i> {self.deviceConfig} </i></td></tr>
|
||||
<tr><td> enabled: </td><td><i> {self.enabled} </i></td></tr>
|
||||
<tr><td> read only: </td><td><i> {self.readOnly} </i></td></tr>
|
||||
</table>
|
||||
"""
|
||||
)
|
||||
|
||||
def add_sources(self, other: HashableDevice):
|
||||
self.source_files.update(other.source_files)
|
||||
|
||||
def add_tags(self, other: HashableDevice):
|
||||
self.deviceTags.update(other.deviceTags)
|
||||
|
||||
def add_names(self, other: HashableDevice):
|
||||
self.names.update(other.names)
|
||||
|
||||
|
||||
class _HashableDeviceSet(set):
|
||||
def __or__(self, value: AbstractSet) -> _HashableDeviceSet:
|
||||
for item in self:
|
||||
if item in value:
|
||||
for other_item in value:
|
||||
if other_item == item:
|
||||
item.add_sources(other_item)
|
||||
item.add_tags(other_item)
|
||||
item.add_names(other_item)
|
||||
for other_item in value:
|
||||
if other_item not in self:
|
||||
self.add(other_item)
|
||||
return self
|
||||
|
||||
|
||||
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."""
|
||||
...
|
||||
|
||||
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."""
|
||||
...
|
||||
|
||||
|
||||
class _ConfigFileBackend(DeviceResourceBackend):
|
||||
def __init__(self) -> None:
|
||||
self._raw_device_set: set[HashableDevice] = self._get_config_from_files(
|
||||
Path(plugin_repo_path()) / plugin_package_name() / "device_configs/"
|
||||
)
|
||||
self._tag_groups = self._get_tag_groups()
|
||||
|
||||
def _get_config_from_files(self, dir: Path):
|
||||
files = glob("*.yaml", root_dir=dir, recursive=True)
|
||||
|
||||
def devices_from_file(file: str):
|
||||
data = yaml_load(str(dir / file))
|
||||
return set(
|
||||
HashableDevice.model_validate(
|
||||
dev | {"name": name, "source_files": {str(dir / file)}}
|
||||
)
|
||||
for name, dev in data.items()
|
||||
)
|
||||
|
||||
return reduce(operator.or_, map(devices_from_file, files))
|
||||
|
||||
def _get_tag_groups(self) -> dict[str, set[HashableDevice]]:
|
||||
return {
|
||||
tag: set(filter(lambda dev: tag in dev.deviceTags, self._raw_device_set))
|
||||
for tag in self.tags()
|
||||
}
|
||||
|
||||
@property
|
||||
def tag_groups(self):
|
||||
return self._tag_groups
|
||||
|
||||
@property
|
||||
def all_devices(self):
|
||||
return self._raw_device_set
|
||||
|
||||
def tags(self) -> set[str]:
|
||||
return reduce(operator.or_, (dev.deviceTags for dev in self._raw_device_set))
|
||||
|
||||
def tag_group(self, tag: str) -> set[HashableDevice]:
|
||||
return self.tag_groups[tag]
|
||||
|
||||
|
||||
def get_backend() -> DeviceResourceBackend:
|
||||
return _ConfigFileBackend()
|
||||
@@ -0,0 +1,163 @@
|
||||
from textwrap import dedent
|
||||
from typing import Callable, NamedTuple
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QSize
|
||||
from qtpy.QtWidgets import QFrame, QHBoxLayout, QLabel, QListWidgetItem, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
|
||||
HashableDevice,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_tag_group_item_ui import (
|
||||
Ui_DeviceTagGroup,
|
||||
)
|
||||
|
||||
DEVICE_HASH_ROLE = 101
|
||||
|
||||
|
||||
class _DeviceEntryWidget(QFrame):
|
||||
_grid_size = QSize(120, 80)
|
||||
|
||||
def __init__(self, device_spec: HashableDevice, parent=None, **kwargs):
|
||||
super().__init__(parent, **kwargs)
|
||||
self._device_spec = device_spec
|
||||
self.included: bool = False
|
||||
|
||||
self.setFrameShape(QFrame.Shape.StyledPanel)
|
||||
self.setFrameShadow(QFrame.Shadow.Raised)
|
||||
|
||||
self._layout = QVBoxLayout()
|
||||
self._layout.setContentsMargins(5, 5, 5, 5)
|
||||
self.setLayout(self._layout)
|
||||
self.setMinimumSize(self._grid_size)
|
||||
|
||||
self.setup_title_layout(device_spec)
|
||||
|
||||
self.setToolTip(device_spec.rich_text())
|
||||
|
||||
self.details = QLabel(f"Tags:\n{', '.join(device_spec.deviceTags)}")
|
||||
self.details.setStyleSheet("QLabel { font-size: 8pt; }")
|
||||
self.details.setWordWrap(True)
|
||||
self._layout.addWidget(self.details)
|
||||
|
||||
def setup_title_layout(self, device_spec: HashableDevice):
|
||||
self._title_layout = QHBoxLayout()
|
||||
self._title_container = QWidget(parent=self)
|
||||
self._title_container.setLayout(self._title_layout)
|
||||
|
||||
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._warning_label = QLabel()
|
||||
self._title_layout.addWidget(self._warning_label)
|
||||
|
||||
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", color="#FFAA00"))
|
||||
self._warning_label.setToolTip(
|
||||
dedent(
|
||||
f"""
|
||||
{f"Device has multiple names! Please check! \n names: {self._device_spec.names}" if len(self._device_spec.names)>1 else ""}
|
||||
{f"Device found in multiple source files! Please check! \n files: {self._device_spec.names}" if len(self._device_spec.names)>1 else ""}
|
||||
"""
|
||||
)
|
||||
)
|
||||
|
||||
@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 DeviceTagGroup(QWidget, Ui_DeviceTagGroup):
|
||||
def __init__(
|
||||
self, parent=None, name: str = "TagGroupTitle", data: set[HashableDevice] = set(), **kwargs
|
||||
):
|
||||
super().__init__(parent=parent, **kwargs)
|
||||
self.setupUi(self)
|
||||
self.device_list.setGridSize(_DeviceEntryWidget._grid_size)
|
||||
self.title.setText(name)
|
||||
self._devices: dict[str, _DeviceEntry] = {}
|
||||
for device in data:
|
||||
self._add_item(device)
|
||||
self.device_list.sortItems()
|
||||
|
||||
def _add_item(self, device: HashableDevice):
|
||||
item = QListWidgetItem(self.device_list)
|
||||
widget = _DeviceEntryWidget(device, self)
|
||||
item.setSizeHint(QSize(widget.width(), widget.height()))
|
||||
self.device_list.setItemWidget(item, widget)
|
||||
self.device_list.addItem(item)
|
||||
self._devices[device.name] = _DeviceEntry(item, widget)
|
||||
|
||||
def 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 resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
self.setMinimumHeight(self.sizeHint().height())
|
||||
self.setMaximumHeight(self.sizeHint().height())
|
||||
|
||||
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 = DeviceTagGroup(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())
|
||||
@@ -0,0 +1,135 @@
|
||||
import math
|
||||
from functools import partial
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QMetaObject, QSize, Qt
|
||||
from qtpy.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QFrame,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QListView,
|
||||
QListWidget,
|
||||
QSizePolicy,
|
||||
QSpacerItem,
|
||||
QToolButton,
|
||||
QVBoxLayout,
|
||||
)
|
||||
|
||||
|
||||
class AutoHeightListWidget(QListWidget):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setViewMode(QListView.ViewMode.IconMode)
|
||||
self.setResizeMode(QListView.ResizeMode.Adjust)
|
||||
self.setWrapping(True)
|
||||
self.setUniformItemSizes(True)
|
||||
self.setMovement(QListView.Movement.Static)
|
||||
self.setAcceptDrops(False)
|
||||
self.setDragEnabled(True)
|
||||
self.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
|
||||
self.setSpacing(5)
|
||||
|
||||
self.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
self.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
|
||||
def resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
self.setMinimumHeight(self._calcSize().height())
|
||||
self.setMaximumHeight(self._calcSize().height())
|
||||
|
||||
def sizeHint(self):
|
||||
return self._calcSize()
|
||||
|
||||
def minimumSizeHint(self):
|
||||
return self._calcSize()
|
||||
|
||||
def _calcSize(self):
|
||||
if self.count() == 0:
|
||||
return super().sizeHint()
|
||||
|
||||
grid = self.gridSize()
|
||||
if not grid.isValid():
|
||||
grid = QSize(100, 100) # fallback
|
||||
|
||||
items_per_row = max(1, self.viewport().width() // grid.width())
|
||||
rows = math.ceil(self.count() / items_per_row)
|
||||
|
||||
height = rows * grid.height() + 2 * self.frameWidth()
|
||||
return QSize(self.viewport().width(), height)
|
||||
|
||||
|
||||
class Ui_DeviceTagGroup(object):
|
||||
def setupUi(self, DeviceTagGroup):
|
||||
if not DeviceTagGroup.objectName():
|
||||
DeviceTagGroup.setObjectName("DeviceTagGroup")
|
||||
DeviceTagGroup.setMinimumWidth(150)
|
||||
self.verticalLayout = QVBoxLayout(DeviceTagGroup)
|
||||
self.verticalLayout.setObjectName("verticalLayout")
|
||||
self.frame = QFrame(DeviceTagGroup)
|
||||
self.frame.setObjectName("frame")
|
||||
self.frame.setFrameShape(QFrame.Shape.StyledPanel)
|
||||
self.frame.setFrameShadow(QFrame.Shadow.Raised)
|
||||
self.verticalLayout_2 = QVBoxLayout(self.frame)
|
||||
self.verticalLayout_2.setObjectName("verticalLayout_2")
|
||||
self.horizontalLayout = QHBoxLayout()
|
||||
self.horizontalLayout.setObjectName("horizontalLayout")
|
||||
|
||||
self.title = QLabel(self.frame)
|
||||
self.title.setObjectName("title")
|
||||
self.horizontalLayout.addWidget(self.title)
|
||||
|
||||
self.n_included = QLabel(self.frame, text="...")
|
||||
self.n_included.setObjectName("n_included")
|
||||
self.horizontalLayout.addWidget(self.n_included)
|
||||
|
||||
self.horizontalSpacer = QSpacerItem(
|
||||
40, 20, QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Minimum
|
||||
)
|
||||
self.horizontalLayout.addItem(self.horizontalSpacer)
|
||||
|
||||
self.delete_tag_button = QToolButton(self.frame)
|
||||
self.delete_tag_button.setObjectName("delete_tag_button")
|
||||
self.horizontalLayout.addWidget(self.delete_tag_button)
|
||||
|
||||
self.remove_from_composition_button = QToolButton(self.frame)
|
||||
self.remove_from_composition_button.setObjectName("remove_from_composition_button")
|
||||
self.horizontalLayout.addWidget(self.remove_from_composition_button)
|
||||
|
||||
self.add_to_composition_button = QToolButton(self.frame)
|
||||
self.add_to_composition_button.setObjectName("add_to_composition_button")
|
||||
self.horizontalLayout.addWidget(self.add_to_composition_button)
|
||||
|
||||
self.remove_all_button = QToolButton(self.frame)
|
||||
self.remove_all_button.setObjectName("remove_all_from_composition_button")
|
||||
self.horizontalLayout.addWidget(self.remove_all_button)
|
||||
|
||||
self.add_all_button = QToolButton(self.frame)
|
||||
self.add_all_button.setObjectName("add_all_to_composition_button")
|
||||
self.horizontalLayout.addWidget(self.add_all_button)
|
||||
|
||||
self.verticalLayout_2.addLayout(self.horizontalLayout)
|
||||
|
||||
self.device_list = AutoHeightListWidget(self.frame)
|
||||
self.device_list.setObjectName("device_list")
|
||||
|
||||
self.verticalLayout_2.addWidget(self.device_list)
|
||||
|
||||
self.verticalLayout.addWidget(self.frame)
|
||||
|
||||
self.set_icons()
|
||||
|
||||
QMetaObject.connectSlotsByName(DeviceTagGroup)
|
||||
|
||||
def set_icons(self):
|
||||
icon = partial(material_icon, size=(15, 15), convert_to_pixmap=False)
|
||||
self.delete_tag_button.setIcon(icon("delete"))
|
||||
self.delete_tag_button.setToolTip("Delete tag group")
|
||||
self.remove_from_composition_button.setIcon(icon("remove"))
|
||||
self.remove_from_composition_button.setToolTip("Remove selected from composition")
|
||||
self.add_to_composition_button.setIcon(icon("add"))
|
||||
self.add_to_composition_button.setToolTip("Add selected to composition")
|
||||
self.remove_all_button.setIcon(icon("chips"))
|
||||
self.remove_all_button.setToolTip("Remove all with this tag from composition")
|
||||
self.add_all_button.setIcon(icon("add_box"))
|
||||
self.add_all_button.setToolTip("Add all with this tag to composition")
|
||||
78
tests/unit_tests/test_available_device_resources.py
Normal file
78
tests/unit_tests/test_available_device_resources.py
Normal file
@@ -0,0 +1,78 @@
|
||||
from copy import copy
|
||||
|
||||
import pytest
|
||||
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
|
||||
HashableDevice,
|
||||
_HashableDeviceSet,
|
||||
)
|
||||
|
||||
TEST_DEVICE_DICT = {
|
||||
"name": "test_device",
|
||||
"deviceClass": "TestDeviceClass",
|
||||
"readoutPriority": "baseline",
|
||||
"enabled": True,
|
||||
}
|
||||
|
||||
|
||||
def _test_device_dict(**kwargs):
|
||||
new = copy(TEST_DEVICE_DICT)
|
||||
new.update(kwargs)
|
||||
return new
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"kwargs_1, kwargs_2, kwargs_3, kwargs_4, n",
|
||||
[
|
||||
({}, {}, {}, {}, 1),
|
||||
({}, {}, {}, {"deviceConfig": {"a": 1}}, 1),
|
||||
({}, {}, {}, {"name": "test_device_2"}, 2),
|
||||
({}, {}, {"name": "test_device_2"}, {"deviceClass": "OtherDeviceClass"}, 3),
|
||||
],
|
||||
)
|
||||
def test_hashable_device_set_merges_equal(kwargs_1, kwargs_2, kwargs_3, kwargs_4, n):
|
||||
item_1 = HashableDevice(**_test_device_dict(**kwargs_1))
|
||||
item_2 = HashableDevice(**_test_device_dict(**kwargs_2))
|
||||
item_3 = HashableDevice(**_test_device_dict(**kwargs_3))
|
||||
item_4 = HashableDevice(**_test_device_dict(**kwargs_4))
|
||||
|
||||
test_set = _HashableDeviceSet((item_1, item_2, item_3, item_4))
|
||||
assert len(test_set) == n
|
||||
|
||||
|
||||
def test_hashable_device_set_or_adds_sources():
|
||||
item_1 = HashableDevice(**_test_device_dict(), source_files={"a", "b"})
|
||||
item_2 = HashableDevice(**_test_device_dict(), source_files={"c", "d"})
|
||||
|
||||
set_1 = _HashableDeviceSet((item_1,))
|
||||
set_2 = _HashableDeviceSet((item_2,))
|
||||
|
||||
combined = set_1 | set_2
|
||||
assert len(combined) == 1
|
||||
assert combined.pop().source_files == {"a", "b", "c", "d"}
|
||||
|
||||
|
||||
def test_hashable_device_set_or_adds_tags():
|
||||
item_1 = HashableDevice(
|
||||
**_test_device_dict(deviceTags={"tag1"}, deviceConfig={"param": "value"}),
|
||||
source_files={"a", "b"},
|
||||
)
|
||||
item_2 = HashableDevice(
|
||||
**_test_device_dict(deviceTags={"tag2"}, deviceConfig={"param": "value"}),
|
||||
source_files={"c", "d"},
|
||||
)
|
||||
item_3 = HashableDevice(
|
||||
**_test_device_dict(deviceTags={"tag3"}, deviceConfig={"param": "other_value"}),
|
||||
source_files={"q"},
|
||||
)
|
||||
|
||||
set_1 = _HashableDeviceSet((item_1,))
|
||||
set_2 = _HashableDeviceSet((item_2,))
|
||||
set_3 = _HashableDeviceSet((item_3,))
|
||||
|
||||
combined = sorted(set_1 | set_2 | set_3, key=lambda hd: hd.deviceConfig["param"])
|
||||
assert len(combined) == 2
|
||||
assert combined[0].source_files == {"q"}
|
||||
assert combined[0].deviceTags == {"tag3"}
|
||||
assert combined[1].source_files == {"a", "b", "c", "d"}
|
||||
assert combined[1].deviceTags == {"tag1", "tag2"}
|
||||
Reference in New Issue
Block a user