XmlImporter.to_nodeid() - Code Metrics - Inspection of "Guid primitive DO NOT MERGE" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#338)

unknown

created 2016-10-19 19:45 UTC

XmlImporter.to_nodeid() D

↳ Parent: XmlImporter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	1
Ratio	7.69 %

Code Coverage

Tests	1
CRAP Score	24.2584

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
dl	1
loc	13
ccs	1
cts	12
cp	0.0833
crap	24.2584
rs	4.9473
c	1
b	0
f	0

"""G
add node defined in XML to address space
format is the one from opc-ua specification
"""
import logging
import sys
import re
import uuid
import dateutil.parser

from opcua import ua
from opcua.common import xmlparser


def ua_type_to_python(val, uatype):
    if uatype.startswith("Int") or uatype.startswith("UInt"):
        return int(val)
    elif uatype in ("Double", "Float"):
        return float(val)
    elif uatype in ("String"):
        return val
    elif uatype in ("Bytes", "Bytes", "ByteString", "ByteArray"):
        if sys.version_info.major > 2:
            return bytes(val, 'utf8')
        else:
            return val
    else:
        raise Exception("uatype nopt handled", uatype, " for val ", val)


def to_python(val, obj, attname):
    if isinstance(obj, ua.NodeId) and attname == "Identifier":
        raise RuntimeError("Error we should parse a NodeId here")
        return ua.NodeId.from_string(val)
    else:
        return ua_type_to_python(val, obj.ua_types[attname])


