XMLParser.__next__() - Code Metrics - Inspection of "[Server] Import XML - Respect depency of nodes (Pa..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#267)

unknown

created 2016-08-03 06:59 UTC

XMLParser.next() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	3

Importance

Changes	1
Bugs	0	Features	0

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

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

import xml.etree.ElementTree as ET


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 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)

        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 in [0, -1]:
                    sorted_nodes.insert(insert, node)
                    sorted_nodes_ids.insert(insert, 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):
        obj = NodeData()
        obj.nodetype = name
        for key, val in child.attrib.items():
            self._set_attr(key, val, obj)
        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)
        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 = val
        elif key == "EventNotifier":
            obj.eventnotifier = 1 if val == "1" else 0
        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 = True if val == "true" else False
        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):
        for val in el:
            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"):
                if val.text in ("True", "true", "1", "on", "On"):
                    obj.value = bool(1)
                else:
                    obj.value = bool(0)
            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 == "ListOfExtensionObject":
                obj.value, obj.valuetype = self._parse_list_of_extension_object(el)
            elif ntag == "ListOfLocalizedText":
                obj.value = self._parse_list_of_localized_text(el)
            else:
                self.logger.info("Value type not implemented: %s", ntag)

    def _get_text(self, el):
        txt = ""
        for text in el.itertext():
            txt += text
        return txt

    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 array with a value of each uax:ExtensionObject/*/* (each element is convert to a netry in a dict.
               also the valuetype is returned. The valuetype is  uax:ExtensionObject/*/tag()
        '''
        value = []
        valuetype = None
        for extension_object_list in el:
            for extension_object in extension_object_list:
                extension_object.find('Body')
                for extension_object_part in extension_object:
                    ntag = self._retag.match(extension_object_part.tag).groups()[1]
                    if ntag == 'Body':
                        data = {}
                        ntag = self._retag.match(extension_object_part.find('*').tag).groups()[1]
                        valuetype = ntag
                        for body_item in extension_object_part.findall('*/*'):
                            ntag = self._retag.match(body_item.tag).groups()[1]

                            child = body_item.find('*')
                            if child is not None:
                                data[ntag] = self._get_text(child)
                            else:
                                data[ntag] = self._get_text(body_item)
                        value.append(data)
        return value, valuetype

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

FreeOpcUa / python-opcua

Pull Request — master (#267)

XMLParser.__next__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

XMLParser.next() A