add tests for data_type_definition stuff...and fix all bugs

asyncua/common/shortcuts.py

@@ -30,3 +30,4 @@ class Shortcuts(object):
         self.HasComponent = Node(server, ObjectIds.HasComponent)
         self.HasProperty = Node(server, ObjectIds.HasProperty)
         self.Organizes = Node(server, ObjectIds.Organizes)
+        self.HasEncoding = Node(server, ObjectIds.HasEncoding)

asyncua/common/structures104.py

@@ -48,7 +48,7 @@ def get_default_value(uatype, enums=None):
         return f"ua.{uatype}()"
 
 
-def make_structure_code(name, sdef):
+def make_structure_code(data_type, name, sdef):
     """
     given a StructureDefinition object, generate Python code
     """
@@ -64,6 +64,8 @@ class {name}:
     {name} structure autogenerated from StructureDefinition object
     '''
 
+    data_type = ua.NodeId({data_type.Identifier}, {data_type.NamespaceIndex})
+
 """
     counter = 0
     if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
@@ -76,6 +78,8 @@ class {name}:
         code += "    }\n\n"
 
     code += '    ua_types = [\n'
+    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
+        code += f"        ('Encoding', 'Byte'),\n"
     uatypes = []
     for field in sdef.Fields:
         prefix = 'ListOf' if field.ValueRank >= 1 else ''
@@ -105,6 +109,8 @@ class {name}:
 """
     if not sdef.Fields:
         code += "      pass"
+    if sdef.StructureType == ua.StructureType.StructureWithOptionalFields:
+        code += f"        self.Encoding = 0\n"
     for field, uatype in uatypes:
         if field.ValueRank >= 1:
             default_value = "[]"
@@ -115,7 +121,7 @@ class {name}:
     return code
 
 
-async def _generate_object(name, sdef, env=None, enum=False):
+async def _generate_object(name, sdef, data_type=None, env=None, enum=False):
     """
     generate Python code and execute in a new environment
     return a dict of structures {name: class}
@@ -137,7 +143,7 @@ async def _generate_object(name, sdef, env=None, enum=False):
     if enum:
         code = make_enum_code(name, sdef)
     else:
-        code = make_structure_code(name, sdef)
+        code = make_structure_code(data_type, name, sdef)
     logger.debug("Executing code: %s", code)
     print("CODE", code)
     exec(code, env)
@@ -145,7 +151,8 @@ async def _generate_object(name, sdef, env=None, enum=False):
 
 
 class DataTypeSorter:
-    def __init__(self, name, desc, sdef):
+    def __init__(self, data_type, name, desc, sdef):
+        self.data_type = data_type
         self.name = name
         self.desc = desc
         self.sdef = sdef
@@ -169,7 +176,7 @@ async def _recursive_parse(server, base_node, dtypes):
         if not sdef:
             continue
         name = clean_name(desc.BrowseName.Name)
-        dtypes.append(DataTypeSorter(name, desc, sdef))
+        dtypes.append(DataTypeSorter(desc.NodeId, name, desc, sdef))
         await _recursive_parse(server, server.get_node(desc.NodeId), dtypes)
 
 
@@ -182,7 +189,7 @@ async def load_data_type_definitions(server, base_node=None):
     dtypes.sort()
     for dts in dtypes:
         try:
-            env = await _generate_object(dts.name, dts.sdef)
+            env = await _generate_object(dts.name, dts.sdef, data_type=dts.data_type)
             ua.register_extension_object(dts.name, dts.encoding_id, env[dts.name], dts.desc.NodeId)
         except NotImplementedError:
             logger.exception("Structure type %s not implemented", dts.sdef)

asyncua/common/xmlimporter.py

@@ -161,7 +161,7 @@ class XmlImporter:
             nodeid.NamespaceIndex = self.namespaces[nodeid.NamespaceIndex]
         return nodeid
 
-    def _get_node(self, obj):
+    def _get_add_node_item(self, obj):
         node = ua.AddNodesItem()
         node.RequestedNewNodeId = self._migrate_ns(obj.nodeid)
         node.BrowseName = self._migrate_ns(obj.browsename)
@@ -195,7 +195,7 @@ class XmlImporter:
                 return ua.NodeId(getattr(ua.ObjectIds, nodeid))
 
     async def add_object(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.ObjectAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -208,7 +208,7 @@ class XmlImporter:
         return res[0].AddedNodeId
 
     async def add_object_type(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.ObjectTypeAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -221,7 +221,7 @@ class XmlImporter:
         return res[0].AddedNodeId
 
     async def add_variable(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.VariableAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -348,7 +348,7 @@ class XmlImporter:
             return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
 
     async def add_variable_type(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.VariableTypeAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -369,7 +369,7 @@ class XmlImporter:
         return res[0].AddedNodeId
 
     async def add_method(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.MethodAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -389,7 +389,7 @@ class XmlImporter:
         return res[0].AddedNodeId
 
     async def add_reference_type(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.ReferenceTypeAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -407,7 +407,7 @@ class XmlImporter:
         return res[0].AddedNodeId
 
     async def add_datatype(self, obj):
-        node = self._get_node(obj)
+        node = self._get_add_node_item(obj)
         attrs = ua.DataTypeAttributes()
         if obj.desc:
             attrs.Description = ua.LocalizedText(obj.desc)
@@ -472,7 +472,7 @@ class XmlImporter:
         if obj.parent:
             sdef.BaseDataType = obj.parent
         for data in obj.refs:
-            if data.reftype == "HasEncoding":
+            if data.reftype == self.server.nodes.HasEncoding.nodeid:
                 # looks likebinary encodingisthe firt one...can someone confirm?
                 sdef.DefaultEncodingId = data.target
                 break

asyncua/ua/uaprotocol_hand.py

@@ -334,7 +334,12 @@ class ReferenceTypeAttributes(auto.ReferenceTypeAttributes):
         self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.IsAbstract | ana.Symmetric | ana.InverseName
 
 
+# FIXME: changes in that class donnot seem to be part of spec as of 1.04
+#not sure what the spec expect, maybe DataTypeDefinition must be set using an extra call...
+# maybe it will be part of spec in 1.05??? no ideas
 class DataTypeAttributes(auto.DataTypeAttributes):
+    auto.DataTypeAttributes.ua_types.append(('DataTypeDefinition', 'ExtensionObject'))
+
     def __init__(self):
         auto.DataTypeAttributes.__init__(self)
         self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.IsAbstract | ana.DataTypeDefinition

asyncua/ua/uatypes.py

@@ -732,7 +732,7 @@ class Variant(FrozenClass):
             self.VariantType = value.VariantType
         if self.VariantType is None:
             self.VariantType = self._guess_type(self.Value)
-        if self.Value is None and not self.is_array and self.VariantType not in (VariantType.Null, VariantType.String, VariantType.DateTime):
+        if self.Value is None and not self.is_array and self.VariantType not in (VariantType.Null, VariantType.String, VariantType.DateTime, VariantType.ExtensionObject):
             raise UaError(f"Non array Variant of type {self.VariantType} cannot have value None")
         if self.Dimensions is None and isinstance(self.Value, (list, tuple)):
             dims = get_shape(self.Value)
@@ -972,7 +972,7 @@ def register_enum(name, nodeid, class_type):
 # decoding and encoding
 extension_objects_by_datatype = {}  #Dict[Datatype, type]
 extension_objects_by_typeid = {}  #Dict[EncodingId, type]
-extension_object_typeids  = {}
+extension_object_typeids = {}
 
 
 def register_extension_object(name, encoding_nodeid, class_type, datatype_nodeid=None):

schemas/generate_protocol_python.py

@@ -116,6 +116,10 @@ class CodeGenerator:
         self.write('"""')
 
         self.write('')
+        #FIXME: next line is a weak way to find out if object is a datatype or not...
+        if not "Parameter" in obj.name and not "Result" in obj.name:
+            self.write(f'data_type = ObjectIds.{obj.name}')
+            self.write('')
         switch_written = False
         for field in obj.fields:
             if field.switchfield is not None:

tests/test_common.py

@@ -987,3 +987,23 @@ async def test_guid_node_id():
 async def test_import_xml_data_type_definition(opc):
     nodes = await opc.opc.import_xml("tests/substructs.xml")
     await opc.opc.load_data_type_definitions()
+    assert hasattr(ua, "MySubstruct")
+    assert hasattr(ua, "MyStruct")
+
+    datatype = opc.opc.get_node(ua.MySubstruct.data_type)
+    sdef = await datatype.read_data_type_definition()
+    assert isinstance(sdef, ua.StructureDefinition)
+    s = ua.MyStruct()
+    s.toto = 0.1
+    ss = ua.MySubstruct()
+    assert ss.titi == 0
+    assert isinstance(ss.structs, list)
+
+    ss.titi = 1
+    ss.structs.append(s)
+    ss.structs.append(s)
+
+    var = await opc.opc.nodes.objects.add_variable(2, "MySubStructVar", ss, datatype=ua.MySubstruct.data_type)
+
+    s2 = await var.read_value()
+    assert s2.structs[1].toto == ss.structs[1].toto == 0.1

tests/test_unit.py

@@ -702,3 +702,18 @@ def test_variant_intenum():
     ase = ua.AxisScaleEnumeration(ua.AxisScaleEnumeration.Linear)  # Just pick an existing IntEnum class
     vAse = ua.Variant(ase)
     assert vAse.VariantType == ua.VariantType.Int32
+
+
+def test_bin_data_type_def():
+    ad = ua.AddNodesItem()
+    ad.ParentNodeId = ua.NodeId(22)
+    dta = ua.DataTypeAttributes()
+    dta.DisplayName = ua.LocalizedText("titi")
+    ad.NodeAttributes = dta
+
+    data = struct_to_binary(ad)
+    ad2 = struct_from_binary(ua.AddNodesItem, ua.utils.Buffer(data))
+    assert ad.ParentNodeId == ad2.ParentNodeId
+    from IPython import embed
+    #embed()
+    assert ad.NodeAttributes.DisplayName == ad2.NodeAttributes.DisplayName