XmlImporter.to_nodeid() - Code Metrics - Inspection of "Support xml address space import with custom data..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 5a5e42...feb0f4 )

by Olivier

created 2016-06-30 11:26 UTC

XmlImporter.to_nodeid() B

↳ Parent: XmlImporter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	5.675

Importance

Changes

Metric	Value
cc	5
dl	0
loc	13
ccs	7
cts	10
cp	0.7
crap	5.675
rs	8.5454
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):
        self.logger.info("Importing XML file %s", xmlpath)
        self.parser = xmlparser.XMLParser(xmlpath)
        for node in self.parser:
            if node.nodetype == 'UAObject':
                self.add_object(node)
            elif node.nodetype == 'UAObjectType':
                self.add_object_type(node)
            elif node.nodetype == 'UAVariable':
                self.add_variable(node)
            elif node.nodetype == 'UAVariableType':
                self.add_variable_type(node)
            elif node.nodetype == 'UAReferenceType':
                self.add_reference(node)
            elif node.nodetype == 'UADataType':
                self.add_datatype(node)
            elif node.nodetype == 'UAMethod':
                self.add_method(node)
            else:
                self.logger.info("Not implemented node type: %s ", node.nodetype)

    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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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 = ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
        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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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
        self.server.add_nodes([node])
        self._add_refs(obj)

    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):
20	1		self.logger.info("Importing XML file %s", xmlpath)
21	1		self.parser = xmlparser.XMLParser(xmlpath)
22	1		for node in self.parser:
23	1		if node.nodetype == 'UAObject':
24	1		self.add_object(node)
25	1		elif node.nodetype == 'UAObjectType':
26			self.add_object_type(node)
27	1		elif node.nodetype == 'UAVariable':
28	1		self.add_variable(node)
29			elif node.nodetype == 'UAVariableType':
30			self.add_variable_type(node)
31			elif node.nodetype == 'UAReferenceType':
32			self.add_reference(node)
33			elif node.nodetype == 'UADataType':
34			self.add_datatype(node)
35			elif node.nodetype == 'UAMethod':
36			self.add_method(node)
37			else:
38			self.logger.info("Not implemented node type: %s ", node.nodetype)
39
40	1		def _get_node(self, obj):
41	1		node = ua.AddNodesItem()
42	1		node.RequestedNewNodeId = ua.NodeId.from_string(obj.nodeid)
43	1		node.BrowseName = ua.QualifiedName.from_string(obj.browsename)
44	1		node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
45	1		if obj.parent:
46	1		node.ParentNodeId = ua.NodeId.from_string(obj.parent)
47	1		if obj.parentlink:
48	1		node.ReferenceTypeId = self.to_nodeid(obj.parentlink)
49	1		if obj.typedef:
50	1		node.TypeDefinition = ua.NodeId.from_string(obj.typedef)
51	1		return node
52
53	1		def to_nodeid(self, nodeid):
54	1		if not nodeid:
55			return ua.NodeId(ua.ObjectIds.String)
56	1		elif "=" in nodeid:
57			return ua.NodeId.from_string(nodeid)
58			elif hasattr(ua.ObjectIds, nodeid):
59	1		return ua.NodeId(getattr(ua.ObjectIds, nodeid))
60			else:
61	1		if nodeid in self.parser.aliases:
62	1		nodeid = self.parser.aliases[nodeid]
63	1		else:
64			nodeid = "i={}".format(getattr(ua.ObjectIds, nodeid))
65			return ua.NodeId.from_string(nodeid)
66	1
67			def add_object(self, obj):
68	1		node = self._get_node(obj)
69			attrs = ua.ObjectAttributes()
70	1		if obj.desc:
71	1		attrs.Description = ua.LocalizedText(obj.desc)
72	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
73	1		attrs.EventNotifier = obj.eventnotifier
74	1		node.NodeAttributes = attrs
75	1		self.server.add_nodes([node])
76	1		self._add_refs(obj)
77	1
78	1		def add_object_type(self, obj):
79	1		node = self._get_node(obj)
80			attrs = ua.ObjectTypeAttributes()
81	1		if obj.desc:
82			attrs.Description = ua.LocalizedText(obj.desc)
83			attrs.DisplayName = ua.LocalizedText(obj.displayname)
84			attrs.IsAbstract = obj.abstract
85			node.NodeAttributes = attrs
86			self.server.add_nodes([node])
87			self._add_refs(obj)
88
89			def add_variable(self, obj):
90			node = self._get_node(obj)
91			attrs = ua.VariableAttributes()
92	1		if obj.desc:
93	1		attrs.Description = ua.LocalizedText(obj.desc)
94	1		attrs.DisplayName = ua.LocalizedText(obj.displayname)
95	1		attrs.DataType = self.to_nodeid(obj.datatype)
96			# if obj.value and len(obj.value) == 1:
97	1		if obj.value is not None:
98	1		attrs.Value = ua.Variant(obj.value, getattr(ua.VariantType, obj.valuetype))
99			if obj.rank:
100	1		attrs.ValueRank = obj.rank
101	1		if obj.accesslevel:
102	1		attrs.AccessLevel = obj.accesslevel
103	1		if obj.useraccesslevel:
104	1		attrs.UserAccessLevel = obj.useraccesslevel
105			if obj.minsample:
106	1		attrs.MinimumSamplingInterval = obj.minsample
107			if obj.dimensions:
108	1		attrs.ArrayDimensions = obj.dimensions
109			node.NodeAttributes = attrs
110	1		self.server.add_nodes([node])
111			self._add_refs(obj)
112	1
113	1	View Code Duplication	def add_variable_type(self, obj):
			0 ignored issues – show Duplication introduced 2016-06-30 08:37 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
