Server support for read array with index range

asyncua/common/ua_utils.py

@@ -2,6 +2,7 @@
 Usefull method and classes not belonging anywhere and depending on asyncua library
 """
 
+from typing import Tuple
 import uuid
 import logging
 from datetime import datetime
@@ -14,6 +15,26 @@
 logger = logging.getLogger('__name__')
 
 
+def ua_index_range_to_list_slice(ua_index_range:str) -> Tuple[int,int]:
+    """Converts an OPC UA NumericRange into a python tuple for list slicing
+    see: https://reference.opcfoundation.org/v104/Core/docs/Part4/7.22/
+    Args:
+        ua_index_range (str): The OPC UA NumericRange
+
+    Returns:
+        Tuple[int,int]: Equivalent python list slicing tuple with low and high bounds
+    """
+    index_range_list = ua_index_range.split(':')
+    low_idx = int(index_range_list[0])
+    if len(index_range_list) == 1:
+        high_idx = int(index_range_list[0])+1
+    elif len(index_range_list) == 2:
+        high_idx = int(index_range_list[1])+1 # python List[n:n] -> empty list, List[n:n+k] -> list of k elements; OPC UA see https://reference.opcfoundation.org/v104/Core/docs/Part4/7.22/
+    else:
+        raise Exception(f'Read with bad index range: {ua_index_range}')
+    return (low_idx,high_idx)
+
+
 def value_to_datavalue(val, varianttype=None):
     """
     convert anyting to a DataValue using varianttype

asyncua/server/address_space.py

@@ -1,4 +1,7 @@
 import asyncio
+import typing
+from typing import Tuple
+from asyncua.ua.uatypes import DataValue, NodeId, StatusCode, Variant
 import pickle
 import shelve
 import logging
@@ -47,7 +50,7 @@ def read(self, params):
         # self.logger.debug("read %s", params)
         res = []
         for readvalue in params.NodesToRead:
-            res.append(self._aspace.read_attribute_value(readvalue.NodeId, readvalue.AttributeId))
+            res.append(self._aspace.read_attribute_value(readvalue.NodeId, readvalue.AttributeId, readvalue.IndexRange))
         return res
 
     async def write(self, params, user=User(role=UserRole.Admin)):
@@ -686,7 +689,18 @@ def __len__(self):
 
         self._nodes = LazyLoadingDict(shelve.open(path, "r"))
 
-    def read_attribute_value(self, nodeid, attr):
+    def read_attribute_value(self, nodeid:NodeId, attr:ua.AttributeIds, index_range: typing.Optional[Tuple[int,int]] = None) -> DataValue:
+        """Reads the Value of the requested Attribute of the given Node ID.
+
+        Args:
+            nodeid (NodeId): The NodeID from which the Attribute's value is to be read
+            attr (ua.AttributeIds): The attribute that is to be read
+            index_range (typing.Optional[Tuple[int,int]], optional): In case the Attribute's value is an array,
+                the range of the array that is actual to be returned. Defaults to None.
+
+        Returns:
+            DataValue: Value of the requested Attribute on the specified NodeID
+        """
         # self.logger.debug("get attr val: %s %s", nodeid, attr)
         if nodeid not in self._nodes:
             dv = ua.DataValue(StatusCode_=ua.StatusCode(ua.StatusCodes.BadNodeIdUnknown))
@@ -696,11 +710,35 @@ def read_attribute_value(self, nodeid, attr):
             dv = ua.DataValue(StatusCode_=ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid))
             return dv
         attval = node.attributes[attr]
+        value = ua.DataValue()
         if attval.value_callback:
