XmlImporter._add_node() - Code Metrics - Inspection of "Xml import / export client side" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#389)

by Olivier

created 2016-12-29 19:40 UTC

XmlImporter._add_node() A

↳ Parent: XmlImporter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
c	0
b	0
f	0
dl	0
loc	6
ccs	5
cts	5
cp	1
crap	2
rs	9.4285

"""
add nodes defined in XML to address space
format is the one from opc-ua specification
"""
import logging
import uuid
from copy import copy

import opcua
from opcua import ua
from opcua.common import xmlparser


class XmlImporter(object):

    def __init__(self, server):
        self.logger = logging.getLogger(__name__)
        self.parser = None
        self.server = server
        self.namespaces = {}
        self.aliases = {}

    def _map_namespaces(self, namespaces_uris):
        """
        creates a mapping between the namespaces in the xml file and in the server.
        if not present the namespace is registered.
        """
        namespaces = {}
        for ns_index, ns_uri in enumerate(namespaces_uris):
            ns_server_index = self.server.register_namespace(ns_uri)
            namespaces[ns_index + 1] = ns_server_index
        return namespaces

    def _map_aliases(self, aliases):
        """
        maps the import aliases to the correct namespaces        
        """
        aliases_mapped = {}
        for alias, node_id in aliases.items():
            aliases_mapped[alias] = self._migrate_ns(self.to_nodeid(node_id))
        return aliases_mapped

    def import_xml(self, xmlpath):
        """
        import xml and return added nodes
        """
        self.logger.info("Importing XML file %s", xmlpath)
        self.parser = xmlparser.XMLParser(xmlpath)

        dnodes = self.parser.get_node_datas()
        dnodes = self.make_objects(dnodes)

        self.namespaces = self._map_namespaces(self.parser.get_used_namespaces())
        self.aliases = self._map_aliases(self.parser.get_aliases())

        nodes_parsed = self._sort_nodes_by_parentid(dnodes)

        nodes = []
        for nodedata in nodes_parsed:  # self.parser:
            if nodedata.nodetype == 'UAObject':
                node = self.add_object(nodedata)
            elif nodedata.nodetype == 'UAObjectType':
                node = self.add_object_type(nodedata)
            elif nodedata.nodetype == 'UAVariable':
                node = self.add_variable(nodedata)
            elif nodedata.nodetype == 'UAVariableType':
                node = self.add_variable_type(nodedata)
            elif nodedata.nodetype == 'UAReferenceType':
                node = self.add_reference_type(nodedata)
            elif nodedata.nodetype == 'UADataType':
                node = self.add_datatype(nodedata)
            elif nodedata.nodetype == 'UAMethod':
                node = self.add_method(nodedata)
            else:
                self.logger.warning("Not implemented node type: %s ", nodedata.nodetype)
                continue
            nodes.append(node)
        return nodes

    def _add_node(self, node):
        print("ADD NODE", node)
        if isinstance(self.server, opcua.server.server.Server):
            return self.server.iserver.isession.add_nodes([node])
        else:
            return self.server.uaclient.add_nodes([node])

    def _add_references(self, refs):
        if isinstance(self.server, opcua.server.server.Server):
            return self.server.iserver.isession.add_references(refs)
        else:
            return self.server.uaclient.add_references(refs)
    
    def make_objects(self, node_datas):
        new_nodes = []
        for ndata in node_datas:
            ndata.nodeid = ua.NodeId.from_string(ndata.nodeid)
            ndata.browsename = ua.QualifiedName.from_string(ndata.browsename)
            if ndata.parent:
                ndata.parent = ua.NodeId.from_string(ndata.parent)
            if ndata.parentlink:
                ndata.parentlink = self.to_nodeid(ndata.parentlink)
            if ndata.typedef:
                ndata.typedef = self.to_nodeid(ndata.typedef)
            new_nodes.append(ndata)
        return new_nodes

    def _migrate_ns(self, nodeid):
        """
        Check if the index of nodeid or browsename  given in the xml model file
        must be converted to a already existing namespace id based on the files
        namespace uri

        :returns: NodeId (str)
        """
        if nodeid.NamespaceIndex in self.namespaces:
            nodeid = copy(nodeid)
            nodeid.NamespaceIndex = self.namespaces[nodeid.NamespaceIndex]
        return nodeid

    def _get_node(self, obj):
        node = ua.AddNodesItem()
        node.RequestedNewNodeId = self._migrate_ns(obj.nodeid)
        node.BrowseName = self._migrate_ns(obj.browsename)
        self.logger.info("Importing xml node (%s, %s) as (%s %s)", obj.browsename, obj.nodeid, node.BrowseName, node.RequestedNewNodeId)
        node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
        if obj.parent:
            node.ParentNodeId = self._migrate_ns(obj.parent)
        if obj.parentlink:
            node.ReferenceTypeId = self._migrate_ns(obj.parentlink)
        if obj.typedef:
            node.TypeDefinition = self._migrate_ns(obj.typedef)
        return node

    def to_nodeid(self, nodeid):
        if isinstance(nodeid, ua.NodeId):
            return nodeid
        elif not nodeid:
            return ua.NodeId(ua.ObjectIds.String)
        elif "=" in nodeid:
            return ua.NodeId.from_string(nodeid)
        elif hasattr(ua.ObjectIds, nodeid):
            return ua.NodeId(getattr(ua.ObjectIds, nodeid))
        else:
            if nodeid in self.aliases:
                return self.aliases[nodeid]
            else:
                return ua.NodeId(getattr(ua.ObjectIds, nodeid))

    def add_object(self, obj):

        node = self._get_node(obj)
        attrs = ua.ObjectAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.EventNotifier = obj.eventnotifier
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def add_object_type(self, obj):

        node = self._get_node(obj)
        attrs = ua.ObjectTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.IsAbstract = obj.abstract
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def add_variable(self, obj):
        node = self._get_node(obj)
        attrs = ua.VariableAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.DataType = self.to_nodeid(obj.datatype)
        if obj.value is not None:
            attrs.Value = self._add_variable_value(obj,)
        if obj.rank:
            attrs.ValueRank = obj.rank
        if obj.accesslevel:
            attrs.AccessLevel = obj.accesslevel
        if obj.useraccesslevel:
            attrs.UserAccessLevel = obj.useraccesslevel
        if obj.minsample:
            attrs.MinimumSamplingInterval = obj.minsample
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def _make_ext_obj(self, obj):
        ext = getattr(ua, obj.objname)()
        for name, val in obj.body:
            if isinstance(val, str):
                raise Exception("Error val should a dict", name, val)
            else:
                for attname, v in val:
                    self._set_attr(ext, attname, v)
        return ext

    def _set_attr(self, obj, attname, val):
        # tow possible values:
        # either we get value directly
        # or a dict if it s an object or a list
        if isinstance(val, str):
            pval = xmlparser.ua_type_to_python(val, obj.ua_types[attname])
            setattr(obj, attname, pval)
        else:
            # so we have either an object or a list...
            obj2 = getattr(obj, attname)
            if isinstance(obj2, ua.NodeId):  # NodeId representation does not follow common rules!!
                for attname2, v2 in val:
                    if attname2 == "Identifier":
                        obj2 = ua.NodeId.from_string(v2)
                        setattr(obj, attname, obj2)
                        break
            elif not isinstance(obj2, ua.NodeId) and not hasattr(obj2, "ua_types"):
                # we probably have a list
                my_list = []
                for vtype, v2 in val:
                    my_list.append(xmlparser.ua_type_to_python(v2, vtype))
                setattr(obj, attname, my_list)
            else:
                for attname2, v2 in val:
                    self._set_attr(obj2, attname2, v2)
                setattr(obj, attname, obj2)

    def _add_variable_value(self, obj):
        """
        Returns the value for a Variable based on the objects value type.
        """
        self.logger.debug("Setting value with type %s and value %s", obj.valuetype, obj.value)
        if obj.valuetype == 'ListOfExtensionObject':
            values = []
            for ext in obj.value:
                extobj = self._make_ext_obj(ext)
                values.append(extobj)
            return values
        elif obj.valuetype == 'ListOfGuid':
            return ua.Variant([
                uuid.UUID(guid) for guid in obj.value
            ], getattr(ua.VariantType, obj.valuetype[6:]))
        elif obj.valuetype.startswith("ListOf"):
            vtype = obj.valuetype[6:]
            if hasattr(ua.ua_binary.Primitives, vtype):
                return ua.Variant(obj.value, getattr(ua.VariantType, vtype))
            else:
                return [getattr(ua, vtype)(v) for v in obj.value]
        elif obj.valuetype == 'ExtensionObject':
            extobj = self._make_ext_obj(obj.value)
            return ua.Variant(extobj, getattr(ua.VariantType, obj.valuetype))
        elif obj.valuetype == 'Guid':
            return ua.Variant(uuid.UUID(obj.value), getattr(ua.VariantType, obj.valuetype))
        elif obj.valuetype == 'LocalizedText':
            ltext = ua.LocalizedText()
            for name, val in obj.value:
                if name == "Text":
                    ltext.Text = val.encode("utf-8")
                else:
                    self.logger.warning("While parsing localizedText value, unkown element: %s with val: %s", name, val)
            return ua.Variant(ltext, ua.VariantType.LocalizedText)
        elif obj.valuetype == 'NodeId':
            return ua.Variant(ua.NodeId.from_string(obj.value))
        else:
            return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))

    def add_variable_type(self, obj):
        node = self._get_node(obj)
        attrs = ua.VariableTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.DataType = self.to_nodeid(obj.datatype)
        if obj.value and len(obj.value) == 1:
            attrs.Value = obj.value[0]
        if obj.rank:
            attrs.ValueRank = obj.rank
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def add_method(self, obj):

        node = self._get_node(obj)
        attrs = ua.MethodAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj.accesslevel:
            attrs.AccessLevel = obj.accesslevel
        if obj.useraccesslevel:
            attrs.UserAccessLevel = obj.useraccesslevel
        if obj.minsample:
            attrs.MinimumSamplingInterval = obj.minsample
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def add_reference_type(self, obj):

        node = self._get_node(obj)
        attrs = ua.ReferenceTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj. inversename:
            attrs.InverseName = ua.LocalizedText(obj.inversename)
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if obj.symmetric:
            attrs.Symmetric = obj.symmetric
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def add_datatype(self, obj):

        node = self._get_node(obj)
        attrs = ua.DataTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        node.NodeAttributes = attrs
        res = self._add_node(node)
        self._add_refs(obj)
        res[0].StatusCode.check()
        return res[0].AddedNodeId

    def _add_refs(self, obj):
        if not obj.refs:
            return
        refs = []
        for data in obj.refs:
            ref = ua.AddReferencesItem()
            ref.IsForward = True
            ref.ReferenceTypeId = self.to_nodeid(data.reftype)
            ref.SourceNodeId = self._migrate_ns(obj.nodeid)
            ref.TargetNodeClass = ua.NodeClass.DataType
            ref.TargetNodeId = self._migrate_ns(self.to_nodeid(data.target))
            refs.append(ref)
        self._add_references(refs)

    def _sort_nodes_by_parentid(self, ndatas):
        """
        Sort the list of nodes according their parent node in order to respect
        the dependency between nodes.

        :param nodes: list of NodeDataObjects
        :returns: list of sorted nodes
        """
        _ndatas = list(ndatas)
        # list of node ids that are already sorted / inserted
        sorted_nodes_ids = []
        # list of sorted nodes (i.e. XML Elements)
        sorted_ndatas = []
        all_node_ids = [data.nodeid for data in ndatas]
        # list of namespace indexes that are relevant for this import
        # we can only respect ordering nodes for namespaces indexes that
        # are defined in the xml file itself. Thus we assume that all other
        # references namespaces are already known to the server and should
        # not create any dependency problems (like "NodeNotFound")
        while len(_ndatas) > 0:
            pop_nodes = []
            for ndata in _ndatas:
                # Insert nodes that
                #   (1) have no parent / parent_ns is None (e.g. namespace 0)
                #   (2) ns is not in list of relevant namespaces
                if ndata.nodeid.NamespaceIndex not in self.namespaces or \
                        ndata.parent is None or \
                        ndata.parent not in all_node_ids:
                    sorted_ndatas.append(ndata)
                    sorted_nodes_ids.append(ndata.nodeid)
                    pop_nodes.append(ndata)
                else:
                    # Check if the nodes parent is already in the list of
                    # inserted nodes
                    if ndata.parent in sorted_nodes_ids:
                        sorted_ndatas.append(ndata)
                        sorted_nodes_ids.append(ndata.nodeid)
                        pop_nodes.append(ndata)
            # Remove inserted nodes from the list
            for ndata in pop_nodes:
                _ndatas.pop(_ndatas.index(ndata))
        return sorted_ndatas


1			"""
2			add nodes defined in XML to address space
3			format is the one from opc-ua specification
4			"""
5	1		import logging
6	1		import uuid
7	1		from copy import copy
8
9	1		import opcua
10			from opcua import ua
11	1		from opcua.common import xmlparser
12	1
13
14			class XmlImporter(object):
15	1
16			def __init__(self, server):
17	1		self.logger = logging.getLogger(__name__)
18	1		self.parser = None
19	1		self.server = server
20	1		self.namespaces = {}
21	1		self.aliases = {}
22	1
23			def _map_namespaces(self, namespaces_uris):
24	1		"""
25			creates a mapping between the namespaces in the xml file and in the server.
26			if not present the namespace is registered.
27			"""
28			namespaces = {}
29	1		for ns_index, ns_uri in enumerate(namespaces_uris):
30	1		ns_server_index = self.server.register_namespace(ns_uri)
31	1		namespaces[ns_index + 1] = ns_server_index
32	1		return namespaces
33	1
34			def _map_aliases(self, aliases):
35	1		"""
36			maps the import aliases to the correct namespaces
37			"""
38			aliases_mapped = {}
39	1		for alias, node_id in aliases.items():
40	1		aliases_mapped[alias] = self._migrate_ns(self.to_nodeid(node_id))
41	1		return aliases_mapped
42	1
43			def import_xml(self, xmlpath):
44	1		"""
45			import xml and return added nodes
46			"""
47			self.logger.info("Importing XML file %s", xmlpath)
48	1		self.parser = xmlparser.XMLParser(xmlpath)
49	1
50			dnodes = self.parser.get_node_datas()
51	1		dnodes = self.make_objects(dnodes)
52	1
53			self.namespaces = self._map_namespaces(self.parser.get_used_namespaces())
54	1		self.aliases = self._map_aliases(self.parser.get_aliases())
55	1
56			nodes_parsed = self._sort_nodes_by_parentid(dnodes)
57	1
58			nodes = []
59	1		for nodedata in nodes_parsed: # self.parser:
60	1		if nodedata.nodetype == 'UAObject':
61	1		node = self.add_object(nodedata)
62	1		elif nodedata.nodetype == 'UAObjectType':
63	1		node = self.add_object_type(nodedata)
64	1		elif nodedata.nodetype == 'UAVariable':
65	1		node = self.add_variable(nodedata)
66	1		elif nodedata.nodetype == 'UAVariableType':
67	1		node = self.add_variable_type(nodedata)
68			elif nodedata.nodetype == 'UAReferenceType':
69	1		node = self.add_reference_type(nodedata)
70			elif nodedata.nodetype == 'UADataType':
71	1		node = self.add_datatype(nodedata)
72	1		elif nodedata.nodetype == 'UAMethod':
73	1		node = self.add_method(nodedata)
74	1		else:
75			self.logger.warning("Not implemented node type: %s ", nodedata.nodetype)
76			continue
77			nodes.append(node)
78	1		return nodes
79	1
80			def _add_node(self, node):
81	1		print("ADD NODE", node)
82	1		if isinstance(self.server, opcua.server.server.Server):
83	1		return self.server.iserver.isession.add_nodes([node])
84	1		else:
85	1		return self.server.uaclient.add_nodes([node])
86	1
87	1		def _add_references(self, refs):
88	1		if isinstance(self.server, opcua.server.server.Server):
89	1		return self.server.iserver.isession.add_references(refs)
90	1		else:
91	1		return self.server.uaclient.add_references(refs)
92	1
93	1		def make_objects(self, node_datas):
94			new_nodes = []
95	1		for ndata in node_datas:
96			ndata.nodeid = ua.NodeId.from_string(ndata.nodeid)
97			ndata.browsename = ua.QualifiedName.from_string(ndata.browsename)
98			if ndata.parent:
99			ndata.parent = ua.NodeId.from_string(ndata.parent)
100			if ndata.parentlink:
101			ndata.parentlink = self.to_nodeid(ndata.parentlink)
102			if ndata.typedef:
103	1		ndata.typedef = self.to_nodeid(ndata.typedef)
104	1		new_nodes.append(ndata)
105	1		return new_nodes
106	1
107			def _migrate_ns(self, nodeid):
108	1		"""
109	1		Check if the index of nodeid or browsename given in the xml model file
110	1		must be converted to a already existing namespace id based on the files
111	1		namespace uri
112	1
113	1		:returns: NodeId (str)
114	1		"""
115	1		if nodeid.NamespaceIndex in self.namespaces:
116	1		nodeid = copy(nodeid)
117	1		nodeid.NamespaceIndex = self.namespaces[nodeid.NamespaceIndex]
118	1		return nodeid
119	1
120	1		def _get_node(self, obj):
121			node = ua.AddNodesItem()
122	1		node.RequestedNewNodeId = self._migrate_ns(obj.nodeid)
123	1		node.BrowseName = self._migrate_ns(obj.browsename)
124			self.logger.info("Importing xml node (%s, %s) as (%s %s)", obj.browsename, obj.nodeid, node.BrowseName, node.RequestedNewNodeId)
125	1		node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
126			if obj.parent:
127	1		node.ParentNodeId = self._migrate_ns(obj.parent)
128	1		if obj.parentlink:
129	1		node.ReferenceTypeId = self._migrate_ns(obj.parentlink)
130	1		if obj.typedef:
131			node.TypeDefinition = self._migrate_ns(obj.typedef)
132	1		return node
133	1
134			def to_nodeid(self, nodeid):
135			if isinstance(nodeid, ua.NodeId):
136			return nodeid
137	1		elif not nodeid:
138	1		return ua.NodeId(ua.ObjectIds.String)
139	1		elif "=" in nodeid:
140	1		return ua.NodeId.from_string(nodeid)
141	1		elif hasattr(ua.ObjectIds, nodeid):
142	1		return ua.NodeId(getattr(ua.ObjectIds, nodeid))
143	1		else:
144	1		if nodeid in self.aliases:
145	1		return self.aliases[nodeid]
146	1		else:
147	1		return ua.NodeId(getattr(ua.ObjectIds, nodeid))
148	1
149		View Code Duplication	def add_object(self, obj):
			0 ignored issues – show Duplication introduced 2016-10-15 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
150	1		node = self._get_node(obj)
151	1		attrs = ua.ObjectAttributes()
152	1		if obj.desc:
153	1		attrs.Description = ua.LocalizedText(obj.desc)
154			attrs.DisplayName = ua.LocalizedText(obj.displayname)
155	1		attrs.EventNotifier = obj.eventnotifier
156	1		node.NodeAttributes = attrs
157	1		res = self._add_node(node)
158	1		self._add_refs(obj)
159	1		res[0].StatusCode.check()
160	1		return res[0].AddedNodeId
161	1
162		View Code Duplication	def add_object_type(self, obj):
			0 ignored issues – show Duplication introduced 2016-10-15 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
163	1		node = self._get_node(obj)
164	1		attrs = ua.ObjectTypeAttributes()
165	1		if obj.desc:
166	1		attrs.Description = ua.LocalizedText(obj.desc)
167	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
168	1		attrs.IsAbstract = obj.abstract
169	1		node.NodeAttributes = attrs
170	1		res = self._add_node(node)
171	1		self._add_refs(obj)
172	1		res[0].StatusCode.check()
173	1		return res[0].AddedNodeId
174	1
175			def add_variable(self, obj):
176	1		node = self._get_node(obj)
177			attrs = ua.VariableAttributes()
178	1		if obj.desc:
179			attrs.Description = ua.LocalizedText(obj.desc)
180	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
181	1		attrs.DataType = self.to_nodeid(obj.datatype)
182	1		if obj.value is not None:
183	1		attrs.Value = self._add_variable_value(obj,)
184	1		if obj.rank:
185	1		attrs.ValueRank = obj.rank
186	1		if obj.accesslevel:
187			attrs.AccessLevel = obj.accesslevel
188	1		if obj.useraccesslevel:
189	1		attrs.UserAccessLevel = obj.useraccesslevel
190	1		if obj.minsample:
191	1		attrs.MinimumSamplingInterval = obj.minsample
192			if obj.dimensions:
193			attrs.ArrayDimensions = obj.dimensions
194	1		node.NodeAttributes = attrs
195	1		res = self._add_node(node)
196	1		self._add_refs(obj)
197			res[0].StatusCode.check()
198	1		return res[0].AddedNodeId
199
200			def _make_ext_obj(self, obj):
201			ext = getattr(ua, obj.objname)()
202	1		for name, val in obj.body:
203	1		if isinstance(val, str):
204	1		raise Exception("Error val should a dict", name, val)
205			else:
206			for attname, v in val:
207	1		self._set_attr(ext, attname, v)
208	1		return ext
209	1
210	1		def _set_attr(self, obj, attname, val):
211	1		# tow possible values:
212	1		# either we get value directly
213	1		# or a dict if it s an object or a list
214	1		if isinstance(val, str):
215			pval = xmlparser.ua_type_to_python(val, obj.ua_types[attname])
216	1		setattr(obj, attname, pval)
217	1		else:
218	1		# so we have either an object or a list...
219	1		obj2 = getattr(obj, attname)
220			if isinstance(obj2, ua.NodeId): # NodeId representation does not follow common rules!!
221	1		for attname2, v2 in val:
222	1		if attname2 == "Identifier":
223	1		obj2 = ua.NodeId.from_string(v2)
224			setattr(obj, attname, obj2)
225	1		break
226			elif not isinstance(obj2, ua.NodeId) and not hasattr(obj2, "ua_types"):
227			# we probably have a list
228			my_list = []
229	1		for vtype, v2 in val:
230	1		my_list.append(xmlparser.ua_type_to_python(v2, vtype))
231	1		setattr(obj, attname, my_list)
232	1		else:
233	1		for attname2, v2 in val:
234	1		self._set_attr(obj2, attname2, v2)
235	1		setattr(obj, attname, obj2)
236	1
237	1		def _add_variable_value(self, obj):
238			"""
239			Returns the value for a Variable based on the objects value type.
240	1		"""
241	1		self.logger.debug("Setting value with type %s and value %s", obj.valuetype, obj.value)
242	1		if obj.valuetype == 'ListOfExtensionObject':
243	1		values = []
244			for ext in obj.value:
245	1		extobj = self._make_ext_obj(ext)
246	1		values.append(extobj)
247	1		return values
248	1		elif obj.valuetype == 'ListOfGuid':
249	1		return ua.Variant([
250	1		uuid.UUID(guid) for guid in obj.value
251	1		], getattr(ua.VariantType, obj.valuetype[6:]))
252	1		elif obj.valuetype.startswith("ListOf"):
253	1		vtype = obj.valuetype[6:]
254	1		if hasattr(ua.ua_binary.Primitives, vtype):
255	1		return ua.Variant(obj.value, getattr(ua.VariantType, vtype))
256			else:
257			return [getattr(ua, vtype)(v) for v in obj.value]
258	1		elif obj.valuetype == 'ExtensionObject':
259	1		extobj = self._make_ext_obj(obj.value)
260	1		return ua.Variant(extobj, getattr(ua.VariantType, obj.valuetype))
261			elif obj.valuetype == 'Guid':
262	1		return ua.Variant(uuid.UUID(obj.value), getattr(ua.VariantType, obj.valuetype))
263			elif obj.valuetype == 'LocalizedText':
264	1		ltext = ua.LocalizedText()
265			for name, val in obj.value:
266			if name == "Text":
267			ltext.Text = val.encode("utf-8")
268			else:
269			self.logger.warning("While parsing localizedText value, unkown element: %s with val: %s", name, val)
270			return ua.Variant(ltext, ua.VariantType.LocalizedText)
271			elif obj.valuetype == 'NodeId':
272			return ua.Variant(ua.NodeId.from_string(obj.value))
273			else:
274			return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
275
276			def add_variable_type(self, obj):
277			node = self._get_node(obj)
278			attrs = ua.VariableTypeAttributes()
279			if obj.desc:
280			attrs.Description = ua.LocalizedText(obj.desc)
281			attrs.DisplayName = ua.LocalizedText(obj.displayname)
282			attrs.DataType = self.to_nodeid(obj.datatype)
283			if obj.value and len(obj.value) == 1:
284			attrs.Value = obj.value[0]
285	1		if obj.rank:
286	1		attrs.ValueRank = obj.rank
287	1		if obj.abstract:
288	1		attrs.IsAbstract = obj.abstract
289	1		if obj.dimensions:
290	1		attrs.ArrayDimensions = obj.dimensions
291	1		node.NodeAttributes = attrs
292			res = self._add_node(node)
293	1		self._add_refs(obj)
294			res[0].StatusCode.check()
295	1		return res[0].AddedNodeId
296
297	1	View Code Duplication	def add_method(self, obj):
			0 ignored issues – show Duplication introduced 2016-08-10 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
