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

Completed

Pull Request — master (#490)

by Olivier

created 2017-09-30 08:03 UTC

header_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
dl	0
loc	10
ccs	0
cts	0
cp	0
crap	12
rs	9.4285
c	1
b	0
f	0

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

import sys
import struct
import logging
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__')


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 _DateTime(object):
    @staticmethod
    def pack(dt):
        epch = ua.datetime_to_win_epoch(dt)
        return Primitives.Int64.pack(epch)

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


class _String(object):
    @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(object):
    @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(object):
    @staticmethod
    def pack(data):
        return b""

    @staticmethod
    def unpack(data):
        return None


class _Guid(object):
    @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(object):
    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]


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(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 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 unpack_uatype_array(vtype, data):
    length = Primitives.Int32.unpack(data)
    if length == -1:
        return None
    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(uatype[6:], val))
        else:
            if has_switch and val is None and name in obj.ua_switches:
                pass
            else:
                packet.append(to_binary(uatype, val))
    return b''.join(packet)


def to_binary(uatype, val):
    """
    Pack a python object to binary given a string defining its type
    """
    if uatype.startswith("ListOf"):
    #if isinstance(val, (list, tuple)):
        return list_to_binary(uatype[6:], val)
    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.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(uatype, val):
    if val is None:
        return Primitives.Int32.pack(-1)
    pack = []
    pack.append(Primitives.Int32.pack(len(val)))
    for el in val:
        pack.append(to_binary(uatype, el))
    return b''.join(pack)


def nodeid_to_binary(nodeid):
    data = None
    if nodeid.NodeIdType == ua.NodeIdType.TwoByte:
        data = struct.pack("<BB", nodeid.NodeIdType.value, nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.FourByte:
        data = struct.pack("<BBH", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.Numeric:
        data = struct.pack("<BHI", nodeid.NodeIdType.value, nodeid.NamespaceIndex, nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.String:
        data = struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
            Primitives.String.pack(nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.ByteString:
        data = struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
            Primitives.Bytes.pack(nodeid.Identifier)
    elif nodeid.NodeIdType == ua.NodeIdType.Guid:
        data = struct.pack("<BH", nodeid.NodeIdType.value, nodeid.NamespaceIndex) + \
               Primitives.Guid.pack(nodeid.Identifier)
    else:
        raise UaError("Unknown NodeIdType: {} for NodeId: {}".format(nodeid.NodeIdType, nodeid))
    # Add NamespaceUri and ServerIndex in case we have an ExpandedNodeId
    if nodeid.NamespaceUri:
        data = bytearray(data)
        data[0] = set_bit(data[0], 7)
        data.extend(Primitives.String.pack(nodeid.NamespaceUri))
    if nodeid.ServerIndex:
        if not isinstance(data, bytearray):
            data = bytearray(data)
        data[0] = set_bit(data[0], 6)
        data.extend(Primitives.UInt32.pack(nodeid.ServerIndex))
    return data


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 struct_to_binary(obj)
    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


def uatcp_to_binary(message_type, message):
    """
    Convert OPC UA TCP message (see OPC UA specs Part 6, 7.1) to binary.
    The only supported types are Hello, Acknowledge and ErrorMessage
    """
    header = ua.Header(message_type, ua.ChunkType.Single)
    binmsg = struct_to_binary(message)
    header.body_size = len(binmsg)
    return header_to_binary(header) + binmsg


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

FreeOpcUa / python-opcua

Pull Request — master (#490)

header_to_binary() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like