mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-03-06 17:02:50 +01:00
feat: add available device resource browser
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
from .available_device_resources import AvailableDeviceResources
|
||||
|
||||
__all__ = ["AvailableDeviceResources"]
|
||||
@@ -0,0 +1,36 @@
|
||||
from typing import Any, Callable, Generator, Iterable, TypeVar
|
||||
|
||||
from qtpy.QtWidgets import QListWidgetItem
|
||||
|
||||
_T = TypeVar("_T")
|
||||
_RT = TypeVar("_RT")
|
||||
|
||||
|
||||
def yield_only_passing(fn: Callable[[_T], _RT], vals: Iterable[_T]) -> Generator[_RT, Any, None]:
|
||||
for v in vals:
|
||||
try:
|
||||
yield fn(v)
|
||||
except BaseException:
|
||||
pass
|
||||
|
||||
|
||||
SORT_KEY_ROLE = 117
|
||||
|
||||
|
||||
class SortableQListWidgetItem(QListWidgetItem):
|
||||
"""Store a sorting string key with .setData(SORT_KEY_ROLE, key) to be able to sort a list with \
|
||||
custom widgets and this item."""
|
||||
|
||||
def __gt__(self, other):
|
||||
if (self_key := self.data(SORT_KEY_ROLE)) is None or (
|
||||
other_key := other.data(SORT_KEY_ROLE)
|
||||
) is None:
|
||||
return False
|
||||
return self_key.lower() > other_key.lower()
|
||||
|
||||
def __lt__(self, other):
|
||||
if (self_key := self.data(SORT_KEY_ROLE)) is None or (
|
||||
other_key := other.data(SORT_KEY_ROLE)
|
||||
) is None:
|
||||
return False
|
||||
return self_key.lower() < other_key.lower()
|
||||
@@ -0,0 +1,200 @@
|
||||
from textwrap import dedent
|
||||
from typing import 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.utils.expandable_frame import ExpandableGroupFrame
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group_ui import (
|
||||
Ui_AvailableDeviceGroup,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
|
||||
HashableDevice,
|
||||
)
|
||||
|
||||
|
||||
def _warning_string(spec: HashableDevice):
|
||||
name_warning = (
|
||||
f"Device defined with multiple names! Please check:\n {'\n '.join(spec.names)}\n"
|
||||
if len(spec.names) > 1
|
||||
else ""
|
||||
)
|
||||
source_warning = (
|
||||
f"Device found in multiple source files! Please check:\n {'\n '.join(spec._source_files)}"
|
||||
if len(spec._source_files) > 1
|
||||
else ""
|
||||
)
|
||||
return f"{name_warning}{source_warning}"
|
||||
|
||||
|
||||
class _DeviceEntryWidget(QFrame):
|
||||
|
||||
def __init__(self, device_spec: HashableDevice, parent=None, **kwargs):
|
||||
super().__init__(parent, **kwargs)
|
||||
self._device_spec = device_spec
|
||||
self.included: bool = False
|
||||
|
||||
self.setFrameShape(QFrame.Shape.StyledPanel)
|
||||
self.setFrameShadow(QFrame.Shadow.Raised)
|
||||
|
||||
self._layout = QVBoxLayout()
|
||||
self._layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.setLayout(self._layout)
|
||||
|
||||
self.setup_title_layout(device_spec)
|
||||
self.check_and_display_warning()
|
||||
|
||||
self.setToolTip(self._rich_text())
|
||||
|
||||
def _rich_text(self):
|
||||
return dedent(
|
||||
f"""
|
||||
<b><u><h2> {self._device_spec.name}: </h2></u></b>
|
||||
<table>
|
||||
<tr><td> description: </td><td><i> {self._device_spec.description} </i></td></tr>
|
||||
<tr><td> config: </td><td><i> {self._device_spec.deviceConfig} </i></td></tr>
|
||||
<tr><td> enabled: </td><td><i> {self._device_spec.enabled} </i></td></tr>
|
||||
<tr><td> read only: </td><td><i> {self._device_spec.readOnly} </i></td></tr>
|
||||
</table>
|
||||
"""
|
||||
)
|
||||
|
||||
def setup_title_layout(self, device_spec: HashableDevice):
|
||||
self._title_layout = QHBoxLayout()
|
||||
self._title_layout.setContentsMargins(0, 0, 0, 0)
|
||||
self._title_container = QWidget(parent=self)
|
||||
self._title_container.setLayout(self._title_layout)
|
||||
|
||||
self._warning_label = QLabel()
|
||||
self._title_layout.addWidget(self._warning_label)
|
||||
|
||||
self.title = QLabel(device_spec.name)
|
||||
self.title.setToolTip(device_spec.name)
|
||||
self.title.setStyleSheet(self.title_style("#FF0000"))
|
||||
self._title_layout.addWidget(self.title)
|
||||
|
||||
self._title_layout.addStretch(1)
|
||||
self._layout.addWidget(self._title_container)
|
||||
|
||||
def check_and_display_warning(self):
|
||||
if len(self._device_spec.names) == 1 and len(self._device_spec._source_files) == 1:
|
||||
self._warning_label.setText("")
|
||||
self._warning_label.setToolTip("")
|
||||
else:
|
||||
self._warning_label.setPixmap(material_icon("warning", size=(12, 12), color="#FFAA00"))
|
||||
self._warning_label.setToolTip(_warning_string(self._device_spec))
|
||||
|
||||
@property
|
||||
def device_hash(self):
|
||||
return hash(self._device_spec)
|
||||
|
||||
def title_style(self, color: str) -> str:
|
||||
return f"QLabel {{ color: {color}; font-weight: bold; font-size: 10pt; }}"
|
||||
|
||||
def setTitle(self, text: str):
|
||||
self.title.setText(text)
|
||||
|
||||
def set_included(self, included: bool):
|
||||
self.included = included
|
||||
self.title.setStyleSheet(self.title_style("#00FF00" if included else "#FF0000"))
|
||||
|
||||
|
||||
class _DeviceEntry(NamedTuple):
|
||||
list_item: QListWidgetItem
|
||||
widget: _DeviceEntryWidget
|
||||
|
||||
|
||||
class AvailableDeviceGroup(ExpandableGroupFrame, Ui_AvailableDeviceGroup):
|
||||
def __init__(
|
||||
self, parent=None, name: str = "TagGroupTitle", data: set[HashableDevice] = set(), **kwargs
|
||||
):
|
||||
super().__init__(parent=parent, **kwargs)
|
||||
self.setupUi(self)
|
||||
self.title_text = name # type: ignore
|
||||
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()
|
||||
|
||||
self.add_to_composition_button.clicked.connect(self.test)
|
||||
|
||||
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 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,
|
||||
)
|
||||
|
||||
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 test(self, *args):
|
||||
print(self.get_selection())
|
||||
|
||||
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())
|
||||
@@ -0,0 +1,66 @@
|
||||
from functools import partial
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from PySide6.QtWidgets import QFrame
|
||||
from qtpy.QtCore import QMetaObject
|
||||
from qtpy.QtWidgets import QLabel, QListWidget, QToolButton, QVBoxLayout
|
||||
|
||||
|
||||
class Ui_AvailableDeviceGroup(object):
|
||||
def setupUi(self, 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.delete_tag_button = QToolButton(AvailableDeviceGroup)
|
||||
self.delete_tag_button.setObjectName("delete_tag_button")
|
||||
title_layout.addWidget(self.delete_tag_button)
|
||||
|
||||
self.remove_from_composition_button = QToolButton(AvailableDeviceGroup)
|
||||
self.remove_from_composition_button.setObjectName("remove_from_composition_button")
|
||||
title_layout.addWidget(self.remove_from_composition_button)
|
||||
|
||||
self.add_to_composition_button = QToolButton(AvailableDeviceGroup)
|
||||
self.add_to_composition_button.setObjectName("add_to_composition_button")
|
||||
title_layout.addWidget(self.add_to_composition_button)
|
||||
|
||||
self.remove_all_button = QToolButton(AvailableDeviceGroup)
|
||||
self.remove_all_button.setObjectName("remove_all_from_composition_button")
|
||||
title_layout.addWidget(self.remove_all_button)
|
||||
|
||||
self.add_all_button = QToolButton(AvailableDeviceGroup)
|
||||
self.add_all_button.setObjectName("add_all_to_composition_button")
|
||||
title_layout.addWidget(self.add_all_button)
|
||||
|
||||
self.device_list = QListWidget(AvailableDeviceGroup)
|
||||
self.device_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
|
||||
self.device_list.setObjectName("device_list")
|
||||
self.device_list.setFrameStyle(0)
|
||||
|
||||
self.verticalLayout.addWidget(self.device_list)
|
||||
|
||||
self.set_icons()
|
||||
|
||||
QMetaObject.connectSlotsByName(AvailableDeviceGroup)
|
||||
|
||||
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")
|
||||
@@ -0,0 +1,83 @@
|
||||
from random import randint
|
||||
from typing import Any, Iterable
|
||||
|
||||
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.available_device_resources._util import (
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
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.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)
|
||||
|
||||
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]):
|
||||
self.device_groups_list.add_item(
|
||||
device_group, self.device_groups_list, device_group, devices, expanded=False
|
||||
)
|
||||
|
||||
def _reset_devices_state(self):
|
||||
for device_group in self.device_groups_list.widgets():
|
||||
device_group.reset_devices_state()
|
||||
|
||||
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)
|
||||
|
||||
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(list)
|
||||
def update_devices_state(self, config_list: list[dict[str, Any]]):
|
||||
self.set_devices_state(yield_only_passing(HashableDevice.model_validate, config_list), True)
|
||||
|
||||
@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())
|
||||
@@ -0,0 +1,55 @@
|
||||
from qtpy.QtCore import QMetaObject, Qt
|
||||
from qtpy.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QComboBox,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QListView,
|
||||
QListWidget,
|
||||
QVBoxLayout,
|
||||
)
|
||||
|
||||
from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group import (
|
||||
AvailableDeviceGroup,
|
||||
)
|
||||
|
||||
|
||||
class Ui_availableDeviceResources(object):
|
||||
def setupUi(self, availableDeviceResources):
|
||||
if not availableDeviceResources.objectName():
|
||||
availableDeviceResources.setObjectName("availableDeviceResources")
|
||||
self.verticalLayout = QVBoxLayout(availableDeviceResources)
|
||||
self.verticalLayout.setObjectName("verticalLayout")
|
||||
|
||||
self.search_layout = QHBoxLayout()
|
||||
self.verticalLayout.addLayout(self.search_layout)
|
||||
self.search_layout.addWidget(QLabel("Filter groups: "))
|
||||
self.search_box = QLineEdit()
|
||||
self.search_layout.addWidget(self.search_box)
|
||||
self.search_layout.addWidget(QLabel("Group by: "))
|
||||
self.grouping_selector = QComboBox()
|
||||
self.search_layout.addWidget(self.grouping_selector)
|
||||
|
||||
self.device_groups_list = ListOfExpandableFrames(
|
||||
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(2)
|
||||
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.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
availableDeviceResources.setMinimumWidth(250)
|
||||
availableDeviceResources.resize(250, availableDeviceResources.height())
|
||||
|
||||
self.verticalLayout.addWidget(self.device_groups_list)
|
||||
|
||||
QMetaObject.connectSlotsByName(availableDeviceResources)
|
||||
@@ -0,0 +1,139 @@
|
||||
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
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
_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()
|
||||
)
|
||||
|
||||
|
||||
# use the last n recovery files
|
||||
_N_RECOVERY_FILES = 3
|
||||
|
||||
|
||||
class _ConfigFileBackend(DeviceResourceBackend):
|
||||
def __init__(self) -> None:
|
||||
self._raw_device_set: set[
|
||||
HashableDevice
|
||||
] = self._get_config_from_backup_files() | 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),
|
||||
),
|
||||
)
|
||||
|
||||
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)))
|
||||
|
||||
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))
|
||||
|
||||
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}
|
||||
Reference in New Issue
Block a user