-            return attval.value_callback()
-        return attval.value
-
-    async def write_attribute_value(self, nodeid, attr, value):
+            value = attval.value_callback()
+        else:
+            value = attval.value
+        if not index_range is None:
+            if not attval.value.Value.is_array:
+                return ua.StatusCode(ua.StatusCodes.BadWriteNotSupported)
+            try:
+                value = DataValue(Variant(value.Value.Value[index_range[0]:index_range[1]]))
+            except Exception as ex:
+                self.logger.warning(f'Read index range failed. IndexRange = {index_range}; Ex: {ex}')
+                dv = ua.DataValue(StatusCode_=ua.StatusCode(ua.StatusCodes.BadIndexRangeInvalid))
+                return dv
+        return value
+
+    async def write_attribute_value(self, nodeid:NodeId, attr:ua.AttributeIds, value:DataValue, index_range: typing.Optional[Tuple[int,int]] = None) -> StatusCode:
+        """Writes the Value of the requested Attribute of the given Node ID
+
+        Args:
+            nodeid (NodeId): The NodeID from which the Attribute's value is to be written
+            attr (ua.AttributeIds): The attribute that is to be written
+            value (DataValue): Value to write
+            index_range (typing.Optional[Tuple[int,int]], optional): In case the Attribute's value is an array,
+                the range of the array that is actual to be written. Defaults to None.
+
+        Returns:
+            StatusCode: StatusCode of the write operation
+        """
         # self.logger.debug("set attr val: %s %s %s", nodeid, attr, value)
         node = self._nodes.get(nodeid, None)
         if node is None:
@@ -713,7 +751,12 @@ async def write_attribute_value(self, nodeid, attr, value):
             return ua.StatusCode(ua.StatusCodes.BadTypeMismatch)
 
         old = attval.value
-        attval.value = value
+        if index_range is None:
+            attval.value = value
+        else:
+            if not attval.value.Value.is_array:
+                return ua.StatusCode(ua.StatusCodes.BadWriteNotSupported)
+            attval.value.Value.Value[index_range[0]:index_range[1]] = value.Value.Value
         cbs = []
         # only send call callback when a value or status code change has happened
         if (old.Value != value.Value) or (old.StatusCode != value.StatusCode):

asyncua/server/internal_server.py

@@ -4,13 +4,15 @@
 """
 
 import asyncio
+from asyncua.ua.uatypes import DataValue, NodeId, StatusCode
+from asyncua.common.ua_utils import ua_index_range_to_list_slice
 from datetime import datetime, timedelta
 from copy import copy
 from struct import unpack_from
 import os
 import logging
 from urllib.parse import urlparse
-from typing import Coroutine
+from typing import Coroutine, Optional, Tuple
 
 from asyncua import ua
 from .user_managers import PermissiveUserManager, UserManager
@@ -292,18 +294,34 @@ def unsubscribe_server_callback(self, event, handle):
         """
         self.callback_service.removeListener(event, handle)
 
-    async def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
+    async def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value, index_range:Optional[str]=None)->StatusCode:
         """
         directly write datavalue to the Attribute, bypassing some checks and structure creation
         so it is a little faster
         """
-        await self.aspace.write_attribute_value(nodeid, attr, datavalue)
-
-    def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
+        if not index_range is None:
+            try:
+                py_list_index_range = ua_index_range_to_list_slice(index_range)
+            except Exception as ex:
+                self.logger.warning(f'Parse index range failed. IndexRange = {index_range}; Ex: {ex}')
+                return ua.StatusCode(ua.StatusCodes.BadIndexRangeInvalid)
+        else:
+            py_list_index_range = None
+        return await self.aspace.write_attribute_value(nodeid, attr, datavalue, py_list_index_range)
+
+    def read_attribute_value(self, nodeid:NodeId, attr=ua.AttributeIds.Value, index_range:Optional[str]=None)->DataValue:
         """
         directly read datavalue of the Attribute
         """
