Fix error messages for sub-device access of setters

bec_lib/bec_lib/device.py

@@ -76,6 +76,28 @@ def __init__(
         super().__init__(alarm, Alarms.MAJOR, handled=False)
 
 
+class NotImplementedOnSubdeviceError(AlarmBase):
+    """Exception raised when a method is not implemented for a subdevice."""
+
+    def __init__(
+        self, device: str, sub_device: str, method: str, additional_info: str | None = None
+    ) -> None:
+        compact_error_message = f"Method '{method}' is not implemented for subdevice '{sub_device}' of device '{device}'. Try to access the root device '{device}' instead."
+        error_message = compact_error_message
+        if additional_info:
+            error_message += f" Additional info: {additional_info}"
+
+        alarm = messages.AlarmMessage(
+            severity=Alarms.MAJOR,
+            info=messages.ErrorInfo(
+                exception_type="NotImplemented",
+                error_message=error_message,
+                compact_error_message=compact_error_message,
+            ),
+        )
+        super().__init__(alarm, Alarms.MAJOR, handled=False)
+
+
 class ScanRequestError(Exception):
     """Exception raised when a scan request is rejected."""
 
@@ -767,6 +789,10 @@ def enabled(self):
     @enabled.setter
     def enabled(self, val):
         # pylint: disable=protected-access
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="enabled"
+            )
         self._update_config({"enabled": val})
         self.root._config["enabled"] = val
 
