XMLParser.__iter__() - Code Metrics - Inspection of "fix and tests for namespaces xml exporter" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#326)

by Olivier

created 2016-10-12 15:59 UTC

XMLParser.iter() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	18
CRAP Score	7.5034

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	7
c	2
b	0
f	0
dl	0
loc	24
ccs	18
cts	23
cp	0.7826
crap	7.5034
rs	5.5

"""
parse xml file from opcua-spec
"""
import logging
import re
import sys

import xml.etree.ElementTree as ET


def _to_bool(val):
    if val in ("True", "true", "on", "On", "1"):
        return True
    else:
        return False



class NodeData(object):

    def __init__(self):
        self.nodetype = None
        self.nodeid = None
        self.browsename = None
        self.displayname = None
        self.symname = None  # FIXME: this param is never used, why?
        self.parent = None
        self.parentlink = None
        self.desc = ""
        self.typedef = None
        self.refs = []
        self.nodeclass = None
        self.eventnotifier = 0

        # variable
        self.datatype = None
        self.rank = -1  # check default value
        self.value = None
        self.valuetype = None
        self.dimensions = None
        self.accesslevel = None
        self.useraccesslevel = None
        self.minsample = None

        # referencetype
        self.inversename = ""
        self.abstract = False 
        self.symmetric = False 

        # datatype
        self.definition = []


class RefStruct(object):

    def __init__(self):
        self.reftype = None
        self.forward = True
        self.target = None


class ExtObj(object):

    def __init__(self):
        self.typeid = None
        self.objname = None
        self.bodytype = None
        self.body = {}