class XmlImporter(object):

    def __init__(self, server):
        self.logger = logging.getLogger(__name__)
        self.parser = None
        self.server = server
        self.namespaces = {}
        self.aliases = {}
        self._re_nodeid = re.compile(r"^ns=(?P<ns>\d+[^;]*);i=(?P<i>\d+)")

    def _map_namespaces(self, namespaces_uris, act_server):
        """
        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 = act_server.register_namespace(ns_uri)
            namespaces[ns_index + 1] = (ns_server_index, ns_uri)
            self.logger.info("namespace offset", ns_index + 1, (ns_server_index, ns_uri))
        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._get_node_id(node_id)
        return aliases_mapped

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

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

        # The ordering of nodes currently only works if namespaces are
        # defined in XML.
        # Also, it is recommended not to use node ids without namespace prefix!
        nodes_parsed = self._sort_nodes_by_parentid(self.parser)

        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 _split_node_id(self, value):
        """
        Split the fq node id into namespace and id part.

        :returns: (namespace, id)
        """
        if not value:
            return (None, value)
        r_match = self._re_nodeid.search(value)
        if r_match:
            return r_match.groups()

        return (None, value)

    def _parse_bname(self, bname):
        """
        Parse a browsename and correct the namespace index.
        """
        if bname.find(':') != -1:
            browse_ns, browse_name = bname.split(':')
            if browse_ns:
                ns_server = self.namespaces.get(int(browse_ns), None)
                if ns_server:
                    return '%d:%s' % (ns_server[0], browse_name)
        return bname

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

        :returns: NodeId (str)
        """
        result = value

        node_ns, node_id = self._split_node_id(value)
        if node_ns:
            ns_server = self.namespaces.get(int(node_ns), None)
            if ns_server:
                result = "ns={};i={}".format(ns_server[0], node_id)
        return result

    def _get_node(self, obj):
        node = ua.AddNodesItem()
        node.RequestedNewNodeId = ua.NodeId.from_string(self._get_node_id(obj.nodeid))
        node.BrowseName = ua.QualifiedName.from_string(self._parse_bname(obj.browsename))
        node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
        if obj.parent:
            node.ParentNodeId = ua.NodeId.from_string(self._get_node_id(obj.parent))
        if obj.parentlink:
            node.ReferenceTypeId = self.to_nodeid(self._get_node_id(obj.parentlink))
        if obj.typedef:
            node.TypeDefinition = ua.NodeId.from_string(self._get_node_id(obj.typedef))
        return node

    def to_nodeid(self, nodeid):
        if 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:
                nodeid = self.aliases[nodeid]
            else:
                nodeid = "i={}".format(getattr(ua.ObjectIds, nodeid))
            return ua.NodeId.from_string(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.server.add_nodes([node])
        self._add_refs(obj)
        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.server.add_nodes([node])
        self._add_refs(obj)
        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 and len(obj.value) == 1:
        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.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def _make_ext_obj(sefl, obj):
        ext = getattr(ua, obj.objname)()
        for name, val in obj.body:
            if type(val) is str:
                raise Exception("Error val should a dict", name, val)
            else:
                for attname, v in val:
                    # tow possible values:
                    # either we get value directly
                    # or a dict if it s an object or a list
                    if type(v) is str:
                        setattr(ext, attname, to_python(v, ext, attname))
                    else:
                        # so we habve either an object or a list...
                        obj2 = getattr(ext, attname)
                        if isinstance(obj2, ua.NodeId):
                            for attname2, v2 in v:
                                if attname2 == "Identifier":
                                    obj2 = ua.NodeId.from_string(v2)
                                    setattr(ext, 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 v:
                                my_list.append(ua_type_to_python(v2, vtype))
                            setattr(ext, attname, my_list)
                        else:
                            for attname2, v2 in v:
                                setattr(obj2, attname2, to_python(v2, obj2, attname2))
                            setattr(ext, attname, obj2)
        return ext

    def _add_variable_value(self, obj):
        """
        Returns the value for a Variable based on the objects valuetype. 
        """
        if obj.valuetype == 'ListOfExtensionObject':
            values = []
            for ext in obj.value:
                extobj = self._make_ext_obj(ext)
                values.append(extobj)
            return values
        elif obj.valuetype.startswith("ListOf"):
            vtype = obj.valuetype[6:]
            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 == 'DateTime':
            return ua.Variant(dateutil.parser.parse(obj.value), getattr(ua.VariantType, obj.valuetype))

        elif obj.valuetype == 'Guid':
            return ua.Variant(uuid.UUID(obj.value), getattr(ua.VariantType, obj.valuetype))
        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.server.add_nodes([node])
        self._add_refs(obj)
        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.server.add_nodes([node])
        self._add_refs(obj)
        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.server.add_nodes([node])
        self._add_refs(obj)
        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.server.add_nodes([node])
        self._add_refs(obj)
        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 = ua.NodeId.from_string(self._get_node_id(obj.nodeid))
            ref.TargetNodeClass = ua.NodeClass.DataType
            ref.TargetNodeId = ua.NodeId.from_string(self._get_node_id(data.target))
            refs.append(ref)
        self.server.add_references(refs)

    # FIX: wrong order of node sorting .. need to find out what is wrong
    def _sort_nodes_by_parentid(self, nodes):
        """
        Sort the list of nodes according theire parent node in order to respect
        the depency between nodes.

        :param nodes: list of NodeDataObjects
        :returns: list of sorted nodes
        """
        _nodes = list(nodes)
        # list of node ids that are already sorted / inserted
        sorted_nodes_ids = []
        # list of sorted nodes (i.e. XML Elements)
        sorted_nodes = []
        # 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")
        relevant_namespaces = [str(ns) for ns in self.namespaces.keys()]
        while len(_nodes) > 0:
            pop_nodes = []
            for node in _nodes:
                insert = None
                # Get the node and parent node namespace and id parts
                node_ns, node_id = self._split_node_id(node.nodeid)
                parent_ns, parent_id = self._split_node_id(node.parent)
                # 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 (parent_ns is None or
                    (len(relevant_namespaces) >= 1 and node_ns not in relevant_namespaces) or
                    parent_id is None):
                    insert = 0
                else:
                    # Check if the nodes parent is already in the list of
                    # inserted nodes
                    if node.parent in sorted_nodes_ids:
                        insert = -1
                if insert == 0:
                    sorted_nodes.insert(insert, node)
                    sorted_nodes_ids.insert(insert, node.nodeid)
                    pop_nodes.append(node)
                elif insert == -1:
                    sorted_nodes.append(node)
                    sorted_nodes_ids.append(node.nodeid)
                    pop_nodes.append(node)

            # Remove inserted nodes from the list
            for node in pop_nodes:
                _nodes.pop(_nodes.index(node))
        return sorted_nodes


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

FreeOpcUa / python-opcua

Pull Request — master (#338)

XmlImporter.to_nodeid() D

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like