@@ -778,19 +804,30 @@ def _update_config(self, update: dict) -> None:
             update (dict): The update dictionary.
 
         """
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name,
+                sub_device=self.dotted_name,
+                method="_update_config",
+                additional_info=f"Received update: {update}",
+            )
         self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: update}
         )
 
-    def get_device_tags(self) -> list:
+    def get_device_tags(self) -> set:
         """get the device tags for this device"""
         # pylint: disable=protected-access
-        return self.root._config.get("deviceTags", [])
+        return self.root._config.get("deviceTags", set())
 
     @typechecked
     def set_device_tags(self, val: Iterable):
         """set the device tags for this device - any duplicates will be discarded"""
         # pylint: disable=protected-access
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="set_device_tags"
+            )
         self.root._config["deviceTags"] = set(val)
         return self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: {"deviceTags": self.root._config["deviceTags"]}}
@@ -800,6 +837,10 @@ def set_device_tags(self, val: Iterable):
     def add_device_tag(self, val: str):
         """add a device tag for this device"""
         # pylint: disable=protected-access
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="add_device_tag"
+            )
         self.root._config["deviceTags"].add(val)
         return self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: {"deviceTags": self.root._config["deviceTags"]}}
@@ -808,6 +849,10 @@ def add_device_tag(self, val: str):
     def remove_device_tag(self, val: str):
         """remove a device tag for this device"""
         # pylint: disable=protected-access
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="remove_device_tag"
+            )
         self.root._config["deviceTags"].remove(val)
         return self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: {"deviceTags": self.root._config["deviceTags"]}}
@@ -816,6 +861,10 @@ def remove_device_tag(self, val: str):
     @property
     def wm(self) -> None:
         """get the current position of a device"""
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="wm"
+            )
         self.parent.devices.wm(self.name)
 
     @property
@@ -829,6 +878,10 @@ def readout_priority(self, val: ReadoutPriority):
         """set the readout priority for this device"""
         if not isinstance(val, ReadoutPriority):
             val = ReadoutPriority(val)
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="readout_priority"
+            )
         # pylint: disable=protected-access
         self.root._config["readoutPriority"] = val
         return self.root.parent.config_helper.send_config_request(
@@ -846,6 +899,10 @@ def on_failure(self, val: OnFailure):
         """set the failure behaviour for this device"""
         if not isinstance(val, OnFailure):
             val = OnFailure(val)
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="on_failure"
+            )
         # pylint: disable=protected-access
         self.root._config["onFailure"] = val
         return self.root.parent.config_helper.send_config_request(
@@ -862,6 +919,10 @@ def read_only(self):
     def read_only(self, value: bool):
         """Whether or not the device is read only"""
         # pylint: disable=protected-access
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="read_only"
+            )
         self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: {"readOnly": value}}
         )
@@ -877,6 +938,10 @@ def software_trigger(self):
     def software_trigger(self, value: bool):
         """Whether or not the device can be software triggered"""
         # pylint: disable=protected-access
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="software_trigger"
+            )
         self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: {"softwareTrigger": value}}
         )
@@ -891,6 +956,10 @@ def user_parameter(self) -> dict:
     @typechecked
     def set_user_parameter(self, val: dict):
         """set the user parameter for this device"""
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="set_user_parameter"
+            )
         self.root.parent.config_helper.send_config_request(
             action="update", config={self.name: {"userParameter": val}}
         )
@@ -1176,6 +1245,10 @@ def limits(self):
 
     @limits.setter
     def limits(self, val: list):
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="limits"
+            )
         self._update_config({"deviceConfig": {"limits": val}})
 
     @property
@@ -1187,6 +1260,10 @@ def low_limit(self):
 
     @low_limit.setter
     def low_limit(self, val: float):
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="low_limit"
+            )
         limits = [val, self.high_limit]
         self._update_config({"deviceConfig": {"limits": limits}})
 
@@ -1199,6 +1276,10 @@ def high_limit(self):
 
     @high_limit.setter
     def high_limit(self, val: float):
+        if self.root != self:
+            raise NotImplementedOnSubdeviceError(
+                device=self.root.name, sub_device=self.dotted_name, method="high_limit"
+            )
         limits = [self.low_limit, val]
         self._update_config({"deviceConfig": {"limits": limits}})
 

bec_lib/tests/test_devices.py

@@ -1,3 +1,4 @@
+from types import SimpleNamespace
 from typing import Any, Callable, Literal
 from unittest import mock
 
@@ -12,6 +13,7 @@
     ComputedSignal,
     Device,
     DeviceBaseWithConfig,
+    NotImplementedOnSubdeviceError,
     Positioner,
     ReadoutPriority,
     RPCError,
@@ -701,10 +703,70 @@ def test_show_all(dm_with_override):
     console.print.assert_called_once()
 
 
+@pytest.fixture()
+def positioner_as_subdevice(dm_with_override):
+    dm_with_override.connector = ConnectorMock("")
+    dev = Device(name="test_device", config=BASIC_CONFIG, parent=dm_with_override)
+    positioner = Positioner(name="positioner1", config=BASIC_CONFIG, parent=dev)
+    return positioner
+
+
+def test_limits_on_sub_device(positioner_as_subdevice):
+    with mock.patch.object(positioner_as_subdevice.root.parent.connector, "get") as mock_get:
+        mock_get.return_value = None
+        assert positioner_as_subdevice.limits == [0, 0]
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        positioner_as_subdevice.limits = [-10, 10]
+
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        positioner_as_subdevice.low_limit = -10
+
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        positioner_as_subdevice.high_limit = 10
+
+
+def test_attribute_access_on_sub_device(positioner_as_subdevice):
+    dev = positioner_as_subdevice
+
+    # Read-only
+    assert dev.read_only == False
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.read_only = True
+
+    # Enabled
+    assert dev.enabled == True
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.enabled = False
+
+    # Readout priority
+    assert dev.readout_priority == ReadoutPriority.MONITORED
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.readout_priority = ReadoutPriority.BASELINE
+
+    # Device tags
+    assert dev.get_device_tags() == set()
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.set_device_tags({"tag1", "tag2"})
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.add_device_tag("tag1")
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.remove_device_tag("tag1")
+
+    # User parameter
+    assert dev.user_parameter == {}
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.set_user_parameter({"param1": 1})
+    with pytest.raises(NotImplementedOnSubdeviceError):
+        dev.update_user_parameter({"param1": 1})
+
+
 @pytest.fixture()
 def adj():
     (adj := AdjustableMixin()).root = mock.MagicMock()
     adj._update_config = mock.MagicMock()
+    # attributes that exist on the DeviceBase
+    adj.name = "test_mixin"
+    adj.root.name = "test_mixin"
     return adj