class XMLParser(object):

    def __init__(self, xmlpath, server):
        self.server = server  # POC
        self.logger = logging.getLogger(__name__)
        self._retag = re.compile(r"(\{.*\})(.*)")
        self.path = xmlpath
        self.aliases = {}

        self.tree = ET.parse(xmlpath)
        self.root = self.tree.getroot()
        self.it = None

        self.namespaces = {}
        self._re_nodeid = re.compile(r"^ns=(?P<ns>\d+[^;]*);i=(?P<i>\d+)")

    def __iter__(self):
        nodes = []
        for child in self.root:
            name = self._retag.match(child.tag).groups()[1]
            if name == "Aliases":
                for el in child:
                    self.aliases[el.attrib["Alias"]] = self._get_node_id(el.text)
            elif name == 'NamespaceUris':
                for ns_index, ns_element in enumerate(child):
                    ns_uri = ns_element.text
                    ns_server_index = self.server.register_namespace(ns_uri)
                    self.namespaces[ns_index + 1] = (ns_server_index, ns_uri)
            else:
                node = self._parse_node(name, child)
                nodes.append(node)

        # 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!
        if self.namespaces:
            nodes = self._sort_nodes_by_parentid(nodes)

        self.it = iter(nodes)
        return self

    def __next__(self):
        while True:
            if sys.version_info[0] < 3:
                child = self.it.next()
            else:
                child = self.it.__next__()
            return child

    def next(self):  # support for python2
        return self.__next__()

    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(i[0]) for i in self.namespaces.values()]
        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 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

    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 _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 _parse_node(self, name, child):
        """
        Parse a XML node and create a NodeData object.
        """
        obj = NodeData()
        obj.nodetype = name
        for key, val in child.attrib.items():
            self._set_attr(key, val, obj)
        self.logger.info("\n     Parsing node: %s %s", obj.nodeid, obj.browsename)
        obj.displayname = obj.browsename  # give a default value to display name
        for el in child:
            self._parse_tag(el, obj)
        return obj

    def _set_attr(self, key, val, obj):
        if key == "NodeId":
            obj.nodeid = self._get_node_id(val)
            print("PARSING", obj.nodeid)
        elif key == "BrowseName":
            obj.browsename = val
        elif key == "SymbolicName":
            obj.symname = val
        elif key == "ParentNodeId":
            obj.parent = self._get_node_id(val)
        elif key == "DataType":
            obj.datatype = val
        elif key == "IsAbstract":
            obj.abstract = _to_bool(val)
        elif key == "Executable":
            obj.executable = _to_bool(val)
        elif key == "EventNotifier":
            obj.eventnotifier = int(val)
        elif key == "ValueRank":
            obj.rank = int(val)
        elif key == "ArrayDimensions":
            obj.dimensions = [int(i) for i in val.split(",")]
        elif key == "MinimumSamplingInterval":
            obj.minsample = int(val)
        elif key == "AccessLevel":
            obj.accesslevel = int(val)
        elif key == "UserAccessLevel":
            obj.useraccesslevel = int(val)
        elif key == "Symmetric":
            obj.symmetric = _to_bool(val)
        else:
            self.logger.info("Attribute not implemented: %s:%s", key, val)

    def _parse_tag(self, el, obj):
        tag = self._retag.match(el.tag).groups()[1]

        if tag == "DisplayName":
            obj.displayname = el.text
        elif tag == "Description":
            obj.desc = el.text
        elif tag == "References":
            self._parse_refs(el, obj)
        elif tag == "Value":
            self._parse_value(el, obj)
        elif tag == "InverseName":
            obj.inversename = el.text
        elif tag == "Definition":
            for field in el:
                obj.definition.append(field)
        else:
            self.logger.info("Not implemented tag: %s", el)

    def _parse_value(self, el, obj):
        self.logger.info("Parsing value")
        for val in el:
            self.logger.info("tag %s", val.tag)
            ntag = self._retag.match(val.tag).groups()[1]
            obj.valuetype = ntag
            if ntag in ("Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64"):
                obj.value = int(val.text)
            elif ntag in ("Float", "Double"):
                obj.value = float(val.text)
            elif ntag in ("Boolean"):
                obj.value = _to_bool(val.text)
            elif ntag in ("ByteString", "String"):
                mytext = val.text
                if mytext is None:  # support importing null strings
                    mytext = ""
                mytext = mytext.replace('\n', '').replace('\r', '')
                # obj.value.append('b"{}"'.format(mytext))
                obj.value = mytext
            elif ntag in ("Guid"):
                self._parse_value(val, obj)
            elif ntag == "ListOfExtensionObject":
                obj.value = self._parse_list_of_extension_object(el)
            elif ntag == "ListOfLocalizedText":
                obj.value = self._parse_list_of_localized_text(el)
            elif ntag == "ExtensionObject":
                obj.value = self._parse_extension_object(el)
            else:
                self.logger.warning("Value type not implemented: '%s'", ntag)

    def _get_text(self, el):
        txtlist = [txt.strip() for txt in el.itertext()]
        return "".join(txtlist)

    def _parse_list_of_localized_text(self, el):
        value = []
        for localized_text_list in el:
            for localized_text in localized_text_list:
                ntag = self._retag.match(localized_text.tag).groups()[1]
                for child in localized_text:
                    ntag = self._retag.match(child.tag).groups()[1]
                    if ntag == 'Text':
                        value.append(self._get_text(child))
        return value

    def _parse_list_of_extension_object(self, el):
        '''
        Parse a uax:ListOfExtensionObject Value
        Return an list of ExtObj
        '''
        value = []
        for extension_object_list in el:
            for extension_object in extension_object_list:
                ext_obj = self._parse_ext_obj(extension_object)
                value.append(ext_obj)
        return value

    def _parse_extension_object(self, el):
        for ext_obj in el:
            return self._parse_ext_obj(ext_obj)

    def _parse_ext_obj(self, el):
        ext = ExtObj()
        for extension_object_part in el:
            ntag = self._retag.match(extension_object_part.tag).groups()[1]
            if ntag == 'TypeId':
                ntag = self._retag.match(extension_object_part.find('*').tag).groups()[1]
                ext.typeid = self._get_text(extension_object_part)
            elif ntag == 'Body':
                ext.objname = self._retag.match(extension_object_part.find('*').tag).groups()[1]
                ext.body = self._parse_body(extension_object_part)
            else:
                print("Uknown ndtag", ntag)
        return ext

    def _parse_body(self, el):
        body = []
        for body_item in el:
            otag = self._retag.match(body_item.tag).groups()[1]
            childs = [i for i in body_item]
            if not childs:
                val = self._get_text(body_item)
            else:
                val = self._parse_body(body_item)
            if val:
                body.append((otag, val))
        return body

    def _parse_refs(self, el, obj):
        for ref in el:
            if ref.attrib["ReferenceType"] == "HasTypeDefinition":
                obj.typedef = self._get_node_id(ref.text)
            elif "IsForward" in ref.attrib and ref.attrib["IsForward"] in ("false", "False"):
                # if obj.parent:
                    # sys.stderr.write("Parent is already set with: "+ obj.parent + " " + ref.text + "\n")
                obj.parent = self._get_node_id(ref.text)
                obj.parentlink = ref.attrib["ReferenceType"]
            else:
                struct = RefStruct()
                if "IsForward" in ref.attrib:
                    struct.forward = ref.attrib["IsForward"]
                struct.target = self._get_node_id(ref.text)
                struct.reftype = ref.attrib["ReferenceType"]
                obj.refs.append(struct)


