|
1
|
|
|
""" |
|
2
|
|
|
JSON encoder / decoder for OPC UA objects |
|
3
|
|
|
This can be useful for serializing data for external system such as a cloud service or a web app |
|
4
|
|
|
""" |
|
5
|
|
|
|
|
6
|
|
|
from json import JSONEncoder, JSONDecoder |
|
7
|
|
|
|
|
8
|
|
|
from opcua import ua |
|
9
|
|
|
from opcua import Node |
|
10
|
|
|
from opcua.common.subscription import DataChangeNotif |
|
11
|
|
|
from opcua.common.events import Event |
|
12
|
|
|
from datetime import datetime |
|
13
|
|
|
|
|
14
|
|
|
# FIXME this JSON utility is not finished, should implement encoding / decoding of useful UA types |
|
15
|
|
|
# FIXME this class should be improved to use JSON schemas to better serialize UA objects |
|
16
|
|
|
|
|
17
|
|
|
|
|
18
|
|
|
class UaJSONEncoder(JSONEncoder): |
|
19
|
|
|
""" |
|
20
|
|
|
JSON encoder utility which is extended to handle opc ua objects |
|
21
|
|
|
note that JSON pair order may not match the python object attribute order |
|
22
|
|
|
""" |
|
23
|
|
|
def default(self, o, sort_keys=True): |
|
24
|
|
|
if isinstance(o, datetime): |
|
25
|
|
|
return { |
|
26
|
|
|
'__type__': 'datetime', |
|
27
|
|
|
'year': o.year, |
|
28
|
|
|
'month': o.month, |
|
29
|
|
|
'day': o.day, |
|
30
|
|
|
'hour': o.hour, |
|
31
|
|
|
'minute': o.minute, |
|
32
|
|
|
'second': o.second, |
|
33
|
|
|
'microsecond': o.microsecond |
|
34
|
|
|
} |
|
35
|
|
|
elif isinstance(o, ua.NodeId): |
|
36
|
|
|
return { |
|
37
|
|
|
'__type__': 'NodeId', |
|
38
|
|
|
'Identifier': str(o.Identifier), |
|
39
|
|
|
'NamespaceIndex': str(o.NamespaceIndex), |
|
40
|
|
|
'NamespaceUri': o.NamespaceUri, |
|
41
|
|
|
'NodeIdType': o.NodeIdType.name, |
|
42
|
|
|
|
|
43
|
|
|
} |
|
44
|
|
|
elif isinstance(o, Node): |
|
45
|
|
|
return { |
|
46
|
|
|
'__type__': 'Node', |
|
47
|
|
|
'NodeId': str(o.nodeid), |
|
48
|
|
|
'BrowseName': o.get_browse_name().Name, |
|
49
|
|
|
'DisplayName': o.get_display_name().Text.decode(encoding='UTF-8'), |
|
50
|
|
|
} |
|
51
|
|
|
elif isinstance(o, ua.DataValue): |
|
52
|
|
|
return { |
|
53
|
|
|
'__type__': 'DataValue', |
|
54
|
|
|
'ServerTimestamp': o.ServerTimestamp.isoformat(' '), |
|
55
|
|
|
'SourceTimestamp': o.SourceTimestamp.isoformat(' '), |
|
56
|
|
|
'StatusCode': o.StatusCode.name, |
|
57
|
|
|
'VariantType': o.Value.VariantType.name, |
|
58
|
|
|
'Value': str(o.Value.Value), |
|
59
|
|
|
} |
|
60
|
|
|
elif isinstance(o, DataChangeNotif): |
|
61
|
|
|
return { |
|
62
|
|
|
'__type__': 'DataChangeNotif', |
|
63
|
|
|
'Node': str(o.subscription_data.node.nodeid), |
|
64
|
|
|
'DisplayName': o.subscription_data.node.get_display_name().Text.decode(encoding='UTF-8'), |
|
65
|
|
|
'ServerTimestamp': o.monitored_item.Value.ServerTimestamp.isoformat(' '), |
|
66
|
|
|
'SourceTimestamp': o.monitored_item.Value.SourceTimestamp.isoformat(' '), |
|
67
|
|
|
'StatusCode': o.monitored_item.Value.StatusCode.name, |
|
68
|
|
|
'VariantType': o.monitored_item.Value.Value.VariantType.name, |
|
69
|
|
|
'Value': str(o.monitored_item.Value.Value.Value), |
|
70
|
|
|
} |
|
71
|
|
|
elif isinstance(o, Event): |
|
72
|
|
|
return { |
|
73
|
|
|
'__type__': 'Event', |
|
74
|
|
|
'SourceNode': str(o.SourceNode), |
|
75
|
|
|
'EventId': o.EventId, |
|
76
|
|
|
'EventType': str(o.EventType), |
|
77
|
|
|
'LocalTime': o.LocalTime.isoformat(' '), |
|
78
|
|
|
'ReceiveTime': o.ReceiveTime.isoformat(' '), |
|
79
|
|
|
'Time': o.Time.isoformat(' '), |
|
80
|
|
|
'Severity': str(o.Severity), |
|
81
|
|
|
'Message': o.Message.Text.decode(encoding='UTF-8'), |
|
82
|
|
|
} |
|
83
|
|
|
else: |
|
84
|
|
|
return JSONEncoder.default(self, o) |
|
85
|
|
|
|
|
86
|
|
|
|
|
87
|
|
|
class UaJSONDecoder(JSONDecoder): |
|
88
|
|
|
""" |
|
89
|
|
|
JSON decoder utility which is extended to handle opc ua objects |
|
90
|
|
|
Limited functionality; user must check what kind of object was decoded externally |
|
91
|
|
|
""" |
|
92
|
|
|
def __init__(self): |
|
93
|
|
|
JSONDecoder.__init__(self, object_hook=self.dict_to_object) |
|
94
|
|
|
|
|
95
|
|
|
def dict_to_object(self, d): |
|
96
|
|
|
if '__type__' not in d: |
|
97
|
|
|
return d |
|
98
|
|
|
|
|
99
|
|
|
o_type = d.pop('__type__') |
|
100
|
|
|
if o_type == 'datetime': |
|
101
|
|
|
return datetime(**d) |
|
102
|
|
|
elif o_type == 'NodeId': |
|
103
|
|
|
node_id_type = d.pop('NodeIdType') |
|
104
|
|
|
|
|
105
|
|
|
if node_id_type == 'Numeric': |
|
106
|
|
|
identifier = int(d.pop('Identifier')) |
|
107
|
|
|
elif node_id_type == 'String': |
|
108
|
|
|
identifier = d.pop('Identifier') |
|
109
|
|
|
elif node_id_type == 'ByteString': |
|
110
|
|
|
identifier = bytes(d.pop('Identifier')) |
|
111
|
|
|
else: |
|
112
|
|
|
raise ValueError("NodeId object missing NodeIdType attribute") |
|
113
|
|
|
|
|
114
|
|
|
return ua.NodeId(identifier, int(d.pop('NamespaceIndex'))) |
|
115
|
|
|
elif o_type == 'Node': |
|
116
|
|
|
raise TypeError("Node should not be decoded because it requires a reference to the server") |
|
117
|
|
|
elif o_type == 'DataValue': |
|
118
|
|
|
variant_type = d.pop('VariantType') |
|
119
|
|
|
if variant_type in ("Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64"): |
|
120
|
|
|
dv = ua.DataValue(ua.Variant(int(d.pop('Value')))) |
|
121
|
|
|
elif variant_type in ("Float", "Double"): |
|
122
|
|
|
dv = ua.DataValue(ua.Variant(float(d.pop('Value')))) |
|
123
|
|
|
elif variant_type in ("Boolean"): |
|
124
|
|
|
dv = ua.DataValue(ua.Variant(bool(d.pop('Value')))) |
|
125
|
|
|
elif variant_type in ("ByteString", "String"): |
|
126
|
|
|
dv = ua.DataValue(ua.Variant(d.pop('Value'))) |
|
127
|
|
|
else: |
|
128
|
|
|
raise TypeError("DataValue did not have a supported variant type") |
|
129
|
|
|
return dv |
|
130
|
|
|
elif o_type == 'DataChangeNotif': |
|
131
|
|
|
raise TypeError("DataChangeNotif should not be decoded") |
|
132
|
|
|
elif o_type == 'Event': |
|
133
|
|
|
raise TypeError("Event should not be decoded") |
|
134
|
|
|
else: |
|
135
|
|
|
# decoder subclass can't handle this object type; put it back in the dict so super throws TypeError |
|
136
|
|
|
d['__type__'] = o_type |
|
137
|
|
|
return d |
|
138
|
|
|
|
FreeOpcUa /
python-opcua
