XmlImporter.add_object() - 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-10 07:17 UTC

XmlImporter.add_object() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	10
Ratio	90.91 %

Code Coverage

Tests	7
CRAP Score	2.108

Importance

Changes

Metric	Value
cc	2
dl	10
loc	11
ccs	7
cts	10
cp	0.7
crap	2.108
rs	9.4285
c	0
b	0
f	0

"""
add node defined in XML to address space
format is the one from opc-ua specification
"""
import logging


from opcua import ua
from opcua.common import xmlparser


def ua_type_to_python(val, uatype):
    if uatype.startswith("Int") or uatype.startswith("UInt"):
        return int(val)
    elif uatype in ("String"):
        return val
    elif uatype in ("Bytes", "Bytes", "ByteString", "ByteArray"):
        return bytes(val, 'utf8')
    else:
        raise Exception("uatype nopt handled", uatype, " for val ", val)


def to_python(val, obj, attname):
    print("TO PYTHON", val, obj, attname)
    if isinstance(obj, ua.NodeId) and attname == "Identifier":
        return ua.NodeId.from_string(val)
    else:
        return ua_type_to_python(val, obj.ua_types[attname])


class XmlImporter(object):

    def __init__(self, server):
        self.logger = logging.getLogger(__name__)
        self.parser = None
        self.server = server

    def import_xml(self, xmlpath, act_server):
        """
        import xml and return added nodes
        """
        self.logger.info("Importing XML file %s", xmlpath)
        self.parser = xmlparser.XMLParser(xmlpath, act_server)
        nodes = []
        for nodedata in self.parser:
            if nodedata.nodetype == 'UAObject':
                node = self.add_object(nodedata)
            elif nodedata.nodetype == 'UAObjectType':
                node = self.add_object_type(nodedata)
            elif nodedata.nodetype == 'UAVariable':
                node = self.add_variable(nodedata)
            elif nodedata.nodetype == 'UAVariableType':
                node = self.add_variable_type(nodedata)
            elif nodedata.nodetype == 'UAReferenceType':
                node = self.add_reference(nodedata)
            elif nodedata.nodetype == 'UADataType':
                node = self.add_datatype(nodedata)
            elif nodedata.nodetype == 'UAMethod':
                node = self.add_method(nodedata)
            else:
                self.logger.warning("Not implemented node type: %s ", nodedata.nodetype)
                continue
            nodes.append(node)
        return nodes

    def _get_node(self, obj):
        node = ua.AddNodesItem()
        node.RequestedNewNodeId = ua.NodeId.from_string(obj.nodeid)
        node.BrowseName = ua.QualifiedName.from_string(obj.browsename)
        node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
        if obj.parent:
            node.ParentNodeId = ua.NodeId.from_string(obj.parent)
        if obj.parentlink:

            node.ReferenceTypeId = self.to_nodeid(obj.parentlink)
        if obj.typedef:
            node.TypeDefinition = ua.NodeId.from_string(obj.typedef)
        return node

    def to_nodeid(self, nodeid):
        if not nodeid:
            return ua.NodeId(ua.ObjectIds.String)
        elif "=" in nodeid:
            return ua.NodeId.from_string(nodeid)
        elif hasattr(ua.ObjectIds, nodeid):
            return ua.NodeId(getattr(ua.ObjectIds, nodeid))

        else:
            if nodeid in self.parser.aliases:
                nodeid = self.parser.aliases[nodeid]
            else:
                nodeid = "i={}".format(getattr(ua.ObjectIds, nodeid))
            return ua.NodeId.from_string(nodeid)

    def add_object(self, obj):
        node = self._get_node(obj)
        attrs = ua.ObjectAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.EventNotifier = obj.eventnotifier
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def add_object_type(self, obj):
        node = self._get_node(obj)
        attrs = ua.ObjectTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.IsAbstract = obj.abstract
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def add_variable(self, obj):
        node = self._get_node(obj)
        attrs = ua.VariableAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.DataType = self.to_nodeid(obj.datatype)
        # if obj.value and len(obj.value) == 1:
        if obj.value is not None:
            attrs.Value = self._add_variable_value(obj, )
        if obj.rank:
            attrs.ValueRank = obj.rank
        if obj.accesslevel:
            attrs.AccessLevel = obj.accesslevel
        if obj.useraccesslevel:
            attrs.UserAccessLevel = obj.useraccesslevel
        if obj.minsample:
            attrs.MinimumSamplingInterval = obj.minsample
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId
    
    def _make_ext_obj(sefl, obj):
        ext = getattr(ua, obj.objname)()
        for name, val in obj.body.items():
            if type(val) is str:
                raise Exception("Error val should a dict", name, val)
            else:
                print("\n", name,"\n")
                for attname, v in val.items():
                    print("\n", attname,"\n")
                    if type(v) is str:
                        print("Setting1 ", ext, attname, v)
                        setattr(ext, attname, to_python(v, ext, attname))
                    else:
                        for attname2, v2 in v.items():
                            obj2 = getattr(ext, attname)
                            print("FØR", obj2)
                            print("Setting1 ", obj2, attname2, v)
                            setattr(obj2, attname2, to_python(v2, obj2, attname2))
                            print("ETTER", getattr(ext, attname))
        return ext

    def _add_variable_value(self, obj):
        """
        Returns the value for a Variable based on the objects valuetype. 
        """
        if obj.valuetype == 'ListOfExtensionObject':
            values = []
            for ext in obj.value:
                extobj = self._make_ext_obj(ext)
                values.append(extobj)
            return values
        elif obj.valuetype.startswith("ListOf"):
            vtype = obj.valuetype[6:]
            return [getattr(ua, vtype)(v) for v in obj.value]
        else:
            return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))

    def add_variable_type(self, obj):
        node = self._get_node(obj)
        attrs = ua.VariableTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        attrs.DataType = self.to_nodeid(obj.datatype)
        if obj.value and len(obj.value) == 1:
            attrs.Value = obj.value[0]
        if obj.rank:
            attrs.ValueRank = obj.rank
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def add_method(self, obj):
        node = self._get_node(obj)
        attrs = ua.MethodAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj.accesslevel:
            attrs.AccessLevel = obj.accesslevel
        if obj.useraccesslevel:

            attrs.UserAccessLevel = obj.useraccesslevel
        if obj.minsample:
            attrs.MinimumSamplingInterval = obj.minsample
        if obj.dimensions:
            attrs.ArrayDimensions = obj.dimensions
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def add_reference(self, obj):
        node = self._get_node(obj)

        attrs = ua.ReferenceTypeAttributes()
        if obj.desc:
            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj. inversename:
            attrs.InverseName = ua.LocalizedText(obj.inversename)
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        if obj.symmetric:
            attrs.Symmetric = obj.symmetric
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def add_datatype(self, obj):
        node = self._get_node(obj)
        attrs = ua.DataTypeAttributes()
        if obj.desc:

            attrs.Description = ua.LocalizedText(obj.desc)
        attrs.DisplayName = ua.LocalizedText(obj.displayname)
        if obj.abstract:
            attrs.IsAbstract = obj.abstract
        node.NodeAttributes = attrs
        res = self.server.add_nodes([node])
        self._add_refs(obj)
        return res[0].AddedNodeId

    def _add_refs(self, obj):
        if not obj.refs:
            return
        refs = []
        for data in obj.refs:
            ref = ua.AddReferencesItem()
            ref.IsForward = True
            ref.ReferenceTypeId = self.to_nodeid(data.reftype)
            ref.SourceNodeId = ua.NodeId.from_string(obj.nodeid)
            ref.TargetNodeClass = ua.NodeClass.DataType
            ref.TargetNodeId = ua.NodeId.from_string(data.target)
            refs.append(ref)
        self.server.add_references(refs)


