_Guid.pack() - Code Metrics - Inspection of "Improve support for custom structures client side" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#494)

by Olivier

created 2017-09-12 07:41 UTC

_Guid.pack() A

↳ Parent: _Guid

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.729

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	14
ccs	1
cts	10
cp	0.1
crap	1.729
rs	9.4285

"""
Binary protocol specific functions and constants
"""

import sys
import struct
import logging
from datetime import datetime, timedelta, tzinfo, MAXYEAR
from calendar import timegm
import uuid

from opcua.ua.uaerrors import UaError


if sys.version_info.major > 2:
    unicode = str

logger = logging.getLogger('__name__')

EPOCH_AS_FILETIME = 116444736000000000  # January 1, 1970 as MS file time
HUNDREDS_OF_NANOSECONDS = 10000000
FILETIME_EPOCH_AS_DATETIME = datetime(1601, 1, 1)


def test_bit(data, offset):
    mask = 1 << offset
    return data & mask


def set_bit(data, offset):
    mask = 1 << offset
    return data | mask


def unset_bit(data, offset):
    mask = 1 << offset
    return data & ~mask


class UTC(tzinfo):
    """
    UTC
    """

    def utcoffset(self, dt):
        return timedelta(0)

    def tzname(self, dt):
        return "UTC"

    def dst(self, dt):
        return timedelta(0)


# method copied from David Buxton <[email protected]> sample code
def datetime_to_win_epoch(dt):
    if (dt.tzinfo is None) or (dt.tzinfo.utcoffset(dt) is None):
        dt = dt.replace(tzinfo=UTC())
    ft = EPOCH_AS_FILETIME + (timegm(dt.timetuple()) * HUNDREDS_OF_NANOSECONDS)
    return ft + (dt.microsecond * 10)


