Skip to content

Instantly share code, notes, and snippets.

@Sekenre
Last active May 28, 2021 17:17
Show Gist options
  • Save Sekenre/ccee9ae253d1b7612b4da57361eff271 to your computer and use it in GitHub Desktop.
Save Sekenre/ccee9ae253d1b7612b4da57361eff271 to your computer and use it in GitHub Desktop.
Embedded CBOR example
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment