XMLParser._get_text() - Code Metrics - Inspection of "[xmlimporter] Support EnumValues (#240)" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 331382...9398d4 )

by Olivier

created 2016-07-11 15:54 UTC

XMLParser._get_text() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
dl	0
loc	5
ccs	5
cts	5
cp	1
crap	2
rs	9.4285
c	0
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):
        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

    def __iter__(self):
        self.it = iter(self.root)
        return self

    def __next__(self):
        while True:
            if sys.version_info[0] < 3:
                child = self.it.next()
            else:
                child = self.it.__next__()
            name = self._retag.match(child.tag).groups()[1]
            if name == "Aliases":
                for el in child:
                    self.aliases[el.attrib["Alias"]] = el.text
            else:
                node = self._parse_node(name, child)
                return node

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

    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 = val
        elif key == "BrowseName":
            obj.browsename = val
        elif key == "SymbolicName":
            obj.symname = val
        elif key == "ParentNodeId":
            obj.parent = val
        elif key == "DataType":
            obj.datatype = val
        elif key == "IsAbstract":
            obj.abstract = val
        elif key == "EventNotifier":
            print("Notfiier", key, val)
            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 = 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 = ref.text
                obj.parentlink = ref.attrib["ReferenceType"]
            else:
                struct = RefStruct()
                if "IsForward" in ref.attrib:
                    struct.forward = ref.attrib["IsForward"]
                struct.target = 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):
