variant_to_etree() - Code Metrics - Inspection of "xml and binary improvement/Cleanup" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#296)

by Olivier

created 2016-09-03 05:36 UTC

variant_to_etree() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	5
rs	9.4285

"""
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 xml.dom import minidom

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 all required aliases to the XML etree; must be done after nodes are added
        self._add_alias_els()

        if uris:
            # add all namespace uris to the XML etree; must be done after aliases are added
            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:
            self.etree.write(xmlpath + "-not-pretty.xml", short_empty_elements=False)
            rough_string = Et.tostring(self.etree.getroot(), 'utf-8')
            reparsed = minidom.parseString(rough_string)
            pretty_string = reparsed.toprettyxml(indent="    ")
            with open(xmlpath, "wt") as f:
                f.write(pretty_string)
        else:
            self.etree.write(xmlpath, short_empty_elements=False)

    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(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_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(encoding='UTF8')
        desc = node.get_description().Text
        print("NODE COMMON", node)
        node_el = Et.SubElement(self.etree.getroot(),
                                nodetype,
                                BrowseName=browsename,
                                NodeId=nodeid)
        if parent is not None:
            node_el.attrib["ParentNodeId"] = parent.nodeid.to_string()
        if desc not in (None, ""):
            node_el.attrib["Description"] = str(desc)
        disp_el = Et.SubElement(node_el, 'DisplayName', )
        disp_el.text = displayname
        return node_el

    def add_etree_object(self, obj):
        """
        Add a UA object element to the XML etree
        """
        obj_el = self._add_node_common("UAObject", obj)
        self._add_ref_els(obj_el, obj)

    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
        obj_el.attrib["IsAbstract"] = abstract
        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()
        el.attrib["DataType"] = dtype_name
        el.attrib["ValueRank"] = str(int(rank))
        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
        # default must of course mange the one in manage_nodes.py
        # and other OPC UA servers
        if accesslevel != ua.AccessLevel.CurrentRead.mask:
            var_el.attrib["AccessLevel"] = str(accesslevel)
        if useraccesslevel != ua.AccessLevel.CurrentRead.mask:
            var_el.attrib["UserAccessLevel"] = str(useraccesslevel)
        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("UAVariable", node)
        self.add_variable_common(node, var_el)

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

        self._add_ref_els(var_el, node)

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

    def add_etree_reference(self, obj):
        obj_el = self._add_node_common("UAReference", obj)
        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:
            uri_el = Et.SubElement(nuris_el, 'Uri')
            uri_el.text = 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_forward = str(ref.IsForward).lower()
            ref_nodeid = ref.NodeId.to_string()
            ref_el = Et.SubElement(refs_el, 'Reference', IsForward=ref_forward, ReferenceType=ref_name)
            ref_el.text = ref_nodeid

            # add any references that gets used to aliases dict; this gets handled later
            self.aliases[ref_name] = ref_nodeid


def value_to_etree(el, dtype_name, dtype, node):
    var = node.get_data_value().Value
    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)
        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
    



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			from xml.dom import minidom
9
10			from opcua import ua
11			from opcua.ua import object_ids as o_ids
12
13
14			class XmlExporter(object):
15
16			def __init__(self, server):
17			self.logger = logging.getLogger(__name__)
18			self.server = server
19			self.aliases = {}
20
21			node_set_attributes = OrderedDict()
22			node_set_attributes['xmlns:xsi'] = 'http://www.w3.org/2001/XMLSchema-instance'
23			node_set_attributes['xmlns:uax'] = 'http://opcfoundation.org/UA/2008/02/Types.xsd'
24			node_set_attributes['xmlns:xsd'] = 'http://www.w3.org/2001/XMLSchema'
25			node_set_attributes['xmlns'] = 'http://opcfoundation.org/UA/2011/03/UANodeSet.xsd'
26
27			self.etree = Et.ElementTree(Et.Element('UANodeSet', node_set_attributes))
28
29			def build_etree(self, node_list, uris=None):
30			"""
31			Create an XML etree object from a list of nodes; custom namespace uris are optional
32			Args:
33			node_list: list of Node objects for export
34			uris: list of namespace uri strings
35
36			Returns:
37			"""
38			self.logger.info('Building XML etree')
39
40			# add all nodes in the list to the XML etree
41			for node in node_list:
42			self.node_to_etree(node)
43
44			# add all required aliases to the XML etree; must be done after nodes are added
45			self._add_alias_els()
46
47			if uris:
48			# add all namespace uris to the XML etree; must be done after aliases are added
49			self._add_namespace_uri_els(uris)
50
51			def write_xml(self, xmlpath, pretty=True):
52			"""
53			Write the XML etree in the exporter object to a file
54			Args:
55			xmlpath: string representing the path/file name
56
57			Returns:
58			"""
59			# try to write the XML etree to a file
60			self.logger.info('Exporting XML file to %s', xmlpath)
61			#from IPython import embed
62			#embed()
63			if pretty:
64			self.etree.write(xmlpath + "-not-pretty.xml", short_empty_elements=False)
65			rough_string = Et.tostring(self.etree.getroot(), 'utf-8')
66			reparsed = minidom.parseString(rough_string)
67			pretty_string = reparsed.toprettyxml(indent=" ")
68			with open(xmlpath, "wt") as f:
69			f.write(pretty_string)
70			else:
71			self.etree.write(xmlpath, short_empty_elements=False)
72
73			def dump_etree(self):
74			"""
75			Dump etree to console for debugging
76			Returns:
77			"""
78			self.logger.info('Dumping XML etree to console')
79			Et.dump(self.etree)
80
81			def node_to_etree(self, node):
82			"""
83			Add the necessary XML sub elements to the etree for exporting the node
84			Args:
85			node: Node object which will be added to XML etree
86
87			Returns:
88			"""
89			node_class = node.get_node_class()
90			print("Exporting: ", node)
91
92			if node_class is ua.NodeClass.Object:
93			self.add_etree_object(node)
94			elif node_class is ua.NodeClass.ObjectType:
95			self.add_etree_object_type(node)
96			elif node_class is ua.NodeClass.Variable:
97			self.add_etree_variable(node)
98			elif node_class is ua.NodeClass.VariableType:
99			self.add_etree_variable_type(node)
100			elif node_class is ua.NodeClass.ReferenceType:
101			self.add_etree_reference(node)
102			elif node_class is ua.NodeClass.DataType:
103			self.add_etree_datatype(node)
104			elif node_class is ua.NodeClass.Method:
105			self.add_etree_method(node)
106			else:
107			self.logger.info("Exporting node class not implemented: %s ", node_class)
108
109			def _add_node_common(self, nodetype, node):
110			browsename = node.get_browse_name().to_string()
111			nodeid = node.nodeid.to_string()
112			parent = node.get_parent()
113			displayname = node.get_display_name().Text.decode(encoding='UTF8')
114			desc = node.get_description().Text
115			print("NODE COMMON", node)
116			node_el = Et.SubElement(self.etree.getroot(),
117			nodetype,
118			BrowseName=browsename,
119			NodeId=nodeid)
120			if parent is not None:
121			node_el.attrib["ParentNodeId"] = parent.nodeid.to_string()
122			if desc not in (None, ""):
123			node_el.attrib["Description"] = str(desc)
124			disp_el = Et.SubElement(node_el, 'DisplayName', )
125			disp_el.text = displayname
126			return node_el
127
128			def add_etree_object(self, obj):
129			"""
130			Add a UA object element to the XML etree
131			"""
132			obj_el = self._add_node_common("UAObject", obj)
133			self._add_ref_els(obj_el, obj)
134
135			def add_etree_object_type(self, node):
136			"""
137			Add a UA object type element to the XML etree
138			"""
139			obj_el = self._add_node_common("UAObjectType", node)
140			abstract = node.get_attribute(ua.AttributeIds.IsAbstract).Value.Value
141			obj_el.attrib["IsAbstract"] = abstract
142			self._add_ref_els(obj_el, node)
143
144			def add_variable_common(self, node, el):
145			dtype = node.get_data_type()
146			if dtype.Identifier in o_ids.ObjectIdNames:
147			dtype_name = o_ids.ObjectIdNames[dtype.Identifier]
148			self.aliases[dtype_name] = dtype.to_string()
149			else:
150			dtype_name = dtype.to_string()
151			rank = node.get_value_rank()
152			el.attrib["DataType"] = dtype_name
153			el.attrib["ValueRank"] = str(int(rank))
154			value_to_etree(el, dtype_name, dtype, node)
155
156			def add_etree_variable(self, node):
157			"""
158			Add a UA variable element to the XML etree
159			"""
160			var_el = self._add_node_common("UAVariable", node)
161			self.add_variable_common(node, var_el)
162
163			accesslevel = node.get_attribute(ua.AttributeIds.AccessLevel).Value.Value
164			useraccesslevel = node.get_attribute(ua.AttributeIds.UserAccessLevel).Value.Value
165
166			# We only write these values if they are different from defaults
167			# default must of course mange the one in manage_nodes.py
168			# and other OPC UA servers
169			if accesslevel != ua.AccessLevel.CurrentRead.mask:
170			var_el.attrib["AccessLevel"] = str(accesslevel)
171			if useraccesslevel != ua.AccessLevel.CurrentRead.mask:
172			var_el.attrib["UserAccessLevel"] = str(useraccesslevel)
173			self._add_ref_els(var_el, node)
174
175			def add_etree_variable_type(self, node):
176			"""
177			Add a UA variable type element to the XML etree
178			"""
179
180			var_el = self._add_node_common("UAVariable", node)
181			self.add_variable_common(node, var_el)
182
183			abstract = node.get_attribute(ua.AttributeIds.IsAbstract)
184			var_el.attrib["IsAbstract"] = abstract
185
186			self._add_ref_els(var_el, node)
187
188			def add_etree_method(self, obj):
189			obj_el = self._add_node_common("UAMethod", obj)
190			self._add_ref_els(obj_el, obj)
191
192			def add_etree_reference(self, obj):
193			obj_el = self._add_node_common("UAReference", obj)
194			self._add_ref_els(obj_el, obj)
195
196			def add_etree_datatype(self, obj):
197			"""
198			Add a UA data type element to the XML etree
199			"""
200			obj_el = self._add_node_common("UADataType", obj)
201			self._add_ref_els(obj_el, obj)
202
203			def _add_namespace_uri_els(self, uris):
204			nuris_el = Et.Element('NamespaceUris')
205
206			for uri in uris:
207			uri_el = Et.SubElement(nuris_el, 'Uri')
208			uri_el.text = uri
209
210			self.etree.getroot().insert(0, nuris_el)
211
212			def _add_alias_els(self):
213			aliases_el = Et.Element('Aliases')
214
215			for k, v in self.aliases.items():
216			ref_el = Et.SubElement(aliases_el, 'Alias', Alias=k)
217			ref_el.text = v
218
219			self.etree.getroot().insert(0, aliases_el)
220
221			def _add_ref_els(self, parent_el, obj):
222			refs = obj.get_references()
223			refs_el = Et.SubElement(parent_el, 'References')
224
225			for ref in refs:
226			if ref.ReferenceTypeId.Identifier in o_ids.ObjectIdNames:
227			ref_name = o_ids.ObjectIdNames[ref.ReferenceTypeId.Identifier]
228			else:
229			ref_name = ref.ReferenceTypeId.to_string()
230			ref_forward = str(ref.IsForward).lower()
231			ref_nodeid = ref.NodeId.to_string()
232			ref_el = Et.SubElement(refs_el, 'Reference', IsForward=ref_forward, ReferenceType=ref_name)
233			ref_el.text = ref_nodeid
234
235			# add any references that gets used to aliases dict; this gets handled later
236			self.aliases[ref_name] = ref_nodeid
237
238
239			def value_to_etree(el, dtype_name, dtype, node):
240			var = node.get_data_value().Value
241			val_el = Et.SubElement(el, 'Value')
242			_value_to_etree(val_el, dtype_name, dtype, var.Value)
243
244
245			def _value_to_etree(el, dtype_name, dtype, val):
246			if isinstance(val, (list, tuple)):
247			list_el = Et.SubElement(el, "uax:ListOf" + dtype_name)
248			for nval in val:
249			_value_to_etree(list_el, dtype_name, dtype, nval)
250			else:
251			if dtype.Identifier is int and dtype.Identifier > 21: # this is an extentionObject:
252			_extobj_to_etree(el, dtype_name, dtype)
253			else:
254			val_el = Et.SubElement(el, "uax:" + dtype_name)
255			val_el.text = str(val)
256
257
258			def _extobj_to_etree(val_el, dtype_name, dtype, val):
259			obj_el = Et.SubElement(val_el, "uax:ExtensionObject")
260			type_el = Et.SubElement(obj_el, "uax:TypeId")
261			id_el = Et.SubElement(type_el, "uax:Identifier")
262			id_el.text = val.TypeId.to_string()
263			body_el = Et.SubElement(obj_el, "uax:Body")
264			struct_el = Et.SubElement(body_el, "uax:" + dtype_name)
265			# FIXME: finish
266
267
268

FreeOpcUa / python-opcua

Pull Request — master (#296)

variant_to_etree() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like