nodeid_from_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

Test Failed

Pull Request — master (#490)

by Olivier

created 2017-09-01 20:21 UTC

nodeid_from_binary() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	90

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	9
c	1
b	0
f	0
dl	0
loc	29
ccs	0
cts	25
cp	0
crap	90
rs	3

"""
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):
    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 = _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.name == "ExtensionObject":
        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":
        return extensionobject_from_binary(data)
    else:
        if hasattr(ua.uatypes, vtype.name):
            klass = getattr(ua.uatypes, vtype.name)
            return from_binary(klass, 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)]


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=None):
    if isinstance(uatype, (str, unicode)) and hasattr(Primitives, uatype):
        st = getattr(Primitives, uatype)
        return st.pack(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):
        return header_to_binary(val)
    elif isinstance(val, ua.Variant):
        return variant_to_binary(val)
    elif uatype == "ExtensionObject":
        #pack val into an extensionobject
        return extensionobject_to_binary(val)
    elif hasattr(val, "ua_types"):
        return struct_to_binary(val)
    else:
        raise ValueError("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.UInt8.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.UInt8.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 = to_binary(obj)
    packet = []
    packet.append(to_binary(TypeId))
    packet.append(Primitives.UInt8.pack(Encoding))
    if Body:
        packet.append(Primitives.Bytes.pack(Body))
    return b''.join(packet)


def from_binary(uatype, data):
    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(Primitives, uatype):
        st = getattr(Primitives, uatype)
        res = st.unpack(data)
        return res
        return st.unpack(data)
    elif uatype == ua.NodeId or uatype in ("NodeId", "ExpandedNodeId"):
        return nodeid_from_binary(data)
    elif uatype == ua.Variant or uatype == "Variant":
        return variant_from_binary(data)
    elif uatype == ua.ExtensionObject or uatype == "ExtensionObject":
        return extensionobject_from_binary(data)
    else:
        return struct_from_binary(uatype, data)


def struct_from_binary(objtype, data):
    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		Int8 = _Primitive1("<b")
227		SByte = Int8
228		Int16 = _Primitive1("<h")
229		Int32 = _Primitive1("<i")
230		Int64 = _Primitive1("<q")
231		UInt8 = _Primitive1("<B")
232		Char = UInt8
233		Byte = UInt8
234		UInt16 = _Primitive1("<H")
235		UInt32 = _Primitive1("<I")
236		UInt64 = _Primitive1("<Q")
237		Boolean = _Primitive1("<?")
238		Double = _Primitive1("<d")
239		Float = _Primitive1("<f")
240
241
242		class Primitives(Primitives1):
243		Null = _Null()
244		String = _String()
245		Bytes = _Bytes()
246		ByteString = _Bytes()
247		CharArray = _String()
248		DateTime = _DateTime()
249		Guid = _Guid()
250
251
252		def pack_uatype_array(vtype, array):
253		if array is None:
254		return b'\xff\xff\xff\xff'
255		length = len(array)
256		b = [pack_uatype(vtype, val) for val in array]
257		b.insert(0, Primitives.Int32.pack(length))
258		return b"".join(b)
259
260
261		def pack_uatype(vtype, value):
262		if hasattr(Primitives, vtype.name):
263		return getattr(Primitives, vtype.name).pack(value)
264		elif vtype.value > 25:
265		return Primitives.Bytes.pack(value)
266		elif vtype.name == "ExtensionObject":
267		return extensionobject_to_binary(value)
268		else:
269		try:
270		return value.to_binary()
271		except AttributeError:
272		raise UaError("{0} could not be packed with value {1}".format(vtype, value))
273
274
275		def unpack_uatype(vtype, data):
276		if hasattr(Primitives, vtype.name):
277		st = getattr(Primitives, vtype.name)
278		return st.unpack(data)
279		elif vtype.value > 25:
280		return Primitives.Bytes.unpack(data)
281		elif vtype.name == "ExtensionObject":
282		return extensionobject_from_binary(data)
283		else:
284		if hasattr(ua.uatypes, vtype.name):
285		klass = getattr(ua.uatypes, vtype.name)
286		return from_binary(klass, data)
287		else:
288		raise UaError("can not unpack unknown vtype {0!s}".format(vtype))
289
290
291		def unpack_uatype_array(vtype, data):
292		if hasattr(Primitives, vtype.name):
293		st = getattr(Primitives, vtype.name)
294		return st.unpack_array(data)
295		else:
296		length = Primitives.Int32.unpack(data)
297		if length == -1:
298		return None
299		else:
300		return [unpack_uatype(vtype, data) for _ in range(length)]
301
302
303		def struct_to_binary(obj):
304		packet = []
305		has_switch = hasattr(obj, "ua_switches")
306		if has_switch:
307		for name, switch in obj.ua_switches.items():
308		member = getattr(obj, name)
309		container_name, idx = switch
310		if member is not None:
311		container_val = getattr(obj, container_name)
312		container_val = container_val \| 1 << idx
313		setattr(obj, container_name, container_val)
314		for name, uatype in obj.ua_types:
315		val = getattr(obj, name)
316		if uatype.startswith("ListOf"):
317		packet.append(list_to_binary(val, uatype[6:]))
318		else:
319		if has_switch and val is None and name in obj.ua_switches:
320		pass
321		else:
322		packet.append(to_binary(val, uatype))
323		return b''.join(packet)
324
325
326		def to_binary(val, uatype=None):
327		if isinstance(uatype, (str, unicode)) and hasattr(Primitives, uatype):
328		st = getattr(Primitives, uatype)
329		return st.pack(val)
330		elif isinstance(val, (IntEnum, Enum)):
331		return Primitives.UInt32.pack(val.value)
332		elif isinstance(val, ua.NodeId):
333		return nodeid_to_binary(val)
334		elif isinstance(val, ua.Header):
335		return header_to_binary(val)
336		elif isinstance(val, ua.Variant):
337		return variant_to_binary(val)
338		elif uatype == "ExtensionObject":
339		#pack val into an extensionobject
340		return extensionobject_to_binary(val)
341		elif hasattr(val, "ua_types"):
342		return struct_to_binary(val)
343		else:
344		raise ValueError("No known way to pack {} of type {} to ua binary".format(val, uatype))
345
346
347		def list_to_binary(val, uatype):
348		if val is None:
349		return Primitives.Int32.pack(-1)
350		else:
351		pack = []
352		pack.append(Primitives.Int32.pack(len(val)))
353		for el in val:
354		pack.append(to_binary(el, uatype))
355		return b''.join(pack)
356
357
358		def nodeid_to_binary(nodeid):
359		if nodeid.NodeIdType == ua.NodeIdType.TwoByte:
360		return struct.pack("<BB", nodeid.NodeIdType.value, nodeid.Identifier)
361		elif nodeid.NodeIdType == ua.NodeIdType.FourByte:
362		return struct.pack("<BBH", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
363		elif nodeid.NodeIdType == ua.NodeIdType.Numeric:
364		return struct.pack("<BHI", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
365		elif nodeid.NodeIdType == ua.NodeIdType.String:
366		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
367		Primitives.String.pack(nodeid.Identifier)
368		elif nodeid.NodeIdType == ua.NodeIdType.ByteString:
369		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
370		Primitives.Bytes.pack(nodeid.Identifier)
371		elif nodeid.NodeIdType == ua.NodeIdType.Guid:
372		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
373		Primitives.Guid.pack(nodeid.Identifier)
374		else:
375		return struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
376		nodeid.Identifier.to_binary()
377		# FIXME: Missing NNamespaceURI and ServerIndex
378
379
380		def nodeid_from_binary(data):
381		nid = ua.NodeId()
382		encoding = ord(data.read(1))
383		nid.NodeIdType = ua.NodeIdType(encoding & 0b00111111)
384
385		if nid.NodeIdType == ua.NodeIdType.TwoByte:
386		nid.Identifier = ord(data.read(1))
387		elif nid.NodeIdType == ua.NodeIdType.FourByte:
388		nid.NamespaceIndex, nid.Identifier = struct.unpack("<BH", data.read(3))
389		elif nid.NodeIdType == ua.NodeIdType.Numeric:
390		nid.NamespaceIndex, nid.Identifier = struct.unpack("<HI", data.read(6))
391		elif nid.NodeIdType == ua.NodeIdType.String:
392		nid.NamespaceIndex = Primitives.UInt16.unpack(data)
393		nid.Identifier = Primitives.String.unpack(data)
394		elif nid.NodeIdType == ua.NodeIdType.ByteString:
395		nid.NamespaceIndex = Primitives.UInt16.unpack(data)
396		nid.Identifier = Primitives.Bytes.unpack(data)
397		elif nid.NodeIdType == ua.NodeIdType.Guid:
398		nid.NamespaceIndex = Primitives.UInt16.unpack(data)
399		nid.Identifier = Primitives.Guid.unpack(data)
400		else:
401		raise UaError("Unknown NodeId encoding: " + str(nid.NodeIdType))
402
403		if test_bit(encoding, 7):
404		nid.NamespaceUri = Primitives.String.unpack(data)
405		if test_bit(encoding, 6):
406		nid.ServerIndex = Primitives.UInt32.unpack(data)
407
408		return nid
409
410
411		def variant_to_binary(var):
412		b = []
413		encoding = var.VariantType.value & 0b111111
414		if var.is_array or isinstance(var.Value, (list, tuple)):
415		var.is_array = True
416		encoding = set_bit(encoding, 7)
417		if var.Dimensions is not None:
418		encoding = set_bit(encoding, 6)
419		b.append(Primitives.UInt8.pack(encoding))
420		b.append(pack_uatype_array(var.VariantType, ua.flatten(var.Value)))
421		if var.Dimensions is not None:
422		b.append(pack_uatype_array(ua.VariantType.Int32, var.Dimensions))
423		else:
424		b.append(Primitives.UInt8.pack(encoding))
425		b.append(pack_uatype(var.VariantType, var.Value))
426
427		return b"".join(b)
428
429
430		def variant_from_binary(data):
431		dimensions = None
432		array = False
433		encoding = ord(data.read(1))
434		int_type = encoding & 0b00111111
435		vtype = ua.datatype_to_varianttype(int_type)
436		if test_bit(encoding, 7):
437		value = unpack_uatype_array(vtype, data)
438		array = True
439		else:
440		value = unpack_uatype(vtype, data)
441		if test_bit(encoding, 6):
442		dimensions = unpack_uatype_array(ua.VariantType.Int32, data)
443		value = reshape(value, dimensions)
444		return ua.Variant(value, vtype, dimensions, is_array=array)
445
446
447		def reshape(flat, dims):
448		subdims = dims[1:]
449		subsize = 1
450		for i in subdims:
451		if i == 0:
452		i = 1
453		subsize *= i
454		while dims[0] * subsize > len(flat):
455		flat.append([])
456		if not subdims or subdims == [0]:
457		return flat
458		return [reshape(flat[i: i + subsize], subdims) for i in range(0, len(flat), subsize)]
459
460
461		def extensionobject_from_binary(data):
462		"""
463		Convert binary-coded ExtensionObject to a Python object.
464		Returns an object, or None if TypeId is zero
465		"""
466		typeid = nodeid_from_binary(data)
467		Encoding = ord(data.read(1))
468		body = None
469		if Encoding & (1 << 0):
470		length = Primitives.Int32.unpack(data)
471		if length < 1:
472		body = Buffer(b"")
473		else:
474		body = data.copy(length)
475		data.skip(length)
476		if typeid.Identifier == 0:
477		return None
478		elif typeid in ua.extension_object_classes:
479		klass = ua.extension_object_classes[typeid]
480		if body is None:
481		raise UaError("parsing ExtensionObject {0} without data".format(klass.__name__))
482		return from_binary(klass, body)
483		else:
484		e = ua.ExtensionObject()
485		e.TypeId = typeid
486		e.Encoding = Encoding
487		if body is not None:
488		e.Body = body.read(len(body))
489		return e
490
491
492		def extensionobject_to_binary(obj):
493		"""
494		Convert Python object to binary-coded ExtensionObject.
495		If obj is None, convert to empty ExtensionObject (TypeId = 0, no Body).
496		Returns a binary string
497		"""
498		if isinstance(obj, ua.ExtensionObject):
499		return obj.to_binary()
500		if obj is None:
501		TypeId = ua.NodeId()
502		Encoding = 0
503		Body = None
504		else:
505		TypeId = ua.extension_object_ids[obj.__class__.__name__]
506		Encoding = 0x01
507		Body = to_binary(obj)
508		packet = []
509		packet.append(to_binary(TypeId))
510		packet.append(Primitives.UInt8.pack(Encoding))
511		if Body:
512		packet.append(Primitives.Bytes.pack(Body))
513		return b''.join(packet)
514
515
516		def from_binary(uatype, data):
517		if isinstance(uatype, (str, unicode)) and uatype.startswith("ListOf"):
518		size = Primitives.Int32.unpack(data)
519		if size == -1:
520		return None
521		res = []
522		for _ in range(size):
523		res.append(from_binary(uatype[6:], data))
524		return res
525		elif isinstance(uatype, (str, unicode)) and hasattr(Primitives, uatype):
526		st = getattr(Primitives, uatype)
527		res = st.unpack(data)
528		return res
529		return st.unpack(data)
530		elif uatype == ua.NodeId or uatype in ("NodeId", "ExpandedNodeId"):
531		return nodeid_from_binary(data)
532		elif uatype == ua.Variant or uatype == "Variant":
533		return variant_from_binary(data)
534		elif uatype == ua.ExtensionObject or uatype == "ExtensionObject":
535		return extensionobject_from_binary(data)
536		else:
537		return struct_from_binary(uatype, data)
538
539
540		def struct_from_binary(objtype, data):
541		if isinstance(objtype, (unicode, str)):
542		objtype = getattr(ua, objtype)
543		if issubclass(objtype, Enum):
544		return objtype(Primitives.UInt32.unpack(data))
545		obj = objtype()
546		for name, uatype in obj.ua_types:
547		# if our member has a swtich and it is not set we skip it
548		if hasattr(obj, "ua_switches") and name in obj.ua_switches:
549		container_name, idx = obj.ua_switches[name]
550		val = getattr(obj, container_name)
551		if not test_bit(val, idx):
552		continue
553		val = from_binary(uatype, data)
554		setattr(obj, name, val)
555		return obj
556
557
558		def header_to_binary(hdr):
559		b = []
560		b.append(struct.pack("<3ss", hdr.MessageType, hdr.ChunkType))
561		size = hdr.body_size + 8
562		if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
563		size += 4
564		b.append(Primitives.UInt32.pack(size))
565		if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
566		b.append(Primitives.UInt32.pack(hdr.ChannelId))
567		return b"".join(b)
568
569
570		def header_from_binary(data):
571		hdr = ua.Header()
572		hdr.MessageType, hdr.ChunkType, hdr.packet_size = struct.unpack("<3scI", data.read(8))
573		hdr.body_size = hdr.packet_size - 8
574		if hdr.MessageType in (ua.MessageType.SecureOpen, ua.MessageType.SecureClose, ua.MessageType.SecureMessage):
575		hdr.body_size -= 4
576		hdr.ChannelId = Primitives.UInt32.unpack(data)
577		return hdr
578
579

FreeOpcUa / python-opcua

Pull Request — master (#490)

nodeid_from_binary() F

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like