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