Completed
Pull Request — master (#322)
by
unknown
04:46
created

XMLParser._parse_bname()   A

Complexity

Conditions 4

Size

Total Lines 11

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 20

Importance

Changes 0
Metric Value
cc 4
dl 0
loc 11
ccs 0
cts 0
cp 0
crap 20
rs 9.2
c 0
b 0
f 0
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_bname(self, bname):
207
        """
208
        Parse a browsename and correct the namespace index.
209
        """
210
        if bname.find(':') != -1:
211
            browse_ns, browse_name = bname.split(':')
212
            if browse_ns:
213
                ns_server = self.namespaces.get(int(browse_ns), None)
214
                if ns_server:
215
                    return '%d:%s' % (ns_server[0], browse_name)
216
        return bname
217
218
    def _parse_node(self, name, child):
219
        """
220
        Parse a XML node and create a NodeData object.
221
        """
222
        obj = NodeData()
223
        obj.nodetype = name
224
        for key, val in child.attrib.items():
225
            self._set_attr(key, val, obj)
226
        self.logger.info("\n     Parsing node: %s %s", obj.nodeid, obj.browsename)
227
        obj.displayname = obj.browsename  # give a default value to display name
228
        for el in child:
229
            self._parse_tag(el, obj)
230
        return obj
231
232
    def _set_attr(self, key, val, obj):
233
        if key == "NodeId":
234
            obj.nodeid = self._get_node_id(val)
235
            print("PARSING", obj.nodeid)
236
        elif key == "BrowseName":
237
            obj.browsename = self._parse_bname(val)
238
        elif key == "SymbolicName":
239
            obj.symname = val
240
        elif key == "ParentNodeId":
241
            obj.parent = self._get_node_id(val)
242
        elif key == "DataType":
243
            obj.datatype = val
244
        elif key == "IsAbstract":
245
            obj.abstract = _to_bool(val)
246
        elif key == "Executable":
247
            obj.executable = _to_bool(val)
248
        elif key == "EventNotifier":
249
            obj.eventnotifier = int(val)
250
        elif key == "ValueRank":
251
            obj.rank = int(val)
252
        elif key == "ArrayDimensions":
253
            obj.dimensions = [int(i) for i in val.split(",")]
254
        elif key == "MinimumSamplingInterval":
255
            obj.minsample = int(val)
256
        elif key == "AccessLevel":
257
            obj.accesslevel = int(val)
258
        elif key == "UserAccessLevel":
259
            obj.useraccesslevel = int(val)
260
        elif key == "Symmetric":
261
            obj.symmetric = _to_bool(val)
262
        else:
263
            self.logger.info("Attribute not implemented: %s:%s", key, val)
264
265
    def _parse_tag(self, el, obj):
266
        tag = self._retag.match(el.tag).groups()[1]
267
268
        if tag == "DisplayName":
269
            obj.displayname = el.text
270
        elif tag == "Description":
271
            obj.desc = el.text
272
        elif tag == "References":
273
            self._parse_refs(el, obj)
274
        elif tag == "Value":
275
            self._parse_value(el, obj)
276
        elif tag == "InverseName":
277
            obj.inversename = el.text
278
        elif tag == "Definition":
279
            for field in el:
280
                obj.definition.append(field)
281
        else:
282
            self.logger.info("Not implemented tag: %s", el)
283
284
    def _parse_value(self, el, obj):
285
        self.logger.info("Parsing value")
286
        for val in el:
287
            self.logger.info("tag %s", val.tag)
288
            ntag = self._retag.match(val.tag).groups()[1]
289
            obj.valuetype = ntag
290
            if ntag in ("Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64"):
291
                obj.value = int(val.text)
292
            elif ntag in ("Float", "Double"):
293
                obj.value = float(val.text)
294
            elif ntag in ("Boolean"):
295
                obj.value = _to_bool(val.text)
296
            elif ntag in ("ByteString", "String"):
297
                mytext = val.text
298
                if mytext is None:  # support importing null strings
299
                    mytext = ""
300
                mytext = mytext.replace('\n', '').replace('\r', '')
301
                # obj.value.append('b"{}"'.format(mytext))
302
                obj.value = mytext
303
            elif ntag in ("Guid"):
304
                self._parse_value(val, obj)
305
            elif ntag == "ListOfExtensionObject":
306
                obj.value = self._parse_list_of_extension_object(el)
307
            elif ntag == "ListOfLocalizedText":
308
                obj.value = self._parse_list_of_localized_text(el)
309
            else:
310
                self.logger.info("Value type not implemented: '%s', %s", ntag)
311
312
    def _get_text(self, el):
313
        txtlist = [txt.strip() for txt in el.itertext()]
314
        return "".join(txtlist)
315
316
    def _parse_list_of_localized_text(self, el):
317
        value = []
318
        for localized_text_list in el:
319
            for localized_text in localized_text_list:
320
                ntag = self._retag.match(localized_text.tag).groups()[1]
321
                for child in localized_text:
322
                    ntag = self._retag.match(child.tag).groups()[1]
323
                    if ntag == 'Text':
324
                        value.append(self._get_text(child))
325
        return value
326
327
    def _parse_list_of_extension_object(self, el):
328
        '''
329
        Parse a uax:ListOfExtensionObject Value
330
        Return an list of ExtObj
331
        '''
332
        value = []
333
        for extension_object_list in el:
334
            for extension_object in extension_object_list:
335
                ext_obj = self._parse_ext_obj(extension_object)
336
                value.append(ext_obj)
337
        return value
338
339
    def _parse_ext_obj(self, el):
340
        ext = ExtObj()
341
        for extension_object_part in el:
342
            ntag = self._retag.match(extension_object_part.tag).groups()[1]
343
            if ntag == 'TypeId':
344
                ntag = self._retag.match(extension_object_part.find('*').tag).groups()[1]
345
                ext.typeid = self._get_text(extension_object_part)
346
            elif ntag == 'Body':
347
                ext.objname = self._retag.match(extension_object_part.find('*').tag).groups()[1]
348
                ext.body = self._parse_body(extension_object_part)
349
                print("body", ext.body)
350
            else:
351
                print("Uknoen ndtag", ntag)
352
        print("ext_obj", ext.objname, ext.typeid, ext.body)
353
        return ext
354
355
    def _parse_body(self, el):
356
        body = {}
357
        for body_item in el:
358
            otag = self._retag.match(body_item.tag).groups()[1]
359
            childs = [i for i in body_item]
360
            if not childs:
361
                val = self._get_text(body_item)
362
            else:
363
                val = self._parse_body(body_item)
364
            if val:
365
                body[otag] = val
366
        return body
367
368
    def _parse_refs(self, el, obj):
369
        for ref in el:
370
            if ref.attrib["ReferenceType"] == "HasTypeDefinition":
371
                obj.typedef = self._get_node_id(ref.text)
372
            elif "IsForward" in ref.attrib and ref.attrib["IsForward"] in ("false", "False"):
373
                # if obj.parent:
374
                    # sys.stderr.write("Parent is already set with: "+ obj.parent + " " + ref.text + "\n")
375
                obj.parent = self._get_node_id(ref.text)
376
                obj.parentlink = ref.attrib["ReferenceType"]
377
            else:
378
                struct = RefStruct()
379
                if "IsForward" in ref.attrib:
380
                    struct.forward = ref.attrib["IsForward"]
381
                struct.target = self._get_node_id(ref.text)
382
                struct.reftype = ref.attrib["ReferenceType"]
383
                obj.refs.append(struct)
384