Embedded CBOR example

embedded.py

from binascii import hexlify
from io import BytesIO

import cbor2


class Embedded:
    def __init__(self, payload=None):
        self.payload = payload

    def __to_cbor__(self, encoder):
        encoder.encode(cbor2.CBORTag(24, encoder.encode_to_bytes(self.payload)))

    @classmethod
    def __from_cbor__(cls, decoder, encoded):
        return cls(decoder.decode_from_bytes(encoded))


class EmbeddedSeq:
    def __init__(self, items=None):
        self.items = items or []

    def __to_cbor__(self, encoder):
        encoded_items = []
        for item in self.items:
            encoded_items.append(encoder.encode_to_bytes(item))
        encoder.encode(cbor2.CBORTag(63, b''.join(encoded_items)))

    @classmethod
    def __from_cbor__(cls, decoder, encoded):
        sequence = []
        with BytesIO(encoded) as fp:
            old_fp = decoder.fp
            decoder.fp = fp
            while True:
                try:
                    item = decoder.decode()
                    sequence.append(item)
                except cbor2.CBORDecodeEOF:
                    break
            decoder.fp = old_fp
        return cls(sequence)


def tag_hook(decoder, tag):
    tag_map = {24: Embedded, 63: EmbeddedSeq}
    if tag.tag in tag_map:
        return tag_map[tag.tag].__from_cbor__(decoder, tag.value)
    else:
        return tag


def default(encoder, value):
    if hasattr(value, '__to_cbor__'):
        value.__to_cbor__(encoder)


embed = Embedded({'status': 'awesome!'})
encoded = cbor2.dumps(embed, default=default)
embed2 = cbor2.loads(encoded, tag_hook=tag_hook)
print(hexlify(encoded))
print(embed.payload, embed2.payload)

embseq = EmbeddedSeq([{'status': n} for n in range(12)])
encodedseq = cbor2.dumps(embseq, default=default)
embseq2 = cbor2.loads(encodedseq, tag_hook=tag_hook)
print(hexlify(encodedseq))
print(embseq.items, embseq2.items)