XmlImporter._add_refs() - Code Metrics - Inspection of "fix typo, avoid reassigning variable in importer" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( ed2734...33fbe2 )

by Olivier

created 2016-08-21 17:44 UTC

XmlImporter._add_refs() A

↳ Parent: XmlImporter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
cc	3
dl	0
loc	13
ccs	0
cts	0
cp	0
crap	12
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


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 _add_variable_value(self, obj):
        """
        Returns the value for a Variable based on the objects valuetype. 
        """
        if obj.valuetype == 'ListOfLocalizedText':
            return ua.Variant([ua.LocalizedText(txt) for txt in obj.value], None)
        elif obj.valuetype == 'EnumValueType':
            values = []
            for ev in obj.value:
                enum_value = ua.EnumValueType()
                enum_value.DisplayName = ua.LocalizedText(ev['DisplayName'])
                enum_value.Description = ua.LocalizedText(ev['Description'])
                enum_value.Value = int(ev['Value'])
                values.append(enum_value)
            return values
        elif obj.valuetype == 'Argument':
            values = []
            for arg in obj.value:
                argument = ua.Argument()
                argument.Name = arg['Name']
                argument.Description = ua.LocalizedText(arg['Description'])
                argument.DataType = self.to_nodeid(arg['DataType'])
                argument.ValueRank = int(arg['ValueRank'])
                argument.ArrayDimensions = arg['ArrayDimensions']
                values.append(argument)
            return values

        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		class XmlImporter(object):
13
14	1		def __init__(self, server):
15	1		self.logger = logging.getLogger(__name__)
16	1		self.parser = None
17	1		self.server = server
18
19	1		def import_xml(self, xmlpath, act_server):
20	1		"""
21	1		import xml and return added nodes
22	1		"""
23	1		self.logger.info("Importing XML file %s", xmlpath)
24	1		self.parser = xmlparser.XMLParser(xmlpath, act_server)
25	1		nodes = []
26			for nodedata in self.parser:
27	1		if nodedata.nodetype == 'UAObject':
28	1		node = self.add_object(nodedata)
29			elif nodedata.nodetype == 'UAObjectType':
30			node = self.add_object_type(nodedata)
31			elif nodedata.nodetype == 'UAVariable':
32			node = self.add_variable(nodedata)
33			elif nodedata.nodetype == 'UAVariableType':
34			node = self.add_variable_type(nodedata)
35			elif nodedata.nodetype == 'UAReferenceType':
36			node = self.add_reference(nodedata)
37			elif nodedata.nodetype == 'UADataType':
38			node = self.add_datatype(nodedata)
39			elif nodedata.nodetype == 'UAMethod':
40	1		node = self.add_method(nodedata)
41	1		else:
42	1		self.logger.warning("Not implemented node type: %s ", nodedata.nodetype)
43	1		continue
44	1		nodes.append(node)
45	1		return nodes
46	1
47	1		def _get_node(self, obj):
48	1		node = ua.AddNodesItem()
49	1		node.RequestedNewNodeId = ua.NodeId.from_string(obj.nodeid)
50	1		node.BrowseName = ua.QualifiedName.from_string(obj.browsename)
51	1		node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
52			if obj.parent:
53	1		node.ParentNodeId = ua.NodeId.from_string(obj.parent)
54	1		if obj.parentlink:
55			node.ReferenceTypeId = self.to_nodeid(obj.parentlink)
56	1		if obj.typedef:
57			node.TypeDefinition = ua.NodeId.from_string(obj.typedef)
58			return node
59	1
60			def to_nodeid(self, nodeid):
61	1		if not nodeid:
62	1		return ua.NodeId(ua.ObjectIds.String)
63	1		elif "=" in nodeid:
64			return ua.NodeId.from_string(nodeid)
65			elif hasattr(ua.ObjectIds, nodeid):
66	1		return ua.NodeId(getattr(ua.ObjectIds, nodeid))
67			else:
68	1		if nodeid in self.parser.aliases:
69			nodeid = self.parser.aliases[nodeid]
70	1		else:
71	1		nodeid = "i={}".format(getattr(ua.ObjectIds, nodeid))
72	1		return ua.NodeId.from_string(nodeid)
73	1	View Code Duplication
			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		def add_object(self, obj):
75	1		node = self._get_node(obj)
76	1		attrs = ua.ObjectAttributes()
77	1		if obj.desc:
78	1		attrs.Description = ua.LocalizedText(obj.desc)
79	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
80			attrs.EventNotifier = obj.eventnotifier
81	1		node.NodeAttributes = attrs
82			res = self.server.add_nodes([node])
83			self._add_refs(obj)
84			return res[0].AddedNodeId
85		View Code Duplication
			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			def add_object_type(self, obj):