57	1	self.logger = logging.getLogger(__name__)
58	1	self._retag = re.compile(r"(\{.\})(.)")
59	1	self.path = xmlpath
60	1	self.aliases = {}
61
62	1	self.tree = ET.parse(xmlpath)
63	1	self.root = self.tree.getroot()
64	1	self.it = None
65
66	1	def __iter__(self):
67	1	self.it = iter(self.root)
68	1	return self
69
70	1	def __next__(self):
71	1	while True:
72	1	if sys.version_info[0] < 3:
73		child = self.it.next()
74		else:
75	1	child = self.it.__next__()
76	1	name = self._retag.match(child.tag).groups()[1]
77	1	if name == "Aliases":
78		for el in child:
79		self.aliases[el.attrib["Alias"]] = el.text
80		else:
81	1	node = self._parse_node(name, child)
82	1	return node
83
84	1	def next(self): # support for python2
85		return self.__next__()
86
87	1	def _parse_node(self, name, child):
88	1	obj = NodeData()
89	1	obj.nodetype = name
90	1	for key, val in child.attrib.items():
91	1	self._set_attr(key, val, obj)
92	1	obj.displayname = obj.browsename # give a default value to display name
93	1	for el in child:
94	1	self._parse_tag(el, obj)
95	1	return obj
96
97	1	def _set_attr(self, key, val, obj):
98	1	if key == "NodeId":
99	1	obj.nodeid = val
100	1	elif key == "BrowseName":
101	1	obj.browsename = val
102	1	elif key == "SymbolicName":
103		obj.symname = val
104	1	elif key == "ParentNodeId":
105		obj.parent = val
106	1	elif key == "DataType":
107	1	obj.datatype = val
108		elif key == "IsAbstract":
109		obj.abstract = val
110		elif key == "EventNotifier":
111		print("Notfiier", key, val)
112		obj.eventnotifier = 1 if val == "1" else 0
113		elif key == "ValueRank":
114		obj.rank = int(val)
115		elif key == "ArrayDimensions":
116		obj.dimensions = [int(i) for i in val.split(",")]
117		elif key == "MinimumSamplingInterval":
118		obj.minsample = int(val)
119		elif key == "AccessLevel":
120		obj.accesslevel = int(val)
121		elif key == "UserAccessLevel":
122		obj.useraccesslevel = int(val)
123		elif key == "Symmetric":
124		obj.symmetric = True if val == "true" else False
125		else:
126		self.logger.info("Attribute not implemented: %s:%s", key, val)
127
128	1	def _parse_tag(self, el, obj):
129	1	tag = self._retag.match(el.tag).groups()[1]
130
131	1	if tag == "DisplayName":
132		obj.displayname = el.text
133	1	elif tag == "Description":
134	1	obj.desc = el.text
135	1	elif tag == "References":
136	1	self._parse_refs(el, obj)
137	1	elif tag == "Value":
138	1	self._parse_value(el, obj)
139		elif tag == "InverseName":
140		obj.inversename = el.text
141		elif tag == "Definition":
142		for field in el:
143		obj.definition.append(field)
144		else:
145		self.logger.info("Not implemented tag: %s", el)
146
147	1	def _parse_value(self, el, obj):
148	1	for val in el:
149	1	ntag = self._retag.match(val.tag).groups()[1]
150	1	obj.valuetype = ntag
151	1	if ntag in ("Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64"):
152	1	obj.value = int(val.text)
153	1	elif ntag in ("Float", "Double"):
154		obj.value = float(val.text)
155	1	elif ntag in ("Boolean"):
156		if val.text in ("True", "true", "1", "on", "On"):
157		obj.value = bool(1)
158		else:
159		obj.value = bool(0)
160	1	elif ntag in ("ByteString", "String"):
161	1	mytext = val.text
162	1	if mytext is None: # support importing null strings
163		mytext = ""
164	1	mytext = mytext.replace('\n', '').replace('\r', '')
165		# obj.value.append('b"{}"'.format(mytext))
166	1	obj.value = mytext
167		elif ntag == "ListOfExtensionObject":
168		obj.value, obj.valuetype = self._parse_list_of_extension_object(el)
169		elif ntag == "ListOfLocalizedText":
170		obj.value = self._parse_list_of_localized_text(el)
171		else:
172		self.logger.info("Value type not implemented: %s", ntag)
173
174	1	def _get_text(self, el):
175	1	txt = ""
176	1	for text in el.itertext():
177	1	txt += text
178	1	return txt
179
180		def _parse_list_of_localized_text(self, el):
181	1	value = []
182	1	for localized_text_list in el:
183		for localized_text in localized_text_list:
184		ntag = self._retag.match(localized_text.tag).groups()[1]
185		for child in localized_text:
186		ntag = self._retag.match(child.tag).groups()[1]
187		if ntag == 'Text':
188		value.append(self._get_text(child))
189		return value
190
191		def _parse_list_of_extension_object(self, el):
192		'''
193		Parse a uax:ListOfExtensionObject Value
194
195		Return an array with a value of each uax:ExtensionObject// (each element is convert to a netry in a dict.
196		also the valuetype is returned. The valuetype is uax:ExtensionObject/*/tag()
197		'''
198		value = []
199		valuetype = None
200		for extension_object_list in el:
201		for extension_object in extension_object_list:
202		extension_object.find('Body')
203		for extension_object_part in extension_object:
204		ntag = self._retag.match(extension_object_part.tag).groups()[1]
205		if ntag == 'Body':
206		data = {}
207		ntag = self._retag.match(extension_object_part.find('*').tag).groups()[1]
208		valuetype = ntag
209		for body_item in extension_object_part.findall('/'):
210		ntag = self._retag.match(body_item.tag).groups()[1]
211
212		child = body_item.find('*')
213		if child is not None:
214		data[ntag] = self._get_text(child)
215		else:
216		data[ntag] = self._get_text(body_item)
217		value.append(data)
218		return value, valuetype
219
220		def _parse_refs(self, el, obj):
221		for ref in el:
222		if ref.attrib["ReferenceType"] == "HasTypeDefinition":
223		obj.typedef = ref.text
224		elif "IsForward" in ref.attrib and ref.attrib["IsForward"] == "false":
225		# if obj.parent:
226		# sys.stderr.write("Parent is already set with: "+ obj.parent + " " + ref.text + "\n")
227		obj.parent = ref.text
228		obj.parentlink = ref.attrib["ReferenceType"]
229		else:
230		struct = RefStruct()
231		if "IsForward" in ref.attrib:
232		struct.forward = ref.attrib["IsForward"]
233		struct.target = ref.text
234		struct.reftype = ref.attrib["ReferenceType"]
235		obj.refs.append(struct)
236

FreeOpcUa / python-opcua

Push — master ( 331382...9398d4 )

XMLParser._get_text() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like