opcua.common.delete_nodes() - Code Metrics - Inspection of "fix delete_nodes and more consistent naming" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#124)

by Olivier

created 2016-01-25 16:40 UTC

opcua.common.delete_nodes() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	3

Metric	Value
cc	3
dl	0
loc	14
ccs	10
cts	10
cp	1
crap	3
rs	9.4285

"""
High level functions to create nodes
"""
from opcua import ua
from opcua.common import node


def _parse_add_args(*args):
    if isinstance(args[0], ua.NodeId):
        return args[0], args[1]
    elif isinstance(args[0], str):
        return ua.NodeId.from_string(args[0]), ua.QualifiedName.from_string(args[1])
    elif isinstance(args[0], int):
        return ua.generate_nodeid(args[0]), ua.QualifiedName(args[1], args[0])
    else:
        raise TypeError("Add methods takes a nodeid and a qualifiedname as argument, received %s" % args)


def create_folder(parent, *args):
    """
    create a child node folder
    arguments are nodeid, browsename
    or namespace index, name
    """
    nodeid, qname = _parse_add_args(*args)
    return node.Node(parent.server, _create_folder(parent.server, parent.nodeid, nodeid, qname))


def create_object(parent, *args):
    """
    create a child node object
    arguments are nodeid, browsename
    or namespace index, name
    """
    nodeid, qname = _parse_add_args(*args)
    return node.Node(parent.server, _create_object(parent.server, parent.nodeid, nodeid, qname))


def create_property(parent, *args):
    """
    create a child node property
    args are nodeid, browsename, value, [variant type]
    or idx, name, value, [variant type]
    """
    nodeid, qname = _parse_add_args(*args[:2])
    val = _to_variant(*args[2:])
    return node.Node(parent.server, _create_variable(parent.server, parent.nodeid, nodeid, qname, val, isproperty=True))


def create_variable(parent, *args):
    """
    create a child node variable
    args are nodeid, browsename, value, [variant type]
    or idx, name, value, [variant type]
    """
    nodeid, qname = _parse_add_args(*args[:2])
    val = _to_variant(*args[2:])
    return node.Node(parent.server, _create_variable(parent.server, parent.nodeid, nodeid, qname, val, isproperty=False))


def create_method(parent, *args):
    """
    create a child method object
    This is only possible on server side!!
    args are nodeid, browsename, method_to_be_called, [input argument types], [output argument types]
    or idx, name, method_to_be_called, [input argument types], [output argument types]
    if argument types is specified, child nodes advertising what arguments the method uses and returns will be created
    a callback is a method accepting the nodeid of the parent as first argument and variants after. returns a list of variants
    """
    nodeid, qname = _parse_add_args(*args[:2])
    callback = args[2]
    if len(args) > 3:
        inputs = args[3]
    if len(args) > 4:
        outputs = args[4]
    return _create_method(parent, nodeid, qname, callback, inputs, outputs)


def _create_folder(server, parentnodeid, nodeid, qname):
    addnode = ua.AddNodesItem()
    addnode.RequestedNewNodeId = nodeid
    addnode.BrowseName = qname
    addnode.NodeClass = ua.NodeClass.Object
    addnode.ParentNodeId = parentnodeid
    addnode.ReferenceTypeId = ua.NodeId.from_string("i=35")
    addnode.TypeDefinition = ua.NodeId.from_string("i=61")
    attrs = ua.ObjectAttributes()
    attrs.Description = ua.LocalizedText(qname.Name)
    attrs.DisplayName = ua.LocalizedText(qname.Name)
    attrs.WriteMask = ua.OpenFileMode.Read
    attrs.UserWriteMask = ua.OpenFileMode.Read
    attrs.EventNotifier = 0
    addnode.NodeAttributes = attrs
    results = server.add_nodes([addnode])
    results[0].StatusCode.check()
    return nodeid


def _create_object(server, parentnodeid, nodeid, qname):
    addnode = ua.AddNodesItem()
    addnode.RequestedNewNodeId = nodeid
    addnode.BrowseName = qname
    addnode.NodeClass = ua.NodeClass.Object
    addnode.ParentNodeId = parentnodeid
    addnode.ReferenceTypeId = ua.NodeId.from_string("i=35")
    addnode.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseObjectType)
    attrs = ua.ObjectAttributes()
    attrs.Description = ua.LocalizedText(qname.Name)
    attrs.DisplayName = ua.LocalizedText(qname.Name)
    attrs.EventNotifier = 0
    attrs.WriteMask = ua.OpenFileMode.Read
    attrs.UserWriteMask = ua.OpenFileMode.Read
    addnode.NodeAttributes = attrs
    results = server.add_nodes([addnode])
    results[0].StatusCode.check()
    return nodeid


def _to_variant(val, vtype=None):
    if isinstance(val, ua.Variant):
        return val
    else:
        return ua.Variant(val, vtype)


def _create_variable(server, parentnodeid, nodeid, qname, val, isproperty=False):
    addnode = ua.AddNodesItem()
    addnode.RequestedNewNodeId = nodeid
    addnode.BrowseName = qname
    addnode.NodeClass = ua.NodeClass.Variable
    addnode.ParentNodeId = parentnodeid
    if isproperty:
        addnode.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasProperty)
        addnode.TypeDefinition = ua.NodeId(ua.ObjectIds.PropertyType)
    else:
        addnode.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent)
        addnode.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseDataVariableType)
    attrs = ua.VariableAttributes()
    attrs.Description = ua.LocalizedText(qname.Name)
    attrs.DisplayName = ua.LocalizedText(qname.Name)
    attrs.DataType = _guess_uatype(val)
    attrs.Value = val
    if isinstance(val, list) or isinstance(val, tuple):
        attrs.ValueRank = ua.ValueRank.OneDimension
    else:
        attrs.ValueRank = ua.ValueRank.Scalar
    #attrs.ArrayDimensions = None
    attrs.WriteMask = ua.OpenFileMode.Read
    attrs.UserWriteMask = ua.OpenFileMode.Read
    attrs.Historizing = 0
    addnode.NodeAttributes = attrs
    results = server.add_nodes([addnode])
    results[0].StatusCode.check()
    return nodeid


def _create_method(parent, nodeid, qname, callback, inputs, outputs):
    addnode = ua.AddNodesItem()
    addnode.RequestedNewNodeId = nodeid
    addnode.BrowseName = qname
    addnode.NodeClass = ua.NodeClass.Method
    addnode.ParentNodeId = parent.nodeid
    addnode.ReferenceTypeId = ua.NodeId.from_string("i=47")
    #node.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseObjectType)
    attrs = ua.MethodAttributes()
    attrs.Description = ua.LocalizedText(qname.Name)
    attrs.DisplayName = ua.LocalizedText(qname.Name)
    attrs.WriteMask = ua.OpenFileMode.Read
    attrs.UserWriteMask = ua.OpenFileMode.Read
    attrs.Executable = True
    attrs.UserExecutable = True
    addnode.NodeAttributes = attrs
    results = parent.server.add_nodes([addnode])
    results[0].StatusCode.check()
    method = node.Node(parent.server, nodeid)
    if inputs:
        create_property(method, ua.generate_nodeid(qname.NamespaceIndex), ua.QualifiedName("InputArguments", 0), [_vtype_to_argument(vtype) for vtype in inputs])
    if outputs:
        create_property(method, ua.generate_nodeid(qname.NamespaceIndex), ua.QualifiedName("OutputArguments", 0), [_vtype_to_argument(vtype) for vtype in outputs])
    parent.server.add_method_callback(method.nodeid, callback)
    return nodeid


def _vtype_to_argument(vtype):
    if isinstance(vtype, ua.Argument):
        return vtype

    arg = ua.Argument()
    v = ua.Variant(None, vtype)
    arg.DataType = _guess_uatype(v)
    return arg


