reactor: cleanup and small logic fix

This commit is contained in:
2025-07-29 11:39:44 +02:00
committed by Christian Appel
parent f88eb8bc82
commit 749766118f
3 changed files with 18 additions and 10 deletions
@@ -23,7 +23,7 @@ from bec_lib.redis_connector import RedisConnector
from bec_server.device_server.device_server import DeviceServer
logger = bec_logger.logger
bec_logger.level = bec_logger.LOGLEVEL.DEBUG
bec_logger.level = bec_logger.LOGLEVEL.INFO
# ---- Forward ophyd logs to bec_logger.logger ----
@@ -140,18 +140,13 @@ def get_device_info(
)
# Collect signals and their metadata
signals = {} # []
if hasattr(obj, "component_names") and connect:
# Read attrs are only available if the device is connected, thus we can only deactivate the device
if hasattr(obj, "read_attrs"):
signal_names = [read_attr_name.replace(".", "_") for read_attr_name in obj.read_attrs]
unique_signal_names = set(signal_names)
if len(unique_signal_names) < len(signal_names):
raise DeviceConfigError(
f"Signal names of {obj.name} must be unique, found duplicates. All signal names: {signal_names}. \n Unique signal names: {unique_signal_names}."
)
signal_names = []
walk = obj.walk_components()
for _ancestor, component_name, comp in walk:
if get_device_base_class(getattr(obj, component_name)) == "signal":
if component_name in protected_names:
raise DeviceConfigError(
f"Signal name {component_name} is protected and cannot be used. Please rename the signal."
@@ -190,12 +185,13 @@ def get_device_info(
}
)
else:
obj_name = signal_obj.name
signals.update(
{
component_name: {
"component_name": component_name,
"signal_class": signal_obj.__class__.__name__,
"obj_name": signal_obj.name,
"obj_name": obj_name,
"kind_int": signal_obj.kind.value,
"kind_str": signal_obj.kind.name,
"doc": doc,
@@ -205,6 +201,13 @@ def get_device_info(
}
}
)
signal_names.append(obj_name)
# Read attrs are only available if the device is connected
unique_signal_names = set(signal_names)
if len(unique_signal_names) < len(signal_names):
raise DeviceConfigError(
f"Signal names of {obj.name} must be unique, found duplicates. All signal names: {signal_names}. \n Unique signal names: {unique_signal_names}."
)
sub_devices = []
if hasattr(obj, "walk_subdevices") and connect:
@@ -72,6 +72,11 @@ class DummyDeviceWithConflictingDuplicateSignalNames(Device):
signal = Cpt(MyDevice, "signal", lazy=True)
signal_custom = Cpt(MyDevice, "custom", lazy=True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Imitate Positioner behavior with conflicting signal names
self.signal_custom.custom.name = self.signal_custom.name
@pytest.mark.parametrize(
"obj",