diff --git a/examples/server-create-custom-structures.py b/examples/server-create-custom-structures.py new file mode 100644 index 000000000..57dc629d3 --- /dev/null +++ b/examples/server-create-custom-structures.py @@ -0,0 +1,94 @@ +from opcua import ua, Server +from opcua.common.type_dictionary_buider import DataTypeDictionaryBuilder, get_ua_class +from IPython import embed + + +class DemoServer: + + def __init__(self): + self.server = Server() + + self.server.set_endpoint('opc.tcp://0.0.0.0:51210/UA/SampleServer') + self.server.set_server_name('Custom structure demo server') + # idx name will be used later for creating the xml used in data type dictionary + self._idx_name = 'http://examples.freeopcua.github.io' + self.idx = self.server.register_namespace(self._idx_name) + + self.dict_builder = DataTypeDictionaryBuilder(self.server, self.idx, self._idx_name, 'MyDictionary') + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + quit() + + def start_server(self): + self.server.start() + + def create_structure(self, name): + # save the created data type + return self.dict_builder.create_data_type(name) + + def complete_creation(self): + self.dict_builder.set_dict_byte_string() + + +if __name__ == '__main__': + + with DemoServer() as ua_server: + # add one basic structure + basic_struct_name = 'basic_structure' + basic_struct = ua_server.create_structure(basic_struct_name) + basic_struct.add_field('ID', ua.VariantType.Int32) + basic_struct.add_field('Gender', ua.VariantType.Boolean) + basic_struct.add_field('Comments', ua.VariantType.String) + + # add an advance structure which uses our basic structure + nested_struct_name = 'nested_structure' + nested_struct = ua_server.create_structure(nested_struct_name) + nested_struct.add_field('Name', ua.VariantType.String) + nested_struct.add_field('Surname', ua.VariantType.String) + # add simple structure as field + nested_struct.add_field('Stuff', basic_struct) + + # this operation will write the OPC dict string to our new data type dictionary + # namely the 'MyDictionary' + ua_server.complete_creation() + + # get the working classes + ua_server.server.load_type_definitions() + + # Create one test structure + basic_var = ua_server.server.nodes.objects.add_variable(ua.NodeId(namespaceidx=ua_server.idx), 'BasicStruct', + ua.Variant(None, ua.VariantType.Null), + datatype=basic_struct.data_type) + + basic_var.set_writable() + basic_msg = get_ua_class(basic_struct_name)() + basic_msg.ID = 3 + basic_msg.Gender = True + basic_msg.Comments = 'Test string' + basic_var.set_value(basic_msg) + + # Create one advance test structure + nested_var = ua_server.server.nodes.objects.add_variable(ua.NodeId(namespaceidx=ua_server.idx), 'NestedStruct', + ua.Variant(None, ua.VariantType.Null), + datatype=nested_struct.data_type) + + nested_var.set_writable() + nested_msg = get_ua_class(nested_struct_name)() + nested_msg.Stuff = basic_msg + nested_msg.Name = 'Max' + nested_msg.Surname = 'Karl' + nested_var.set_value(nested_msg) + + ua_server.start_server() + + # see the xml value in our customized dictionary 'MyDictionary', only for debugging use + print(getattr(ua_server.dict_builder, '_type_dictionary').get_dict_value()) + + # values can be write back and retrieved with the codes below. + basic_result = basic_var.get_value() + nested_result = nested_var.get_value() + + embed() diff --git a/opcua/common/type_dictionary_buider.py b/opcua/common/type_dictionary_buider.py new file mode 100644 index 000000000..7305c28c1 --- /dev/null +++ b/opcua/common/type_dictionary_buider.py @@ -0,0 +1,265 @@ +from opcua import ua +from enum import Enum + +import xml.etree.ElementTree as Et +import re + +# Indicates which type should be OPC build in types +_ua_build_in_types = [ua_type for ua_type in ua.VariantType.__members__ if ua_type != 'ExtensionObject'] + + +def _repl_func(m): + """ + taken from + https://stackoverflow.com/questions/1549641/how-to-capitalize-the-first-letter-of-each-word-in-a-string-python + """ + return m.group(1) + m.group(2).upper() + + +def _to_camel_case(name): + """ + Create python class name from an arbitrary string to CamelCase string + e.g. actionlib/TestAction -> ActionlibTestAction + turtle_actionlib/ShapeActionFeedback -> TurtleActionlibShapeActionFeedback + """ + name = re.sub(r'[^a-zA-Z0-9]+', ' ', name) + name = re.sub('(^|\s)(\S)', _repl_func, name) + name = name.replace(' ', '') + return name + + +class OPCTypeDictionaryBuilder: + + def __init__(self, idx_name): + """ + :param idx_name: name of the name space + types in dict is created as opc:xxx, otherwise as tns:xxx + """ + head_attributes = {'xmlns:xsi': 'http://www.w3.org/2001/XMLSchema-instance', 'xmlns:tns': idx_name, + 'DefaultByteOrder': 'LittleEndian', 'xmlns:opc': 'http://opcfoundation.org/BinarySchema/', + 'xmlns:ua': 'http://opcfoundation.org/UA/', 'TargetNamespace': idx_name} + + self.etree = Et.ElementTree(Et.Element('opc:TypeDictionary', head_attributes)) + + name_space = Et.SubElement(self.etree.getroot(), 'opc:Import') + name_space.attrib['Namespace'] = 'http://opcfoundation.org/UA/' + + self._structs_dict = {} + self._build_in_list = _ua_build_in_types + + def _process_type(self, data_type): + if data_type in self._build_in_list: + data_type = 'opc:' + data_type + else: + data_type = 'tns:' + _to_camel_case(data_type) + return data_type + + def _add_field(self, variable_name, data_type, struct_name): + data_type = self._process_type(data_type) + field = Et.SubElement(self._structs_dict[struct_name], 'opc:Field') + field.attrib['Name'] = variable_name + field.attrib['TypeName'] = data_type + + def _add_array_field(self, variable_name, data_type, struct_name): + data_type = self._process_type(data_type) + array_len = 'NoOf' + variable_name + field = Et.SubElement(self._structs_dict[struct_name], 'opc:Field') + field.attrib['Name'] = array_len + field.attrib['TypeName'] = 'opc:Int32' + field = Et.SubElement(self._structs_dict[struct_name], 'opc:Field') + field.attrib['Name'] = variable_name + field.attrib['TypeName'] = data_type + field.attrib['LengthField'] = array_len + + def add_field(self, variable_name, data_type, struct_name, is_array=False): + if isinstance(data_type, Enum): + data_type = data_type.name + if is_array: + self._add_array_field(variable_name, data_type, struct_name) + else: + self._add_field(variable_name, data_type, struct_name) + + def append_struct(self, name): + appended_struct = Et.SubElement(self.etree.getroot(), 'opc:StructuredType') + appended_struct.attrib['BaseType'] = 'ua:ExtensionObject' + appended_struct.attrib['Name'] = _to_camel_case(name) + self._structs_dict[name] = appended_struct + return appended_struct + + def get_dict_value(self): + self.indent(self.etree.getroot()) + # For debugging + # Et.dump(self.etree.getroot()) + return Et.tostring(self.etree.getroot(), encoding='utf-8') + + def indent(self, elem, level=0): + i = '\n' + level * ' ' + if len(elem): + if not elem.text or not elem.text.strip(): + elem.text = i + ' ' + if not elem.tail or not elem.tail.strip(): + elem.tail = i + for elem in elem: + self.indent(elem, level + 1) + if not elem.tail or not elem.tail.strip(): + elem.tail = i + else: + if level and (not elem.tail or not elem.tail.strip()): + elem.tail = i + + +def _reference_generator(source_id, target_id, reference_type, is_forward=True): + ref = ua.AddReferencesItem() + ref.IsForward = is_forward + ref.ReferenceTypeId = reference_type + ref.SourceNodeId = source_id + ref.TargetNodeClass = ua.NodeClass.DataType + ref.TargetNodeId = target_id + return ref + + +class DataTypeDictionaryBuilder: + + def __init__(self, server, idx, idx_name, dict_name): + self._server = server + self._session_server = server.get_root_node().server + self._idx = idx + # Risk of bugs using a fixed number without checking + self._id_counter = 8000 + self.dict_id = self._add_dictionary(dict_name) + self._type_dictionary = OPCTypeDictionaryBuilder(idx_name) + + def _add_dictionary(self, name): + dictionary_node_id = self._nodeid_generator() + node = ua.AddNodesItem() + node.RequestedNewNodeId = dictionary_node_id + node.BrowseName = ua.QualifiedName(name, self._idx) + node.NodeClass = ua.NodeClass.Variable + node.ParentNodeId = ua.NodeId(ua.ObjectIds.OPCBinarySchema_TypeSystem, 0) + node.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent, 0) + node.TypeDefinition = ua.NodeId(ua.ObjectIds.DataTypeDictionaryType, 0) + attrs = ua.VariableAttributes() + attrs.DisplayName = ua.LocalizedText(name) + attrs.DataType = ua.NodeId(ua.ObjectIds.ByteString) + # Value should be set after all data types created by calling set_dict_byte_string + attrs.Value = ua.Variant(None, ua.VariantType.Null) + attrs.ValueRank = -1 + node.NodeAttributes = attrs + self._session_server.add_nodes([node]) + + return dictionary_node_id + + def _nodeid_generator(self): + self._id_counter += 1 + return ua.NodeId(self._id_counter, namespaceidx=self._idx, nodeidtype=ua.NodeIdType.Numeric) + + def _link_nodes(self, linked_obj_node_id, data_type_node_id, description_node_id): + """link the three node by their node ids according to UA standard""" + refs = [ + # add reverse reference to BaseDataType -> Structure + _reference_generator(data_type_node_id, ua.NodeId(ua.ObjectIds.Structure, 0), + ua.NodeId(ua.ObjectIds.HasSubtype, 0), False), + # add reverse reference to created data type + _reference_generator(linked_obj_node_id, data_type_node_id, + ua.NodeId(ua.ObjectIds.HasEncoding, 0), False), + # add HasDescription link to dictionary description + _reference_generator(linked_obj_node_id, description_node_id, + ua.NodeId(ua.ObjectIds.HasDescription, 0)), + # add reverse HasDescription link + _reference_generator(description_node_id, linked_obj_node_id, + ua.NodeId(ua.ObjectIds.HasDescription, 0), False), + # add link to the type definition node + _reference_generator(linked_obj_node_id, ua.NodeId(ua.ObjectIds.DataTypeEncodingType, 0), + ua.NodeId(ua.ObjectIds.HasTypeDefinition, 0)), + # add has type definition link + _reference_generator(description_node_id, ua.NodeId(ua.ObjectIds.DataTypeDescriptionType, 0), + ua.NodeId(ua.ObjectIds.HasTypeDefinition, 0)), + # add forward link of dict to description item + _reference_generator(self.dict_id, description_node_id, + ua.NodeId(ua.ObjectIds.HasComponent, 0)), + # add reverse link to dictionary + _reference_generator(description_node_id, self.dict_id, + ua.NodeId(ua.ObjectIds.HasComponent, 0), False)] + self._session_server.add_references(refs) + + def _create_data_type(self, type_name): + name = _to_camel_case(type_name) + # apply for new node id + data_type_node_id = self._nodeid_generator() + description_node_id = self._nodeid_generator() + bind_obj_node_id = self._nodeid_generator() + + # create data type node + dt_node = ua.AddNodesItem() + dt_node.RequestedNewNodeId = data_type_node_id + dt_node.BrowseName = ua.QualifiedName(name, self._idx) + dt_node.NodeClass = ua.NodeClass.DataType + dt_node.ParentNodeId = ua.NodeId(ua.ObjectIds.Structure, 0) + dt_node.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasSubtype, 0) + dt_attributes = ua.DataTypeAttributes() + dt_attributes.DisplayName = ua.LocalizedText(type_name) + dt_node.NodeAttributes = dt_attributes + + # create description node + desc_node = ua.AddNodesItem() + desc_node.RequestedNewNodeId = description_node_id + desc_node.BrowseName = ua.QualifiedName(name, self._idx) + desc_node.NodeClass = ua.NodeClass.Variable + desc_node.ParentNodeId = self.dict_id + desc_node.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent, 0) + desc_node.TypeDefinition = ua.NodeId(ua.ObjectIds.DataTypeDescriptionType, 0) + desc_attributes = ua.VariableAttributes() + desc_attributes.DisplayName = ua.LocalizedText(type_name) + desc_attributes.DataType = ua.NodeId(ua.ObjectIds.String) + desc_attributes.Value = ua.Variant(name, ua.VariantType.String) + desc_attributes.ValueRank = -1 + desc_node.NodeAttributes = desc_attributes + + # create object node which the loaded python class should link to + obj_node = ua.AddNodesItem() + obj_node.RequestedNewNodeId = bind_obj_node_id + obj_node.BrowseName = ua.QualifiedName('Default Binary', 0) + obj_node.NodeClass = ua.NodeClass.Object + obj_node.ParentNodeId = data_type_node_id + obj_node.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasEncoding, 0) + obj_node.TypeDefinition = ua.NodeId(ua.ObjectIds.DataTypeEncodingType, 0) + obj_attributes = ua.ObjectAttributes() + obj_attributes.DisplayName = ua.LocalizedText('Default Binary') + obj_attributes.EventNotifier = 0 + obj_node.NodeAttributes = obj_attributes + + self._session_server.add_nodes([dt_node, desc_node, obj_node]) + self._link_nodes(bind_obj_node_id, data_type_node_id, description_node_id) + + self._type_dictionary.append_struct(type_name) + return StructNode(self, data_type_node_id, type_name) + + def create_data_type(self, type_name): + return self._create_data_type(type_name) + + def add_field(self, variable_name, data_type, struct_name, is_array=False): + self._type_dictionary.add_field(variable_name, data_type, struct_name, is_array) + + def set_dict_byte_string(self): + dict_node = self._server.get_node(self.dict_id) + value = self._type_dictionary.get_dict_value() + dict_node.set_value(value, ua.VariantType.ByteString) + + +class StructNode: + + def __init__(self, type_dict, data_type, name): + self._type_dict = type_dict + self.data_type = data_type + self.name = name + pass + + def add_field(self, field_name, data_type, is_array=False): + # nested structure could directly use simple structure as field + if isinstance(data_type, StructNode): + data_type = data_type.name + self._type_dict.add_field(field_name, data_type, self.name, is_array) + + +def get_ua_class(ua_class_name): + return getattr(ua, _to_camel_case(ua_class_name)) diff --git a/tests/tests.py b/tests/tests.py index 677116897..23f422f14 100644 --- a/tests/tests.py +++ b/tests/tests.py @@ -13,6 +13,7 @@ from tests_history import TestHistory, TestHistorySQL, TestHistoryLimits, TestHistorySQLLimits from tests_crypto_connect import TestCryptoConnect from tests_uaerrors import TestUaErrors +from tests_custom_structures import TypeDictionaryBuilderTest if __name__ == '__main__': diff --git a/tests/tests_custom_structures.py b/tests/tests_custom_structures.py new file mode 100644 index 000000000..5f94a058b --- /dev/null +++ b/tests/tests_custom_structures.py @@ -0,0 +1,393 @@ +import unittest +import logging +import xml.etree.ElementTree as Et +from opcua import ua, Server +import opcua.common.type_dictionary_buider +from opcua.common.type_dictionary_buider import OPCTypeDictionaryBuilder, DataTypeDictionaryBuilder +from opcua.common.type_dictionary_buider import get_ua_class, StructNode + +port_num = 48540 +idx_name = 'http://test.freeopcua.github.io' + + +def to_camel_case(name): + func = getattr(opcua.common.type_dictionary_buider, '_to_camel_case') + return func(name) + + +def reference_generator(source_id, target_id, reference_type, is_forward=True): + func = getattr(opcua.common.type_dictionary_buider, '_reference_generator') + return func(source_id, target_id, reference_type, is_forward) + + +def set_up_test_tree(): + ext_head_attributes = {'xmlns:xsi': 'http://www.w3.org/2001/XMLSchema-instance', 'xmlns:tns': idx_name, + 'DefaultByteOrder': 'LittleEndian', 'xmlns:opc': 'http://opcfoundation.org/BinarySchema/', + 'xmlns:ua': 'http://opcfoundation.org/UA/', 'TargetNamespace': idx_name} + + test_etree = Et.ElementTree(Et.Element('opc:TypeDictionary', ext_head_attributes)) + name_space = Et.SubElement(test_etree.getroot(), 'opc:Import') + name_space.attrib['Namespace'] = 'http://opcfoundation.org/UA/' + return test_etree + + +class TypeDictionaryBuilderTest(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.srv = Server() + cls.srv.set_endpoint('opc.tcp://127.0.0.1:{0:d}'.format(port_num)) + cls.idx = cls.srv.register_namespace(idx_name) + cls.srv.start() + + @classmethod + def tearDownClass(cls): + cls.srv.stop() + + def setUp(self): + self.test_etree = set_up_test_tree() + self.opc_type_builder = OPCTypeDictionaryBuilder(idx_name) + self.dict_builder = DataTypeDictionaryBuilder(self.srv, self.idx, idx_name, 'TestDict') + self.init_counter = getattr(self.dict_builder, '_id_counter') + + def tearDown(self): + curr_counter = getattr(self.dict_builder, '_id_counter') + trash_nodes = [] + for counter in range(self.init_counter, curr_counter + 1): + trash_nodes.append(self.srv.get_node('ns={0};i={1}'.format(self.idx, str(counter)))) + self.srv.delete_nodes(trash_nodes) + + def test_camel_case_1(self): + case = 'TurtleActionlibShapeActionFeedback' + result = to_camel_case('turtle_actionlib/ShapeActionFeedback') + self.assertEquals(result, case) + + def test_camel_case_2(self): + case = 'HelloWorldFffD' + result = to_camel_case('Hello#world+fff_**?&&d') + self.assertEquals(result, case) + + def test_opc_type_dict_process_type_opc(self): + case = 'opc:Boolean' + result = getattr(self.opc_type_builder, '_process_type')('Boolean') + self.assertEquals(result, case) + + def test_opc_type_dict_process_type_tns(self): + case = 'tns:CustomizedStruct' + result = getattr(self.opc_type_builder, '_process_type')('CustomizedStruct') + self.assertEquals(result, case) + + def test_opc_type_dict_append_struct_1(self): + case = {'BaseType': 'ua:ExtensionObject', + 'Name': 'CustomizedStruct'} + result = self.opc_type_builder.append_struct('CustomizedStruct') + self.assertEquals(result.attrib, case) + + def test_opc_type_dict_append_struct_2(self): + case = {'BaseType': 'ua:ExtensionObject', + 'Name': 'CustomizedStruct'} + result = self.opc_type_builder.append_struct('customized_#?+`struct') + self.assertEquals(result.attrib, case) + + def test_opc_type_dict_add_field_1(self): + structure_name = 'CustomizedStruct' + self.opc_type_builder.append_struct(structure_name) + self.opc_type_builder.add_field('id', ua.VariantType.Boolean, structure_name) + case = {'TypeName': 'opc:Boolean', + 'Name': 'id'} + struct_dict = getattr(self.opc_type_builder, '_structs_dict') + result = list(struct_dict[structure_name])[0] + self.assertEquals(result.attrib, case) + + def test_opc_type_dict_add_field_2(self): + structure_name = 'CustomizedStruct' + self.opc_type_builder.append_struct(structure_name) + self.opc_type_builder.add_field('id', 'Boolean', structure_name) + case = {'TypeName': 'opc:Boolean', + 'Name': 'id'} + struct_dict = getattr(self.opc_type_builder, '_structs_dict') + result = list(struct_dict[structure_name])[0] + self.assertEquals(result.attrib, case) + + def test_opc_type_dict_add_field_3(self): + structure_name = 'CustomizedStruct' + self.opc_type_builder.append_struct(structure_name) + self.opc_type_builder.add_field('id', ua.VariantType.Boolean, structure_name, is_array=True) + case = [{'TypeName': 'opc:Int32', + 'Name': 'NoOfid'}, + {'TypeName': 'opc:Boolean', + 'LengthField': 'NoOfid', + 'Name': 'id'}] + struct_dict = getattr(self.opc_type_builder, '_structs_dict') + result = [item.attrib for item in list(struct_dict[structure_name])] + self.assertTrue(len(case), len(result)) + for item in case: + self.assertTrue(item in result) + + def test_opc_type_dict_get_dict_value(self): + structure_name = 'CustomizedStruct' + self.opc_type_builder.append_struct(structure_name) + # external tree operation + appended_struct = Et.SubElement(self.test_etree.getroot(), 'opc:StructuredType') + appended_struct.attrib['BaseType'] = 'ua:ExtensionObject' + appended_struct.attrib['Name'] = to_camel_case(structure_name) + + self.opc_type_builder.add_field('id', ua.VariantType.Boolean, structure_name) + # external tree operation + field = Et.SubElement(appended_struct, 'opc:Field') + field.attrib['Name'] = 'id' + field.attrib['TypeName'] = 'opc:Boolean' + case = Et.tostring(self.test_etree.getroot(), encoding='utf-8').decode("utf-8").replace(' ', '') + result = self.opc_type_builder.get_dict_value().decode("utf-8").replace(' ', '').replace('\n', '') + self.assertEqual(result, case) + + def test_reference_generator_1(self): + id1 = ua.NodeId(1, namespaceidx=2, nodeidtype=ua.NodeIdType.Numeric) + id2 = ua.NodeId(2, namespaceidx=2, nodeidtype=ua.NodeIdType.Numeric) + ref = ua.NodeId(ua.ObjectIds.HasEncoding, 0) + result = reference_generator(id1, id2, ref) + self.assertTrue(result.IsForward) + self.assertEqual(result.ReferenceTypeId, ref) + self.assertEqual(result.SourceNodeId, id1) + self.assertEqual(result.TargetNodeClass, ua.NodeClass.DataType) + self.assertEqual(result.TargetNodeId, id2) + + def test_reference_generator_2(self): + id1 = ua.NodeId(1, namespaceidx=2, nodeidtype=ua.NodeIdType.Numeric) + id2 = ua.NodeId(2, namespaceidx=2, nodeidtype=ua.NodeIdType.Numeric) + ref = ua.NodeId(ua.ObjectIds.HasEncoding, 0) + result = reference_generator(id1, id2, ref, False) + self.assertFalse(result.IsForward) + self.assertEqual(result.ReferenceTypeId, ref) + self.assertEqual(result.SourceNodeId, id1) + self.assertEqual(result.TargetNodeClass, ua.NodeClass.DataType) + self.assertEqual(result.TargetNodeId, id2) + + def test_data_type_dict_general(self): + self.assertIsNotNone(self.dict_builder.dict_id) + self.assertIsNotNone(getattr(self.dict_builder, '_type_dictionary')) + + def test_data_type_dict_nodeid_generator(self): + nodeid_generator = getattr(self.dict_builder, '_nodeid_generator') + result = nodeid_generator() + self.assertTrue(isinstance(result, ua.NodeId)) + self.assertTrue(str(result.Identifier).isdigit()) + self.assertEqual(result.NamespaceIndex, self.idx) + setattr(self.dict_builder, '_id_counter', self.init_counter) + + def test_data_type_dict_add_dictionary(self): + add_dictionary = getattr(self.dict_builder, '_add_dictionary') + dict_name = 'TestDict' + dict_node = self.srv.get_node(add_dictionary(dict_name)) + self.assertEqual(dict_node.get_browse_name(), ua.QualifiedName(dict_name, self.idx)) + self.assertEqual(dict_node.get_node_class(), ua.NodeClass.Variable) + self.assertEqual(dict_node.get_parent().nodeid, ua.NodeId(ua.ObjectIds.OPCBinarySchema_TypeSystem, 0)) + self.assertEqual(ua.NodeId(ua.ObjectIds.HasComponent, 0), + dict_node.get_references(refs=ua.ObjectIds.HasComponent)[0].ReferenceTypeId) + self.assertEqual(dict_node.get_type_definition(), ua.NodeId(ua.ObjectIds.DataTypeDictionaryType, 0)) + self.assertEqual(dict_node.get_display_name(), ua.LocalizedText(dict_name)) + self.assertEqual(dict_node.get_data_type(), ua.NodeId(ua.ObjectIds.ByteString)) + self.assertEqual(dict_node.get_value_rank(), -1) + + def test_data_type_dict_create_data_type(self): + type_name = 'CustomizedStruct' + created_type = self.dict_builder.create_data_type(type_name) + self.assertTrue(isinstance(created_type, StructNode)) + # Test data type node + type_node = self.srv.get_node(created_type.data_type) + self.assertEqual(type_node.get_browse_name(), ua.QualifiedName(type_name, self.idx)) + self.assertEqual(type_node.get_node_class(), ua.NodeClass.DataType) + self.assertEqual(type_node.get_parent().nodeid, ua.NodeId(ua.ObjectIds.Structure, 0)) + self.assertEqual(ua.NodeId(ua.ObjectIds.HasSubtype, 0), + type_node.get_references(refs=ua.ObjectIds.HasSubtype)[0].ReferenceTypeId) + self.assertEqual(type_node.get_display_name(), ua.LocalizedText(type_name)) + + # Test description node + desc_node = self.srv.get_node(self.dict_builder.dict_id).get_children()[0] + self.assertEqual(desc_node.get_browse_name(), ua.QualifiedName(type_name, self.idx)) + self.assertEqual(desc_node.get_node_class(), ua.NodeClass.Variable) + self.assertEqual(desc_node.get_parent().nodeid, self.dict_builder.dict_id) + self.assertEqual(ua.NodeId(ua.ObjectIds.HasComponent, 0), + desc_node.get_references(refs=ua.ObjectIds.HasComponent)[0].ReferenceTypeId) + self.assertEqual(desc_node.get_type_definition(), ua.NodeId(ua.ObjectIds.DataTypeDescriptionType, 0)) + + self.assertEqual(desc_node.get_display_name(), ua.LocalizedText(type_name)) + self.assertEqual(desc_node.get_data_type(), ua.NodeId(ua.ObjectIds.String)) + self.assertEqual(desc_node.get_value(), type_name) + self.assertEqual(desc_node.get_value_rank(), -1) + + # Test object node + obj_node = type_node.get_children(refs=ua.ObjectIds.HasEncoding)[0] + self.assertEqual(obj_node.get_browse_name(), ua.QualifiedName('Default Binary', 0)) + self.assertEqual(obj_node.get_node_class(), ua.NodeClass.Object) + self.assertEqual(obj_node.get_references(refs=ua.ObjectIds.HasEncoding)[0].NodeId, type_node.nodeid) + self.assertEqual(ua.NodeId(ua.ObjectIds.HasEncoding, 0), + obj_node.get_references(refs=ua.ObjectIds.HasEncoding)[0].ReferenceTypeId) + self.assertEqual(obj_node.get_type_definition(), ua.NodeId(ua.ObjectIds.DataTypeEncodingType, 0)) + self.assertEqual(obj_node.get_display_name(), ua.LocalizedText('Default Binary')) + self.assertEqual(len(obj_node.get_event_notifier()), 0) + + # Test links, three were tested above + struct_node = self.srv.get_node(ua.NodeId(ua.ObjectIds.Structure, 0)) + struct_children = struct_node.get_children(refs=ua.ObjectIds.HasSubtype) + self.assertTrue(type_node in struct_children) + dict_node = self.srv.get_node(self.dict_builder.dict_id) + dict_children = dict_node.get_children(refs=ua.ObjectIds.HasComponent) + self.assertTrue(desc_node in dict_children) + self.assertTrue(obj_node in type_node.get_children(ua.ObjectIds.HasEncoding)) + self.assertTrue(desc_node in obj_node.get_children(refs=ua.ObjectIds.HasDescription)) + self.assertEqual(obj_node.nodeid, desc_node.get_references(refs=ua.ObjectIds.HasDescription, + direction=ua.BrowseDirection.Inverse)[0].NodeId) + + def test_data_type_dict_set_dict_byte_string(self): + structure_name = 'CustomizedStruct' + self.dict_builder.create_data_type(structure_name) + self.dict_builder.add_field('id', ua.VariantType.Int32, structure_name) + self.dict_builder.set_dict_byte_string() + # external tree operation + appended_struct = Et.SubElement(self.test_etree.getroot(), 'opc:StructuredType') + appended_struct.attrib['BaseType'] = 'ua:ExtensionObject' + appended_struct.attrib['Name'] = to_camel_case(structure_name) + + # external tree operation + field = Et.SubElement(appended_struct, 'opc:Field') + field.attrib['Name'] = 'id' + field.attrib['TypeName'] = 'opc:Int32' + case = Et.tostring(self.test_etree.getroot(), encoding='utf-8').decode("utf-8").replace(' ', '') + result_string = self.srv.get_node(self.dict_builder.dict_id).get_value().decode("utf-8") + result = result_string.replace(' ', '').replace('\n', '') + self.assertEqual(result, case) + + def test_data_type_dict_add_field_1(self): + struct_name = 'CustomizedStruct' + self.dict_builder.create_data_type(struct_name) + self.dict_builder.add_field('id', ua.VariantType.Int32, struct_name) + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + struct = get_ua_class(struct_name) + self.assertEqual(struct.ua_types[0][0], 'id') + self.assertEqual(struct.ua_types[0][1], 'Int32') + struct_instance = struct() + self.assertEqual(struct_instance.id, 0) + + def test_data_type_dict_add_field_2(self): + struct_name = 'AnotherCustomizedStruct' + self.dict_builder.create_data_type(struct_name) + self.dict_builder.add_field('id', ua.VariantType.Int32, struct_name, is_array=True) + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + struct = get_ua_class(struct_name) + self.assertEqual(struct.ua_types[0][0], 'id') + self.assertEqual(struct.ua_types[0][1], 'ListOfInt32') + struct_instance = struct() + self.assertTrue(isinstance(struct_instance.id, list)) + + def test_struct_node_general(self): + struct_name = 'CustomizedStruct' + struct_node = self.dict_builder.create_data_type(struct_name) + self.assertEqual(getattr(struct_node, '_type_dict'), self.dict_builder) + self.assertTrue(isinstance(struct_node.data_type, ua.NodeId)) + self.assertEqual(struct_node.name, struct_name) + + def test_struct_node_add_field(self): + struct_name = 'CustomizedStruct' + struct_node = self.dict_builder.create_data_type(struct_name) + struct_node.add_field('id', ua.VariantType.Int32) + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + struct = get_ua_class(struct_name) + self.assertEqual(struct.ua_types[0][0], 'id') + self.assertEqual(struct.ua_types[0][1], 'Int32') + struct_instance = struct() + self.assertEqual(struct_instance.id, 0) + + def test_get_ua_class_1(self): + struct_name = 'CustomizedStruct' + struct_node = self.dict_builder.create_data_type(struct_name) + struct_node.add_field('id', ua.VariantType.Int32) + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + try: + self.assertIsNotNone(get_ua_class(struct_name)) + except AttributeError: + pass + + def test_get_ua_class_2(self): + struct_name = '*c*u_stom-ized&Stru#ct' + struct_node = self.dict_builder.create_data_type(struct_name) + struct_node.add_field('id', ua.VariantType.Int32) + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + try: + self.assertIsNotNone(get_ua_class(struct_name)) + except AttributeError: + pass + + def test_functional_basic(self): + basic_struct_name = 'basic_structure' + basic_struct = self.dict_builder.create_data_type(basic_struct_name) + basic_struct.add_field('ID', ua.VariantType.Int32) + basic_struct.add_field('Gender', ua.VariantType.Boolean) + basic_struct.add_field('Comments', ua.VariantType.String) + + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + + basic_var = self.srv.nodes.objects.add_variable(ua.NodeId(namespaceidx=self.idx), 'BasicStruct', + ua.Variant(None, ua.VariantType.Null), + datatype=basic_struct.data_type) + + basic_msg = get_ua_class(basic_struct_name)() + basic_msg.ID = 3 + basic_msg.Gender = True + basic_msg.Comments = 'Test string' + basic_var.set_value(basic_msg) + + basic_result = basic_var.get_value() + self.assertEqual(basic_result, basic_msg) + + def test_functional_advance(self): + basic_struct_name = 'base_structure' + basic_struct = self.dict_builder.create_data_type(basic_struct_name) + basic_struct.add_field('ID', ua.VariantType.Int32) + basic_struct.add_field('Gender', ua.VariantType.Boolean) + basic_struct.add_field('Comments', ua.VariantType.String) + + nested_struct_name = 'nested_structure' + nested_struct = self.dict_builder.create_data_type(nested_struct_name) + nested_struct.add_field('Name', ua.VariantType.String) + nested_struct.add_field('Surname', ua.VariantType.String) + nested_struct.add_field('Stuff', basic_struct) + + self.dict_builder.set_dict_byte_string() + self.srv.load_type_definitions() + + basic_var = self.srv.nodes.objects.add_variable(ua.NodeId(namespaceidx=self.idx), 'BaseStruct', + ua.Variant(None, ua.VariantType.Null), + datatype=basic_struct.data_type) + + basic_msg = get_ua_class(basic_struct_name)() + basic_msg.ID = 3 + basic_msg.Gender = True + basic_msg.Comments = 'Test string' + basic_var.set_value(basic_msg) + + nested_var = self.srv.nodes.objects.add_variable(ua.NodeId(namespaceidx=self.idx), 'NestedStruct', + ua.Variant(None, ua.VariantType.Null), + datatype=nested_struct.data_type) + + nested_msg = get_ua_class(nested_struct_name)() + nested_msg.Stuff = basic_msg + nested_msg.Name = 'Max' + nested_msg.Surname = 'Karl' + nested_var.set_value(nested_msg) + + basic_result = basic_var.get_value() + self.assertEqual(basic_result, basic_msg) + nested_result = nested_var.get_value() + self.assertEqual(nested_result, nested_msg) + + +if __name__ == '__main__': + logging.basicConfig(level=logging.WARNING) + unittest.main(verbosity=3)