def win_epoch_to_datetime(epch):
    try:
        return FILETIME_EPOCH_AS_DATETIME + timedelta(microseconds=epch // 10)
    except OverflowError:
        # FILETIMEs after 31 Dec 9999 can't be converted to datetime
        logger.warning("datetime overflow: %s", epch)
        return datetime(MAXYEAR, 12, 31, 23, 59, 59, 999999)


def build_array_format_py2(prefix, length, fmtchar):
    return prefix + str(length) + fmtchar


def build_array_format_py3(prefix, length, fmtchar):
    return prefix + str(length) + chr(fmtchar)


if sys.version_info.major < 3:
    build_array_format = build_array_format_py2
else:
    build_array_format = build_array_format_py3


class _Primitive(object):

    def pack_array(self, array):
        if array is None:
            return b'\xff\xff\xff\xff'
        length = len(array)
        b = [self.pack(val) for val in array]
        b.insert(0, Primitives.Int32.pack(length))
        return b"".join(b)

    def unpack_array(self, data):
        length = Primitives.Int32.unpack(data)
        if length == -1:
            return None
        elif length == 0:
            return []
        else:
            return [self.unpack(data) for _ in range(length)]


class _DateTime(_Primitive):

    @staticmethod
    def pack(dt):
        epch = datetime_to_win_epoch(dt)
        return Primitives.Int64.pack(epch)

    @staticmethod
    def unpack(data):
        epch = Primitives.Int64.unpack(data)
        return win_epoch_to_datetime(epch)


class _String(_Primitive):

    @staticmethod
    def pack(string):
        if string is None:
            return Primitives.Int32.pack(-1)
        if isinstance(string, unicode):
            string = string.encode('utf-8')
        length = len(string)
        return Primitives.Int32.pack(length) + string

    @staticmethod
    def unpack(data):
        b = _Bytes.unpack(data)
        if sys.version_info.major < 3:
            return b
        else:
            if b is None:
                return b
            return b.decode("utf-8")


class _Bytes(_Primitive):

    @staticmethod
    def pack(data):
        return _String.pack(data)

    @staticmethod
    def unpack(data):
        length = Primitives.Int32.unpack(data)
        if length == -1:
            return None
        return data.read(length)


class _Null(_Primitive):

    @staticmethod
    def pack(data):
        return b""

    @staticmethod
    def unpack(data):
        return None


class _Guid(_Primitive):

    @staticmethod
    def pack(guid):
        # convert python UUID 6 field format to OPC UA 4 field format
        f1 = Primitives.UInt32.pack(guid.time_low)
        f2 = Primitives.UInt16.pack(guid.time_mid)
        f3 = Primitives.UInt16.pack(guid.time_hi_version)
        f4a = Primitives.Byte.pack(guid.clock_seq_hi_variant)
        f4b = Primitives.Byte.pack(guid.clock_seq_low)
        f4c = struct.pack('>Q', guid.node)[2:8]  # no primitive .pack available for 6 byte int
        f4 = f4a+f4b+f4c
        # concat byte fields
        b = f1+f2+f3+f4

        return b

    @staticmethod
    def unpack(data):
        # convert OPC UA 4 field format to python UUID bytes
        f1 = struct.pack('>I', Primitives.UInt32.unpack(data))
        f2 = struct.pack('>H', Primitives.UInt16.unpack(data))
        f3 = struct.pack('>H', Primitives.UInt16.unpack(data))
        f4 = data.read(8)
        # concat byte fields
        b = f1 + f2 + f3 + f4

        return uuid.UUID(bytes=b)


class _Primitive1(_Primitive):
    def __init__(self, fmt):
        self.struct = struct.Struct(fmt)
        self.size = self.struct.size
        self.format = self.struct.format

    def pack(self, data):
        return struct.pack(self.format, data)

    def unpack(self, data):
        return struct.unpack(self.format, data.read(self.size))[0]
    
    #def pack_array(self, array):
        #"""
        #Basically the same as the method in _Primitive but MAYBE a bit more efficient....
        #"""
        #if array is None:
            #return b'\xff\xff\xff\xff'
        #length = len(array)
        #if length == 0:
            #return b'\x00\x00\x00\x00'
        #if length == 1:
            #return b'\x01\x00\x00\x00' + self.pack(array[0])
        #return struct.pack(build_array_format("<i", length, self.format[1]), length, *array)


class Primitives1(object):
    Int8 = _Primitive1("<b")
    SByte = Int8
    Int16 = _Primitive1("<h")
    Int32 = _Primitive1("<i")
    Int64 = _Primitive1("<q")
    UInt8 = _Primitive1("<B")
    Char = UInt8
    Byte = UInt8
    UInt16 = _Primitive1("<H")
    UInt32 = _Primitive1("<I")
    UInt64 = _Primitive1("<Q")
    Boolean = _Primitive1("<?")
    Double = _Primitive1("<d")
    Float = _Primitive1("<f")


class Primitives(Primitives1):
    Null = _Null()
    String = _String()
    Bytes = _Bytes()
    ByteString = _Bytes()
    CharArray = _Bytes()
    DateTime = _DateTime()
    Guid = _Guid()


def pack_uatype_array(vtype, array):
    if array is None:
        return b'\xff\xff\xff\xff'
    length = len(array)
    b = [pack_uatype(vtype, val) for val in array]
    b.insert(0, Primitives.Int32.pack(length))
    return b"".join(b)


def pack_uatype(vtype, value):
    if hasattr(Primitives, vtype.name):
        return getattr(Primitives, vtype.name).pack(value)
    elif vtype.value > 25:
        return Primitives.Bytes.pack(value)
    elif vtype.name == "ExtensionObject":
        # dependency loop: classes in uaprotocol_auto use Variant defined in this file,
        # but Variant can contain any object from uaprotocol_auto as ExtensionObject.
        # Using local import to avoid import loop
        from opcua.ua import extensionobject_to_binary
        return extensionobject_to_binary(value)
    else:
        try:
            return value.to_binary()
        except AttributeError:
            raise UaError("{0} could not be packed with value {1}".format(vtype, value))


def unpack_uatype(vtype, data):
    if hasattr(Primitives, vtype.name):
        st = getattr(Primitives, vtype.name)
        return st.unpack(data)
    elif vtype.value > 25:
        return Primitives.Bytes.unpack(data)
    elif vtype.name == "ExtensionObject":
        # dependency loop: classes in uaprotocol_auto use Variant defined in this file,
        # but Variant can contain any object from uaprotocol_auto as ExtensionObject.
        # Using local import to avoid import loop
        from opcua.ua import extensionobject_from_binary
        return extensionobject_from_binary(data)
    else:
        from opcua.ua import uatypes
        if hasattr(uatypes, vtype.name):
            klass = getattr(uatypes, vtype.name)
            return klass.from_binary(data)
        else:
            raise UaError("can not unpack unknown vtype {0!s}".format(vtype))


def unpack_uatype_array(vtype, data):
    if hasattr(Primitives, vtype.name):
        st = getattr(Primitives, vtype.name)
        return st.unpack_array(data)
    else:
        length = Primitives.Int32.unpack(data)
        if length == -1:
            return None
        else:
            return [unpack_uatype(vtype, data) for _ in range(length)]




1		"""
2		Binary protocol specific functions and constants
3		"""
4
5	1	import sys
6	1	import struct
7	1	import logging
8	1	from datetime import datetime, timedelta, tzinfo, MAXYEAR
9	1	from calendar import timegm
10	1	import uuid
11
12	1	from opcua.ua.uaerrors import UaError
13
14
15	1	if sys.version_info.major > 2:
16		unicode = str
17
18	1	logger = logging.getLogger('__name__')
19
20	1	EPOCH_AS_FILETIME = 116444736000000000 # January 1, 1970 as MS file time
21	1	HUNDREDS_OF_NANOSECONDS = 10000000
22	1	FILETIME_EPOCH_AS_DATETIME = datetime(1601, 1, 1)
23
24
25	1	def test_bit(data, offset):
26		mask = 1 << offset
27		return data & mask
28
29
30	1	def set_bit(data, offset):
31		mask = 1 << offset
32		return data \| mask
33
34
35	1	def unset_bit(data, offset):
36		mask = 1 << offset
37		return data & ~mask
38
39
40	1	class UTC(tzinfo):
41		"""
42		UTC
43		"""
44
45	1	def utcoffset(self, dt):
46		return timedelta(0)
47
48	1	def tzname(self, dt):
49		return "UTC"
50
51	1	def dst(self, dt):
52		return timedelta(0)
53
54
55		# method copied from David Buxton <[email protected]> sample code
56	1	def datetime_to_win_epoch(dt):
57		if (dt.tzinfo is None) or (dt.tzinfo.utcoffset(dt) is None):
58		dt = dt.replace(tzinfo=UTC())
59		ft = EPOCH_AS_FILETIME + (timegm(dt.timetuple()) * HUNDREDS_OF_NANOSECONDS)
60		return ft + (dt.microsecond * 10)
61
62
63	1	def win_epoch_to_datetime(epch):
64		try:
65		return FILETIME_EPOCH_AS_DATETIME + timedelta(microseconds=epch // 10)
66		except OverflowError:
67		# FILETIMEs after 31 Dec 9999 can't be converted to datetime
68		logger.warning("datetime overflow: %s", epch)
69		return datetime(MAXYEAR, 12, 31, 23, 59, 59, 999999)
70
71
72	1	def build_array_format_py2(prefix, length, fmtchar):
73		return prefix + str(length) + fmtchar
74
75
76	1	def build_array_format_py3(prefix, length, fmtchar):
77		return prefix + str(length) + chr(fmtchar)
78
79
80	1	if sys.version_info.major < 3:
81	1	build_array_format = build_array_format_py2
82		else:
83		build_array_format = build_array_format_py3
84
85
86	1	class _Primitive(object):
87
88	1	def pack_array(self, array):
89		if array is None:
90		return b'\xff\xff\xff\xff'
91		length = len(array)
92		b = [self.pack(val) for val in array]
93		b.insert(0, Primitives.Int32.pack(length))
94		return b"".join(b)
95
96	1	def unpack_array(self, data):
97		length = Primitives.Int32.unpack(data)
98		if length == -1:
99		return None
100		elif length == 0:
101		return []
102		else:
103		return [self.unpack(data) for _ in range(length)]
104
105
106	1	class _DateTime(_Primitive):
107
108	1	@staticmethod
109		def pack(dt):
110		epch = datetime_to_win_epoch(dt)
111		return Primitives.Int64.pack(epch)
112
113	1	@staticmethod
114		def unpack(data):
115		epch = Primitives.Int64.unpack(data)
116		return win_epoch_to_datetime(epch)
117
118
119	1	class _String(_Primitive):
120
121	1	@staticmethod
122		def pack(string):
123		if string is None:
124		return Primitives.Int32.pack(-1)
125		if isinstance(string, unicode):
126		string = string.encode('utf-8')
127		length = len(string)
128		return Primitives.Int32.pack(length) + string
129
130	1	@staticmethod
131		def unpack(data):
132		b = _Bytes.unpack(data)
133		if sys.version_info.major < 3:
134		return b
135		else:
136		if b is None:
137		return b
138		return b.decode("utf-8")
139
140
141	1	class _Bytes(_Primitive):
142
143	1	@staticmethod
144		def pack(data):
145		return _String.pack(data)
146
147	1	@staticmethod
148		def unpack(data):
149		length = Primitives.Int32.unpack(data)
150		if length == -1:
151		return None
152		return data.read(length)
153
154
155	1	class _Null(_Primitive):
156
157	1	@staticmethod
158		def pack(data):
159		return b""
160
161	1	@staticmethod
162		def unpack(data):
163		return None
164
165
166	1	class _Guid(_Primitive):
167
168	1	@staticmethod
169		def pack(guid):
170		# convert python UUID 6 field format to OPC UA 4 field format
171		f1 = Primitives.UInt32.pack(guid.time_low)
172		f2 = Primitives.UInt16.pack(guid.time_mid)
173		f3 = Primitives.UInt16.pack(guid.time_hi_version)
174		f4a = Primitives.Byte.pack(guid.clock_seq_hi_variant)
175		f4b = Primitives.Byte.pack(guid.clock_seq_low)
176		f4c = struct.pack('>Q', guid.node)[2:8] # no primitive .pack available for 6 byte int
177		f4 = f4a+f4b+f4c
178		# concat byte fields
179		b = f1+f2+f3+f4
180
181		return b
182
183	1	@staticmethod
184		def unpack(data):
185		# convert OPC UA 4 field format to python UUID bytes
186		f1 = struct.pack('>I', Primitives.UInt32.unpack(data))
187		f2 = struct.pack('>H', Primitives.UInt16.unpack(data))
188		f3 = struct.pack('>H', Primitives.UInt16.unpack(data))
189		f4 = data.read(8)
190		# concat byte fields
191		b = f1 + f2 + f3 + f4
192
193		return uuid.UUID(bytes=b)
194
195
196	1	class _Primitive1(_Primitive):
197	1	def __init__(self, fmt):
198	1	self.struct = struct.Struct(fmt)
199	1	self.size = self.struct.size
200	1	self.format = self.struct.format
201
202	1	def pack(self, data):
203		return struct.pack(self.format, data)
204
205	1	def unpack(self, data):
206		return struct.unpack(self.format, data.read(self.size))[0]
207
208		#def pack_array(self, array):
209		#"""
210		#Basically the same as the method in _Primitive but MAYBE a bit more efficient....
211		#"""
212		#if array is None:
213		#return b'\xff\xff\xff\xff'
214		#length = len(array)
215		#if length == 0:
216		#return b'\x00\x00\x00\x00'
217		#if length == 1:
218		#return b'\x01\x00\x00\x00' + self.pack(array[0])
219		#return struct.pack(build_array_format("<i", length, self.format[1]), length, *array)
220
221
222	1	class Primitives1(object):
223	1	Int8 = _Primitive1("<b")
224	1	SByte = Int8
225	1	Int16 = _Primitive1("<h")
226	1	Int32 = _Primitive1("<i")
227	1	Int64 = _Primitive1("<q")
228	1	UInt8 = _Primitive1("<B")
229	1	Char = UInt8
230	1	Byte = UInt8
231	1	UInt16 = _Primitive1("<H")
232	1	UInt32 = _Primitive1("<I")
233	1	UInt64 = _Primitive1("<Q")
234	1	Boolean = _Primitive1("<?")
235	1	Double = _Primitive1("<d")
236	1	Float = _Primitive1("<f")
237
238
239	1	class Primitives(Primitives1):
240	1	Null = _Null()
241	1	String = _String()
242	1	Bytes = _Bytes()
243	1	ByteString = _Bytes()
244	1	CharArray = _Bytes()
245	1	DateTime = _DateTime()
246	1	Guid = _Guid()
247
248
249	1	def pack_uatype_array(vtype, array):
250		if array is None:
251		return b'\xff\xff\xff\xff'
252		length = len(array)
253		b = [pack_uatype(vtype, val) for val in array]
254		b.insert(0, Primitives.Int32.pack(length))
255		return b"".join(b)
256
257
258	1	def pack_uatype(vtype, value):
259		if hasattr(Primitives, vtype.name):
260		return getattr(Primitives, vtype.name).pack(value)
261		elif vtype.value > 25:
262		return Primitives.Bytes.pack(value)
263		elif vtype.name == "ExtensionObject":
264		# dependency loop: classes in uaprotocol_auto use Variant defined in this file,
265		# but Variant can contain any object from uaprotocol_auto as ExtensionObject.
266		# Using local import to avoid import loop
267		from opcua.ua import extensionobject_to_binary
268		return extensionobject_to_binary(value)
269		else:
270		try:
271		return value.to_binary()
272		except AttributeError:
273		raise UaError("{0} could not be packed with value {1}".format(vtype, value))
274
275
276	1	def unpack_uatype(vtype, data):
277		if hasattr(Primitives, vtype.name):
278		st = getattr(Primitives, vtype.name)
279		return st.unpack(data)
280		elif vtype.value > 25:
281		return Primitives.Bytes.unpack(data)
282		elif vtype.name == "ExtensionObject":
283		# dependency loop: classes in uaprotocol_auto use Variant defined in this file,
284		# but Variant can contain any object from uaprotocol_auto as ExtensionObject.
285		# Using local import to avoid import loop
286		from opcua.ua import extensionobject_from_binary
287		return extensionobject_from_binary(data)
288		else:
289		from opcua.ua import uatypes
290		if hasattr(uatypes, vtype.name):
291		klass = getattr(uatypes, vtype.name)
292		return klass.from_binary(data)
293		else:
294		raise UaError("can not unpack unknown vtype {0!s}".format(vtype))
295
296
297	1	def unpack_uatype_array(vtype, data):
298		if hasattr(Primitives, vtype.name):
299		st = getattr(Primitives, vtype.name)
300		return st.unpack_array(data)
301		else:
302		length = Primitives.Int32.unpack(data)
303		if length == -1:
304		return None
305		else:
306		return [unpack_uatype(vtype, data) for _ in range(length)]
307
308
309

FreeOpcUa / python-opcua

Pull Request — master (#494)

_Guid.pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like