get_nodes_of_namespace() - Code Metrics - Inspection of "Fix "Avoid mutable default arguments" issue" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#369)

by Olivier

created 2016-11-19 20:17 UTC

get_nodes_of_namespace() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	9.4867

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	9
c	1
b	0
f	0
dl	0
loc	25
ccs	9
cts	11
cp	0.8182
crap	9.4867
rs	3

"""
Usefull method and classes not belonging anywhere and depending on opcua library
"""

from dateutil import parser
from datetime import datetime
from enum import Enum, IntEnum
import uuid

from opcua import ua
from opcua.ua.uaerrors import UaError


def val_to_string(val):
    """
    convert a python object or python-opcua object to a string
    which should be easy to understand for human
    easy to modify, and not too hard to parse back ....not easy
    meant for UI or command lines

    """
    if isinstance(val, (list, tuple)):
        res = []
        for v in val:
            res.append(val_to_string(v))
        return "[" + ", ".join(res) + "]"

    if hasattr(val, "to_string"):
        val = val.to_string()
    elif isinstance(val, ua.StatusCode):
        val = val.name
    elif isinstance(val, (Enum, IntEnum)):
        val = val.name
    elif isinstance(val, ua.DataValue):
        val = variant_to_string(val.Value)
    elif isinstance(val, str):
        pass
    elif isinstance(val, bytes):
        val = str(val)
    elif isinstance(val, datetime):
        val = val.isoformat()
    elif isinstance(val, (int, float)):
        val = str(val)
    else:
        # FIXME: Some types are probably missing!
        val = str(val)
    return val


def variant_to_string(var):
    """
    convert a variant to a string which should be easy to understand for human
    easy to modify, and not too hard to parse back ....not easy
    meant for UI or command lines
    """
    return val_to_string(var.Value)


def string_to_val(string, vtype):
    """
    Convert back a string to a python or python-opcua object
    Note: no error checking is done here, supplying null strings could raise exceptions (datetime and guid)
    """
    string = string.strip()
    if string.startswith("["):
        string = string[1:-1]
        var = []
        for s in string.split(","):
            s = s.strip()
            val = string_to_val(s, vtype)
            var.append(val)
        return var

    if vtype == ua.VariantType.Null:
        val = None
    elif vtype == ua.VariantType.Boolean:
        if string in ("True", "true", "on", "On", "1"):
            val = True
        else:
            val = False
    elif vtype in (ua.VariantType.Int16, ua.VariantType.Int32, ua.VariantType.Int64):
            val = int(string)
    elif vtype in (ua.VariantType.UInt16, ua.VariantType.UInt32, ua.VariantType.UInt64):
        val = int(string)
    elif vtype in (ua.VariantType.Float, ua.VariantType.Double):
        val = float(string)
    elif vtype in (ua.VariantType.String, ua.VariantType.XmlElement):
        val = string
    elif vtype in (ua.VariantType.SByte, ua.VariantType.ByteString):
        val = bytes(string)
    elif vtype in (ua.VariantType.NodeId, ua.VariantType.ExpandedNodeId):
        val = ua.NodeId.from_string(string)
    elif vtype == ua.VariantType.QualifiedName:
        val = ua.QualifiedName.from_string(string)
    elif vtype == ua.VariantType.DateTime:
        val = parser.parse(string)
    elif vtype == ua.VariantType.LocalizedText:
        val = ua.LocalizedText(string)
    elif vtype == ua.VariantType.StatusCode:
        val = ua.StatusCode(string)
    elif vtype == ua.VariantType.Guid:
        val = uuid.UUID(string)
    else:
        # FIXME: Some types are probably missing!
        raise NotImplementedError
    return val


def string_to_variant(string, vtype):
    """
    convert back a string to an ua.Variant
    """
    return ua.Variant(string_to_val(string, vtype), vtype)


def get_node_children(node, nodes=None):
    """
    Get recursively all children of a node
    """
    if nodes is None:
        nodes = [node]
    for child in node.get_children():
        nodes.append(child)
        get_node_children(child, nodes)
    return nodes


def get_node_subtypes(node, nodes=None):
    if nodes is None:
        nodes = [node]
    for child in node.get_children(refs=ua.ObjectIds.HasSubtype):
        nodes.append(child)
        get_node_subtypes(child, nodes)
    return nodes


