XmlExporter.add_etree_method() - Code Metrics - Inspection of "Merge pull request #303 from Pro/hotfix/di_schema" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 6c18d6...f8fb14 )

by Olivier

created 2016-09-22 10:30 UTC

XmlExporter.add_etree_method() A

↳ Parent: XmlExporter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	1	Features	1

Metric	Value
cc	3
dl	0
loc	10
rs	9.4285
c	4
b	1
f	1

"""
from a list of nodes in the address space, build an XML file
format is the one from opc-ua specification
"""
import logging
from collections import OrderedDict
import xml.etree.ElementTree as Et

from opcua import ua
from opcua.ua import object_ids as o_ids


class XmlExporter(object):

    def __init__(self, server):
        self.logger = logging.getLogger(__name__)
        self.server = server
        self.aliases = {}

        node_set_attributes = OrderedDict()
        node_set_attributes['xmlns:xsi'] = 'http://www.w3.org/2001/XMLSchema-instance'
        node_set_attributes['xmlns:uax'] = 'http://opcfoundation.org/UA/2008/02/Types.xsd'
        node_set_attributes['xmlns:xsd'] = 'http://www.w3.org/2001/XMLSchema'
        node_set_attributes['xmlns'] = 'http://opcfoundation.org/UA/2011/03/UANodeSet.xsd'

        self.etree = Et.ElementTree(Et.Element('UANodeSet', node_set_attributes))

    def build_etree(self, node_list, uris=None):
        """
        Create an XML etree object from a list of nodes; custom namespace uris are optional
        Args:
            node_list: list of Node objects for export
            uris: list of namespace uri strings

        Returns:
        """
        self.logger.info('Building XML etree')

        # add all nodes in the list to the XML etree
        for node in node_list:
            self.node_to_etree(node)

        # add aliases to the XML etree
        self._add_alias_els()

        if uris:
            # add namespace uris to the XML etree
            self._add_namespace_uri_els(uris)

    def write_xml(self, xmlpath, pretty=True):
        """
        Write the XML etree in the exporter object to a file
        Args:
            xmlpath: string representing the path/file name

        Returns:
        """
        # try to write the XML etree to a file
        self.logger.info('Exporting XML file to %s', xmlpath)
        #from IPython import embed
        #embed()
        if pretty:
            indent(self.etree.getroot())
            self.etree.write(xmlpath, 
                             short_empty_elements=False, 
                             encoding='utf-8', 
                             xml_declaration=True
                            )
        else:
            self.etree.write(xmlpath, 
                             short_empty_elements=False, 
                             encoding='utf-8', 
                             xml_declaration=True)

    def dump_etree(self):
        """
        Dump etree to console for debugging
        Returns:
        """
        self.logger.info('Dumping XML etree to console')
        Et.dump(self.etree)

    def node_to_etree(self, node):
        """
        Add the necessary XML sub elements to the etree for exporting the node
        Args:
            node: Node object which will be added to XML etree

        Returns:
        """
        node_class = node.get_node_class()
        print("Exporting: ", node)

        if node_class is ua.NodeClass.Object:
            self.add_etree_object(node)
        elif node_class is ua.NodeClass.ObjectType:
            self.add_etree_object_type(node)
        elif node_class is ua.NodeClass.Variable:
            self.add_etree_variable(node)
        elif node_class is ua.NodeClass.VariableType:
            self.add_etree_variable_type(node)
        elif node_class is ua.NodeClass.ReferenceType:
            self.add_etree_reference_type(node)
        elif node_class is ua.NodeClass.DataType:
            self.add_etree_datatype(node)
        elif node_class is ua.NodeClass.Method:
            self.add_etree_method(node)
        else:
            self.logger.info("Exporting node class not implemented: %s ", node_class)

    def _add_sub_el(self, el, name, text):
        child_el = Et.SubElement(el, name)
        child_el.text = text
        return child_el

    def _add_node_common(self, nodetype, node):
        browsename = node.get_browse_name().to_string()
        nodeid = node.nodeid.to_string()
        parent = node.get_parent()
        displayname = node.get_display_name().Text.decode('utf-8')
        desc = node.get_description().Text
        print("NODE COMMON", node)
        node_el = Et.SubElement(self.etree.getroot(), nodetype)
        node_el.attrib["NodeId"] = nodeid
        node_el.attrib["BrowseName"] = browsename
        if parent is not None:
            node_class = node.get_node_class()
            if node_class in (ua.NodeClass.Object, ua.NodeClass.Variable, ua.NodeClass.Method):
                node_el.attrib["ParentNodeId"] = parent.nodeid.to_string()
        self._add_sub_el(node_el, 'DisplayName', displayname)
        if desc not in (None, ""):
            self._add_sub_el(node_el, 'Description', desc.decode('utf-8'))
        #FIXME: add WriteMask and UserWriteMask
        return node_el

    def add_etree_object(self, node):
        """
        Add a UA object element to the XML etree
        """
        obj_el = self._add_node_common("UAObject", node)
        var = node.get_attribute(ua.AttributeIds.EventNotifier)
        if var.Value.Value != 0:
            obj_el.attrib["EventNotifier"] = str(var.Value.Value)
        self._add_ref_els(obj_el, node)

    def add_etree_object_type(self, node):
        """
        Add a UA object type element to the XML etree
        """
        obj_el = self._add_node_common("UAObjectType", node)
        abstract = node.get_attribute(ua.AttributeIds.IsAbstract).Value.Value
        if abstract:
            obj_el.attrib["IsAbstract"] = 'true'
        self._add_ref_els(obj_el, node)

    def add_variable_common(self, node, el):
        dtype = node.get_data_type()
        if dtype.Identifier in o_ids.ObjectIdNames:
            dtype_name = o_ids.ObjectIdNames[dtype.Identifier]
            self.aliases[dtype_name] = dtype.to_string()
        else:
            dtype_name = dtype.to_string()
        rank = node.get_value_rank()
        if rank != -1:
            el.attrib["ValueRank"] = str(int(rank))
        #var = node.get_attribute(ua.AttributeIds.ArrayDimensions())
        #self._addobj_el.attrib["ArrayDimensions"] = str(var.Value.Value)
        el.attrib["DataType"] = dtype_name
        value_to_etree(el, dtype_name, dtype, node)

    def add_etree_variable(self, node):
        """
        Add a UA variable element to the XML etree
        """
        var_el = self._add_node_common("UAVariable", node)
        self.add_variable_common(node, var_el)

        accesslevel = node.get_attribute(ua.AttributeIds.AccessLevel).Value.Value
        useraccesslevel = node.get_attribute(ua.AttributeIds.UserAccessLevel).Value.Value

        # We only write these values if they are different from defaults
        # Not sure where default is defined....
        if accesslevel not in (0, ua.AccessLevel.CurrentRead.mask):
            var_el.attrib["AccessLevel"] = str(accesslevel)
        if useraccesslevel not in (0, ua.AccessLevel.CurrentRead.mask):
            var_el.attrib["UserAccessLevel"] = str(useraccesslevel)

        var = node.get_attribute(ua.AttributeIds.MinimumSamplingInterval)
        if var.Value.Value:
            var_el.attrib["MinimumSamplingInterval"] = str(var.Value.Value)
        var = node.get_attribute(ua.AttributeIds.Historizing)
        if var.Value.Value:
            var_el.attrib["Historizing"] = 'true'
        self._add_ref_els(var_el, node)

    def add_etree_variable_type(self, node):
        """
        Add a UA variable type element to the XML etree
        """

        var_el = self._add_node_common("UAVariableType", node)
        self.add_variable_common(node, var_el)

        abstract = node.get_attribute(ua.AttributeIds.IsAbstract)
        if abstract:
            var_el.attrib["IsAbstract"] = abstract

        self._add_ref_els(var_el, node)

    def add_etree_method(self, node):
        obj_el = self._add_node_common("UAMethod", node)

        var = node.get_attribute(ua.AttributeIds.Executable)
        if var.Value.Value is False:
            obj_el.attrib["Executable"] = "false"
        var = node.get_attribute(ua.AttributeIds.UserExecutable)
        if var.Value.Value is False:
            obj_el.attrib["UserExecutable"] = "false"
        self._add_ref_els(obj_el, node)

    def add_etree_reference_type(self, obj):
        obj_el = self._add_node_common("UAReferenceType", obj)
        var = obj.get_attribute(ua.AttributeIds.InverseName)
        if var is not None and var.Value.Value is not None:
            self._add_sub_el(obj_el, 'InverseName', var.Value.Value.Text.decode('utf-8'))
        self._add_ref_els(obj_el, obj)

    def add_etree_datatype(self, obj):
        """
        Add a UA data type element to the XML etree
        """
        obj_el = self._add_node_common("UADataType", obj)
        self._add_ref_els(obj_el, obj)

    def _add_namespace_uri_els(self, uris):
        nuris_el = Et.Element('NamespaceUris')

        for uri in uris:
            self._add_sub_el(nuris_el, 'Uri', uri)

        self.etree.getroot().insert(0, nuris_el)

    def _add_alias_els(self):
        aliases_el = Et.Element('Aliases')

        for k, v in self.aliases.items():
            ref_el = Et.SubElement(aliases_el, 'Alias', Alias=k)
            ref_el.text = v

        self.etree.getroot().insert(0, aliases_el)

    def _add_ref_els(self, parent_el, obj):
        refs = obj.get_references()
        refs_el = Et.SubElement(parent_el, 'References')

        for ref in refs:
            if ref.ReferenceTypeId.Identifier in o_ids.ObjectIdNames:
                ref_name = o_ids.ObjectIdNames[ref.ReferenceTypeId.Identifier]
            else:
                ref_name = ref.ReferenceTypeId.to_string()
            ref_el = Et.SubElement(refs_el, 'Reference')
            ref_el.attrib['ReferenceType'] = ref_name
            if not ref.IsForward:
                ref_el.attrib['IsForward'] = 'false'
            ref_el.text = ref.NodeId.to_string()

            self.aliases[ref_name] = ref.ReferenceTypeId.to_string()


