bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-12-31 03:01:18 +01:00
Code Activity

fix: check plugin exists before loading

Browse Source
This commit is contained in:
perl_d
2025-09-15 10:40:56 +02:00
committed by Jan Wyzula
parent 33c4527da9
commit 0bb693a062
1 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
bec_widgets/widgets/control/device_manager/components/available_device_resources/device_resource_backend.py
View File

@@ -12,7 +12,7 @@ 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
from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path, plugins_installed
logger = bec_logger.logger
@@ -81,17 +81,21 @@ _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._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:]
if len(last_n_files) == 0:
return set()
return reduce(
operator.or_,
map(
@@ -127,7 +131,7 @@ class _ConfigFileBackend(DeviceResourceBackend):
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))
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]