nodeid_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-08-28 18:54 UTC

nodeid_to_binary() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	56

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	7
dl	0
loc	19
ccs	0
cts	0
cp	0
crap	56
rs	7.3333
c	1
b	0
f	0

"""
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:
        from opcua.ua import uatypes
        if hasattr(uatypes, vtype.name):
            klass = getattr(uatypes, vtype.name)
            return klass.from_binary(data)
        else:
            raise UaError("can not unpack unknown vtype {0!s}".format(vtype))


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


"""
def to_binary(obj):
    if isintance(obj, NodeId):
        return nodeid_to_binary(obj)
    elif hasattr(val, "ua_types"):
        return to_binary(val)
    else:
        raise ValueError("Do not know how to convert {} to binary: {}".format(type(obj), obj))
"""


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
            print("Test for", name, member)
            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:
        print(name, uatype)
        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:
                print("append", val, uatype)
                packet.append(to_binary(val, uatype))
    return b''.join(packet)


def to_binary(val, uatype=None):
    print("TB", val, uatype)
    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 hasattr(val, "ua_types"):
        return struct_to_binary(val)
    else:
        raise ValueError("Cannot 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 klass.from_binary(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):
    print("FROM BIN", uatype, data)
    if isinstance(uatype, (str, unicode)) and uatype.startswith("ListOf"):
        size = Primitives.Int32.unpack(data)
        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 == "NodeId":
        return nodeid_from_binary(data)
    elif uatype == ua.Variant or uatype == "Variant":
        return variant_from_binary(data)
    else:
        return struct_from_binary(uatype, data)


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

FreeOpcUa / python-opcua

Pull Request — master (#490)

nodeid_to_binary() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like