298			node = self._get_node(obj)
299	1		attrs = ua.MethodAttributes()
300	1		if obj.desc:
301	1		attrs.Description = ua.LocalizedText(obj.desc)
302	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
303	1		if obj.accesslevel:
304			attrs.AccessLevel = obj.accesslevel
305	1		if obj.useraccesslevel:
306			attrs.UserAccessLevel = obj.useraccesslevel
307			if obj.minsample:
308			attrs.MinimumSamplingInterval = obj.minsample
309			if obj.dimensions:
310			attrs.ArrayDimensions = obj.dimensions
311			node.NodeAttributes = attrs
312			res = self._add_node(node)
313			self._add_refs(obj)
314			res[0].StatusCode.check()
315			return res[0].AddedNodeId
316
317		View Code Duplication	def add_reference_type(self, obj):
			0 ignored issues – show Duplication introduced 2016-12-29 19:44 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
318			node = self._get_node(obj)
319			attrs = ua.ReferenceTypeAttributes()
320			if obj.desc:
321			attrs.Description = ua.LocalizedText(obj.desc)
322			attrs.DisplayName = ua.LocalizedText(obj.displayname)
323	1		if obj. inversename:
324	1		attrs.InverseName = ua.LocalizedText(obj.inversename)
325	1		if obj.abstract:
326	1		attrs.IsAbstract = obj.abstract
327	1		if obj.symmetric:
328	1		attrs.Symmetric = obj.symmetric
329	1		node.NodeAttributes = attrs
330			res = self._add_node(node)
331	1		self._add_refs(obj)
332	1		res[0].StatusCode.check()
333	1		return res[0].AddedNodeId
334	1
335	1	View Code Duplication	def add_datatype(self, obj):
			0 ignored issues – show Duplication introduced 2016-10-15 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
