instantiate: control which optional nodes are created

asyncua/common/instantiate_util.py

@@ -4,7 +4,7 @@
 
 import logging
 
-from typing import Union
+from typing import Union, List
 
 from asyncua import ua
 from .ua_utils import get_node_supertypes, is_child_present
@@ -19,13 +19,24 @@ async def is_abstract(node_type) -> bool:
     return result.Value.Value
 
 
-async def instantiate(parent, node_type, nodeid: ua.NodeId=None, bname: Union[str, ua.QualifiedName]=None, dname: ua.LocalizedText=None, idx: int=0, instantiate_optional: bool=True):
+async def instantiate(parent, node_type, nodeid: ua.NodeId=None, bname: Union[str, ua.QualifiedName]=None, dname: ua.LocalizedText=None, idx: int=0, instantiate_optional: bool=True,
+                      instantiate_optional_list: List[Union[str, ua.QualifiedName]]=None):
     """
     instantiate a node type under a parent node.
     nodeid and browse name of new node can be specified, or just namespace index
     If they exists children of the node type, such as components, variables and
     properties are also instantiated
+    instantiate_optional: instantiate all optional nodes
+    instantiate_optional_list: list of optinal nodes to instantiate
     """
+    instante_list = []
+    if instantiate_optional_list is not None:
+        for optional in instantiate_optional_list:
+            if isinstance(optional, ua.QualifiedName):
+                instante_list.append(optional)
+            else:
+                instante_list.append(ua.QualifiedName.from_string(optional))
+        instantiate_optional = False
     rdesc = await _rdesc_from_node(parent, node_type)
     rdesc.TypeDefinition = node_type.nodeid
     if rdesc.NodeClass in (ua.NodeClass.DataType, ua.NodeClass.ReferenceType, ua.NodeClass.ObjectType, ua.NodeClass.ReferenceType):
@@ -48,7 +59,8 @@ async def instantiate(parent, node_type, nodeid: ua.NodeId=None, bname: Union[st
         nodeid,
         bname,
         dname=dname,
-        instantiate_optional=instantiate_optional)
+        instantiate_optional=instantiate_optional,
+        instantiate_optional_list=instante_list)
     return [make_node(parent.session, nid) for nid in nodeids]
 
 
@@ -60,7 +72,8 @@ async def _instantiate_node(session,
                             bname,
                             dname=None,
                             recursive=True,
-                            instantiate_optional=True):
+                            instantiate_optional=True,
+                            instantiate_optional_list=None):
     """
     instantiate a node type under parent
     """