-        return self.aspace.read_attribute_value(nodeid, attr)  
+        if not index_range is None:
+            try:
+                py_list_index_range = ua_index_range_to_list_slice(index_range)
+            except Exception as ex:
+                self.logger.warning(f'Parse index range failed. IndexRange = {index_range}; Ex: {ex}')
+                return ua.DataValue(StatusCode_=ua.StatusCode(ua.StatusCodes.BadIndexRangeInvalid))
+        else:
+            py_list_index_range = None
+        return self.aspace.read_attribute_value(nodeid, attr, py_list_index_range)  
 
     def set_user_manager(self, user_manager):
         """

asyncua/server/internal_session.py

@@ -1,6 +1,9 @@
+from asyncua.ua.uaprotocol_auto import ReadParameters, WriteParameters, WriteValue
+from asyncua.ua.uatypes import DataValue, NodeId, StatusCode
+from asyncua.ua.attribute_ids import AttributeIds
 import logging
 from enum import Enum
-from typing import Coroutine, Iterable, Optional
+from typing import Coroutine, Iterable, List, Optional
 
 from asyncua import ua
 from ..common.callback import CallbackType, ServerItemCallback
@@ -102,29 +105,78 @@ def activate_session(self, params, peer_certificate):
         self.logger.info("Activated internal session %s for user %s", self.name, self.user)
         return result
 
-    async def read(self, params):
+    async def read(self, params:ReadParameters)->List[DataValue]:
+        """Reads a set of nodes to read
+
+        Args:
+            params (ReadParameters): Parameters with nodes to be read
+
+        Returns:
+            List[DataValue]: List of values read
+        """
         if self.user is None:
             user = User()
         else:
             user = self.user
         await self.iserver.callback_service.dispatch(CallbackType.PreRead,
                                                      ServerItemCallback(params, None, user))
-        results = self.iserver.attribute_service.read(params)
+        results = [
+                    self.iserver.read_attribute_value(node_to_read.NodeId, node_to_read.AttributeId, node_to_read.IndexRange)
+                    for node_to_read in params.NodesToRead
+                ]
         await self.iserver.callback_service.dispatch(CallbackType.PostRead,
                                                      ServerItemCallback(params, results, user))
         return results
 
     async def history_read(self, params) -> Coroutine:
         return await self.iserver.history_manager.read_history(params)
 
-    async def write(self, params):
+    def check_user_access_to_node_to_write(self, user:User, node_to_write:WriteValue)->bool:
+        """Checks if the given user has the access permissions for the WriteValue
+
+        Args:
+            user (User): The user making the access request
+            node_to_write (WriteValue): Value to write
+
+        Returns:
+            bool: True when the user has the permissions
+        """
+        if user.role != UserRole.Admin:
+            if node_to_write.AttributeId != ua.AttributeIds.Value:
+                return False
+            al = self.iserver.read_attribute_value(node_to_write.NodeId, ua.AttributeIds.AccessLevel)
+            ual = self.iserver.read_attribute_value(node_to_write.NodeId, ua.AttributeIds.UserAccessLevel)
+            if (
+                not al.StatusCode.is_good()
+                or not ua.ua_binary.test_bit(al.Value.Value, ua.AccessLevel.CurrentWrite)
+                or not ua.ua_binary.test_bit(ual.Value.Value, ua.AccessLevel.CurrentWrite)
+            ):
+                return False
+        return True
+
+    async def write(self, params:WriteParameters)->List[StatusCode]:
+        """Writes a set of Nodes to write
+
+        Args:
+            params (WriteParameters): Parameters with nodes to be written
+
+        Returns:
+            List[StatusCode]: Status codes of the write operation of each of the nodes
+        """
         if self.user is None:
             user = User()
         else:
             user = self.user
         await self.iserver.callback_service.dispatch(CallbackType.PreWrite,
                                                      ServerItemCallback(params, None, user))
-        write_result = await self.iserver.attribute_service.write(params, user=user)
+        write_result = []
+        for node_to_write in params.NodesToWrite:
+            if not self.check_user_access_to_node_to_write(user, node_to_write):
+                write_result.append(ua.StatusCode(ua.StatusCodes.BadUserAccessDenied))
+            else:
+                write_result.append(
+                    await self.iserver.write_attribute_value(node_to_write.NodeId, node_to_write.Value, node_to_write.AttributeId, node_to_write.IndexRange)
+                )
         await self.iserver.callback_service.dispatch(CallbackType.PostWrite,
                                                      ServerItemCallback(params, write_result, user))
         return write_result

asyncua/server/server.py

@@ -3,6 +3,7 @@
 """
 
 import asyncio
+from asyncua.ua.uatypes import DataValue, StatusCode
 import logging
 from datetime import timedelta, datetime
 from urllib.parse import urlparse
@@ -663,15 +664,15 @@ async def load_enums(self) -> Coroutine:
         _logger.warning("Deprecated since spec 1.04, call load_data_type_definitions")
         return await load_enums(self)
 
-    async def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value):
+    async def write_attribute_value(self, nodeid, datavalue, attr=ua.AttributeIds.Value, index_range:Optional[str]=None)->StatusCode:
         """
         directly write datavalue to the Attribute, bypasing some checks and structure creation
         so it is a little faster
         """
-        return await self.iserver.write_attribute_value(nodeid, datavalue, attr)
+        return await self.iserver.write_attribute_value(nodeid, datavalue, attr, index_range)
 
-    def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value):
+    def read_attribute_value(self, nodeid, attr=ua.AttributeIds.Value, index_range:Optional[str]=None)->DataValue:
         """
         directly read datavalue of the Attribute
         """