def value_to_etree(el, dtype_name, dtype, node):
    var = node.get_data_value().Value
    if var.Value is not None:
        val_el = Et.SubElement(el, 'Value')
        _value_to_etree(val_el, dtype_name, dtype, var.Value)


def _value_to_etree(el, dtype_name, dtype, val):
    if isinstance(val, (list, tuple)):
        list_el = Et.SubElement(el, "uax:ListOf" + dtype_name)
        for nval in val:
            _value_to_etree(list_el, dtype_name, dtype, nval)
    else:
        if dtype.Identifier is int and dtype.Identifier > 21:  # this is an extentionObject:
            _extobj_to_etree(el, dtype_name, dtype, val)
        else:
            val_el = Et.SubElement(el, "uax:" + dtype_name)
            val_el.text = str(val)


def _extobj_to_etree(val_el, dtype_name, dtype, val):
    obj_el = Et.SubElement(val_el, "uax:ExtensionObject")
    type_el = Et.SubElement(obj_el, "uax:TypeId")
    id_el = Et.SubElement(type_el, "uax:Identifier")
    id_el.text = val.TypeId.to_string()
    body_el = Et.SubElement(obj_el, "uax:Body")
    struct_el = Et.SubElement(body_el, "uax:" + dtype_name)
    # FIXME: finish
   