def get_node_supertypes(node, includeitself=False, skipbase=True):
    """
    return get all subtype parents of node recursive
    :param node: can be a ua.Node or ua.NodeId
    :param includeitself: include also node to the list
    :param skipbase don't include the toplevel one
    :returns list of ua.Node, top parent first 
    """
    parents = []
    if includeitself:
        parents.append(node)
    parents.extend(_get_node_supertypes(node))
    if skipbase and len(parents) > 1:
        parents = parents[:-1]

    return parents


def _get_node_supertypes(node):
    """
    recursive implementation of get_node_derived_from_types
    """
    basetypes = []
    parent = get_node_supertype(node)
    if parent:
        basetypes.append(parent)
        basetypes.extend(_get_node_supertypes(parent))

    return basetypes


def get_node_supertype(node):
    """
    return node supertype or None
    """
    supertypes = node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype,
                                           direction=ua.BrowseDirection.Inverse,
                                           includesubtypes=True)
    if supertypes:
        return supertypes[0]
    else:
        return None


def is_child_present(node, browsename):
    """
    return if a browsename is present a child from the provide node
    :param node: node wherein to find the browsename
    :param browsename: browsename to search
    :returns returne True if the browsename is present else False 
    """
    child_descs = node.get_children_descriptions()
    for child_desc in child_descs:
        if child_desc.BrowseName == browsename:
            return True

    return False


def data_type_to_variant_type(dtype_node):
    """
    Given a Node datatype, find out the variant type to encode
    data. This is not exactly straightforward...
    """
    base = get_base_data_type(dtype_node)

    if base.nodeid.Identifier != 29:
        return ua.VariantType(base.nodeid.Identifier)
    else:
        # we have an enumeration, we need to look at child to find type
        descs = dtype_node.get_children_descriptions()
        bnames = [d.BrowseName.Name for d in descs]
        if "EnumStrings" in bnames:
            return ua.VariantType.LocalizedText
        elif "EnumValues" in bnames:
            return ua.VariantType.ExtensionObject
        else:
            raise ua.UaError("Enumeration must have a child node describing its type and values")


def get_base_data_type(datatype):
    """
    Looks up the base datatype of the provided datatype Node
    The base datatype is either:
    A primitive type (ns=0, i<=21) or a complex one (ns=0 i>21 and i<=30) like Enum and Struct.
    
    Args:
        datatype: NodeId of a datype of a variable
    Returns:
        NodeId of datatype base or None in case base datype can not be determined
    """
    base = datatype
    while base:
        if base.nodeid.NamespaceIndex == 0 and isinstance(base.nodeid.Identifier, int) and base.nodeid.Identifier <= 30:
            return base
        base = get_node_supertype(base)
    raise ua.UaError("Datatype must be a subtype of builtin types %s" % datatype)


def get_nodes_of_namespace(server, namespaces=None):
    """
    Get the nodes of one or more namespaces .      
    Args:
        server: opc ua server to use
        namespaces: list of string uri or int indexes of the namespace to export
    Returns:
        List of nodes that are part of the provided namespaces
    """
    if namespaces is None:
        namespaces = []
    ns_available = server.get_namespace_array()

    if not namespaces:
        namespaces = ns_available[1:]
    elif isinstance(namespaces, (str, int)):
        namespaces = [namespaces]

    # make sure all namespace are indexes (if needed convert strings to indexes)
    namespace_indexes = [n if isinstance(n, int) else ns_available.index(n) for n in namespaces]

    # filter nodeis based on the provide namespaces and convert the nodeid to a node
    nodes = [server.get_node(nodeid) for nodeid in server.iserver.aspace.keys()
             if nodeid.NamespaceIndex != 0 and nodeid.NamespaceIndex in namespace_indexes]
    return nodes