-        return self.iserver.read_attribute_value(nodeid, attr)
+        return self.iserver.read_attribute_value(nodeid, attr, index_range)

tests/test_client.py

@@ -1,4 +1,5 @@
 
+from asyncua.ua.uatypes import DataValue, Variant
 import logging
 import pytest
 
@@ -123,5 +124,77 @@ async def test_browse_nodes(server, client):
     assert isinstance(results[1][0], Node)
     assert isinstance(results[0][1], ua.BrowseResult)
     assert isinstance(results[1][1], ua.BrowseResult)
-
-
+
+
+async def test_read_attribute_value_without_index_range_reads_full_array(server, admin_client, client):
+    objects = admin_client.nodes.objects
+    array_data = [i for i in range(100)]
+    v = await objects.add_variable(3, 'MyArray', array_data)
+    await v.write_value(array_data)
+    v_client = client.get_node(v.nodeid)
+
+    # calling read_attribute without giving an index range must read the full array
+    array_data_read = (await v_client.read_attribute(ua.AttributeIds.Value)).Value.Value
+    assert len(set(array_data).intersection(array_data_read)) == len(array_data)
+
+
+async def test_read_attribute_value_with_index_range_reads_subarray(server, admin_client, client):
+    objects = admin_client.nodes.objects
+    array_data = [i for i in range(100)]
+    v = await objects.add_variable(3, 'MyArray', array_data)
+    await v.write_value(array_data)
+    v_client = client.get_node(v.nodeid)
+
+    # calling read_attribute giving an index range must read the sub array,
+    # in this case 10th element until 19th element, meaning 10 elements
+    array_data_read = (await v_client.read_attribute(ua.AttributeIds.Value, '10:19')).Value.Value
+    assert len(set(array_data).intersection(array_data_read)) == 10
+
+async def test_read_attribute_value_with_single_value_in_index_range_reads_subarray(server, admin_client, client):
+    objects = admin_client.nodes.objects
+    array_data = [i for i in range(100)]
+    v = await objects.add_variable(3, 'MyArray', array_data)
+    await v.write_value(array_data)
+    v_client = client.get_node(v.nodeid)
+
+    # calling read_attribute giving an index range must read the sub array,
+    # in this case 10th element until 19th element, meaning 10 elements
+    array_data_read = (await v_client.read_attribute(ua.AttributeIds.Value, '30')).Value.Value
+    assert len(array_data_read) == 1
+    assert array_data[30] == array_data_read[0]
+
+
+async def test_write_attribute_value_without_index_range_writes_full_array(server, admin_client, client):
+    objects = admin_client.nodes.objects
+    array_data = [i for i in range(100)]
+    v = await objects.add_variable(3, 'MyArray', array_data)
+    await v.set_writable()
+    await v.write_value(array_data)
+    v_ro = client.get_node(v.nodeid)
+
+    # calling write_attribute without giving an index range must read the full array
+    array_data_write = [i+100 for i in range(100)]
+    await v_ro.write_attribute(ua.AttributeIds.Value, DataValue(Variant(array_data_write)))
+    array_data_read = (await v_ro.read_attribute(ua.AttributeIds.Value)).Value.Value
+    assert len(set(array_data_write).intersection(array_data_read)) == len(array_data_write)
+
+
+async def test_write_attribute_value_with_index_range_writes_sub_array(server, admin_client, client):
+    objects = admin_client.nodes.objects
+    array_data = [i for i in range(100)]
+    v = await objects.add_variable(3, 'MyArray', array_data)
+    await v.set_writable()
+    await v.write_value(array_data)
+    v_ro = client.get_node(v.nodeid)
+
+    # calling write_attribute with an index range and an array os same length must write the sub array in 
+    # the position specified by the index range
+    array_data_write = [i+100 for i in range(9)]
+    await v_ro.write_attribute(ua.AttributeIds.Value, DataValue(Variant(array_data_write)), '10:20')
+    array_data_read = (await v_ro.read_attribute(ua.AttributeIds.Value)).Value.Value
+    # assert the sub array was written
+    assert len(set(array_data_write).intersection(array_data_read)) == len(array_data_write)
+
+    # it was written in the correct location
+    sub_array = array_data_read[10:21]
+    assert len(set(array_data_write).intersection(sub_array)) == len(array_data_write)