def indent(elem, level=0):
    '''
    copy and paste from http://effbot.org/zone/element-lib.htm#prettyprint
    it basically walks your tree and adds spaces and newlines so the tree is
    printed in a nice way
    '''
    i = "\n" + level*"  "
    if len(elem):
        if not elem.text or not elem.text.strip():
            elem.text = i + "  "
        if not elem.tail or not elem.tail.strip():
            elem.tail = i
        for elem in elem:
            indent(elem, level+1)
        if not elem.tail or not elem.tail.strip():
            elem.tail = i
    else:
        if level and (not elem.tail or not elem.tail.strip()):
            elem.tail = i


1			"""
2			from a list of nodes in the address space, build an XML file
3			format is the one from opc-ua specification
4			"""
5			import logging
6			from collections import OrderedDict
7			import xml.etree.ElementTree as Et
8
9			from opcua import ua
10			from opcua.ua import object_ids as o_ids
11
12
13			class XmlExporter(object):
14
15			def __init__(self, server):
16			self.logger = logging.getLogger(__name__)
17			self.server = server
18			self.aliases = {}
19
20			node_set_attributes = OrderedDict()
21			node_set_attributes['xmlns:xsi'] = 'http://www.w3.org/2001/XMLSchema-instance'
22			node_set_attributes['xmlns:uax'] = 'http://opcfoundation.org/UA/2008/02/Types.xsd'
23			node_set_attributes['xmlns:xsd'] = 'http://www.w3.org/2001/XMLSchema'
24			node_set_attributes['xmlns'] = 'http://opcfoundation.org/UA/2011/03/UANodeSet.xsd'
25
26			self.etree = Et.ElementTree(Et.Element('UANodeSet', node_set_attributes))
27
28			def build_etree(self, node_list, uris=None):
29			"""
30			Create an XML etree object from a list of nodes; custom namespace uris are optional
31			Args:
32			node_list: list of Node objects for export
33			uris: list of namespace uri strings
34
35			Returns:
36			"""
37			self.logger.info('Building XML etree')
38
39			# add all nodes in the list to the XML etree
40			for node in node_list:
41			self.node_to_etree(node)
42
43			# add aliases to the XML etree
44			self._add_alias_els()
45
46			if uris:
47			# add namespace uris to the XML etree
48			self._add_namespace_uri_els(uris)
49
50			def write_xml(self, xmlpath, pretty=True):
51			"""
52			Write the XML etree in the exporter object to a file
53			Args:
54			xmlpath: string representing the path/file name
55
56			Returns:
57			"""
58			# try to write the XML etree to a file
59			self.logger.info('Exporting XML file to %s', xmlpath)
60			#from IPython import embed
61			#embed()
62			if pretty:
63			indent(self.etree.getroot())
64			self.etree.write(xmlpath,
65			short_empty_elements=False,
66			encoding='utf-8',
67			xml_declaration=True
68			)
69			else:
70			self.etree.write(xmlpath,
71			short_empty_elements=False,
72			encoding='utf-8',
73			xml_declaration=True)
74
75			def dump_etree(self):
76			"""
77			Dump etree to console for debugging
78			Returns:
79			"""
80			self.logger.info('Dumping XML etree to console')
81			Et.dump(self.etree)
82
83			def node_to_etree(self, node):
84			"""
85			Add the necessary XML sub elements to the etree for exporting the node
86			Args:
87			node: Node object which will be added to XML etree
88
89			Returns:
90			"""
91			node_class = node.get_node_class()
92			print("Exporting: ", node)
93
94			if node_class is ua.NodeClass.Object:
95			self.add_etree_object(node)
96			elif node_class is ua.NodeClass.ObjectType:
97			self.add_etree_object_type(node)
98			elif node_class is ua.NodeClass.Variable:
99			self.add_etree_variable(node)
100			elif node_class is ua.NodeClass.VariableType:
101			self.add_etree_variable_type(node)
102			elif node_class is ua.NodeClass.ReferenceType:
103			self.add_etree_reference_type(node)
104			elif node_class is ua.NodeClass.DataType:
105			self.add_etree_datatype(node)
106			elif node_class is ua.NodeClass.Method:
107			self.add_etree_method(node)
108			else:
109			self.logger.info("Exporting node class not implemented: %s ", node_class)
110
111			def _add_sub_el(self, el, name, text):
112			child_el = Et.SubElement(el, name)
113			child_el.text = text
114			return child_el
115
116			def _add_node_common(self, nodetype, node):
117			browsename = node.get_browse_name().to_string()
118			nodeid = node.nodeid.to_string()
119			parent = node.get_parent()
120			displayname = node.get_display_name().Text.decode('utf-8')
121			desc = node.get_description().Text
122			print("NODE COMMON", node)
123			node_el = Et.SubElement(self.etree.getroot(), nodetype)
124			node_el.attrib["NodeId"] = nodeid
125			node_el.attrib["BrowseName"] = browsename
126			if parent is not None:
127			node_class = node.get_node_class()
128			if node_class in (ua.NodeClass.Object, ua.NodeClass.Variable, ua.NodeClass.Method):
129			node_el.attrib["ParentNodeId"] = parent.nodeid.to_string()
130			self._add_sub_el(node_el, 'DisplayName', displayname)
131			if desc not in (None, ""):
132			self._add_sub_el(node_el, 'Description', desc.decode('utf-8'))
133			#FIXME: add WriteMask and UserWriteMask
134			return node_el
135
136			def add_etree_object(self, node):
137			"""
138			Add a UA object element to the XML etree
139			"""
140			obj_el = self._add_node_common("UAObject", node)
141			var = node.get_attribute(ua.AttributeIds.EventNotifier)
142			if var.Value.Value != 0:
143			obj_el.attrib["EventNotifier"] = str(var.Value.Value)
144			self._add_ref_els(obj_el, node)
145
146			def add_etree_object_type(self, node):
147			"""
148			Add a UA object type element to the XML etree
149			"""
150			obj_el = self._add_node_common("UAObjectType", node)
151			abstract = node.get_attribute(ua.AttributeIds.IsAbstract).Value.Value
152			if abstract:
153			obj_el.attrib["IsAbstract"] = 'true'
154			self._add_ref_els(obj_el, node)
155
156			def add_variable_common(self, node, el):
157			dtype = node.get_data_type()
158			if dtype.Identifier in o_ids.ObjectIdNames:
159			dtype_name = o_ids.ObjectIdNames[dtype.Identifier]
160			self.aliases[dtype_name] = dtype.to_string()
161			else:
162			dtype_name = dtype.to_string()
163			rank = node.get_value_rank()
164			if rank != -1:
165			el.attrib["ValueRank"] = str(int(rank))
166			#var = node.get_attribute(ua.AttributeIds.ArrayDimensions())
167			#self._addobj_el.attrib["ArrayDimensions"] = str(var.Value.Value)
168			el.attrib["DataType"] = dtype_name
169			value_to_etree(el, dtype_name, dtype, node)
170
171			def add_etree_variable(self, node):
172			"""
173			Add a UA variable element to the XML etree
174			"""
175			var_el = self._add_node_common("UAVariable", node)
176			self.add_variable_common(node, var_el)
177
178			accesslevel = node.get_attribute(ua.AttributeIds.AccessLevel).Value.Value
179			useraccesslevel = node.get_attribute(ua.AttributeIds.UserAccessLevel).Value.Value
180
181			# We only write these values if they are different from defaults
182			# Not sure where default is defined....
183			if accesslevel not in (0, ua.AccessLevel.CurrentRead.mask):
184			var_el.attrib["AccessLevel"] = str(accesslevel)
185			if useraccesslevel not in (0, ua.AccessLevel.CurrentRead.mask):
186			var_el.attrib["UserAccessLevel"] = str(useraccesslevel)
187
188			var = node.get_attribute(ua.AttributeIds.MinimumSamplingInterval)
189			if var.Value.Value:
190			var_el.attrib["MinimumSamplingInterval"] = str(var.Value.Value)
191			var = node.get_attribute(ua.AttributeIds.Historizing)
192			if var.Value.Value:
193			var_el.attrib["Historizing"] = 'true'
194			self._add_ref_els(var_el, node)
195
196			def add_etree_variable_type(self, node):
197			"""
198			Add a UA variable type element to the XML etree
199			"""
200
201			var_el = self._add_node_common("UAVariableType", node)
202			self.add_variable_common(node, var_el)
203
204			abstract = node.get_attribute(ua.AttributeIds.IsAbstract)
205			if abstract:
206			var_el.attrib["IsAbstract"] = abstract
207
208			self._add_ref_els(var_el, node)
209
210			def add_etree_method(self, node):
211			obj_el = self._add_node_common("UAMethod", node)
212
213			var = node.get_attribute(ua.AttributeIds.Executable)
214			if var.Value.Value is False:
215			obj_el.attrib["Executable"] = "false"
216			var = node.get_attribute(ua.AttributeIds.UserExecutable)
217			if var.Value.Value is False:
218			obj_el.attrib["UserExecutable"] = "false"
219			self._add_ref_els(obj_el, node)
220
221			def add_etree_reference_type(self, obj):
222			obj_el = self._add_node_common("UAReferenceType", obj)
223			var = obj.get_attribute(ua.AttributeIds.InverseName)
224			if var is not None and var.Value.Value is not None:
225			self._add_sub_el(obj_el, 'InverseName', var.Value.Value.Text.decode('utf-8'))
226			self._add_ref_els(obj_el, obj)
227
228			def add_etree_datatype(self, obj):
229			"""
230			Add a UA data type element to the XML etree
231			"""
232			obj_el = self._add_node_common("UADataType", obj)
233			self._add_ref_els(obj_el, obj)
234
235			def _add_namespace_uri_els(self, uris):
236			nuris_el = Et.Element('NamespaceUris')
237
238			for uri in uris:
239			self._add_sub_el(nuris_el, 'Uri', uri)
240
241			self.etree.getroot().insert(0, nuris_el)
242
243			def _add_alias_els(self):
244			aliases_el = Et.Element('Aliases')
245
246			for k, v in self.aliases.items():
247			ref_el = Et.SubElement(aliases_el, 'Alias', Alias=k)
248			ref_el.text = v
249
250			self.etree.getroot().insert(0, aliases_el)
251
252			def _add_ref_els(self, parent_el, obj):
253			refs = obj.get_references()
254			refs_el = Et.SubElement(parent_el, 'References')
255
256			for ref in refs:
257			if ref.ReferenceTypeId.Identifier in o_ids.ObjectIdNames:
258			ref_name = o_ids.ObjectIdNames[ref.ReferenceTypeId.Identifier]
259			else:
260			ref_name = ref.ReferenceTypeId.to_string()
261			ref_el = Et.SubElement(refs_el, 'Reference')
262			ref_el.attrib['ReferenceType'] = ref_name
263			if not ref.IsForward:
264			ref_el.attrib['IsForward'] = 'false'
265			ref_el.text = ref.NodeId.to_string()
266
267			self.aliases[ref_name] = ref.ReferenceTypeId.to_string()
268
269
270			def value_to_etree(el, dtype_name, dtype, node):
271			var = node.get_data_value().Value
272			if var.Value is not None:
273			val_el = Et.SubElement(el, 'Value')
274			_value_to_etree(val_el, dtype_name, dtype, var.Value)
275
276
277			def _value_to_etree(el, dtype_name, dtype, val):
278			if isinstance(val, (list, tuple)):
279			list_el = Et.SubElement(el, "uax:ListOf" + dtype_name)
280			for nval in val:
281			_value_to_etree(list_el, dtype_name, dtype, nval)
282			else:
283			if dtype.Identifier is int and dtype.Identifier > 21: # this is an extentionObject:
284			_extobj_to_etree(el, dtype_name, dtype, val)
285			else:
286			val_el = Et.SubElement(el, "uax:" + dtype_name)
287			val_el.text = str(val)
288
289
290			def _extobj_to_etree(val_el, dtype_name, dtype, val):
291			obj_el = Et.SubElement(val_el, "uax:ExtensionObject")
292			type_el = Et.SubElement(obj_el, "uax:TypeId")
293			id_el = Et.SubElement(type_el, "uax:Identifier")
294			id_el.text = val.TypeId.to_string()
295			body_el = Et.SubElement(obj_el, "uax:Body")
296			struct_el = Et.SubElement(body_el, "uax:" + dtype_name)
297			# FIXME: finish
298
299
300			def indent(elem, level=0):
301			'''
302			copy and paste from http://effbot.org/zone/element-lib.htm#prettyprint
303			it basically walks your tree and adds spaces and newlines so the tree is
304			printed in a nice way
305			'''
306			i = "\n" + level*" "
307			if len(elem):
308			if not elem.text or not elem.text.strip():
309			elem.text = i + " "
310			if not elem.tail or not elem.tail.strip():
311			elem.tail = i
312			for elem in elem:
313			indent(elem, level+1)
314			if not elem.tail or not elem.tail.strip():
315			elem.tail = i
316			else:
317			if level and (not elem.tail or not elem.tail.strip()):
318			elem.tail = i
319

FreeOpcUa / python-opcua

Push — master ( 6c18d6...f8fb14 )

XmlExporter.add_etree_method() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like