"""Support for the OPACK serialization format. Notes: * Absolute time (0x06) is not implemented (can unpack as integer, not pack) * Pack implementation does not implement UID referencing * Likely other cases missing """ from datetime import datetime # pylint: disable=too-many-branches,too-many-return-statements,too-many-statements import struct from typing import Dict, Tuple, Type from uuid import UUID _SIZED_INT_TYPES: Dict[int, Type] = {} def _sized_int(value: int, size: int) -> int: """Return an int subclass with a size attribute. This preserves the original encoded size, and allows re-encoding to the same size. """ if size in _SIZED_INT_TYPES: type_ = _SIZED_INT_TYPES[size] else: type_ = type(f"int_{size}b", (int,), {"size": size}) _SIZED_INT_TYPES[size] = type_ return type_(value) def pack(data: object) -> bytes: """Pack data structure with OPACK and return bytes.""" return _pack(data, []) def _pack(data, object_list): packed_bytes = None if data is None: packed_bytes = b"\x04" elif isinstance(data, bool): packed_bytes = bytes([1 if data else 2]) elif isinstance(data, UUID): packed_bytes = b"\x05" + data.bytes elif isinstance(data, datetime): raise NotImplementedError("absolute time") elif isinstance(data, int): size_hint = getattr(data, "size", None) # if created with _sized_int() if data < 0x28 and not size_hint: packed_bytes = bytes([data + 8]) elif (data <= 0xFF and not size_hint) or size_hint == 1: packed_bytes = bytes([0x30]) + data.to_bytes(1, byteorder="little") elif (data <= 0xFFFF and not size_hint) or size_hint == 2: packed_bytes = bytes([0x31]) + data.to_bytes(2, byteorder="little") elif (data <= 0xFFFFFFFF and not size_hint) or size_hint == 4: packed_bytes = bytes([0x32]) + data.to_bytes(4, byteorder="little") elif data <= 0xFFFFFFFFFFFFFFFF: packed_bytes = bytes([0x33]) + data.to_bytes(8, byteorder="little") elif isinstance(data, float): packed_bytes = b"\x36" + struct.pack("= 0xF: packed_bytes += b"\x03" elif isinstance(data, dict): packed_bytes = bytes([0xE0 + min(len(data), 0xF)]) + b"".join( _pack(k, object_list) + _pack(v, object_list) for k, v in data.items() ) if len(data) >= 0xF: packed_bytes += b"\x03" else: raise TypeError(str(type(data))) # Reuse if in object list, otherwise add it to list if packed_bytes in object_list: object_index = object_list.index(packed_bytes) if object_index < 0x21: packed_bytes = bytes([0xA0 + object_index]) elif object_index <= 0xFF: packed_bytes = bytes([0xC1]) + object_index.to_bytes(1, byteorder="little") elif object_index <= 0xFFFF: packed_bytes = bytes([0xC2]) + object_index.to_bytes(2, byteorder="little") elif object_index <= 0xFFFFFFFF: packed_bytes = bytes([0xC3]) + object_index.to_bytes(4, byteorder="little") elif object_index <= 0xFFFFFFFFFFFFFFFF: packed_bytes = bytes([0xC4]) + object_index.to_bytes(8, byteorder="little") elif len(packed_bytes) > 1: object_list.append(packed_bytes) return packed_bytes def unpack(data: bytes) -> Tuple[object, bytes]: """Unpack raw OPACK data into python objects.""" return _unpack(data, []) def _unpack(data, object_list): value = None remaining = None add_to_object_list = True if data[0] == 0x01: value, remaining = True, data[1:] add_to_object_list = False elif data[0] == 0x02: value, remaining = False, data[1:] add_to_object_list = False elif data[0] == 0x04: value, remaining = None, data[1:] add_to_object_list = False elif data[0] == 0x05: value = UUID(bytes=data[1:17]) remaining = data[17:] elif data[0] == 0x06: # TODO: Dummy implementation: only parse as integer value, remaining = int.from_bytes(data[1:9], byteorder="little"), data[9:] elif 0x08 <= data[0] <= 0x2F: value, remaining = data[0] - 8, data[1:] add_to_object_list = False elif data[0] == 0x35: value, remaining = struct.unpack("