1			"""
2			add node defined in XML to address space
3			format is the one from opc-ua specification
4			"""
5	1		import logging
6
7
8	1		from opcua import ua
9	1		from opcua.common import xmlparser
10
11
12	1		def ua_type_to_python(val, uatype):
13			if uatype.startswith("Int") or uatype.startswith("UInt"):
14	1		return int(val)
15	1		elif uatype in ("String"):
16	1		return val
17	1		elif uatype in ("Bytes", "Bytes", "ByteString", "ByteArray"):
18			return bytes(val, 'utf8')
19	1		else:
20	1		raise Exception("uatype nopt handled", uatype, " for val ", val)
21	1
22	1
23	1		def to_python(val, obj, attname):
24	1		print("TO PYTHON", val, obj, attname)
25	1		if isinstance(obj, ua.NodeId) and attname == "Identifier":
26			return ua.NodeId.from_string(val)
27	1		else:
28	1		return ua_type_to_python(val, obj.ua_types[attname])
29
30
31			class XmlImporter(object):
32
33			def __init__(self, server):
34			self.logger = logging.getLogger(__name__)
35			self.parser = None
36			self.server = server
37
38			def import_xml(self, xmlpath, act_server):
39			"""
40	1		import xml and return added nodes
41	1		"""
42	1		self.logger.info("Importing XML file %s", xmlpath)
43	1		self.parser = xmlparser.XMLParser(xmlpath, act_server)
44	1		nodes = []
45	1		for nodedata in self.parser:
46	1		if nodedata.nodetype == 'UAObject':
47	1		node = self.add_object(nodedata)
48	1		elif nodedata.nodetype == 'UAObjectType':
49	1		node = self.add_object_type(nodedata)
50	1		elif nodedata.nodetype == 'UAVariable':
51	1		node = self.add_variable(nodedata)
52			elif nodedata.nodetype == 'UAVariableType':
53	1		node = self.add_variable_type(nodedata)
54	1		elif nodedata.nodetype == 'UAReferenceType':
55			node = self.add_reference(nodedata)
56	1		elif nodedata.nodetype == 'UADataType':
57			node = self.add_datatype(nodedata)
58			elif nodedata.nodetype == 'UAMethod':
59	1		node = self.add_method(nodedata)
60			else:
61	1		self.logger.warning("Not implemented node type: %s ", nodedata.nodetype)
62	1		continue
63	1		nodes.append(node)
64			return nodes
65
66	1		def _get_node(self, obj):
67			node = ua.AddNodesItem()
68	1		node.RequestedNewNodeId = ua.NodeId.from_string(obj.nodeid)
69			node.BrowseName = ua.QualifiedName.from_string(obj.browsename)
70	1		node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
71	1		if obj.parent:
72	1		node.ParentNodeId = ua.NodeId.from_string(obj.parent)
73	1	View Code Duplication	if obj.parentlink:
			0 ignored issues – show Duplication introduced 2016-08-10 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