@@ -111,7 +124,7 @@ async def _instantiate_node(session,
                         # exclude nodes with optional ModellingRule if requested
                     if refs[0].nodeid in (ua.NodeId(ua.ObjectIds.ModellingRule_Optional), ua.NodeId(ua.ObjectIds.ModellingRule_OptionalPlaceholder)):
                         # instatiate optionals
-                        if not instantiate_optional:
+                        if not instantiate_optional and c_rdesc.BrowseName not in instantiate_optional_list:
                             _logger.info("Instantiate: Skip optional node %s as part of %s", c_rdesc.BrowseName,
                                 addnode.BrowseName)
                             continue
@@ -125,7 +138,8 @@ async def _instantiate_node(session,
                             c_rdesc,
                             nodeid=ua.NodeId(Identifier=inst_nodeid, NamespaceIndex=res.AddedNodeId.NamespaceIndex),
                             bname=c_rdesc.BrowseName,
-                            instantiate_optional=instantiate_optional
+                            instantiate_optional=instantiate_optional,
+                            instantiate_optional_list=instantiate_optional_list
                         )
                     else:
                         nodeids = await _instantiate_node(
@@ -135,7 +149,8 @@ async def _instantiate_node(session,
                             c_rdesc,
                             nodeid=ua.NodeId(NamespaceIndex=res.AddedNodeId.NamespaceIndex),
                             bname=c_rdesc.BrowseName,
-                            instantiate_optional=instantiate_optional
+                            instantiate_optional=instantiate_optional,
+                            instantiate_optional_list=instantiate_optional_list
                         )
                     added_nodes.extend(nodeids)
     return added_nodes

asyncua/common/manage_nodes.py

@@ -53,7 +53,7 @@ async def create_folder(parent, nodeid, bname):
     )
 
 
-async def create_object(parent, nodeid, bname, objecttype=None, instantiate_optional=True):
+async def create_object(parent, nodeid, bname, objecttype=None, instantiate_optional=True, instantiate_optional_list=None):
     """
     create a child node object
     arguments are nodeid, browsename, [objecttype]
@@ -64,7 +64,7 @@ async def create_object(parent, nodeid, bname, objecttype=None, instantiate_opti
     if objecttype is not None:
         objecttype = make_node(parent.session, objecttype)
         dname = ua.LocalizedText(qname.Name)
-        nodes = await instantiate(parent, objecttype, nodeid, bname=qname, dname=dname, instantiate_optional=instantiate_optional)
+        nodes = await instantiate(parent, objecttype, nodeid, bname=qname, dname=dname, instantiate_optional=instantiate_optional, instantiate_optional_list=instantiate_optional_list)
         return nodes[0]
     else:
         return make_node(

asyncua/common/node.py

@@ -685,8 +685,8 @@ async def set_modelling_rule(self, mandatory: bool):
     async def add_folder(self, nodeid, bname):
         return await create_folder(self, nodeid, bname)
 
-    async def add_object(self, nodeid, bname, objecttype=None, instantiate_optional=True):
-        return await create_object(self, nodeid, bname, objecttype, instantiate_optional)
+    async def add_object(self, nodeid, bname, objecttype=None, instantiate_optional=True, instantiate_optional_list=None):
+        return await create_object(self, nodeid, bname, objecttype, instantiate_optional, instantiate_optional_list)
 
     async def add_variable(self, nodeid, bname, val, varianttype=None, datatype=None):
         return await create_variable(self, nodeid, bname, val, varianttype, datatype)

asyncua/sync.py

@@ -138,7 +138,7 @@ def copy_node(parent, node, nodeid=None, recursive=True):
 
 
 @syncfunc(aio_func=common.instantiate_util.instantiate)
-def instantiate(parent, node_type, nodeid=None, bname=None, dname=None, idx=0, instantiate_optional=True):
+def instantiate(parent, node_type, nodeid=None, bname=None, dname=None, idx=0, instantiate_optional=True, instantiate_optional_list: List[Union[str, ua.QualifiedName]]=None):
     pass
 
 

tests/test_common.py

@@ -1029,6 +1029,7 @@ async def test_instantiate_string_nodeid(opc):
     assert "Running" == await prop.read_value()
     assert prop.nodeid != prop_t.nodeid
     await opc.opc.delete_nodes([dev_t])
+    await opc.opc.delete_nodes(nodes)
 
 
 async def test_instantiate_abstract(opc):
@@ -1037,6 +1038,33 @@ async def test_instantiate_abstract(opc):
         _ = await instantiate(opc.opc.nodes.objects, finit_statemachine_type, bname="2:TestFiniteStateMachine")
 
 
+async def test_optional_list(opc):
+    # Create device type
+    dev_t = await opc.opc.nodes.base_object_type.add_object_type(0, "MyDeviceOpt")
+    v_t = await dev_t.add_variable(0, "sensor", 1.0)
+    await v_t.set_modelling_rule(True)
+    p_t = await dev_t.add_property(0, "sensor_id", "0340")
+    await p_t.set_modelling_rule(False)
+    s_i = await dev_t.add_property(0, "sensor_info", "xx")
+    await s_i.set_modelling_rule(False)
+    ctrl_t = await dev_t.add_object(2, "controller")
+    await ctrl_t.set_modelling_rule(False)
+    prop_t = await ctrl_t.add_property(0, "state", "Running")
+    await prop_t.set_modelling_rule(False)
+    prop_t = await ctrl_t.add_property(0, "info", "ab")
+    await prop_t.set_modelling_rule(False)
+    nodes = await instantiate(opc.opc.nodes.objects, dev_t, nodeid=ua.NodeId("InstDeviceOpt", 2, ua.NodeIdType.String),
+                              bname="2:InstDeviceOpt", instantiate_optional_list=["sensor_id", ua.QualifiedName("controller", 2)])
+    mydevice = nodes[0]
+    await mydevice.get_child(["0:sensor_id"])
+    await mydevice.get_child([ua.QualifiedName("controller", 2)])
+    with pytest.raises(ua.UaError):
+        ch = await mydevice.get_child(["0:sensor_info"])
+        assert ch is None
+    await opc.opc.delete_nodes(nodes)
+    await opc.opc.delete_nodes([dev_t])
+
+
 async def test_variable_with_datatype(opc):
     v1 = await opc.opc.nodes.objects.add_variable(
         3, 'VariableEnumType1', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType)