_create_folder() - Code Metrics - Inspection of "Added support for Inverse references" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#165)

by Denis

created 2016-04-23 22:40 UTC

_create_folder() A

↳ Parent: _create_object()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	18
Ratio	100 %

Code Coverage

Tests	18
CRAP Score	1

Metric	Value
cc	1
dl	18
loc	18
rs	9.4285
ccs	18
cts	18
cp	1
crap	1

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


def _parse_add_args(*args):
    try:
        if isinstance(args[0], int):
            nodeid = ua.NodeId(0, int(args[0]))
            qname = ua.QualifiedName(args[1], int(args[0]))
            return nodeid, qname
        if isinstance(args[0], ua.NodeId):
            nodeid = args[0]
        elif isinstance(args[0], str):
            nodeid = ua.NodeId.from_string(args[0])
        else:
            raise RuntimeError()
        if isinstance(args[1], ua.QualifiedName):
            qname = args[1]
        elif isinstance(args[1], str):
            qname = ua.QualifiedName.from_string(args[1])
        else:
            raise RuntimeError()
        return nodeid, qname
    except ua.UaError:
        raise
    except Exception as ex:
        raise TypeError("This method takes either a namespace index and a string as argument or a nodeid and a qualifiedname. Received arguments {} and got exception {}".format(args, ex))


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_object(parent.server, parent.nodeid, nodeid, qname, ua.ObjectIds.FolderType))


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, ua.ObjectIds.BaseObjectType))


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]
    else:
        inputs = []
    if len(args) > 4:
        outputs = args[4]
    else:
        outputs = []
    return node.Node(parent.server, _create_method(parent, nodeid, qname, callback, inputs, outputs))


def _create_object(server, parentnodeid, nodeid, qname, objecttype):
    addnode = ua.AddNodesItem()
    addnode.RequestedNewNodeId = nodeid
    addnode.BrowseName = qname
    addnode.NodeClass = ua.NodeClass.Object
    addnode.ParentNodeId = parentnodeid
    #TODO: maybe move to address_space.py and implement for all node types?
    if node.Node(server, parentnodeid).get_type_definition() == ua.ObjectIds.FolderType:
        addnode.ReferenceTypeId = ua.NodeId(ua.ObjectIds.Organizes)
    else:
        addnode.ReferenceTypeId = ua.NodeId(ua.ObjectIds.HasComponent)
    addnode.TypeDefinition = ua.NodeId(objecttype)
    attrs = ua.ObjectAttributes()
    attrs.Description = ua.LocalizedText(qname.Name)
    attrs.DisplayName = ua.LocalizedText(qname.Name)
    attrs.EventNotifier = 0
    attrs.WriteMask = 0
    attrs.UserWriteMask = 0
    addnode.NodeAttributes = attrs
    results = server.add_nodes([addnode])
    results[0].StatusCode.check()
    return results[0].AddedNodeId


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 = 0
    attrs.UserWriteMask = 0
    attrs.Historizing = 0
    attrs.AccessLevel = ua.AccessLevelMask.CurrentRead
    attrs.UserAccessLevel = ua.AccessLevelMask.CurrentRead
    addnode.NodeAttributes = attrs
    results = server.add_nodes([addnode])
    results[0].StatusCode.check()
    return results[0].AddedNodeId


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(ua.ObjectIds.HasComponent)
    #node.TypeDefinition = ua.NodeId(ua.ObjectIds.BaseObjectType)
    attrs = ua.MethodAttributes()
    attrs.Description = ua.LocalizedText(qname.Name)
    attrs.DisplayName = ua.LocalizedText(qname.Name)
    attrs.WriteMask = 0
    attrs.UserWriteMask = 0
    attrs.Executable = True
    attrs.UserExecutable = True
    addnode.NodeAttributes = attrs
    results = parent.server.add_nodes([addnode])
    results[0].StatusCode.check()
    method = node.Node(parent.server, results[0].AddedNodeId)
    if inputs:
        create_property(method, ua.NodeId(), ua.QualifiedName("InputArguments", 0), [_vtype_to_argument(vtype) for vtype in inputs])
    if outputs:
        create_property(method, ua.NodeId(), ua.QualifiedName("OutputArguments", 0), [_vtype_to_argument(vtype) for vtype in outputs])
    parent.server.add_method_callback(method.nodeid, callback)
    return results[0].AddedNodeId


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

FreeOpcUa / python-opcua

Pull Request — master (#165)

_create_folder() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like