Source code for caspia.node.binary

import struct


def decode_list(data):
    items = []
    while True:
        length, = struct.unpack('<H', data[0:2])
        item = data[2:2 + length]
        data = data[2 + length:]
        if length == 0:
            break
        items.append(item)
    return items, data


def encode_list(lst):
    data = bytes()
    for item in lst:
        data += struct.pack('<H', len(item))
        data += item
    data += struct.pack('<H', 0)  # zero termination
    return data


def decode_dictionary(data, keyfmt='<B'):
    keylen = struct.Struct(keyfmt).size
    dictionary = {}
    items, data = decode_list(data)
    for item in items:
        if len(item) < keylen:
            msg = f'Invalid binary dictionary. Entry must be at least {keylen} bytes long.'
            raise ValueError(msg)
        key, = struct.unpack(keyfmt, item[0:keylen])
        dictionary[key] = item[keylen:]
    return dictionary, data


def encode_dictionary(dictionary, keyfmt='<B'):
    items = []
    for key, value in dictionary.items():
        items.append(struct.pack(keyfmt, key) + value)
    return encode_list(items)