add set_attribute_value_callback() to Server

asyncua/server/address_space.py

@@ -9,7 +9,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from functools import partial
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
     from typing import Callable, Dict, List, Union, Tuple, Generator
@@ -38,12 +38,12 @@ class AttributeValue(object):
     The class holds the value(s) of an attribute and callbacks.
     """
     def __init__(self, value: ua.DataValue):
-        self.value = value
-        self.value_callback: Union[Callable[[], ua.DataValue], None] = None
+        self.value: Optional[ua.DataValue] = value
+        self.value_callback: Optional[Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue]] = None
         self.datachange_callbacks = {}
 
     def __str__(self) -> str:
-        return f"AttributeValue({self.value})"
+        return f"AttributeValue({self.value})" if not self.value_callback else f"AttributeValue({self.value_callback})"
 
     __repr__ = __str__
 
@@ -769,9 +769,10 @@ def read_attribute_value(self, nodeid: ua.NodeId, attr: ua.AttributeIds) -> ua.D
             dv = ua.DataValue(StatusCode_=ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid))
             return dv
         attval = node.attributes[attr]
+        # TODO: async support by using asyncio.iscoroutinefunction()
         if attval.value_callback:
-            return attval.value_callback()
-        return attval.value
+            return attval.value_callback(nodeid, attr)
+        return attval.value  # type: ignore[return-value] # .value must be filled
 
     async def write_attribute_value(self, nodeid: ua.NodeId, attr: ua.AttributeIds, value: ua.DataValue) -> ua.StatusCode:
         # self.logger.debug("set attr val: %s %s %s", nodeid, attr, value)
@@ -790,6 +791,7 @@ async def write_attribute_value(self, nodeid: ua.NodeId, attr: ua.AttributeIds,
             return ua.StatusCode(ua.StatusCodes.BadTypeMismatch)
 
         attval.value = value
+        attval.value_callback = None
 
         for k, v in attval.datachange_callbacks.items():
             try:
@@ -800,6 +802,9 @@ async def write_attribute_value(self, nodeid: ua.NodeId, attr: ua.AttributeIds,
         return ua.StatusCode()
 
     def _is_expected_variant_type(self, value: ua.DataValue, attval: AttributeValue, node: NodeData) -> bool:
+        if attval.value is None:
+            return True  # None data value can be overwritten anytime.
+
         # FIXME Type hinting reveals that it is possible that Value (Optional) is None which would raise an exception
         vtype = attval.value.Value.VariantType  # type: ignore[union-attr]
         if vtype == ua.VariantType.Null:
@@ -822,6 +827,26 @@ def _is_expected_variant_type(self, value: ua.DataValue, attval: AttributeValue,
         )
         return False
 
+    def set_attribute_value_callback(
+        self,
+        nodeid: ua.NodeId,
+        attr: ua.AttributeIds,
+        callback: Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue],
+    ) -> ua.StatusCode:
+        node = self._nodes.get(nodeid, None)
+        if node is None:
+            return ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown)
+        attval = node.attributes.get(attr, None)
+        if attval is None:
+            return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
+
+        attval.value = None
+        attval.value_callback = callback
+
+        # Note: It does not trigger the datachange_callbacks unlike write_attribute_value.
+
+        return ua.StatusCode()
+
     def add_datachange_callback(self, nodeid: ua.NodeId, attr: ua.AttributeIds, callback: Callable) -> Tuple[ua.StatusCode, int]:
         # self.logger.debug("set attr callback: %s %s %s", nodeid, attr, callback)
         if nodeid not in self._nodes:

asyncua/server/internal_server.py

@@ -2,7 +2,7 @@
 Internal server implementing opcu-ua interface.
 Can be used on server side or to implement binary/https opc-ua servers
 """
-from typing import Optional
+from typing import Callable, Optional
 import asyncio
 from datetime import datetime, timedelta
 from copy import copy
@@ -67,7 +67,7 @@ def __init__(self, user_manager: UserManager = None):
             _logger.info("No user manager specified. Using default permissive manager instead.")
             user_manager = PermissiveUserManager()
         self.user_manager = user_manager
-        self.certificate_validator: Optional[CertificateValidatorMethod]= None
+        self.certificate_validator: Optional[CertificateValidatorMethod] = None
         """hook to validate a certificate, raises a ServiceError when not valid"""
         # create a session to use on server side
         self.isession = InternalSession(
@@ -341,6 +341,18 @@ async def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Va
         """
         await self.aspace.write_attribute_value(nodeid, attr, datavalue)
 
+    def set_attribute_value_callback(
+        self,
+        nodeid: ua.NodeId,
+        callback: Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue],
+        attr=ua.AttributeIds.Value
+    ) -> None:
+        """
+        Set a callback function to the Attribute that returns a value for read_attribute_value() instead of the
+        written value. Note that it does not trigger the datachange_callbacks unlike write_attribute_value().
+        """
+        self.aspace.set_attribute_value_callback(nodeid, attr, callback)
+
     def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
         """
         directly read datavalue of the Attribute

asyncua/server/server.py

@@ -9,7 +9,7 @@
 from datetime import timedelta, datetime
 import socket
 from urllib.parse import urlparse
-from typing import Optional, Tuple, Union
+from typing import Callable, Optional, Tuple, Union
 from pathlib import Path
 
 from asyncua import ua
@@ -836,6 +836,18 @@ async def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Va
         """
         return await self.iserver.write_attribute_value(nodeid, datavalue, attr)
 
+    def set_attribute_value_callback(
+        self,
+        nodeid: ua.NodeId,
+        callback: Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue],
+        attr=ua.AttributeIds.Value
+    ) -> None:
+        """
+        Set a callback function to the Attribute that returns a value for read_attribute_value() instead of the
+        written value. Note that it does not trigger the datachange_callbacks unlike write_attribute_value().
+        """
+        self.iserver.set_attribute_value_callback(nodeid, callback, attr)
+
     def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
         """
         directly read datavalue of the Attribute

asyncua/sync.py

@@ -11,8 +11,7 @@
 from pathlib import Path
 from threading import Thread, Condition
 import logging
-from typing import Any, Dict, Iterable, List, Sequence, Set, Tuple, Type, Union, Optional, overload
-
+from typing import Any, Callable, Dict, Iterable, List, Sequence, Set, Tuple, Type, Union, Optional, overload
 
 if sys.version_info >= (3, 8):
     from typing import Literal
@@ -635,6 +634,14 @@ def load_data_type_definitions(self, node=None):
     def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
         pass
 
+    def set_attribute_value_callback(
+        self,
+        nodeid: ua.NodeId,
+        callback: Callable[[ua.NodeId, ua.AttributeIds], ua.DataValue],
+        attr=ua.AttributeIds.Value
+    ) -> None:
+        self.aio_obj.set_attribute_value_callback(nodeid, callback, attr)
+
     def create_subscription(self, period, handler):
         coro = self.aio_obj.create_subscription(period, _SubHandler(self.tloop, handler))
         aio_sub = self.tloop.post(coro)

tests/test_server.py

@@ -664,6 +664,26 @@ async def test_server_read_write_attribute_value(server: Server):
     await server.delete_nodes([node])
 
 
+async def test_server_read_set_attribute_value_callback(server: Server):
+    node = await server.get_objects_node().add_variable(0, "0:TestVar", 0, varianttype=ua.VariantType.Int64)
+    dv = server.read_attribute_value(node.nodeid, attr=ua.AttributeIds.Value)
+    assert dv.Value.Value == 0
+
+    def callback(nodeid, attr):
+        return ua.DataValue(Value=ua.Variant(Value=5, VariantType=ua.VariantType.Int64))
+
+    server.set_attribute_value_callback(node.nodeid, callback, attr=ua.AttributeIds.Value)
+    dv = server.read_attribute_value(node.nodeid, attr=ua.AttributeIds.Value)
+    assert dv.Value.Value == 5
+
+    dv = ua.DataValue(Value=ua.Variant(Value=10, VariantType=ua.VariantType.Int64))
+    await server.write_attribute_value(node.nodeid, dv, attr=ua.AttributeIds.Value)
+    dv = server.read_attribute_value(node.nodeid, attr=ua.AttributeIds.Value)
+    assert dv.Value.Value == 10
+
+    await server.delete_nodes([node])
+
+
 @pytest.fixture(scope="function")
 def restore_transport_limits_server(server: Server):
     # Restore limits after test