UaJSONEncoder.default() - Code Metrics - Inspection of "JSON Module" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#222)

unknown

created 2016-06-21 22:13 UTC

UaJSONEncoder.default() C

↳ Parent: UaJSONEncoder

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	7
c	1
b	0
f	0
dl	0
loc	62
rs	6.2295

How to fix Long Method

"""
JSON encoder / decoder for OPC UA objects
This can be useful for serializing data for external system such as a cloud service or a web app
"""

from json import JSONEncoder, JSONDecoder

from opcua import ua
from opcua import Node
from opcua.common.subscription import DataChangeNotif
from opcua.common.events import Event
from datetime import datetime

# FIXME this JSON utility is not finished, should implement encoding / decoding of useful UA types
# FIXME this class should be improved to use JSON schemas to better serialize UA objects


class UaJSONEncoder(JSONEncoder):
    """
    JSON encoder utility which is extended to handle opc ua objects
    note that JSON pair order may not match the python object attribute order
    """
    def default(self, o, sort_keys=True):
        if isinstance(o, datetime):
            return {
                '__type__': 'datetime',
                'year': o.year,
                'month': o.month,
                'day': o.day,
                'hour': o.hour,
                'minute': o.minute,
                'second': o.second,
                'microsecond': o.microsecond
            }
        elif isinstance(o, ua.NodeId):
            return {
                '__type__': 'NodeId',
                'Identifier': str(o.Identifier),
                'NamespaceIndex': str(o.NamespaceIndex),
                'NamespaceUri': o.NamespaceUri,
                'NodeIdType': o.NodeIdType.name,

            }
        elif isinstance(o, Node):
            return {
                '__type__': 'Node',
                'NodeId': str(o.nodeid),
                'BrowseName': o.get_browse_name().Name,
                'DisplayName': o.get_display_name().Text.decode(encoding='UTF-8'),
            }
        elif isinstance(o, ua.DataValue):
            return {
                '__type__': 'DataValue',
                'ServerTimestamp': o.ServerTimestamp.isoformat(' '),
                'SourceTimestamp': o.SourceTimestamp.isoformat(' '),
                'StatusCode': o.StatusCode.name,
                'VariantType': o.Value.VariantType.name,
                'Value': str(o.Value.Value),
            }
        elif isinstance(o, DataChangeNotif):
            return {
                '__type__': 'DataChangeNotif',
                'Node': str(o.subscription_data.node.nodeid),
                'DisplayName': o.subscription_data.node.get_display_name().Text.decode(encoding='UTF-8'),
                'ServerTimestamp': o.monitored_item.Value.ServerTimestamp.isoformat(' '),
                'SourceTimestamp': o.monitored_item.Value.SourceTimestamp.isoformat(' '),
                'StatusCode': o.monitored_item.Value.StatusCode.name,
                'VariantType': o.monitored_item.Value.Value.VariantType.name,
                'Value': str(o.monitored_item.Value.Value.Value),
            }
        elif isinstance(o, Event):
            return {
                '__type__': 'Event',
                'SourceNode': str(o.SourceNode),
                'EventId': o.EventId,
                'EventType': str(o.EventType),
                'LocalTime': o.LocalTime.isoformat(' '),
                'ReceiveTime': o.ReceiveTime.isoformat(' '),
                'Time': o.Time.isoformat(' '),
                'Severity': str(o.Severity),
                'Message': o.Message.Text.decode(encoding='UTF-8'),
            }
        else:
            return JSONEncoder.default(self, o)


class UaJSONDecoder(JSONDecoder):
    """
    JSON decoder utility which is extended to handle opc ua objects
    Limited functionality; user must check what kind of object was decoded externally
    """
    def __init__(self):
        JSONDecoder.__init__(self, object_hook=self.dict_to_object)

    def dict_to_object(self, d):
        if '__type__' not in d:
            return d

        o_type = d.pop('__type__')
        if o_type == 'datetime':
            return datetime(**d)
        elif o_type == 'NodeId':
            node_id_type = d.pop('NodeIdType')

            if node_id_type == 'Numeric':
                identifier = int(d.pop('Identifier'))
            elif node_id_type == 'String':
                identifier = d.pop('Identifier')
            elif node_id_type == 'ByteString':
                identifier = bytes(d.pop('Identifier'))
            else:
                raise ValueError("NodeId object missing NodeIdType attribute")

            return ua.NodeId(identifier, int(d.pop('NamespaceIndex')))
        elif o_type == 'Node':
            raise TypeError("Node should not be decoded because it requires a reference to the server")
        elif o_type == 'DataValue':
            variant_type = d.pop('VariantType')
            if variant_type in ("Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64"):
                dv = ua.DataValue(ua.Variant(int(d.pop('Value'))))
            elif variant_type in ("Float", "Double"):
                dv = ua.DataValue(ua.Variant(float(d.pop('Value'))))
            elif variant_type in ("Boolean"):
                dv = ua.DataValue(ua.Variant(bool(d.pop('Value'))))
            elif variant_type in ("ByteString", "String"):
                dv = ua.DataValue(ua.Variant(d.pop('Value')))
            else:
                raise TypeError("DataValue did not have a supported variant type")
            return dv
        elif o_type == 'DataChangeNotif':
            raise TypeError("DataChangeNotif should not be decoded")
        elif o_type == 'Event':
            raise TypeError("Event should not be decoded")
        else:
            #  decoder subclass can't handle this object type; put it back in the dict so super throws TypeError
            d['__type__'] = o_type
            return d


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

Pull Request — master (#222)

UaJSONEncoder.default() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like