1		"""
2		parse xml file from opcua-spec
3		"""
4	1	import logging
5	1	import re
6	1	import sys
7
8	1	import xml.etree.ElementTree as ET
9
10
11	1	def _to_bool(val):
12		if val in ("True", "true", "on", "On", "1"):
13	1	return True
14	1	else:
15	1	return False
16	1
17	1
18	1
19	1	class NodeData(object):
20	1
21	1	def __init__(self):
22	1	self.nodetype = None
23	1	self.nodeid = None
24	1	self.browsename = None
25	1	self.displayname = None
26		self.symname = None # FIXME: this param is never used, why?
27		self.parent = None
28	1	self.parentlink = None
29	1	self.desc = ""
30	1	self.typedef = None
31	1	self.refs = []
32	1	self.nodeclass = None
33	1	self.eventnotifier = 0
34	1
35	1	# variable
36		self.datatype = None
37		self.rank = -1 # check default value
38	1	self.value = None
39	1	self.valuetype = None
40	1	self.dimensions = None
41		self.accesslevel = None
42		self.useraccesslevel = None
43	1	self.minsample = None
44
45		# referencetype
46	1	self.inversename = ""
47		self.abstract = False
48	1	self.symmetric = False
49
50		# datatype
51		self.definition = []
52
53
54	1	class RefStruct(object):
55
56	1	def __init__(self):
57	1	self.reftype = None
58	1	self.forward = True
59	1	self.target = None
60	1
61
62	1	class ExtObj(object):
63	1
64	1	def __init__(self):
65		self.typeid = None
66	1	self.objname = None
67	1	self.bodytype = None
68	1	self.body = {}
69
70	1
71	1	class XMLParser(object):
72	1
73		def __init__(self, xmlpath, server):
74		self.server = server # POC
75	1	self.logger = logging.getLogger(__name__)
76	1	self._retag = re.compile(r"(\{.\})(.)")
77	1	self.path = xmlpath
78		self.aliases = {}
79
80		self.tree = ET.parse(xmlpath)
81	1	self.root = self.tree.getroot()
82	1	self.it = None
83
84	1	self.namespaces = {}
85		self._re_nodeid = re.compile(r"^ns=(?P<ns>\d+[^;]*);i=(?P<i>\d+)")
86
87	1	def __iter__(self):
88	1	nodes = []
89	1	for child in self.root:
90	1	name = self._retag.match(child.tag).groups()[1]
91	1	if name == "Aliases":
92	1	for el in child:
93	1	self.aliases[el.attrib["Alias"]] = self._get_node_id(el.text)
94	1	elif name == 'NamespaceUris':
95	1	for ns_index, ns_element in enumerate(child):
96		ns_uri = ns_element.text
97	1	ns_server_index = self.server.register_namespace(ns_uri)
98	1	self.namespaces[ns_index + 1] = (ns_server_index, ns_uri)
99	1	else:
100	1	node = self._parse_node(name, child)
101	1	nodes.append(node)
102	1
103		# The ordering of nodes currently only works if namespaces are
104	1	# defined in XML.
105		# Also, it is recommended not to use node ids without namespace prefix!
106	1	if self.namespaces:
107	1	nodes = self._sort_nodes_by_parentid(nodes)
108
109		self.it = iter(nodes)
110		return self
111
112		def __next__(self):
113		while True:
114		if sys.version_info[0] < 3:
115		child = self.it.next()
116		else:
117		child = self.it.__next__()
118		return child
119
120		def next(self): # support for python2
121		return self.__next__()
122
123		def _sort_nodes_by_parentid(self, nodes):
124		"""
125		Sort the list of nodes according theire parent node in order to respect
126		the depency between nodes.
127
128	1	:param nodes: list of NodeDataObjects
129	1	:returns: list of sorted nodes
130		"""
131	1	_nodes = list(nodes)
132		# list of node ids that are already sorted / inserted
133	1	sorted_nodes_ids = []
134	1	# list of sorted nodes (i.e. XML Elements)
135	1	sorted_nodes = []
136	1	# list of namespace indexes that are relevant for this import
137	1	# we can only respect ordering nodes for namespaces indexes that
138	1	# are defined in the xml file itself. Thus we assume that all other
139		# references namespaces are already known to the server and should
140		# not create any dependency problems (like "NodeNotFound")
141		relevant_namespaces = [str(i[0]) for i in self.namespaces.values()]
142		while len(_nodes) > 0:
143		pop_nodes = []
144		for node in _nodes:
145		insert = None
146		# Get the node and parent node namespace and id parts
147	1	node_ns, node_id = self._split_node_id(node.nodeid)
148	1	parent_ns, parent_id = self._split_node_id(node.parent)
149	1
150	1	# Insert nodes that
151	1	# (1) have no parent / parent_ns is None (e.g. namespace 0)
152	1	# (2) ns is not in list of relevant namespaces
153	1	if (parent_ns is None or node_ns not in relevant_namespaces or
154		parent_id is None):
155	1	insert = 0
156		else:
157		# Check if the nodes parent is already in the list of
158		# inserted nodes
159		if node.parent in sorted_nodes_ids:
160	1	insert = -1
161	1
162	1	if insert == 0:
163		sorted_nodes.insert(insert, node)
164	1	sorted_nodes_ids.insert(insert, node.nodeid)
165		pop_nodes.append(node)
166	1	elif insert == -1:
167		sorted_nodes.append(node)
168		sorted_nodes_ids.append(node.nodeid)
169		pop_nodes.append(node)
170
171		# Remove inserted nodes from the list
172		for node in pop_nodes:
173		_nodes.pop(_nodes.index(node))
174	1	return sorted_nodes
175	1
176	1	def _split_node_id(self, value):
177	1	"""
178	1	Split the fq node id into namespace and id part.
179
180		:returns: (namespace, id)
181	1	"""
182	1	if not value:
183		return (None, value)
184		r_match = self._re_nodeid.search(value)
185		if r_match:
186		return r_match.groups()
187
188		return (None, value)
189
190		def _get_node_id(self, value):
191		"""
192		Check if the nodeid given in the xml model file must be converted
193		to a already existing namespace id based on the files namespace uri
194
195		:returns: NodeId (str)
196		"""
197		result = value
198
199		node_ns, node_id = self._split_node_id(value)
200		if node_ns:
201		ns_server = self.namespaces.get(int(node_ns), None)
202		if ns_server:
203		result = "ns={};i={}".format(ns_server[0], node_id)
204		return result
205
206		def _parse_node(self, name, child):
207		"""
208		Parse a XML node and create a NodeData object.
209		"""
210		obj = NodeData()
211		obj.nodetype = name
212		for key, val in child.attrib.items():
213		self._set_attr(key, val, obj)
214		self.logger.info("\n Parsing node: %s %s", obj.nodeid, obj.browsename)
215		obj.displayname = obj.browsename # give a default value to display name
216		for el in child:
217		self._parse_tag(el, obj)
218		return obj
219
220		def _set_attr(self, key, val, obj):
221		if key == "NodeId":
222		obj.nodeid = self._get_node_id(val)
223		print("PARSING", obj.nodeid)
224		elif key == "BrowseName":
225		obj.browsename = val
226		elif key == "SymbolicName":
227		obj.symname = val
228		elif key == "ParentNodeId":
229		obj.parent = self._get_node_id(val)
230		elif key == "DataType":
231		obj.datatype = val
232		elif key == "IsAbstract":
233		obj.abstract = _to_bool(val)
234		elif key == "Executable":
235		obj.executable = _to_bool(val)
236		elif key == "EventNotifier":
237		obj.eventnotifier = int(val)
238		elif key == "ValueRank":
239		obj.rank = int(val)
240		elif key == "ArrayDimensions":
241		obj.dimensions = [int(i) for i in val.split(",")]
242		elif key == "MinimumSamplingInterval":
243		obj.minsample = int(val)
244		elif key == "AccessLevel":
245		obj.accesslevel = int(val)
246		elif key == "UserAccessLevel":
247		obj.useraccesslevel = int(val)
248		elif key == "Symmetric":
249		obj.symmetric = _to_bool(val)
250		else:
251		self.logger.info("Attribute not implemented: %s:%s", key, val)
252
253		def _parse_tag(self, el, obj):
254		tag = self._retag.match(el.tag).groups()[1]
255
256		if tag == "DisplayName":
257		obj.displayname = el.text
258		elif tag == "Description":
259		obj.desc = el.text
260		elif tag == "References":
261		self._parse_refs(el, obj)
262		elif tag == "Value":
263		self._parse_value(el, obj)
264		elif tag == "InverseName":
265		obj.inversename = el.text
266		elif tag == "Definition":
267		for field in el:
268		obj.definition.append(field)
269		else:
270		self.logger.info("Not implemented tag: %s", el)
271
272		def _parse_value(self, el, obj):
273		self.logger.info("Parsing value")
274		for val in el:
275		self.logger.info("tag %s", val.tag)
276		ntag = self._retag.match(val.tag).groups()[1]
277		obj.valuetype = ntag
278		if ntag in ("Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64"):
279		obj.value = int(val.text)
280		elif ntag in ("Float", "Double"):
281		obj.value = float(val.text)
282		elif ntag in ("Boolean"):
283		obj.value = _to_bool(val.text)
284		elif ntag in ("ByteString", "String"):
285		mytext = val.text
286		if mytext is None: # support importing null strings
287		mytext = ""
288		mytext = mytext.replace('\n', '').replace('\r', '')
289		# obj.value.append('b"{}"'.format(mytext))
290		obj.value = mytext
291		elif ntag in ("Guid"):
292		self._parse_value(val, obj)
293		elif ntag == "ListOfExtensionObject":
294		obj.value = self._parse_list_of_extension_object(el)
295		elif ntag == "ListOfLocalizedText":
296		obj.value = self._parse_list_of_localized_text(el)
297		elif ntag == "ExtensionObject":
298		obj.value = self._parse_extension_object(el)
299		else:
300		self.logger.warning("Value type not implemented: '%s'", ntag)
301
302		def _get_text(self, el):
303		txtlist = [txt.strip() for txt in el.itertext()]
304		return "".join(txtlist)
305
306		def _parse_list_of_localized_text(self, el):
307		value = []
308		for localized_text_list in el:
309		for localized_text in localized_text_list:
310		ntag = self._retag.match(localized_text.tag).groups()[1]
311		for child in localized_text:
312		ntag = self._retag.match(child.tag).groups()[1]
313		if ntag == 'Text':
314		value.append(self._get_text(child))
315		return value
316
317		def _parse_list_of_extension_object(self, el):
318		'''
319		Parse a uax:ListOfExtensionObject Value
320		Return an list of ExtObj
321		'''
322		value = []
323		for extension_object_list in el:
324		for extension_object in extension_object_list:
325		ext_obj = self._parse_ext_obj(extension_object)
326		value.append(ext_obj)
327		return value
328
329		def _parse_extension_object(self, el):
330		for ext_obj in el:
331		return self._parse_ext_obj(ext_obj)
332
333		def _parse_ext_obj(self, el):
334		ext = ExtObj()
335		for extension_object_part in el:
336		ntag = self._retag.match(extension_object_part.tag).groups()[1]
337		if ntag == 'TypeId':
338		ntag = self._retag.match(extension_object_part.find('*').tag).groups()[1]
339		ext.typeid = self._get_text(extension_object_part)
340		elif ntag == 'Body':
341		ext.objname = self._retag.match(extension_object_part.find('*').tag).groups()[1]
342		ext.body = self._parse_body(extension_object_part)
343		else:
344		print("Uknown ndtag", ntag)
345		return ext
346
347		def _parse_body(self, el):
348		body = []
349		for body_item in el:
350		otag = self._retag.match(body_item.tag).groups()[1]
351		childs = [i for i in body_item]
352		if not childs:
353		val = self._get_text(body_item)
354		else:
355		val = self._parse_body(body_item)
356		if val:
357		body.append((otag, val))
358		return body
359
360		def _parse_refs(self, el, obj):
361		for ref in el:
362		if ref.attrib["ReferenceType"] == "HasTypeDefinition":
363		obj.typedef = self._get_node_id(ref.text)
364		elif "IsForward" in ref.attrib and ref.attrib["IsForward"] in ("false", "False"):
365		# if obj.parent:
366		# sys.stderr.write("Parent is already set with: "+ obj.parent + " " + ref.text + "\n")
367		obj.parent = self._get_node_id(ref.text)
368		obj.parentlink = ref.attrib["ReferenceType"]
369		else:
370		struct = RefStruct()
371		if "IsForward" in ref.attrib:
372		struct.forward = ref.attrib["IsForward"]
373		struct.target = self._get_node_id(ref.text)
374		struct.reftype = ref.attrib["ReferenceType"]
375		obj.refs.append(struct)
376

FreeOpcUa / python-opcua

Pull Request — master (#326)

XMLParser.__iter__() C

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

XMLParser.iter() C