list_to_binary() - Code Metrics - Inspection of "replace to/from_binary() methods by global to_bina..." - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#490)

by Olivier

created 2017-09-04 19:06 UTC

list_to_binary() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
c	1
b	0
f	0
dl	0
loc	9
ccs	0
cts	8
cp	0
crap	12
rs	9.6666

"""
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 enum import IntEnum, Enum

from opcua.ua.uaerrors import UaError
from opcua.common.utils import Buffer
from opcua import ua


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):
    SByte = _Primitive1("<b")
    Int16 = _Primitive1("<h")
    Int32 = _Primitive1("<i")
    Int64 = _Primitive1("<q")
    Byte = _Primitive1("<B")
    Char = Byte
    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 = _String()
    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 == ua.VariantType.ExtensionObject:
        return extensionobject_to_binary(value)
    elif vtype in (ua.VariantType.NodeId, ua.VariantType.ExpandedNodeId):
        return nodeid_to_binary(value)
    elif vtype == ua.VariantType.Variant:
        return variant_to_binary(value)
    else:
        return struct_to_binary(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 == ua.VariantType.ExtensionObject:
        return extensionobject_from_binary(data)
    elif vtype in (ua.VariantType.NodeId, ua.VariantType.ExpandedNodeId):
        return nodeid_from_binary(data)
    elif vtype == ua.VariantType.Variant:
        return variant_from_binary(data)
    else:
        if hasattr(ua, vtype.name):
            klass = getattr(ua, vtype.name)
            return struct_from_binary(klass, data)
        else:
            raise UaError("Cannot unpack unknown variant type {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)]


def struct_to_binary(obj):
    packet = []
    has_switch = hasattr(obj, "ua_switches")
    if has_switch:
        for name, switch in obj.ua_switches.items():
            member = getattr(obj, name)
            container_name, idx = switch
            if member is not None:
                container_val = getattr(obj, container_name)
                container_val = container_val | 1 << idx
                setattr(obj, container_name, container_val)
    for name, uatype in obj.ua_types:
        val = getattr(obj, name)
        if uatype.startswith("ListOf"):
            packet.append(list_to_binary(val, uatype[6:]))
        else:
            if has_switch and val is None and name in obj.ua_switches:
                pass
            else:
                packet.append(to_binary(val, uatype))
    return b''.join(packet)


def to_binary(val, uatype):
    """
    try hard to pack a python object to binary
    """
    if isinstance(val, (list, tuple)):
        length = len(val)
        b = [to_binary(el) for el in val]
        b.insert(0, Primitives.Int32.pack(length))
        return b"".join(b)
    elif isinstance(uatype, (str, unicode)) and hasattr(ua.VariantType, uatype):
        vtype = getattr(ua.VariantType, uatype)
        return pack_uatype(vtype, val)
    elif isinstance(val, (IntEnum, Enum)):
        return Primitives.UInt32.pack(val.value)
    elif isinstance(val, ua.NodeId):
        return nodeid_to_binary(val)
    elif isinstance(val, ua.Header): # should have this here or use directly correct method when necessary??
        return header_to_binary(val) #
    elif isinstance(val, ua.Variant):
        return variant_to_binary(val)
    elif hasattr(val, "ua_types"):
        return struct_to_binary(val)
    else:
        raise UaError("No known way to pack {} of type {} to ua binary".format(val, uatype))


def list_to_binary(val, uatype):
    if val is None:
        return Primitives.Int32.pack(-1)
    else:
        pack = []
        pack.append(Primitives.Int32.pack(len(val)))
        for el in val:
            pack.append(to_binary(el, uatype))
        return b''.join(pack)


def nodeid_to_binary(nodeid):
    if nodeid.NodeIdType == ua.NodeIdType.TwoByte:
        return struct.pack("<BB", nodeid.NodeIdType.value, nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.FourByte:
        return struct.pack("<BBH", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.Numeric:
        return struct.pack("<BHI", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.String:
        return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
            Primitives.String.pack(nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.ByteString:
        return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
            Primitives.Bytes.pack(nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.Guid:
        return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
               Primitives.Guid.pack(nodeid.Identifier)
    else:
        return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
            nodeid.Identifier.to_binary()
    # FIXME: Missing NNamespaceURI and ServerIndex


def nodeid_from_binary(data):
    nid = ua.NodeId()
    encoding = ord(data.read(1))
    nid.NodeIdType = ua.NodeIdType(encoding & 0b00111111)

    if nid.NodeIdType == ua.NodeIdType.TwoByte:
        nid.Identifier = ord(data.read(1))
    elif nid.NodeIdType == ua.NodeIdType.FourByte:
        nid.NamespaceIndex, nid.Identifier = struct.unpack("<BH", data.read(3))
    elif nid.NodeIdType == ua.NodeIdType.Numeric:
        nid.NamespaceIndex, nid.Identifier = struct.unpack("<HI", data.read(6))
    elif nid.NodeIdType == ua.NodeIdType.String:
        nid.NamespaceIndex = Primitives.UInt16.unpack(data)
        nid.Identifier = Primitives.String.unpack(data)
    elif nid.NodeIdType == ua.NodeIdType.ByteString:
        nid.NamespaceIndex = Primitives.UInt16.unpack(data)
        nid.Identifier = Primitives.Bytes.unpack(data)
    elif nid.NodeIdType == ua.NodeIdType.Guid:
        nid.NamespaceIndex = Primitives.UInt16.unpack(data)
        nid.Identifier = Primitives.Guid.unpack(data)
    else:
        raise UaError("Unknown NodeId encoding: " + str(nid.NodeIdType))

    if test_bit(encoding, 7):
        nid.NamespaceUri = Primitives.String.unpack(data)
    if test_bit(encoding, 6):
        nid.ServerIndex = Primitives.UInt32.unpack(data)

    return nid


def variant_to_binary(var):
    b = []
    encoding = var.VariantType.value & 0b111111
    if var.is_array or isinstance(var.Value, (list, tuple)):
        var.is_array = True
        encoding = set_bit(encoding, 7)
        if var.Dimensions is not None:
            encoding = set_bit(encoding, 6)
        b.append(Primitives.Byte.pack(encoding))
        b.append(pack_uatype_array(var.VariantType, ua.flatten(var.Value)))
        if var.Dimensions is not None:
            b.append(pack_uatype_array(ua.VariantType.Int32, var.Dimensions))
    else:
        b.append(Primitives.Byte.pack(encoding))
        b.append(pack_uatype(var.VariantType, var.Value))

    return b"".join(b)


def variant_from_binary(data):
    dimensions = None
    array = False
    encoding = ord(data.read(1))
    int_type = encoding & 0b00111111
    vtype = ua.datatype_to_varianttype(int_type)
    if test_bit(encoding, 7):
        value = unpack_uatype_array(vtype, data)
        array = True
    else:
        value = unpack_uatype(vtype, data)
    if test_bit(encoding, 6):
        dimensions = unpack_uatype_array(ua.VariantType.Int32, data)
        value = reshape(value, dimensions)
    return ua.Variant(value, vtype, dimensions, is_array=array)


def reshape(flat, dims):
    subdims = dims[1:]
    subsize = 1
    for i in subdims:
        if i == 0:
            i = 1
        subsize *= i
    while dims[0] * subsize > len(flat):
        flat.append([])
    if not subdims or subdims == [0]:
        return flat
    return [reshape(flat[i: i + subsize], subdims) for i in range(0, len(flat), subsize)]


def extensionobject_from_binary(data):
    """
    Convert binary-coded ExtensionObject to a Python object.
    Returns an object, or None if TypeId is zero
    """
    typeid = nodeid_from_binary(data)
    Encoding = ord(data.read(1))
    body = None
    if Encoding & (1 << 0):
        length = Primitives.Int32.unpack(data)
        if length < 1:
            body = Buffer(b"")
        else:
            body = data.copy(length)
            data.skip(length)
    if typeid.Identifier == 0:
        return None
    elif typeid in ua.extension_object_classes:
        klass = ua.extension_object_classes[typeid]
        if body is None:
            raise UaError("parsing ExtensionObject {0} without data".format(klass.__name__))
        return from_binary(klass, body)
    else:
        e = ua.ExtensionObject()
        e.TypeId = typeid
        e.Encoding = Encoding
        if body is not None:
            e.Body = body.read(len(body))
        return e


def extensionobject_to_binary(obj):
    """
    Convert Python object to binary-coded ExtensionObject.
    If obj is None, convert to empty ExtensionObject (TypeId = 0, no Body).
    Returns a binary string
    """
    if isinstance(obj, ua.ExtensionObject):
        return obj.to_binary()
    if obj is None:
        TypeId = ua.NodeId()
        Encoding = 0
        Body = None
    else:
        TypeId = ua.extension_object_ids[obj.__class__.__name__]
        Encoding = 0x01
        Body = struct_to_binary(obj)
    packet = []
    packet.append(nodeid_to_binary(TypeId))
    packet.append(Primitives.Byte.pack(Encoding))
    if Body:
        packet.append(Primitives.Bytes.pack(Body))
    return b''.join(packet)


def from_binary(uatype, data):
    """
    unpack data given an uatype as a string or a python class having a ua_types memeber
    """
    if isinstance(uatype, (str, unicode)) and uatype.startswith("ListOf"):
        size = Primitives.Int32.unpack(data)
        if size == -1:
            return None
        res = []
        for _ in range(size):
            res.append(from_binary(uatype[6:], data))
        return res
    elif isinstance(uatype, (str, unicode)) and hasattr(ua.VariantType, uatype):
        vtype = getattr(ua.VariantType, uatype)
        return unpack_uatype(vtype, data)
    else:
        return struct_from_binary(uatype, data)


def struct_from_binary(objtype, data):
    """
    unpack an ua struct. Arguments are an objtype as Python class or string
    """
    if isinstance(objtype, (unicode, str)):
        objtype = getattr(ua, objtype)
    if issubclass(objtype, Enum):
        return objtype(Primitives.UInt32.unpack(data))
    obj = objtype()
    for name, uatype in obj.ua_types:
        # if our member has a swtich and it is not set we skip it
        if hasattr(obj, "ua_switches") and name in obj.ua_switches:
            container_name, idx = obj.ua_switches[name]
            val = getattr(obj, container_name)
            if not test_bit(val, idx):
                continue
        val = from_binary(uatype, data) 
        setattr(obj, name, val)
    return obj


def header_to_binary(hdr):
    b = []
    b.append(struct.pack("<3ss", hdr.MessageType, hdr.ChunkType))
    size = hdr.body_size + 8
    if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
        size += 4
    b.append(Primitives.UInt32.pack(size))
    if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
        b.append(Primitives.UInt32.pack(hdr.ChannelId))
    return b"".join(b)


def header_from_binary(data):
    hdr = ua.Header()
    hdr.MessageType, hdr.ChunkType, hdr.packet_size = struct.unpack("<3scI", data.read(8))
    hdr.body_size = hdr.packet_size - 8
    if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
        hdr.body_size -= 4
        hdr.ChannelId = Primitives.UInt32.unpack(data)
    return hdr



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	1	from enum import IntEnum, Enum
12
13	1	from opcua.ua.uaerrors import UaError
14	1	from opcua.common.utils import Buffer
15	1	from opcua import ua
16
17
18		if sys.version_info.major > 2:
19		unicode = str
20
21		logger = logging.getLogger('__name__')
22
23		EPOCH_AS_FILETIME = 116444736000000000 # January 1, 1970 as MS file time
24		HUNDREDS_OF_NANOSECONDS = 10000000
25		FILETIME_EPOCH_AS_DATETIME = datetime(1601, 1, 1)
26
27
28		def test_bit(data, offset):
29		mask = 1 << offset
30		return data & mask
31
32
33		def set_bit(data, offset):
34		mask = 1 << offset
35		return data \| mask
36
37
38		def unset_bit(data, offset):
39		mask = 1 << offset
40		return data & ~mask
41
42
43		class UTC(tzinfo):
44		"""
45		UTC
46		"""
47
48		def utcoffset(self, dt):
49		return timedelta(0)
50
51		def tzname(self, dt):
52		return "UTC"
53
54		def dst(self, dt):
55		return timedelta(0)
56
57
58		# method copied from David Buxton <[email protected]> sample code
59		def datetime_to_win_epoch(dt):
60		if (dt.tzinfo is None) or (dt.tzinfo.utcoffset(dt) is None):
61		dt = dt.replace(tzinfo=UTC())
62		ft = EPOCH_AS_FILETIME + (timegm(dt.timetuple()) * HUNDREDS_OF_NANOSECONDS)
63		return ft + (dt.microsecond * 10)
64
65
66		def win_epoch_to_datetime(epch):
67		try:
68		return FILETIME_EPOCH_AS_DATETIME + timedelta(microseconds=epch // 10)
69		except OverflowError:
70		# FILETIMEs after 31 Dec 9999 can't be converted to datetime
71		logger.warning("datetime overflow: %s", epch)
72		return datetime(MAXYEAR, 12, 31, 23, 59, 59, 999999)
73
74
75		def build_array_format_py2(prefix, length, fmtchar):
76		return prefix + str(length) + fmtchar
77
78
79		def build_array_format_py3(prefix, length, fmtchar):
80		return prefix + str(length) + chr(fmtchar)
81
82
83		if sys.version_info.major < 3:
84		build_array_format = build_array_format_py2
85		else:
86		build_array_format = build_array_format_py3
87
88
89		class _Primitive(object):
90
91		def pack_array(self, array):
92		if array is None:
93		return b'\xff\xff\xff\xff'
94		length = len(array)
95		b = [self.pack(val) for val in array]
96		b.insert(0, Primitives.Int32.pack(length))
97		return b"".join(b)
98
99		def unpack_array(self, data):
100		length = Primitives.Int32.unpack(data)
101		if length == -1:
102		return None
103		elif length == 0:
104		return []
105		else:
106		return [self.unpack(data) for _ in range(length)]
107
108
109		class _DateTime(_Primitive):
110
111		@staticmethod
112		def pack(dt):
113		epch = datetime_to_win_epoch(dt)
114		return Primitives.Int64.pack(epch)
115
116		@staticmethod
117		def unpack(data):
118		epch = Primitives.Int64.unpack(data)
119		return win_epoch_to_datetime(epch)
120
121
122		class _String(_Primitive):
123
124		@staticmethod
125		def pack(string):
126		if string is None:
127		return Primitives.Int32.pack(-1)
128		if isinstance(string, unicode):
129		string = string.encode('utf-8')
130		length = len(string)
131		return Primitives.Int32.pack(length) + string
132
133		@staticmethod
134		def unpack(data):
135		b = _Bytes.unpack(data)
136		if sys.version_info.major < 3:
137		return b
138		else:
139		if b is None:
140		return b
141		return b.decode("utf-8")
142
143
144		class _Bytes(_Primitive):
145
146		@staticmethod
147		def pack(data):
148		return _String.pack(data)
149
150		@staticmethod
151		def unpack(data):
152		length = Primitives.Int32.unpack(data)
153		if length == -1:
154		return None
155		return data.read(length)
156
157
158		class _Null(_Primitive):
159
160		@staticmethod
161		def pack(data):
162		return b""
163
164		@staticmethod
165		def unpack(data):
166		return None
167
168
169		class _Guid(_Primitive):
170
171		@staticmethod
172		def pack(guid):
173		# convert python UUID 6 field format to OPC UA 4 field format
174		f1 = Primitives.UInt32.pack(guid.time_low)
175		f2 = Primitives.UInt16.pack(guid.time_mid)
176		f3 = Primitives.UInt16.pack(guid.time_hi_version)
177		f4a = Primitives.Byte.pack(guid.clock_seq_hi_variant)
178		f4b = Primitives.Byte.pack(guid.clock_seq_low)
179		f4c = struct.pack('>Q', guid.node)[2:8] # no primitive .pack available for 6 byte int
180		f4 = f4a+f4b+f4c
181		# concat byte fields
182		b = f1+f2+f3+f4
183
184		return b
185
186		@staticmethod
187		def unpack(data):
188		# convert OPC UA 4 field format to python UUID bytes
189		f1 = struct.pack('>I', Primitives.UInt32.unpack(data))
190		f2 = struct.pack('>H', Primitives.UInt16.unpack(data))
191		f3 = struct.pack('>H', Primitives.UInt16.unpack(data))
192		f4 = data.read(8)
193		# concat byte fields
194		b = f1 + f2 + f3 + f4
195
196		return uuid.UUID(bytes=b)
197
198
199		class _Primitive1(_Primitive):
200		def __init__(self, fmt):
201		self.struct = struct.Struct(fmt)
202		self.size = self.struct.size
203		self.format = self.struct.format
204
205		def pack(self, data):
206		return struct.pack(self.format, data)
207
208		def unpack(self, data):
209		return struct.unpack(self.format, data.read(self.size))[0]
210
211		#def pack_array(self, array):
212		#"""
213		#Basically the same as the method in _Primitive but MAYBE a bit more efficient....
214		#"""
215		#if array is None:
216		#return b'\xff\xff\xff\xff'
217		#length = len(array)
218		#if length == 0:
219		#return b'\x00\x00\x00\x00'
220		#if length == 1:
221		#return b'\x01\x00\x00\x00' + self.pack(array[0])
222		#return struct.pack(build_array_format("<i", length, self.format[1]), length, *array)
223
224
225		class Primitives1(object):
226		SByte = _Primitive1("<b")
227		Int16 = _Primitive1("<h")
228		Int32 = _Primitive1("<i")
229		Int64 = _Primitive1("<q")
230		Byte = _Primitive1("<B")
231		Char = Byte
232		UInt16 = _Primitive1("<H")
233		UInt32 = _Primitive1("<I")
234		UInt64 = _Primitive1("<Q")
235		Boolean = _Primitive1("<?")
236		Double = _Primitive1("<d")
237		Float = _Primitive1("<f")
238
239
240		class Primitives(Primitives1):
241		Null = _Null()
242		String = _String()
243		Bytes = _Bytes()
244		ByteString = _Bytes()
245		CharArray = _String()
246		DateTime = _DateTime()
247		Guid = _Guid()
248
249
250		def pack_uatype_array(vtype, array):
251		if array is None:
252		return b'\xff\xff\xff\xff'
253		length = len(array)
254		b = [pack_uatype(vtype, val) for val in array]
255		b.insert(0, Primitives.Int32.pack(length))
256		return b"".join(b)
257
258
259		def pack_uatype(vtype, value):
260		if hasattr(Primitives, vtype.name):
261		return getattr(Primitives, vtype.name).pack(value)
262		elif vtype.value > 25:
263		return Primitives.Bytes.pack(value)
264		elif vtype == ua.VariantType.ExtensionObject:
265		return extensionobject_to_binary(value)
266		elif vtype in (ua.VariantType.NodeId, ua.VariantType.ExpandedNodeId):
267		return nodeid_to_binary(value)
268		elif vtype == ua.VariantType.Variant:
269		return variant_to_binary(value)
270		else:
271		return struct_to_binary(value)
272
273
274		def unpack_uatype(vtype, data):
275		if hasattr(Primitives, vtype.name):
276		st = getattr(Primitives, vtype.name)
277		return st.unpack(data)
278		elif vtype.value > 25:
279		return Primitives.Bytes.unpack(data)
280		elif vtype == ua.VariantType.ExtensionObject:
281		return extensionobject_from_binary(data)
282		elif vtype in (ua.VariantType.NodeId, ua.VariantType.ExpandedNodeId):
283		return nodeid_from_binary(data)
284		elif vtype == ua.VariantType.Variant:
285		return variant_from_binary(data)
286		else:
287		if hasattr(ua, vtype.name):
288		klass = getattr(ua, vtype.name)
289		return struct_from_binary(klass, data)
290		else:
291		raise UaError("Cannot unpack unknown variant type {0!s}".format(vtype))
292
293
294		def unpack_uatype_array(vtype, data):
295		if hasattr(Primitives, vtype.name):
296		st = getattr(Primitives, vtype.name)
297		return st.unpack_array(data)
298		else:
299		length = Primitives.Int32.unpack(data)
300		if length == -1:
301		return None
302		else:
303		return [unpack_uatype(vtype, data) for _ in range(length)]
304
305
306		def struct_to_binary(obj):
307		packet = []
308		has_switch = hasattr(obj, "ua_switches")
309		if has_switch:
310		for name, switch in obj.ua_switches.items():
311		member = getattr(obj, name)
312		container_name, idx = switch
313		if member is not None:
314		container_val = getattr(obj, container_name)
315		container_val = container_val \| 1 << idx
316		setattr(obj, container_name, container_val)
317		for name, uatype in obj.ua_types:
318		val = getattr(obj, name)
319		if uatype.startswith("ListOf"):
320		packet.append(list_to_binary(val, uatype[6:]))
321		else:
322		if has_switch and val is None and name in obj.ua_switches:
323		pass
324		else:
325		packet.append(to_binary(val, uatype))
326		return b''.join(packet)
327
328
329		def to_binary(val, uatype):
330		"""
331		try hard to pack a python object to binary
332		"""
333		if isinstance(val, (list, tuple)):
334		length = len(val)
335		b = [to_binary(el) for el in val]
336		b.insert(0, Primitives.Int32.pack(length))
337		return b"".join(b)
338		elif isinstance(uatype, (str, unicode)) and hasattr(ua.VariantType, uatype):
339		vtype = getattr(ua.VariantType, uatype)
340		return pack_uatype(vtype, val)
341		elif isinstance(val, (IntEnum, Enum)):
342		return Primitives.UInt32.pack(val.value)
343		elif isinstance(val, ua.NodeId):
344		return nodeid_to_binary(val)
345		elif isinstance(val, ua.Header): # should have this here or use directly correct method when necessary??
346		return header_to_binary(val) #
347		elif isinstance(val, ua.Variant):
348		return variant_to_binary(val)
349		elif hasattr(val, "ua_types"):
350		return struct_to_binary(val)
351		else:
352		raise UaError("No known way to pack {} of type {} to ua binary".format(val, uatype))
353
354
355		def list_to_binary(val, uatype):
356		if val is None:
357		return Primitives.Int32.pack(-1)
358		else:
359		pack = []
360		pack.append(Primitives.Int32.pack(len(val)))
361		for el in val:
362		pack.append(to_binary(el, uatype))
363		return b''.join(pack)
364
365
366		def nodeid_to_binary(nodeid):
367		if nodeid.NodeIdType == ua.NodeIdType.TwoByte:
368		return struct.pack("<BB", nodeid.NodeIdType.value, nodeid.Identifier)
369		elif nodeid.NodeIdType == ua.NodeIdType.FourByte:
370		return struct.pack("<BBH", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
371		elif nodeid.NodeIdType == ua.NodeIdType.Numeric:
372		return struct.pack("<BHI", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
373		elif nodeid.NodeIdType == ua.NodeIdType.String:
374		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
375		Primitives.String.pack(nodeid.Identifier)
376		elif nodeid.NodeIdType == ua.NodeIdType.ByteString:
377		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
378		Primitives.Bytes.pack(nodeid.Identifier)
379		elif nodeid.NodeIdType == ua.NodeIdType.Guid:
380		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
381		Primitives.Guid.pack(nodeid.Identifier)
382		else:
383		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
384		nodeid.Identifier.to_binary()
385		# FIXME: Missing NNamespaceURI and ServerIndex
386
387
388		def nodeid_from_binary(data):
389		nid = ua.NodeId()
390		encoding = ord(data.read(1))
391		nid.NodeIdType = ua.NodeIdType(encoding & 0b00111111)
392
393		if nid.NodeIdType == ua.NodeIdType.TwoByte:
394		nid.Identifier = ord(data.read(1))
395		elif nid.NodeIdType == ua.NodeIdType.FourByte:
396		nid.NamespaceIndex, nid.Identifier = struct.unpack("<BH", data.read(3))
397		elif nid.NodeIdType == ua.NodeIdType.Numeric:
398		nid.NamespaceIndex, nid.Identifier = struct.unpack("<HI", data.read(6))
399		elif nid.NodeIdType == ua.NodeIdType.String:
400		nid.NamespaceIndex = Primitives.UInt16.unpack(data)
401		nid.Identifier = Primitives.String.unpack(data)
402		elif nid.NodeIdType == ua.NodeIdType.ByteString:
403		nid.NamespaceIndex = Primitives.UInt16.unpack(data)
404		nid.Identifier = Primitives.Bytes.unpack(data)
405		elif nid.NodeIdType == ua.NodeIdType.Guid:
406		nid.NamespaceIndex = Primitives.UInt16.unpack(data)
407		nid.Identifier = Primitives.Guid.unpack(data)
408		else:
409		raise UaError("Unknown NodeId encoding: " + str(nid.NodeIdType))
410
411		if test_bit(encoding, 7):
412		nid.NamespaceUri = Primitives.String.unpack(data)
413		if test_bit(encoding, 6):
414		nid.ServerIndex = Primitives.UInt32.unpack(data)
415
416		return nid
417
418
419		def variant_to_binary(var):
420		b = []
421		encoding = var.VariantType.value & 0b111111
422		if var.is_array or isinstance(var.Value, (list, tuple)):
423		var.is_array = True
424		encoding = set_bit(encoding, 7)
425		if var.Dimensions is not None:
426		encoding = set_bit(encoding, 6)
427		b.append(Primitives.Byte.pack(encoding))
428		b.append(pack_uatype_array(var.VariantType, ua.flatten(var.Value)))
429		if var.Dimensions is not None:
430		b.append(pack_uatype_array(ua.VariantType.Int32, var.Dimensions))
431		else:
432		b.append(Primitives.Byte.pack(encoding))
433		b.append(pack_uatype(var.VariantType, var.Value))
434
435		return b"".join(b)
436
437
438		def variant_from_binary(data):
439		dimensions = None
440		array = False
441		encoding = ord(data.read(1))
442		int_type = encoding & 0b00111111
443		vtype = ua.datatype_to_varianttype(int_type)
444		if test_bit(encoding, 7):
445		value = unpack_uatype_array(vtype, data)
446		array = True
447		else:
448		value = unpack_uatype(vtype, data)
449		if test_bit(encoding, 6):
450		dimensions = unpack_uatype_array(ua.VariantType.Int32, data)
451		value = reshape(value, dimensions)
452		return ua.Variant(value, vtype, dimensions, is_array=array)
453
454
455		def reshape(flat, dims):
456		subdims = dims[1:]
457		subsize = 1
458		for i in subdims:
459		if i == 0:
460		i = 1
461		subsize *= i
462		while dims[0] * subsize > len(flat):
463		flat.append([])
464		if not subdims or subdims == [0]:
465		return flat
466		return [reshape(flat[i: i + subsize], subdims) for i in range(0, len(flat), subsize)]
467
468
469		def extensionobject_from_binary(data):
470		"""
471		Convert binary-coded ExtensionObject to a Python object.
472		Returns an object, or None if TypeId is zero
473		"""
474		typeid = nodeid_from_binary(data)
475		Encoding = ord(data.read(1))
476		body = None
477		if Encoding & (1 << 0):
478		length = Primitives.Int32.unpack(data)
479		if length < 1:
480		body = Buffer(b"")
481		else:
482		body = data.copy(length)
483		data.skip(length)
484		if typeid.Identifier == 0:
485		return None
486		elif typeid in ua.extension_object_classes:
487		klass = ua.extension_object_classes[typeid]
488		if body is None:
489		raise UaError("parsing ExtensionObject {0} without data".format(klass.__name__))
490		return from_binary(klass, body)
491		else:
492		e = ua.ExtensionObject()
493		e.TypeId = typeid
494		e.Encoding = Encoding
495		if body is not None:
496		e.Body = body.read(len(body))
497		return e
498
499
500		def extensionobject_to_binary(obj):
501		"""
502		Convert Python object to binary-coded ExtensionObject.
503		If obj is None, convert to empty ExtensionObject (TypeId = 0, no Body).
504		Returns a binary string
505		"""
506		if isinstance(obj, ua.ExtensionObject):
507		return obj.to_binary()
508		if obj is None:
509		TypeId = ua.NodeId()
510		Encoding = 0
511		Body = None
512		else:
513		TypeId = ua.extension_object_ids[obj.__class__.__name__]
514		Encoding = 0x01
515		Body = struct_to_binary(obj)
516		packet = []
517		packet.append(nodeid_to_binary(TypeId))
518		packet.append(Primitives.Byte.pack(Encoding))
519		if Body:
520		packet.append(Primitives.Bytes.pack(Body))
521		return b''.join(packet)
522
523
524		def from_binary(uatype, data):
525		"""
526		unpack data given an uatype as a string or a python class having a ua_types memeber
527		"""
528		if isinstance(uatype, (str, unicode)) and uatype.startswith("ListOf"):
529		size = Primitives.Int32.unpack(data)
530		if size == -1:
531		return None
532		res = []
533		for _ in range(size):
534		res.append(from_binary(uatype[6:], data))
535		return res
536		elif isinstance(uatype, (str, unicode)) and hasattr(ua.VariantType, uatype):
537		vtype = getattr(ua.VariantType, uatype)
538		return unpack_uatype(vtype, data)
539		else:
540		return struct_from_binary(uatype, data)
541
542
543		def struct_from_binary(objtype, data):
544		"""
545		unpack an ua struct. Arguments are an objtype as Python class or string
546		"""
547		if isinstance(objtype, (unicode, str)):
548		objtype = getattr(ua, objtype)
549		if issubclass(objtype, Enum):
550		return objtype(Primitives.UInt32.unpack(data))
551		obj = objtype()
552		for name, uatype in obj.ua_types:
553		# if our member has a swtich and it is not set we skip it
554		if hasattr(obj, "ua_switches") and name in obj.ua_switches:
555		container_name, idx = obj.ua_switches[name]
556		val = getattr(obj, container_name)
557		if not test_bit(val, idx):
558		continue
559		val = from_binary(uatype, data)
560		setattr(obj, name, val)
561		return obj
562
563
564		def header_to_binary(hdr):
565		b = []
566		b.append(struct.pack("<3ss", hdr.MessageType, hdr.ChunkType))
567		size = hdr.body_size + 8
568		if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
569		size += 4
570		b.append(Primitives.UInt32.pack(size))
571		if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
572		b.append(Primitives.UInt32.pack(hdr.ChannelId))
573		return b"".join(b)
574
575
576		def header_from_binary(data):
577		hdr = ua.Header()
578		hdr.MessageType, hdr.ChunkType, hdr.packet_size = struct.unpack("<3scI", data.read(8))
579		hdr.body_size = hdr.packet_size - 8
580		if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
581		hdr.body_size -= 4
582		hdr.ChannelId = Primitives.UInt32.unpack(data)
583		return hdr
584
585

FreeOpcUa / python-opcua

Pull Request — master (#490)

list_to_binary() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like