114	1		node = self._get_node(obj)
115			attrs = ua.VariableTypeAttributes()
116	1		if obj.desc:
117			attrs.Description = ua.LocalizedText(obj.desc)
118			attrs.DisplayName = ua.LocalizedText(obj.displayname)
119			attrs.DataType = self.to_nodeid(obj.datatype)
120			if obj.value and len(obj.value) == 1:
121			attrs.Value = obj.value[0]
122			if obj.rank:
123			attrs.ValueRank = obj.rank
124			if obj.abstract:
125			attrs.IsAbstract = obj.abstract
126			if obj.dimensions:
127			attrs.ArrayDimensions = obj.dimensions
128			node.NodeAttributes = attrs
129			self.server.add_nodes([node])
130			self._add_refs(obj)
131
132		View Code Duplication	def add_method(self, obj):
			0 ignored issues – show Duplication introduced 2016-06-30 08:37 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
133			node = self._get_node(obj)
134			attrs = ua.MethodAttributes()
135	1		if obj.desc:
136			attrs.Description = ua.LocalizedText(obj.desc)
137			attrs.DisplayName = ua.LocalizedText(obj.displayname)
138			if obj.accesslevel:
139			attrs.AccessLevel = obj.accesslevel
140			if obj.useraccesslevel:
141			attrs.UserAccessLevel = obj.useraccesslevel
142			if obj.minsample:
143			attrs.MinimumSamplingInterval = obj.minsample
144			if obj.dimensions:
145			attrs.ArrayDimensions = obj.dimensions
146			node.NodeAttributes = attrs
147			self.server.add_nodes([node])
148			self._add_refs(obj)
149
150			def add_reference(self, obj):
151			node = self._get_node(obj)
152			attrs = ua.ReferenceTypeAttributes()
153	1		if obj.desc:
154			attrs.Description = ua.LocalizedText(obj.desc)
155			attrs.DisplayName = ua.LocalizedText(obj.displayname)
156			if obj. inversename:
157			attrs.InverseName = ua.LocalizedText(obj.inversename)
158			if obj.abstract:
159			attrs.IsAbstract = obj.abstract
160			if obj.symmetric:
161			attrs.Symmetric = obj.symmetric
162			node.NodeAttributes = attrs
163			self.server.add_nodes([node])
164			self._add_refs(obj)
165
166			def add_datatype(self, obj):
167			node = self._get_node(obj)
168			attrs = ua.DataTypeAttributes()
169	1		if obj.desc:
170			attrs.Description = ua.LocalizedText(obj.desc)
171			attrs.DisplayName = ua.LocalizedText(obj.displayname)
172			if obj.abstract:
173			attrs.IsAbstract = obj.abstract
174			node.NodeAttributes = attrs
175			self.server.add_nodes([node])
176			self._add_refs(obj)
177
178			def _add_refs(self, obj):
179			if not obj.refs:
180			return
181	1		refs = []
182	1		for data in obj.refs:
183	1		ref = ua.AddReferencesItem()
184			ref.IsForward = True
185			ref.ReferenceTypeId = self.to_nodeid(data.reftype)
186			ref.SourceNodeId = ua.NodeId.from_string(obj.nodeid)
187			ref.TargetNodeClass = ua.NodeClass.DataType
188			ref.TargetNodeId = ua.NodeId.from_string(data.target)
189			refs.append(ref)
190			self.server.add_references(refs)
191

FreeOpcUa / python-opcua

Push — master ( 5a5e42...feb0f4 )

XmlImporter.to_nodeid() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like