XmlExporter.add_variable_common() - Code Metrics - Inspection of "also export namespace of referenced nodes" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#327)

by Olivier

created 2016-10-15 08:28 UTC

XmlExporter.add_variable_common() A

↳ Parent: XmlExporter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	5
Bugs	1	Features	1

Metric	Value
cc	3
dl	0
loc	14
rs	9.4285
c	5
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 copy import copy

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 = {}
        self._addr_idx_to_xml_idx = {}

        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
        Namespaces used by nodes are allways exported for consistency.
        Args:
            node_list: list of Node objects for export
            uris: list of namespace uri strings

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

        self._add_namespaces(node_list, uris)

        # 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()

    def _add_namespaces(self, nodes, uris):
        idxs = self._get_ns_idxs_of_nodes(nodes)

        ns_array = self.server.get_namespace_array()

        # now add index of provided uris if necessary
        if uris:
            self._add_idxs_from_uris(idxs, uris, ns_array)

        # now create a dict of idx_in_address_space to idx_in_exported_file
        self._addr_idx_to_xml_idx = self._make_idx_dict(idxs, ns_array) 
        print("DICT", self._addr_idx_to_xml_idx)
        ns_to_export = [ns_array[i] for i in self._addr_idx_to_xml_idx.keys() if i != 0]
        print("NS_TO_EXPORT", ns_to_export)

        # write namespaces to xml
        self._add_namespace_uri_els(ns_to_export)

    def _make_idx_dict(self, idxs, ns_array):
        print("MAKE", idxs, ns_array)
        idxs.sort()
        addr_idx_to_xml_idx = {0: 0}
        for xml_idx, addr_idx in enumerate(idxs):
            if addr_idx >= len(ns_array):
                break
            addr_idx_to_xml_idx[addr_idx] = xml_idx + 1
        print(" ADDR_TO_ XML", self._addr_idx_to_xml_idx)
        return addr_idx_to_xml_idx

    def _get_ns_idxs_of_nodes(self, nodes):
        """
        get a list of all indexes used or references by nodes
        """
        idxs = []
        for node in nodes:
            node_idxs = [node.nodeid.NamespaceIndex]
            node_idxs.append(node.get_browse_name().NamespaceIndex)
            node_idxs.extend(ref.NodeId.NamespaceIndex for ref in node.get_references())
            node_idxs = list(set(node_idxs))  # remove duplicates
            for i in node_idxs:
                if i != 0 and i not in idxs:
                    idxs.append(i)
        print("IDXS", idxs)
        return idxs

    def _add_idxs_from_uris(self, idxs, uris, ns_array):
        for uri in uris:
            if uri in ns_array:
                i = ns_array.index(uri)
                if i not in idxs:
                    idxs.append(i)


    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,
                             encoding='utf-8',
                             xml_declaration=True
                            )
        else:
            self.etree.write(xmlpath,
                             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 _node_to_string(self, nodeid):
        if not isinstance(nodeid, ua.NodeId):
            nodeid = nodeid.nodeid

        if nodeid.NamespaceIndex in self._addr_idx_to_xml_idx:
            nodeid = copy(nodeid)
            nodeid.NamespaceIndex = self._addr_idx_to_xml_idx[nodeid.NamespaceIndex]
        return nodeid.to_string()

    def _bname_to_string(self, bname):
        if bname.NamespaceIndex in self._addr_idx_to_xml_idx:
            bname = copy(bname)
            bname.NamespaceIndex = self._addr_idx_to_xml_idx[bname.NamespaceIndex]
        return bname.to_string()

    def _add_node_common(self, nodetype, node):
        browsename = node.get_browse_name()
        nodeid = node.nodeid
        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"] = self._node_to_string(nodeid)
        node_el.attrib["BrowseName"] = self._bname_to_string(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"] = self._node_to_string(parent)
        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(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.Value.Value:
            var_el.attrib["IsAbstract"] = "true"

        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 = self._node_to_string(ref.NodeId)

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


def member_to_etree(el, name, dtype, val):
    member_el = Et.SubElement(el, "uax:" + name)
    if isinstance(val, (list, tuple)):
        for v in val:
            _value_to_etree(member_el, ua.ObjectIdNames[dtype.Identifier], dtype, v)
    else:
        _val_to_etree(member_el, dtype, val)


def _val_to_etree(el, dtype, val):
    if val is None:
        val = ""
    if dtype == ua.NodeId(ua.ObjectIds.NodeId):
        id_el = Et.SubElement(el, "uax:Identifier")
        id_el.text = val.to_string()
    elif not hasattr(val, "ua_types"):
        if isinstance(val, bytes):
            el.text = val.decode("utf-8")
        else:
            el.text = str(val)
    else:
        for name, vtype in val.ua_types.items():
            member_to_etree(el, name, ua.NodeId(getattr(ua.ObjectIds, vtype)), getattr(val, name))


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, type_name, dtype, val):
    if not val:
        return
    if isinstance(val, (list, tuple)):
        if dtype.Identifier > 21:  # this is an extentionObject:
            elname = "uax:ListOfExtensionObject"
        else:
            elname = "uax:ListOf" + type_name
        list_el = Et.SubElement(el, elname)
        for nval in val:
            _value_to_etree(list_el, type_name, dtype, nval)
    else:
        if dtype.Identifier > 21:  # this is an extentionObject:
            _extobj_to_etree(el, type_name, dtype, val)
        else:
            val_el = Et.SubElement(el, "uax:" + type_name)
            _val_to_etree(val_el, dtype, val)
           

def _extobj_to_etree(val_el, 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 = dtype.to_string()
    body_el = Et.SubElement(obj_el, "uax:Body")
    struct_el = Et.SubElement(body_el, "uax:" + name)
    for name, vtype in val.ua_types.items():
        member_to_etree(struct_el, name, ua.NodeId(getattr(ua.ObjectIds, vtype)), getattr(val, name))


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			from copy import copy
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			self._addr_idx_to_xml_idx = {}
21
22			node_set_attributes = OrderedDict()
23			node_set_attributes['xmlns:xsi'] = 'http://www.w3.org/2001/XMLSchema-instance'
24			node_set_attributes['xmlns:uax'] = 'http://opcfoundation.org/UA/2008/02/Types.xsd'
25			node_set_attributes['xmlns:xsd'] = 'http://www.w3.org/2001/XMLSchema'
26			node_set_attributes['xmlns'] = 'http://opcfoundation.org/UA/2011/03/UANodeSet.xsd'
27
28			self.etree = Et.ElementTree(Et.Element('UANodeSet', node_set_attributes))
29
30			def build_etree(self, node_list, uris=None):
31			"""
32			Create an XML etree object from a list of nodes; custom namespace uris are optional
33			Namespaces used by nodes are allways exported for consistency.
34			Args:
35			node_list: list of Node objects for export
36			uris: list of namespace uri strings
37
38			Returns:
39			"""
40			self.logger.info('Building XML etree')
41
42			self._add_namespaces(node_list, uris)
43
44			# add all nodes in the list to the XML etree
45			for node in node_list:
46			self.node_to_etree(node)
47
48			# add aliases to the XML etree
49			self._add_alias_els()
50
51			def _add_namespaces(self, nodes, uris):
52			idxs = self._get_ns_idxs_of_nodes(nodes)
53
54			ns_array = self.server.get_namespace_array()
55
56			# now add index of provided uris if necessary
57			if uris:
58			self._add_idxs_from_uris(idxs, uris, ns_array)
59
60			# now create a dict of idx_in_address_space to idx_in_exported_file
61			self._addr_idx_to_xml_idx = self._make_idx_dict(idxs, ns_array)
62			print("DICT", self._addr_idx_to_xml_idx)
63			ns_to_export = [ns_array[i] for i in self._addr_idx_to_xml_idx.keys() if i != 0]
64			print("NS_TO_EXPORT", ns_to_export)
65
66			# write namespaces to xml
67			self._add_namespace_uri_els(ns_to_export)
68
69			def _make_idx_dict(self, idxs, ns_array):
70			print("MAKE", idxs, ns_array)
71			idxs.sort()
72			addr_idx_to_xml_idx = {0: 0}
73			for xml_idx, addr_idx in enumerate(idxs):
74			if addr_idx >= len(ns_array):
75			break
76			addr_idx_to_xml_idx[addr_idx] = xml_idx + 1
77			print(" ADDR_TO_ XML", self._addr_idx_to_xml_idx)
78			return addr_idx_to_xml_idx
79
80			def _get_ns_idxs_of_nodes(self, nodes):
81			"""
82			get a list of all indexes used or references by nodes
83			"""
84			idxs = []
85			for node in nodes:
86			node_idxs = [node.nodeid.NamespaceIndex]
87			node_idxs.append(node.get_browse_name().NamespaceIndex)
88			node_idxs.extend(ref.NodeId.NamespaceIndex for ref in node.get_references())
89			node_idxs = list(set(node_idxs)) # remove duplicates
90			for i in node_idxs:
91			if i != 0 and i not in idxs:
92			idxs.append(i)
93			print("IDXS", idxs)
94			return idxs
95
96			def _add_idxs_from_uris(self, idxs, uris, ns_array):
97			for uri in uris:
98			if uri in ns_array:
99			i = ns_array.index(uri)
100			if i not in idxs:
101			idxs.append(i)
102
103
104			def write_xml(self, xmlpath, pretty=True):
105			"""
106			Write the XML etree in the exporter object to a file
107			Args:
108			xmlpath: string representing the path/file name
109
110			Returns:
111			"""
112			# try to write the XML etree to a file
113			self.logger.info('Exporting XML file to %s', xmlpath)
114			#from IPython import embed
115			# embed()
116			if pretty:
117			indent(self.etree.getroot())
118			self.etree.write(xmlpath,
119			encoding='utf-8',
120			xml_declaration=True
121			)
122			else:
123			self.etree.write(xmlpath,
124			encoding='utf-8',
125			xml_declaration=True
126			)
127
128			def dump_etree(self):
129			"""
130			Dump etree to console for debugging
131			Returns:
132			"""
133			self.logger.info('Dumping XML etree to console')
134			Et.dump(self.etree)
135
136			def node_to_etree(self, node):
137			"""
138			Add the necessary XML sub elements to the etree for exporting the node
139			Args:
140			node: Node object which will be added to XML etree
141
142			Returns:
143			"""
144			node_class = node.get_node_class()
145			print("Exporting: ", node)
146
147			if node_class is ua.NodeClass.Object:
148			self.add_etree_object(node)
149			elif node_class is ua.NodeClass.ObjectType:
150			self.add_etree_object_type(node)
151			elif node_class is ua.NodeClass.Variable:
152			self.add_etree_variable(node)
153			elif node_class is ua.NodeClass.VariableType:
154			self.add_etree_variable_type(node)
155			elif node_class is ua.NodeClass.ReferenceType:
156			self.add_etree_reference_type(node)
157			elif node_class is ua.NodeClass.DataType:
158			self.add_etree_datatype(node)
159			elif node_class is ua.NodeClass.Method:
160			self.add_etree_method(node)
161			else:
162			self.logger.info("Exporting node class not implemented: %s ", node_class)
163
164			def _add_sub_el(self, el, name, text):
165			child_el = Et.SubElement(el, name)
166			child_el.text = text
167			return child_el
168
169			def _node_to_string(self, nodeid):
170			if not isinstance(nodeid, ua.NodeId):
171			nodeid = nodeid.nodeid
172
173			if nodeid.NamespaceIndex in self._addr_idx_to_xml_idx:
174			nodeid = copy(nodeid)
175			nodeid.NamespaceIndex = self._addr_idx_to_xml_idx[nodeid.NamespaceIndex]
176			return nodeid.to_string()
177
178			def _bname_to_string(self, bname):
179			if bname.NamespaceIndex in self._addr_idx_to_xml_idx:
180			bname = copy(bname)
181			bname.NamespaceIndex = self._addr_idx_to_xml_idx[bname.NamespaceIndex]
182			return bname.to_string()
183
184			def _add_node_common(self, nodetype, node):
185			browsename = node.get_browse_name()
186			nodeid = node.nodeid
187			parent = node.get_parent()
188			displayname = node.get_display_name().Text.decode('utf-8')
189			desc = node.get_description().Text
190			print("NODE COMMON", node)
191			node_el = Et.SubElement(self.etree.getroot(), nodetype)
192			node_el.attrib["NodeId"] = self._node_to_string(nodeid)
193			node_el.attrib["BrowseName"] = self._bname_to_string(browsename)
194			if parent is not None:
195			node_class = node.get_node_class()
196			if node_class in (ua.NodeClass.Object, ua.NodeClass.Variable, ua.NodeClass.Method):
197			node_el.attrib["ParentNodeId"] = self._node_to_string(parent)
198			self._add_sub_el(node_el, 'DisplayName', displayname)
199			if desc not in (None, ""):
200			self._add_sub_el(node_el, 'Description', desc.decode('utf-8'))
201			# FIXME: add WriteMask and UserWriteMask
202			return node_el
203
204			def add_etree_object(self, node):
205			"""
206			Add a UA object element to the XML etree
207			"""
208			obj_el = self._add_node_common("UAObject", node)
209			var = node.get_attribute(ua.AttributeIds.EventNotifier)
210			if var.Value.Value != 0:
211			obj_el.attrib["EventNotifier"] = str(var.Value.Value)
212			self._add_ref_els(obj_el, node)
213
214			def add_etree_object_type(self, node):
215			"""
216			Add a UA object type element to the XML etree
217			"""
218			obj_el = self._add_node_common("UAObjectType", node)
219			abstract = node.get_attribute(ua.AttributeIds.IsAbstract).Value.Value
220			if abstract:
221			obj_el.attrib["IsAbstract"] = 'true'
222			self._add_ref_els(obj_el, node)
223
224			def add_variable_common(self, node, el):
225			dtype = node.get_data_type()
226			if dtype.Identifier in o_ids.ObjectIdNames:
227			dtype_name = o_ids.ObjectIdNames[dtype.Identifier]
228			self.aliases[dtype_name] = dtype.to_string()
229			else:
230			dtype_name = dtype.to_string()
231			rank = node.get_value_rank()
232			if rank != -1:
233			el.attrib["ValueRank"] = str(rank)
234			#var = node.get_attribute(ua.AttributeIds.ArrayDimensions())
235			#self._addobj_el.attrib["ArrayDimensions"] = str(var.Value.Value)
236			el.attrib["DataType"] = dtype_name
237			value_to_etree(el, dtype_name, dtype, node)
238
239			def add_etree_variable(self, node):
240			"""
241			Add a UA variable element to the XML etree
242			"""
243			var_el = self._add_node_common("UAVariable", node)
244			self.add_variable_common(node, var_el)
245
246			accesslevel = node.get_attribute(ua.AttributeIds.AccessLevel).Value.Value
247			useraccesslevel = node.get_attribute(ua.AttributeIds.UserAccessLevel).Value.Value
248
249			# We only write these values if they are different from defaults
250			# Not sure where default is defined....
251			if accesslevel not in (0, ua.AccessLevel.CurrentRead.mask):
252			var_el.attrib["AccessLevel"] = str(accesslevel)
253			if useraccesslevel not in (0, ua.AccessLevel.CurrentRead.mask):
254			var_el.attrib["UserAccessLevel"] = str(useraccesslevel)
255
256			var = node.get_attribute(ua.AttributeIds.MinimumSamplingInterval)
257			if var.Value.Value:
258			var_el.attrib["MinimumSamplingInterval"] = str(var.Value.Value)
259			var = node.get_attribute(ua.AttributeIds.Historizing)
260			if var.Value.Value:
261			var_el.attrib["Historizing"] = 'true'
262			self._add_ref_els(var_el, node)
263
264			def add_etree_variable_type(self, node):
265			"""
266			Add a UA variable type element to the XML etree
267			"""
268
269			var_el = self._add_node_common("UAVariableType", node)
270			self.add_variable_common(node, var_el)
271
272			abstract = node.get_attribute(ua.AttributeIds.IsAbstract)
273			if abstract.Value.Value:
274			var_el.attrib["IsAbstract"] = "true"
275
276			self._add_ref_els(var_el, node)
277
278			def add_etree_method(self, node):
279			obj_el = self._add_node_common("UAMethod", node)
280
281			var = node.get_attribute(ua.AttributeIds.Executable)
282			if var.Value.Value is False:
283			obj_el.attrib["Executable"] = "false"
284			var = node.get_attribute(ua.AttributeIds.UserExecutable)
285			if var.Value.Value is False:
286			obj_el.attrib["UserExecutable"] = "false"
287			self._add_ref_els(obj_el, node)
288
289			def add_etree_reference_type(self, obj):
290			obj_el = self._add_node_common("UAReferenceType", obj)
291			var = obj.get_attribute(ua.AttributeIds.InverseName)
292			if var is not None and var.Value.Value is not None:
293			self._add_sub_el(obj_el, 'InverseName', var.Value.Value.Text.decode('utf-8'))
294			self._add_ref_els(obj_el, obj)
295
296			def add_etree_datatype(self, obj):
297			"""
298			Add a UA data type element to the XML etree
299			"""
300			obj_el = self._add_node_common("UADataType", obj)
301			self._add_ref_els(obj_el, obj)
302
303			def _add_namespace_uri_els(self, uris):
304			nuris_el = Et.Element('NamespaceUris')
305
306			for uri in uris:
307			self._add_sub_el(nuris_el, 'Uri', uri)
308
309			self.etree.getroot().insert(0, nuris_el)
310
311			def _add_alias_els(self):
312			aliases_el = Et.Element('Aliases')
313
314			for k, v in self.aliases.items():
315			ref_el = Et.SubElement(aliases_el, 'Alias', Alias=k)
316			ref_el.text = v
317
318			self.etree.getroot().insert(0, aliases_el)
319
320			def _add_ref_els(self, parent_el, obj):
321			refs = obj.get_references()
322			refs_el = Et.SubElement(parent_el, 'References')
323
324			for ref in refs:
325			if ref.ReferenceTypeId.Identifier in o_ids.ObjectIdNames:
326			ref_name = o_ids.ObjectIdNames[ref.ReferenceTypeId.Identifier]
327			else:
328			ref_name = ref.ReferenceTypeId.to_string()
329			ref_el = Et.SubElement(refs_el, 'Reference')
330			ref_el.attrib['ReferenceType'] = ref_name
331			if not ref.IsForward:
332			ref_el.attrib['IsForward'] = 'false'
333			ref_el.text = self._node_to_string(ref.NodeId)
334
335			self.aliases[ref_name] = ref.ReferenceTypeId.to_string()
336
337
338			def member_to_etree(el, name, dtype, val):
339			member_el = Et.SubElement(el, "uax:" + name)
340			if isinstance(val, (list, tuple)):
341			for v in val:
342			_value_to_etree(member_el, ua.ObjectIdNames[dtype.Identifier], dtype, v)
343			else:
344			_val_to_etree(member_el, dtype, val)
345
346
347			def _val_to_etree(el, dtype, val):
348			if val is None:
349			val = ""
350			if dtype == ua.NodeId(ua.ObjectIds.NodeId):
351			id_el = Et.SubElement(el, "uax:Identifier")
352			id_el.text = val.to_string()
353			elif not hasattr(val, "ua_types"):
354			if isinstance(val, bytes):
355			el.text = val.decode("utf-8")
356			else:
357			el.text = str(val)
358			else:
359			for name, vtype in val.ua_types.items():
360			member_to_etree(el, name, ua.NodeId(getattr(ua.ObjectIds, vtype)), getattr(val, name))
361
362
363			def value_to_etree(el, dtype_name, dtype, node):
364			var = node.get_data_value().Value
365			if var.Value is not None:
366			val_el = Et.SubElement(el, 'Value')
367			_value_to_etree(val_el, dtype_name, dtype, var.Value)
368
369
370			def _value_to_etree(el, type_name, dtype, val):
371			if not val:
372			return
373			if isinstance(val, (list, tuple)):
374			if dtype.Identifier > 21: # this is an extentionObject:
375			elname = "uax:ListOfExtensionObject"
376			else:
377			elname = "uax:ListOf" + type_name
378			list_el = Et.SubElement(el, elname)
379			for nval in val:
380			_value_to_etree(list_el, type_name, dtype, nval)
381			else:
382			if dtype.Identifier > 21: # this is an extentionObject:
383			_extobj_to_etree(el, type_name, dtype, val)
384			else:
385			val_el = Et.SubElement(el, "uax:" + type_name)
386			_val_to_etree(val_el, dtype, val)
387
388
389			def _extobj_to_etree(val_el, name, dtype, val):
390			obj_el = Et.SubElement(val_el, "uax:ExtensionObject")
391			type_el = Et.SubElement(obj_el, "uax:TypeId")
392			id_el = Et.SubElement(type_el, "uax:Identifier")
393			id_el.text = dtype.to_string()
394			body_el = Et.SubElement(obj_el, "uax:Body")
395			struct_el = Et.SubElement(body_el, "uax:" + name)
396			for name, vtype in val.ua_types.items():
397			member_to_etree(struct_el, name, ua.NodeId(getattr(ua.ObjectIds, vtype)), getattr(val, name))
398
399
400			def indent(elem, level=0):
401			'''
402			copy and paste from http://effbot.org/zone/element-lib.htm#prettyprint
403			it basically walks your tree and adds spaces and newlines so the tree is
404			printed in a nice way
405			'''
406			i = "\n" + level * " "
407			if len(elem):
408			if not elem.text or not elem.text.strip():
409			elem.text = i + " "
410			if not elem.tail or not elem.tail.strip():
411			elem.tail = i
412			for elem in elem:
413			indent(elem, level + 1)
414			if not elem.tail or not elem.tail.strip():
415			elem.tail = i
416			else:
417			if level and (not elem.tail or not elem.tail.strip()):
418			elem.tail = i
419

FreeOpcUa / python-opcua

Pull Request — master (#327)

XmlExporter.add_variable_common() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like