336			node = self._get_node(obj)
337	1		attrs = ua.DataTypeAttributes()
338	1		if obj.desc:
339	1		attrs.Description = ua.LocalizedText(obj.desc)
340	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
341	1		if obj.abstract:
342	1		attrs.IsAbstract = obj.abstract
343	1		node.NodeAttributes = attrs
344	1		res = self._add_node(node)
345	1		self._add_refs(obj)
346	1		res[0].StatusCode.check()
347	1		return res[0].AddedNodeId
348	1
349	1		def _add_refs(self, obj):
350			if not obj.refs:
351	1		return
352			refs = []
353			for data in obj.refs:
354			ref = ua.AddReferencesItem()
355			ref.IsForward = True
356			ref.ReferenceTypeId = self.to_nodeid(data.reftype)
357			ref.SourceNodeId = self._migrate_ns(obj.nodeid)
358			ref.TargetNodeClass = ua.NodeClass.DataType
359	1		ref.TargetNodeId = self._migrate_ns(self.to_nodeid(data.target))
360			refs.append(ref)
361	1		self._add_references(refs)
362
363	1		def _sort_nodes_by_parentid(self, ndatas):
364	1		"""
365			Sort the list of nodes according their parent node in order to respect
366			the dependency between nodes.
367
368			:param nodes: list of NodeDataObjects
369			:returns: list of sorted nodes
370	1		"""
371	1		_ndatas = list(ndatas)
372	1		# list of node ids that are already sorted / inserted
373			sorted_nodes_ids = []
374			# list of sorted nodes (i.e. XML Elements)
375			sorted_ndatas = []
376	1		all_node_ids = [data.nodeid for data in ndatas]
377			# list of namespace indexes that are relevant for this import
378			# we can only respect ordering nodes for namespaces indexes that
379	1		# are defined in the xml file itself. Thus we assume that all other
380	1		# references namespaces are already known to the server and should
381	1		# not create any dependency problems (like "NodeNotFound")
382			while len(_ndatas) > 0:
383			pop_nodes = []
384			for ndata in _ndatas:
385	1		# Insert nodes that
386	1		# (1) have no parent / parent_ns is None (e.g. namespace 0)
387	1		# (2) ns is not in list of relevant namespaces
388	1		if ndata.nodeid.NamespaceIndex not in self.namespaces or \
389			ndata.parent is None or \
390	1		ndata.parent not in all_node_ids:
391	1		sorted_ndatas.append(ndata)
392	1		sorted_nodes_ids.append(ndata.nodeid)
393			pop_nodes.append(ndata)
394			else:
395			# Check if the nodes parent is already in the list of
396			# inserted nodes
397			if ndata.parent in sorted_nodes_ids:
398			sorted_ndatas.append(ndata)
399			sorted_nodes_ids.append(ndata.nodeid)
400			pop_nodes.append(ndata)
401			# Remove inserted nodes from the list
402			for ndata in pop_nodes:
403			_ndatas.pop(_ndatas.index(ndata))
404			return sorted_ndatas
405

FreeOpcUa / python-opcua

Pull Request — master (#389)

XmlImporter._add_node() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like