74	1		node.ReferenceTypeId = self.to_nodeid(obj.parentlink)
75	1		if obj.typedef:
76	1		node.TypeDefinition = ua.NodeId.from_string(obj.typedef)
77	1		return node
78	1
79	1		def to_nodeid(self, nodeid):
80			if not nodeid:
81	1		return ua.NodeId(ua.ObjectIds.String)
82			elif "=" in nodeid:
83			return ua.NodeId.from_string(nodeid)
84			elif hasattr(ua.ObjectIds, nodeid):
85		View Code Duplication	return ua.NodeId(getattr(ua.ObjectIds, nodeid))
			0 ignored issues – show Duplication introduced 2016-08-10 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
86			else:
87			if nodeid in self.parser.aliases:
88			nodeid = self.parser.aliases[nodeid]
89			else:
90			nodeid = "i={}".format(getattr(ua.ObjectIds, nodeid))
91			return ua.NodeId.from_string(nodeid)
92	1
93	1		def add_object(self, obj):
94	1		node = self._get_node(obj)
95	1		attrs = ua.ObjectAttributes()
96			if obj.desc:
97	1		attrs.Description = ua.LocalizedText(obj.desc)
98	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
99			attrs.EventNotifier = obj.eventnotifier
100	1		node.NodeAttributes = attrs
101	1		res = self.server.add_nodes([node])
102	1		self._add_refs(obj)
103	1		return res[0].AddedNodeId
104	1
105			def add_object_type(self, obj):
106	1		node = self._get_node(obj)
107			attrs = ua.ObjectTypeAttributes()
108	1		if obj.desc:
109			attrs.Description = ua.LocalizedText(obj.desc)
110	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
111			attrs.IsAbstract = obj.abstract
112	1		node.NodeAttributes = attrs
113	1		res = self.server.add_nodes([node])
114	1		self._add_refs(obj)
115			return res[0].AddedNodeId
116	1
117			def add_variable(self, obj):
118			node = self._get_node(obj)
119			attrs = ua.VariableAttributes()
120			if obj.desc:
121			attrs.Description = ua.LocalizedText(obj.desc)
122			attrs.DisplayName = ua.LocalizedText(obj.displayname)
123			attrs.DataType = self.to_nodeid(obj.datatype)
124			# if obj.value and len(obj.value) == 1:
125			if obj.value is not None:
126			attrs.Value = self._add_variable_value(obj, )
127			if obj.rank:
128			attrs.ValueRank = obj.rank
129			if obj.accesslevel:
130			attrs.AccessLevel = obj.accesslevel
131			if obj.useraccesslevel:
132			attrs.UserAccessLevel = obj.useraccesslevel
133			if obj.minsample:
134			attrs.MinimumSamplingInterval = obj.minsample
135	1		if obj.dimensions:
136			attrs.ArrayDimensions = obj.dimensions
137			node.NodeAttributes = attrs
138			res = self.server.add_nodes([node])
139			self._add_refs(obj)
140			return res[0].AddedNodeId
141
142			def _make_ext_obj(sefl, obj):
143			ext = getattr(ua, obj.objname)()
144			for name, val in obj.body.items():
145			if type(val) is str:
146			raise Exception("Error val should a dict", name, val)
147			else:
148			print("\n", name,"\n")
149			for attname, v in val.items():
150			print("\n", attname,"\n")
151			if type(v) is str:
152			print("Setting1 ", ext, attname, v)
153	1		setattr(ext, attname, to_python(v, ext, attname))
154			else:
155			for attname2, v2 in v.items():
156			obj2 = getattr(ext, attname)
157			print("FØR", obj2)
158			print("Setting1 ", obj2, attname2, v)
159			setattr(obj2, attname2, to_python(v2, obj2, attname2))
160			print("ETTER", getattr(ext, attname))
161			return ext
162
163			def _add_variable_value(self, obj):
164			"""
165			Returns the value for a Variable based on the objects valuetype.
166			"""
167			if obj.valuetype == 'ListOfExtensionObject':
168			values = []
169	1		for ext in obj.value:
170			extobj = self._make_ext_obj(ext)
171			values.append(extobj)
172			return values
173			elif obj.valuetype.startswith("ListOf"):
174			vtype = obj.valuetype[6:]
175			return [getattr(ua, vtype)(v) for v in obj.value]
176			else:
177			return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
178
179			def add_variable_type(self, obj):
180			node = self._get_node(obj)
181	1		attrs = ua.VariableTypeAttributes()
182	1		if obj.desc:
183	1		attrs.Description = ua.LocalizedText(obj.desc)
184			attrs.DisplayName = ua.LocalizedText(obj.displayname)
185			attrs.DataType = self.to_nodeid(obj.datatype)
186			if obj.value and len(obj.value) == 1:
187			attrs.Value = obj.value[0]
188			if obj.rank:
189			attrs.ValueRank = obj.rank
190			if obj.abstract:
191			attrs.IsAbstract = obj.abstract
192			if obj.dimensions:
193			attrs.ArrayDimensions = obj.dimensions
194			node.NodeAttributes = attrs
195			res = self.server.add_nodes([node])
196			self._add_refs(obj)
197			return res[0].AddedNodeId
198
199			def add_method(self, obj):
200			node = self._get_node(obj)
201			attrs = ua.MethodAttributes()
202			if obj.desc:
203			attrs.Description = ua.LocalizedText(obj.desc)
204			attrs.DisplayName = ua.LocalizedText(obj.displayname)
205			if obj.accesslevel:
206			attrs.AccessLevel = obj.accesslevel
207		View Code Duplication	if obj.useraccesslevel:
			0 ignored issues – show Duplication introduced 2016-08-10 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