def _guess_uatype(variant):
    if variant.VariantType == ua.VariantType.ExtensionObject:
        if variant.Value is None:
            raise ua.UaError("Cannot guess DataType from Null ExtensionObject")
        if type(variant.Value) in (list, tuple):
            if len(variant.Value) == 0:
                raise ua.UaError("Cannot guess DataType from Null ExtensionObject")
            extobj = variant.Value[0]
        else:
            extobj = variant.Value
        classname = extobj.__class__.__name__
        return ua.NodeId(getattr(ua.ObjectIds, classname))
    else:
        return ua.NodeId(getattr(ua.ObjectIds, variant.VariantType.name))


def delete_nodes(server, nodes, recursive=False):
    """
    Delete specified nodes. Optionally delete recursively all nodes with a 
    downward hierachic references to the node
    """
    nodestodelete = []
    if recursive:
        nodes += _add_childs(nodes)
    for mynode in nodes:
        it = ua.DeleteNodesItem()
        it.NodeId = mynode.nodeid
        it.DeleteTargetReferences = True
        nodestodelete.append(it)
    return server.delete_nodes(nodestodelete)


def _add_childs(nodes):
    results = []
    for mynode in nodes[:]:
        results += mynode.get_children()
    return results




1		"""
2		High level functions to create nodes
3		"""
4	1	from opcua import ua
5	1	from opcua.common import node
6
7
8	1	def _parse_add_args(*args):
9	1	if isinstance(args[0], ua.NodeId):
10	1	return args[0], args[1]
11	1	elif isinstance(args[0], str):
12	1	return ua.NodeId.from_string(args[0]), ua.QualifiedName.from_string(args[1])
13	1	elif isinstance(args[0], int):
14	1	return ua.generate_nodeid(args[0]), ua.QualifiedName(args[1], args[0])
15		else:
16		raise TypeError("Add methods takes a nodeid and a qualifiedname as argument, received %s" % args)
17
18
19	1	def create_folder(parent, *args):
20		"""
21		create a child node folder
22		arguments are nodeid, browsename
23		or namespace index, name
24		"""
25	1	nodeid, qname = _parse_add_args(*args)
26	1	return node.Node(parent.server, _create_folder(parent.server, parent.nodeid, nodeid, qname))
27
28
29	1	def create_object(parent, *args):
30		"""
31		create a child node object
32		arguments are nodeid, browsename
33		or namespace index, name
34		"""
35	1	nodeid, qname = _parse_add_args(*args)
36	1	return node.Node(parent.server, _create_object(parent.server, parent.nodeid, nodeid, qname))
37
38
39	1	def create_property(parent, *args):
40		"""
41		create a child node property
42		args are nodeid, browsename, value, [variant type]
43		or idx, name, value, [variant type]
44		"""
45	1	nodeid, qname = _parse_add_args(*args[:2])
46	1	val = _to_variant(*args[2:])
47	1	return node.Node(parent.server, _create_variable(parent.server, parent.nodeid, nodeid, qname, val, isproperty=True))
48
49
50	1	def create_variable(parent, *args):
51		"""
52		create a child node variable
53		args are nodeid, browsename, value, [variant type]
54		or idx, name, value, [variant type]
55		"""
56	1	nodeid, qname = _parse_add_args(*args[:2])
57	1	val = _to_variant(*args[2:])
58	1	return node.Node(parent.server, _create_variable(parent.server, parent.nodeid, nodeid, qname, val, isproperty=False))
59
60
61	1	def create_method(parent, *args):
62		"""
63		create a child method object
64		This is only possible on server side!!
65		args are nodeid, browsename, method_to_be_called, [input argument types], [output argument types]
66		or idx, name, method_to_be_called, [input argument types], [output argument types]
67		if argument types is specified, child nodes advertising what arguments the method uses and returns will be created
68		a callback is a method accepting the nodeid of the parent as first argument and variants after. returns a list of variants
69		"""
70	1	nodeid, qname = _parse_add_args(*args[:2])
71	1	callback = args[2]
72	1	if len(args) > 3:
73	1	inputs = args[3]
74	1	if len(args) > 4:
75	1	outputs = args[4]
76	1	return _create_method(parent, nodeid, qname, callback, inputs, outputs)
77
78
79	1	def _create_folder(server, parentnodeid, nodeid, qname):
80	1	addnode = ua.AddNodesItem()
81	1	addnode.RequestedNewNodeId = nodeid
82	1	addnode.BrowseName = qname
83	1	addnode.NodeClass = ua.NodeClass.Object
84	1	addnode.ParentNodeId = parentnodeid
85	1	addnode.ReferenceTypeId = ua.NodeId.from_string("i=35")
86	1	addnode.TypeDefinition = ua.NodeId.from_string("i=61")
87	1	attrs = ua.ObjectAttributes()
88	1	attrs.Description = ua.LocalizedText(qname.Name)
89	1	attrs.DisplayName = ua.LocalizedText(qname.Name)
90	1	attrs.WriteMask = ua.OpenFileMode.Read
91	1	attrs.UserWriteMask = ua.OpenFileMode.Read
92	1	attrs.EventNotifier = 0
93	1	addnode.NodeAttributes = attrs
94	1	results = server.add_nodes([addnode])
95	1	results[0].StatusCode.check()
96	1	return nodeid
97
98
99	1	def _create_object(server, parentnodeid, nodeid, qname):
100	1	addnode = ua.AddNodesItem()
101	1	addnode.RequestedNewNodeId = nodeid
102	1	addnode.BrowseName = qname
103	1	addnode.NodeClass = ua.NodeClass.Object
104	1	addnode.ParentNodeId = parentnodeid
105	1	addnode.ReferenceTypeId = ua.NodeId.from_string("i=35")
106	1	addnode.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseObjectType)
107	1	attrs = ua.ObjectAttributes()
108	1	attrs.Description = ua.LocalizedText(qname.Name)
109	1	attrs.DisplayName = ua.LocalizedText(qname.Name)
110	1	attrs.EventNotifier = 0
111	1	attrs.WriteMask = ua.OpenFileMode.Read
112	1	attrs.UserWriteMask = ua.OpenFileMode.Read
113	1	addnode.NodeAttributes = attrs
114	1	results = server.add_nodes([addnode])
115	1	results[0].StatusCode.check()
116	1	return nodeid
117
118
119	1	def _to_variant(val, vtype=None):
120	1	if isinstance(val, ua.Variant):
121	1	return val
122		else:
123	1	return ua.Variant(val, vtype)
124
125
126	1	def _create_variable(server, parentnodeid, nodeid, qname, val, isproperty=False):
127	1	addnode = ua.AddNodesItem()
128	1	addnode.RequestedNewNodeId = nodeid
129	1	addnode.BrowseName = qname
130	1	addnode.NodeClass = ua.NodeClass.Variable
131	1	addnode.ParentNodeId = parentnodeid
132	1	if isproperty:
133	1	addnode.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasProperty)
134	1	addnode.TypeDefinition = ua.NodeId(ua.ObjectIds.PropertyType)
135		else:
136	1	addnode.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent)
137	1	addnode.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseDataVariableType)
138	1	attrs = ua.VariableAttributes()
139	1	attrs.Description = ua.LocalizedText(qname.Name)
140	1	attrs.DisplayName = ua.LocalizedText(qname.Name)
141	1	attrs.DataType = _guess_uatype(val)
142	1	attrs.Value = val
143	1	if isinstance(val, list) or isinstance(val, tuple):
144		attrs.ValueRank = ua.ValueRank.OneDimension
145		else:
146	1	attrs.ValueRank = ua.ValueRank.Scalar
147		#attrs.ArrayDimensions = None
148	1	attrs.WriteMask = ua.OpenFileMode.Read
149	1	attrs.UserWriteMask = ua.OpenFileMode.Read
150	1	attrs.Historizing = 0
151	1	addnode.NodeAttributes = attrs
152	1	results = server.add_nodes([addnode])
153	1	results[0].StatusCode.check()
154	1	return nodeid
155
156
157	1	def _create_method(parent, nodeid, qname, callback, inputs, outputs):
158	1	addnode = ua.AddNodesItem()
159	1	addnode.RequestedNewNodeId = nodeid
160	1	addnode.BrowseName = qname
161	1	addnode.NodeClass = ua.NodeClass.Method
162	1	addnode.ParentNodeId = parent.nodeid
163	1	addnode.ReferenceTypeId = ua.NodeId.from_string("i=47")
164		#node.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseObjectType)
165	1	attrs = ua.MethodAttributes()
166	1	attrs.Description = ua.LocalizedText(qname.Name)
167	1	attrs.DisplayName = ua.LocalizedText(qname.Name)
168	1	attrs.WriteMask = ua.OpenFileMode.Read
169	1	attrs.UserWriteMask = ua.OpenFileMode.Read
170	1	attrs.Executable = True
171	1	attrs.UserExecutable = True
172	1	addnode.NodeAttributes = attrs
173	1	results = parent.server.add_nodes([addnode])
174	1	results[0].StatusCode.check()
175	1	method = node.Node(parent.server, nodeid)
176	1	if inputs:
177	1	create_property(method, ua.generate_nodeid(qname.NamespaceIndex), ua.QualifiedName("InputArguments", 0), [_vtype_to_argument(vtype) for vtype in inputs])
178	1	if outputs:
179	1	create_property(method, ua.generate_nodeid(qname.NamespaceIndex), ua.QualifiedName("OutputArguments", 0), [_vtype_to_argument(vtype) for vtype in outputs])
180	1	parent.server.add_method_callback(method.nodeid, callback)
181	1	return nodeid
182
183
184	1	def _vtype_to_argument(vtype):
185	1	if isinstance(vtype, ua.Argument):
186		return vtype
187
188	1	arg = ua.Argument()
189	1	v = ua.Variant(None, vtype)
190	1	arg.DataType = _guess_uatype(v)
191	1	return arg
192
193
194	1	def _guess_uatype(variant):
195	1	if variant.VariantType == ua.VariantType.ExtensionObject:
196	1	if variant.Value is None:
197		raise ua.UaError("Cannot guess DataType from Null ExtensionObject")
198	1	if type(variant.Value) in (list, tuple):
199	1	if len(variant.Value) == 0:
200		raise ua.UaError("Cannot guess DataType from Null ExtensionObject")
201	1	extobj = variant.Value[0]
202		else:
203		extobj = variant.Value
204	1	classname = extobj.__class__.__name__
205	1	return ua.NodeId(getattr(ua.ObjectIds, classname))
206		else:
207	1	return ua.NodeId(getattr(ua.ObjectIds, variant.VariantType.name))
208
209
210	1	def delete_nodes(server, nodes, recursive=False):
211		"""
212		Delete specified nodes. Optionally delete recursively all nodes with a
213		downward hierachic references to the node
214		"""
215	1	nodestodelete = []
216	1	if recursive:
217	1	nodes += _add_childs(nodes)
218	1	for mynode in nodes:
219	1	it = ua.DeleteNodesItem()
220	1	it.NodeId = mynode.nodeid
221	1	it.DeleteTargetReferences = True
222	1	nodestodelete.append(it)
223	1	return server.delete_nodes(nodestodelete)
224
225
226	1	def _add_childs(nodes):
227	1	results = []
228	1	for mynode in nodes[:]:
229	1	results += mynode.get_children()
230	1	return results
231
232
233

FreeOpcUa / python-opcua

Pull Request — master (#124)

opcua.common.delete_nodes() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like