1		"""
2		Usefull method and classes not belonging anywhere and depending on opcua library
3		"""
4
5	1	from dateutil import parser
6	1	from datetime import datetime
7	1	from enum import Enum, IntEnum
8	1	import uuid
9
10	1	from opcua import ua
11	1	from opcua.ua.uaerrors import UaError
12
13
14	1	def val_to_string(val):
15		"""
16		convert a python object or python-opcua object to a string
17		which should be easy to understand for human
18		easy to modify, and not too hard to parse back ....not easy
19		meant for UI or command lines
20
21		"""
22	1	if isinstance(val, (list, tuple)):
23	1	res = []
24	1	for v in val:
25	1	res.append(val_to_string(v))
26	1	return "[" + ", ".join(res) + "]"
27
28	1	if hasattr(val, "to_string"):
29	1	val = val.to_string()
30	1	elif isinstance(val, ua.StatusCode):
31		val = val.name
32	1	elif isinstance(val, (Enum, IntEnum)):
33		val = val.name
34	1	elif isinstance(val, ua.DataValue):
35		val = variant_to_string(val.Value)
36	1	elif isinstance(val, str):
37	1	pass
38	1	elif isinstance(val, bytes):
39		val = str(val)
40	1	elif isinstance(val, datetime):
41		val = val.isoformat()
42	1	elif isinstance(val, (int, float)):
43	1	val = str(val)
44		else:
45		# FIXME: Some types are probably missing!
46		val = str(val)
47	1	return val
48
49
50	1	def variant_to_string(var):
51		"""
52		convert a variant to a string which should be easy to understand for human
53		easy to modify, and not too hard to parse back ....not easy
54		meant for UI or command lines
55		"""
56		return val_to_string(var.Value)
57
58
59	1	def string_to_val(string, vtype):
60		"""
61		Convert back a string to a python or python-opcua object
62		Note: no error checking is done here, supplying null strings could raise exceptions (datetime and guid)
63		"""
64	1	string = string.strip()
65	1	if string.startswith("["):
66	1	string = string[1:-1]
67	1	var = []
68	1	for s in string.split(","):
69	1	s = s.strip()
70	1	val = string_to_val(s, vtype)
71	1	var.append(val)
72	1	return var
73
74	1	if vtype == ua.VariantType.Null:
75		val = None
76	1	elif vtype == ua.VariantType.Boolean:
77		if string in ("True", "true", "on", "On", "1"):
78		val = True
79		else:
80		val = False
81	1	elif vtype in (ua.VariantType.Int16, ua.VariantType.Int32, ua.VariantType.Int64):
82		val = int(string)
83	1	elif vtype in (ua.VariantType.UInt16, ua.VariantType.UInt32, ua.VariantType.UInt64):
84	1	val = int(string)
85	1	elif vtype in (ua.VariantType.Float, ua.VariantType.Double):
86	1	val = float(string)
87	1	elif vtype in (ua.VariantType.String, ua.VariantType.XmlElement):
88	1	val = string
89	1	elif vtype in (ua.VariantType.SByte, ua.VariantType.ByteString):
90		val = bytes(string)
91	1	elif vtype in (ua.VariantType.NodeId, ua.VariantType.ExpandedNodeId):
92	1	val = ua.NodeId.from_string(string)
93	1	elif vtype == ua.VariantType.QualifiedName:
94	1	val = ua.QualifiedName.from_string(string)
95	1	elif vtype == ua.VariantType.DateTime:
96	1	val = parser.parse(string)
97	1	elif vtype == ua.VariantType.LocalizedText:
98	1	val = ua.LocalizedText(string)
99	1	elif vtype == ua.VariantType.StatusCode:
100	1	val = ua.StatusCode(string)
101		elif vtype == ua.VariantType.Guid:
102		val = uuid.UUID(string)
103		else:
104		# FIXME: Some types are probably missing!
105		raise NotImplementedError
106	1	return val
107
108
109	1	def string_to_variant(string, vtype):
110		"""
111		convert back a string to an ua.Variant
112		"""
113		return ua.Variant(string_to_val(string, vtype), vtype)
114
115
116	1	def get_node_children(node, nodes=None):
117		"""
118		Get recursively all children of a node
119		"""
120		if nodes is None:
121		nodes = [node]
122		for child in node.get_children():
123		nodes.append(child)
124		get_node_children(child, nodes)
125		return nodes
126
127
128	1	def get_node_subtypes(node, nodes=None):
129	1	if nodes is None:
130	1	nodes = [node]
131	1	for child in node.get_children(refs=ua.ObjectIds.HasSubtype):
132	1	nodes.append(child)
133	1	get_node_subtypes(child, nodes)
134	1	return nodes
135
136
137	1	def get_node_supertypes(node, includeitself=False, skipbase=True):
138		"""
139		return get all subtype parents of node recursive
140		:param node: can be a ua.Node or ua.NodeId
141		:param includeitself: include also node to the list
142		:param skipbase don't include the toplevel one
143		:returns list of ua.Node, top parent first
144		"""
145	1	parents = []
146	1	if includeitself:
147	1	parents.append(node)
148	1	parents.extend(_get_node_supertypes(node))
149	1	if skipbase and len(parents) > 1:
150	1	parents = parents[:-1]
151
152	1	return parents
153
154
155	1	def _get_node_supertypes(node):
156		"""
157		recursive implementation of get_node_derived_from_types
158		"""
159	1	basetypes = []
160	1	parent = get_node_supertype(node)
161	1	if parent:
162	1	basetypes.append(parent)
163	1	basetypes.extend(_get_node_supertypes(parent))
164
165	1	return basetypes
166
167
168	1	def get_node_supertype(node):
169		"""
170		return node supertype or None
171		"""
172	1	supertypes = node.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype,
173		direction=ua.BrowseDirection.Inverse,
174		includesubtypes=True)
175	1	if supertypes:
176	1	return supertypes[0]
177		else:
178	1	return None
179
180
181	1	def is_child_present(node, browsename):
182		"""
183		return if a browsename is present a child from the provide node
184		:param node: node wherein to find the browsename
185		:param browsename: browsename to search
186		:returns returne True if the browsename is present else False
187		"""
188	1	child_descs = node.get_children_descriptions()
189	1	for child_desc in child_descs:
190	1	if child_desc.BrowseName == browsename:
191		return True
192
193	1	return False
194
195
196	1	def data_type_to_variant_type(dtype_node):
197		"""
198		Given a Node datatype, find out the variant type to encode
199		data. This is not exactly straightforward...
200		"""
201	1	base = get_base_data_type(dtype_node)
202
203	1	if base.nodeid.Identifier != 29:
204	1	return ua.VariantType(base.nodeid.Identifier)
205		else:
206		# we have an enumeration, we need to look at child to find type
207		descs = dtype_node.get_children_descriptions()
208		bnames = [d.BrowseName.Name for d in descs]
209		if "EnumStrings" in bnames:
210		return ua.VariantType.LocalizedText
211		elif "EnumValues" in bnames:
212		return ua.VariantType.ExtensionObject
213		else:
214		raise ua.UaError("Enumeration must have a child node describing its type and values")
215
216
217	1	def get_base_data_type(datatype):
218		"""
219		Looks up the base datatype of the provided datatype Node
220		The base datatype is either:
221		A primitive type (ns=0, i<=21) or a complex one (ns=0 i>21 and i<=30) like Enum and Struct.
222
223		Args:
224		datatype: NodeId of a datype of a variable
225		Returns:
226		NodeId of datatype base or None in case base datype can not be determined
227		"""
228	1	base = datatype
229	1	while base:
230	1	if base.nodeid.NamespaceIndex == 0 and isinstance(base.nodeid.Identifier, int) and base.nodeid.Identifier <= 30:
231	1	return base
232	1	base = get_node_supertype(base)
233		raise ua.UaError("Datatype must be a subtype of builtin types %s" % datatype)
234
235
236	1	def get_nodes_of_namespace(server, namespaces=None):
237		"""
238		Get the nodes of one or more namespaces .
239		Args:
240		server: opc ua server to use
241		namespaces: list of string uri or int indexes of the namespace to export
242		Returns:
243		List of nodes that are part of the provided namespaces
244		"""
245	1	if namespaces is None:
246		namespaces = []
247	1	ns_available = server.get_namespace_array()
248
249	1	if not namespaces:
250		namespaces = ns_available[1:]
251	1	elif isinstance(namespaces, (str, int)):
252	1	namespaces = [namespaces]
253
254		# make sure all namespace are indexes (if needed convert strings to indexes)
255	1	namespace_indexes = [n if isinstance(n, int) else ns_available.index(n) for n in namespaces]
256
257		# filter nodeis based on the provide namespaces and convert the nodeid to a node
258	1	nodes = [server.get_node(nodeid) for nodeid in server.iserver.aspace.keys()
259		if nodeid.NamespaceIndex != 0 and nodeid.NamespaceIndex in namespace_indexes]
260		return nodes
261

FreeOpcUa / python-opcua

Pull Request — master (#369)

get_nodes_of_namespace() F

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like