208			attrs.UserAccessLevel = obj.useraccesslevel
209			if obj.minsample:
210			attrs.MinimumSamplingInterval = obj.minsample
211			if obj.dimensions:
212			attrs.ArrayDimensions = obj.dimensions
213			node.NodeAttributes = attrs
214			res = self.server.add_nodes([node])
215			self._add_refs(obj)
216			return res[0].AddedNodeId
217
218			def add_reference(self, obj):
219		View Code Duplication	node = self._get_node(obj)
			0 ignored issues – show Duplication introduced 2016-08-10 08:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
220			attrs = ua.ReferenceTypeAttributes()
221			if obj.desc:
222			attrs.Description = ua.LocalizedText(obj.desc)
223			attrs.DisplayName = ua.LocalizedText(obj.displayname)
224			if obj. inversename:
225			attrs.InverseName = ua.LocalizedText(obj.inversename)
226			if obj.abstract:
227			attrs.IsAbstract = obj.abstract
228			if obj.symmetric:
229			attrs.Symmetric = obj.symmetric
230			node.NodeAttributes = attrs
231			res = self.server.add_nodes([node])
232			self._add_refs(obj)
233			return res[0].AddedNodeId
234
235			def add_datatype(self, obj):
236			node = self._get_node(obj)
237			attrs = ua.DataTypeAttributes()
238		View Code Duplication	if obj.desc:
			0 ignored issues – show Duplication introduced 2016-09-05 18:21 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
239			attrs.Description = ua.LocalizedText(obj.desc)
240			attrs.DisplayName = ua.LocalizedText(obj.displayname)
241			if obj.abstract:
242			attrs.IsAbstract = obj.abstract
243			node.NodeAttributes = attrs
244			res = self.server.add_nodes([node])
245			self._add_refs(obj)
246			return res[0].AddedNodeId
247
248			def _add_refs(self, obj):
249			if not obj.refs:
250			return
251			refs = []
252			for data in obj.refs:
253			ref = ua.AddReferencesItem()
254			ref.IsForward = True
255			ref.ReferenceTypeId = self.to_nodeid(data.reftype)
256			ref.SourceNodeId = ua.NodeId.from_string(obj.nodeid)
257			ref.TargetNodeClass = ua.NodeClass.DataType
258			ref.TargetNodeId = ua.NodeId.from_string(data.target)
259			refs.append(ref)
260			self.server.add_references(refs)
261

FreeOpcUa / python-opcua

Pull Request — master (#296)

XmlImporter.add_object() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like