Completed
Pull Request — master (#323)
by Olivier
03:43
created

XMLParser._parse_extension_object()   A

Complexity

Conditions 2

Size

Total Lines 3

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 6

Importance

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