byte_length = len(items)
return struct.pack('!BII', item_tag, byte_length, item_length) + items
+def _serialize_key(o):
+ o = tags.autotag(o)
+ assert o.tag in tags.STRING_TAGS
+ return struct.pack('!B', o.tag) + _BINARY_SERIALIZERS[o.tag](o.value)
+
def _binary_serialize_dict(d):
item_length = 0
serialized = b''
key_serializer = _BINARY_SERIALIZERS[tags.UTF8]
for key, value in d.items():
- assert isinstance(key, str)
item_length += 1
- serialized += key_serializer(key) + serialize(value)
+ serialized += _serialize_key(key) + serialize(value)
byte_length = len(serialized)
return struct.pack('!II', byte_length, item_length) + serialized
tags.INT8: _pack_format_string_to_binary_serializer('!b'),
tags.INT16: _pack_format_string_to_binary_serializer('!h'),
tags.INT32: _pack_format_string_to_binary_serializer('!i'),
- tags.FLOAT: _pack_format_string_to_binary_serializer('!f'),
- tags.DOUBLE: _pack_format_string_to_binary_serializer('!d'),
tags.BINARY: _encoder_to_binary_serializer(lambda b: b),
tags.UTF8: _encoder_to_binary_serializer(lambda s: s.encode('utf-8')),
tags.UTF16: _encoder_to_binary_serializer(lambda s: s.encode('utf-16')),
}
def serialize(o):
- o = tags._tag(o)
+ o = tags.autotag(o)
return struct.pack('!B', o.tag) + _BINARY_SERIALIZERS[o.tag](o.value)
_BYTE_SIZES_TO_UNPACK_FORMATS = {
tags.INT16: make_integer_parser(2),
tags.INT32: make_integer_parser(4),
tags.INT64: make_integer_parser(8),
- tags.DOUBLE: binary64_parser,
tags.BINARY: make_string_parser(lambda b : b),
tags.UTF8: make_string_parser(lambda b : b.decode('utf-8')),
tags.UTF16: make_string_parser(lambda b : b.decode('utf-16')),
return serializer
-def _serialize_float(f):
- return '{}f'.format(f)
-
-def _serialize_double(d):
- return '{}d'.format(d)
-
def _serialize_binary(b):
return '"{}"b'.format(binascii.hexlify(b).decode('ascii'))
tags.INT16: _integer_size_to_string_serializer(16),
tags.INT32: _integer_size_to_string_serializer(32),
tags.INT64: _integer_size_to_string_serializer(64),
- tags.FLOAT: _serialize_float,
- tags.DOUBLE: _serialize_double,
tags.BINARY: _serialize_binary,
tags.UTF8: _utf_encoding_to_serializer('utf8'),
tags.UTF16: _utf_encoding_to_serializer('utf16'),
return integer_parser
-_BINARY32_MATCHER = re.compile(r'(-?\d+\.\d+)f')
-_BINARY64_MATCHER = re.compile(r'(-?\d+\.\d+)d')
-
-@_consume_leading_whitespace
-def _binary32_parser(s):
- match = _BINARY32_MATCHER.match(s)
-
- if match:
- # TODO Validate that the float is in range
- return _shared.ParseResult(
- success = True,
- value = float(match.group(1)),
- remaining = s[match.end():],
- )
-
- return _shared._FAILED_PARSE_RESULT
-
-@_consume_leading_whitespace
-def _binary64_parser(s):
- match = _BINARY64_MATCHER.match(s)
-
- if match:
- # TODO Validate that the double is in range
- return _shared.ParseResult(
- success = True,
- value = float(match.group(1)),
- remaining = s[match.end():],
- )
-
- return _shared._FAILED_PARSE_RESULT
-
_BINARY_MATCHER = re.compile(r'"([\da-f]*)"b')
@_consume_leading_whitespace
_make_integer_parser(16),
_make_integer_parser(32),
_make_integer_parser(64),
- _binary32_parser,
- _binary64_parser,
_binary_parser,
_make_utf_parser('utf8'),
_make_utf_parser('utf16'),
INT16 = 0x11
INT32 = 0x12
INT64 = 0x13
-FLOAT = 0x20
-DOUBLE = 0x21
+# These are to be supported in the future
+# FLOAT = 0x20
+# DOUBLE = 0x21
BINARY = 0x30
UTF8 = 0x31
UTF16 = 0x32
LIST = 0x40
DICTIONARY = 0x41
+STRING_TAGS = set([UTF8, UTF16, UTF32])
+
DEFAULT_INTEGER_ENCODING = INT32
-DEFAULT_DECIMAL_ENCODING = DOUBLE
DEFAULT_STRING_ENCODING = UTF8
TaggedObject = collections.namedtuple('TaggedObject', ['tag', 'value'])
_TYPES_TO_TAGS = {
int: DEFAULT_INTEGER_ENCODING,
- float: DEFAULT_DECIMAL_ENCODING,
bytes: BINARY,
str: DEFAULT_STRING_ENCODING,
list: LIST,
raise TooWideError("Integer {} is too wide to be serialized")
- if isinstance(o, float):
- raise Exception('Unsupported type {}'.format(type(o)))
-
if isinstance(o, str):
# TODO Support SMALLEST for preferred string tag
return TaggedObject(tag = preferred_string_tag, value = o)
+ if isinstance(o, bytes):
+ return TaggedObject(tag = BINARY, value = o)
+
if isinstance(o, list):
return TaggedObject(
tag = LIST,
+# -*- coding: utf-8 -*-
import collections
import unittest
self.assertEqual(binary.serialize(1), b'\x12\x00\x00\x00\x01')
self.assertEqual(binary.serialize(2147483647), b'\x12\x7f\xff\xff\xff')
- def test_serializes_floats_into_binary64_with_network_byte_order(self):
- self.assertEqual(binary.serialize(1.0), b'\x21\x3f\xf0\x00\x00\x00\x00\x00\x00')
- self.assertEqual(binary.serialize(2.0), b'\x21\x40\x00\x00\x00\x00\x00\x00\x00')
- self.assertEqual(binary.serialize(-2.0), b'\x21\xc0\x00\x00\x00\x00\x00\x00\x00')
- self.assertEqual(binary.serialize(0.5), b'\x21\x3f\xe0\x00\x00\x00\x00\x00\x00')
- self.assertEqual(binary.serialize(2.0 ** -1074), b'\x21\x00\x00\x00\x00\x00\x00\x00\x01')
- self.assertEqual(binary.serialize(2.0 ** -1022), b'\x21\x00\x10\x00\x00\x00\x00\x00\x00')
- self.assertEqual(binary.serialize(0.0), b'\x21\x00\x00\x00\x00\x00\x00\x00\x00')
-
def test_serializes_binary(self):
self.assertEqual(binary.serialize(b'\xde\xad\xbe\xef'), b'\x30\x00\x00\x00\x04\xde\xad\xbe\xef')
self.assertEqual(binary.serialize([]), b'\x40\x00\x00\x00\x00\x00\x00\x00\x00\x00')
self.assertEqual(binary.serialize([1,2,3]), b'\x40\x12\x00\x00\x00\x0c\x00\x00\x00\x03\x00\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00\x03')
self.assertEqual(binary.serialize(['Hello, world', 'Goodnight, moon']), b'\x40\x31\x00\x00\x00#\x00\x00\x00\x02\x00\x00\x00\x0cHello, world\x00\x00\x00\x0fGoodnight, moon')
- self.assertEqual(binary.serialize([1.618, 2.718, 3.142]), b'\x40\x21\x00\x00\x00\x18\x00\x00\x00\x03?\xf9\xe3S\xf7\xce\xd9\x17@\x05\xbev\xc8\xb49X@\t"\xd0\xe5`A\x89')
def test_serializes_dictionary(self):
self.assertEqual(binary.serialize({}), b'\x41\x00\x00\x00\x00\x00\x00\x00\x00')
self.assertEqual(binary.serialize(collections.OrderedDict([
('foo',42),
- ('bar',3.14),
- ('baz','qux'),
- ])), b'A\x00\x00\x00+\x00\x00\x00\x03\x00\x00\x00\x03foo\x12\x00\x00\x00*\x00\x00\x00\x03bar!@\t\x1e\xb8Q\xeb\x85\x1f\x00\x00\x00\x03baz1\x00\x00\x00\x03qux')
+ ('bar','baz'),
+ ])), b'A\x00\x00\x00\x1d\x00\x00\x00\x021\x00\x00\x00\x03foo\x12\x00\x00\x00*1\x00\x00\x00\x03bar1\x00\x00\x00\x03baz')
class TestBinaryDeserialize(unittest.TestCase):
def test_deserializes_null(self):
self.assertEqual(binary.deserialize(b'\x13\x00\x00\x00\x00\x00\x00\x00\x01'), 1)
self.assertEqual(binary.deserialize(b'\x13\x7f\xff\xff\xff\xff\xff\xff\xff'), 9223372036854775807)
- def test_deserializes_binary64_as_float(self):
- self.assertEqual(binary.deserialize(b'\x21\x3f\xf0\x00\x00\x00\x00\x00\x00'), 1.0)
- self.assertEqual(binary.deserialize(b'\x21\x40\x00\x00\x00\x00\x00\x00\x00'), 2.0)
- self.assertEqual(binary.deserialize(b'\x21\xc0\x00\x00\x00\x00\x00\x00\x00'), -2.0)
- self.assertEqual(binary.deserialize(b'\x21\x3f\xe0\x00\x00\x00\x00\x00\x00'), 0.5)
- self.assertEqual(binary.deserialize(b'\x21\x00\x00\x00\x00\x00\x00\x00\x01'), 2.0 ** -1074)
- self.assertEqual(binary.deserialize(b'\x21\x00\x10\x00\x00\x00\x00\x00\x00'), 2.0 ** -1022)
- self.assertEqual(binary.deserialize(b'\x21\x00\x00\x00\x00\x00\x00\x00\x00'), 0.0)
-
def test_deserializes_binary(self):
self.assertEqual(binary.deserialize(b'\x30\x00\x00\x00\x04\xde\xad\xbe\xef'), b'\xde\xad\xbe\xef')
self.assertEqual(string.serialize(tags.TaggedObject(tags.INT64, -1)), '-1i64')
self.assertEqual(string.serialize(tags.TaggedObject(tags.INT64, 42)), '42i64')
- def test_serializes_float(self):
- self.assertEqual(string.serialize(tags.TaggedObject(tags.FLOAT, 1.0)), '1.0f')
-
- def test_serializes_double(self):
- self.assertEqual(string.serialize(tags.TaggedObject(tags.DOUBLE, 1.0)), '1.0d')
-
def test_serializes_binary(self):
self.assertEqual(string.serialize(tags.TaggedObject(tags.BINARY, b'\xde\xad\xbe\xef')), '"deadbeef"b')
self.assertEqual(10, string.deserialize(' \t\n10i64'))
self.assertEqual(-1, string.deserialize(' \t\n-1i64'))
- def test_deserializes_float(self):
- self.assertEqual(1.0, string.deserialize('1.0f'))
-
- def test_deserializes_float_with_leading_whitspace(self):
- self.assertEqual(1.0, string.deserialize(' \t\n1.0f'))
-
- def test_deserializes_double(self):
- self.assertEqual(1.0, string.deserialize('1.0d'))
-
- def test_deserializes_double_with_leading_whitespace(self):
- self.assertEqual(1.0, string.deserialize(' \t\n1.0d'))
-
def test_deserializes_binary(self):
self.assertEqual(
b'\xde\xad\xbe\xef',
tags.TaggedObject(tag = tags.UTF32, value = 'Hello, world'),
)
+ def test_tags_bytes(self):
+ self.assertEqual(
+ tags.autotag(b'\xde\xad\xbe\xef'),
+ tags.TaggedObject(tag = tags.BINARY, value = b'\xde\xad\xbe\xef'),
+ )
+
def test_tags_list(self):
self.assertEqual(
tags.autotag([1,2,3]),