87			node = self._get_node(obj)
88			attrs = ua.ObjectTypeAttributes()
89			if obj.desc:
90			attrs.Description = ua.LocalizedText(obj.desc)
91			attrs.DisplayName = ua.LocalizedText(obj.displayname)
92	1		attrs.IsAbstract = obj.abstract
93	1		node.NodeAttributes = attrs
94	1		res = self.server.add_nodes([node])
95	1		self._add_refs(obj)
96			return res[0].AddedNodeId
97	1
98	1		def add_variable(self, obj):
99			node = self._get_node(obj)
100	1		attrs = ua.VariableAttributes()
101	1		if obj.desc:
102	1		attrs.Description = ua.LocalizedText(obj.desc)
103	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
104	1		attrs.DataType = self.to_nodeid(obj.datatype)
105			# if obj.value and len(obj.value) == 1:
106	1		if obj.value is not None:
107			attrs.Value = self._add_variable_value(obj, )
108	1		if obj.rank:
109			attrs.ValueRank = obj.rank
110	1		if obj.accesslevel:
111			attrs.AccessLevel = obj.accesslevel
112	1		if obj.useraccesslevel:
113	1		attrs.UserAccessLevel = obj.useraccesslevel
114	1		if obj.minsample:
115			attrs.MinimumSamplingInterval = obj.minsample
116	1		if obj.dimensions:
117			attrs.ArrayDimensions = obj.dimensions
118			node.NodeAttributes = attrs
119			res = self.server.add_nodes([node])
120			self._add_refs(obj)
121			return res[0].AddedNodeId
122
123			def _add_variable_value(self, obj):
124			"""
125			Returns the value for a Variable based on the objects valuetype.
126			"""
127			if obj.valuetype == 'ListOfLocalizedText':
128			return ua.Variant([ua.LocalizedText(txt) for txt in obj.value], None)
129			elif obj.valuetype == 'EnumValueType':
130			values = []
131			for ev in obj.value:
132			enum_value = ua.EnumValueType()
133			enum_value.DisplayName = ua.LocalizedText(ev['DisplayName'])
134			enum_value.Description = ua.LocalizedText(ev['Description'])
135	1		enum_value.Value = int(ev['Value'])
136			values.append(enum_value)
137			return values
138			elif obj.valuetype == 'Argument':
139			values = []
140			for arg in obj.value:
141			argument = ua.Argument()
142			argument.Name = arg['Name']
143			argument.Description = ua.LocalizedText(arg['Description'])
144			argument.DataType = self.to_nodeid(arg['DataType'])
145			argument.ValueRank = int(arg['ValueRank'])
146			argument.ArrayDimensions = arg['ArrayDimensions']
147			values.append(argument)
148			return values
149
150			return ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
151
152
153	1		def add_variable_type(self, obj):
154			node = self._get_node(obj)
155			attrs = ua.VariableTypeAttributes()
156			if obj.desc:
157			attrs.Description = ua.LocalizedText(obj.desc)
158			attrs.DisplayName = ua.LocalizedText(obj.displayname)
159			attrs.DataType = self.to_nodeid(obj.datatype)
160			if obj.value and len(obj.value) == 1:
161			attrs.Value = obj.value[0]
162			if obj.rank:
163			attrs.ValueRank = obj.rank
164			if obj.abstract:
165			attrs.IsAbstract = obj.abstract
166			if obj.dimensions:
167			attrs.ArrayDimensions = obj.dimensions
168			node.NodeAttributes = attrs
169	1		res = self.server.add_nodes([node])
170			self._add_refs(obj)
171			return res[0].AddedNodeId
172
173			def add_method(self, obj):
174			node = self._get_node(obj)
175			attrs = ua.MethodAttributes()
176			if obj.desc:
177			attrs.Description = ua.LocalizedText(obj.desc)
178			attrs.DisplayName = ua.LocalizedText(obj.displayname)
179			if obj.accesslevel:
180			attrs.AccessLevel = obj.accesslevel
181	1		if obj.useraccesslevel:
182	1		attrs.UserAccessLevel = obj.useraccesslevel
183	1		if obj.minsample:
184			attrs.MinimumSamplingInterval = obj.minsample
185			if obj.dimensions:
186			attrs.ArrayDimensions = obj.dimensions
187			node.NodeAttributes = attrs
188			res = self.server.add_nodes([node])
189			self._add_refs(obj)
190			return res[0].AddedNodeId
191
192			def add_reference(self, obj):
193			node = self._get_node(obj)
194			attrs = ua.ReferenceTypeAttributes()
195			if obj.desc:
196			attrs.Description = ua.LocalizedText(obj.desc)
197			attrs.DisplayName = ua.LocalizedText(obj.displayname)
198			if obj. inversename:
199			attrs.InverseName = ua.LocalizedText(obj.inversename)
200			if obj.abstract:
201			attrs.IsAbstract = obj.abstract
202			if obj.symmetric:
203			attrs.Symmetric = obj.symmetric
204			node.NodeAttributes = attrs
205			res = self.server.add_nodes([node])
206			self._add_refs(obj)
207		View Code Duplication	return res[0].AddedNodeId
			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
209			def add_datatype(self, obj):
210			node = self._get_node(obj)
211			attrs = ua.DataTypeAttributes()
212			if obj.desc:
213			attrs.Description = ua.LocalizedText(obj.desc)
214			attrs.DisplayName = ua.LocalizedText(obj.displayname)
215			if obj.abstract:
216			attrs.IsAbstract = obj.abstract
217			node.NodeAttributes = attrs
218			res = self.server.add_nodes([node])
219			self._add_refs(obj)
220			return res[0].AddedNodeId
221
222			def _add_refs(self, obj):
223			if not obj.refs:
224			return
225			refs = []
226			for data in obj.refs:
227			ref = ua.AddReferencesItem()
228			ref.IsForward = True
229			ref.ReferenceTypeId = self.to_nodeid(data.reftype)
230			ref.SourceNodeId = ua.NodeId.from_string(obj.nodeid)
231			ref.TargetNodeClass = ua.NodeClass.DataType
232			ref.TargetNodeId = ua.NodeId.from_string(data.target)
233			refs.append(ref)
234			self.server.add_references(refs)
235

FreeOpcUa / python-opcua

Push — master ( ed2734...33fbe2 )

XmlImporter._add_refs() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like