code.kerkeslager.com Git - sandbox/blob - serial/serial/text.py

   1 import binascii
   2 import re
   3
   4 from . import tags
   5
   6 def _make_literal_serializer(expected_value, literal):
   7     def serializer(to):
   8         assert to.instance is expected_value
   9         return literal
  10
  11     return serializer
  12
  13 def _make_integer_serializer(lower_bound, upper_bound, suffix):
  14     def _serializer(to):
  15         assert lower_bound <= to.instance and to.instance < upper_bound
  16         return '{}{}'.format(to.instance, suffix)
  17
  18     return _serializer
  19
  20 def _make_unsigned_integer_serializer(bit_length):
  21     return _make_integer_serializer(0, 2 << (bit_length - 1), 'u{}'.format(bit_length))
  22
  23 def _make_signed_integer_serializer(bit_length):
  24     upper_bound = 2 << (bit_length - 2)
  25     lower_bound = -upper_bound
  26     return _make_integer_serializer(lower_bound, upper_bound, 'i{}'.format(bit_length))
  27
  28 def _serialize_binary(to):
  29     return 'bin"{}"'.format(binascii.hexlify(to.instance).decode('ascii'))
  30
  31 _ESCAPES = {
  32     '\\': '\\\\',
  33     '"': '\\"',
  34 }
  35
  36 def _escape_character(ch):
  37     return _ESCAPES.get(ch, ch)
  38
  39 def _escape(s):
  40     return ''.join(_escape_character(ch) for ch in s)
  41
  42 def _make_string_serializer(prefix):
  43     def serializer(to):
  44         assert isinstance(to.instance, str)
  45         return '{}"{}"'.format(prefix, _escape(to.instance))
  46
  47     return serializer
  48
  49 def _indent(s, depth = 2):
  50     return '\n'.join(' ' * depth + line for line in s.split('\n'))
  51
  52 def _serialize_list(to):
  53     assert isinstance(to.instance, list)
  54
  55     return '[\n' + _indent(',\n'.join(serialize(i) for i in to.instance)) + '\n]'
  56
  57 def _serialize_key_value_pair(kvp):
  58     assert isinstance(kvp, tuple)
  59     key, value = kvp
  60
  61     assert key.tag in [tags.UTF8, tags.UTF16, tags.UTF32]
  62
  63     return '{}: {}'.format(serialize(key), serialize(value))
  64
  65 def _serialize_object(to):
  66     assert isinstance(to.instance, list)
  67
  68     return '{\n' + _indent(',\n'.join(_serialize_key_value_pair(kvp) for kvp in to.instance)) + '\n}'
  69     raise Exception('Not implemented')
  70
  71 _SERIALIZERS = {
  72     tags.NULL: _make_literal_serializer(None, 'null'),
  73     tags.TRUE: _make_literal_serializer(True, 'true'),
  74     tags.FALSE: _make_literal_serializer(False, 'false'),
  75     tags.UINT8: _make_unsigned_integer_serializer(8),
  76     tags.UINT16: _make_unsigned_integer_serializer(16),
  77     tags.UINT32: _make_unsigned_integer_serializer(32),
  78     tags.UINT64: _make_unsigned_integer_serializer(64),
  79     tags.INT8: _make_signed_integer_serializer(8),
  80     tags.INT16: _make_signed_integer_serializer(16),
  81     tags.INT32: _make_signed_integer_serializer(32),
  82     tags.INT64: _make_signed_integer_serializer(64),
  83     tags.BINARY: _serialize_binary,
  84     tags.UTF8: _make_string_serializer('utf8'),
  85     tags.UTF16: _make_string_serializer('utf16'),
  86     tags.UTF32: _make_string_serializer('utf32'),
  87     tags.LIST: _serialize_list,
  88     tags.OBJECT: _serialize_object,
  89 }
  90
  91 def serialize(to):
  92     return _SERIALIZERS[to.tag](to)
  93
  94 def _make_literal_deserializer(tag, instance, literal):
  95     def _deserializer(s):
  96         if s.startswith(literal):
  97             return True, tags.TaggedObject(tag = tag, instance = instance), s[len(literal):]
  98
  99         return False, None, None
 100
 101     return _deserializer
 102
 103 def _make_regex_deserializer(tag, decoder, regex):
 104     matcher = re.compile(regex).match
 105
 106     def _deserializer(s):
 107         match = matcher(s)
 108
 109         if match is None:
 110             return False, None, None
 111
 112         return True, tags.TaggedObject(tag = tag, instance = decoder(match)), s[match.end():]
 113
 114     return _deserializer
 115
 116 def _make_unsigned_int_deserializer(tag, bit_length):
 117     bound = 2 << (bit_length - 1)
 118
 119     def _decoder(match):
 120         result = int(match.group(1))
 121         assert result < bound
 122         return result
 123
 124     return _make_regex_deserializer(tag, _decoder, r'(\d+)' + 'u{}'.format(bit_length))
 125
 126 def _make_signed_int_deserializer(tag, bit_length):
 127     upper_bound = 2 << (bit_length - 2)
 128     lower_bound = -upper_bound
 129
 130     def _decoder(match):
 131         result = int(match.group(1))
 132         assert lower_bound <= result and result < upper_bound
 133         return result
 134
 135     return _make_regex_deserializer(tag, _decoder, r'(-?\d+)' + 'i{}'.format(bit_length))
 136
 137 _BINARY_MATCHER = re.compile(r'bin"([\da-f]*)"').match
 138
 139 def _deserialize_binary(s):
 140     match = _BINARY_MATCHER(s)
 141
 142     if match is None:
 143         return False, None, None
 144
 145     result = tags.TaggedObject(
 146         tag = tags.BINARY,
 147         instance = binascii.unhexlify(match.group(1)),
 148     )
 149
 150     return True, result, s[match.end():]
 151
 152 def _make_string_matcher(prefix):
 153     return re.compile(prefix + r'"(([^"]|\\.)*)"').match
 154
 155 _UNESCAPE_CHARACTERS = {
 156     '\\': '\\',
 157     '"': '"',
 158 }
 159
 160 def _unescape_character(ch):
 161     return _UNESCAPE_CHARACTERS[ch]
 162
 163 def _unescape(s):
 164     characters = []
 165     escaping = False
 166
 167     for ch in s:
 168         if escaping:
 169             characters.append(_unescape_character(ch))
 170             escaping = False
 171
 172         elif ch == '\\':
 173             escaping = True
 174
 175         else:
 176             characters.append(ch)
 177
 178     return ''.join(characters)
 179
 180 def _make_string_deserializer(tag, prefix):
 181     matcher = _make_string_matcher(prefix)
 182
 183     def deserializer(s):
 184         match = matcher(s)
 185
 186         if match is None:
 187             return False, None, None
 188
 189         result = tags.TaggedObject(
 190             tag = tag,
 191             instance = _unescape(match.group(1)),
 192         )
 193
 194         return True, result, s[match.end():]
 195
 196     return deserializer
 197
 198 def _deserialize_list(s):
 199     if not s.startswith('['):
 200         return False, None, None
 201
 202     instance = []
 203
 204     s = s[1:].lstrip()
 205
 206     succeeded, result, s = _deserialize_one(s)
 207
 208     # TODO Handle empty lists
 209
 210     if succeeded:
 211         instance.append(result)
 212
 213     s = s.lstrip()
 214
 215     while not s.startswith(']'):
 216         assert s.startswith(',')
 217         s = s[1:].lstrip()
 218
 219         succeeded, result, s = _deserialize_one(s)
 220
 221         # TODO Handle trailing commas
 222         assert succeeded
 223         instance.append(result)
 224         s = s.lstrip()
 225
 226     assert s.startswith(']')
 227     return True, tags.TaggedObject(tag = tags.LIST, instance = instance), s[1:]
 228
 229 def _deserialize_object(s):
 230     if not s.startswith('{'):
 231         return False, None, None
 232
 233     instance = []
 234
 235     s = s[1:].lstrip()
 236
 237     # TODO Handle empty objects
 238
 239     succeeded, key, s = _deserialize_one(s)
 240
 241     assert succeeded
 242     assert key.tag in [tags.UTF8, tags.UTF16, tags.UTF32]
 243
 244     s = s.lstrip()
 245     assert s.startswith(':')
 246
 247     s = s[1:].lstrip()
 248
 249     succeeded, value, s = _deserialize_one(s)
 250
 251     assert succeeded
 252     instance.append((key, value))
 253
 254     s = s.lstrip()
 255
 256     while s.startswith(','):
 257         succeeded, key, s = _deserialize_one(s)
 258
 259         assert succeeded
 260         assert key.tag in [tags.UTF8, tags.UTF16, tags.UTF32]
 261
 262         s = s.lstrip()
 263         assert s.startswith(':')
 264
 265         s = s[1:].lstrip()
 266
 267         succeeded, value, s = _deserialize_one(s)
 268
 269         assert succeeded
 270         instance.append((key, value))
 271
 272         s = s.lstrip()
 273
 274     assert s.startswith('}')
 275     return True, tags.TaggedObject(tag = tags.LIST, instance = instance), s[1:]
 276
 277 _DESERIALIZERS = [
 278     _make_literal_deserializer(tags.NULL, None, 'null'),
 279     _make_literal_deserializer(tags.TRUE, True, 'true'),
 280     _make_literal_deserializer(tags.FALSE, False, 'false'),
 281     _make_unsigned_int_deserializer(tags.UINT8, 8),
 282     _make_unsigned_int_deserializer(tags.UINT16, 16),
 283     _make_unsigned_int_deserializer(tags.UINT32, 32),
 284     _make_unsigned_int_deserializer(tags.UINT64, 64),
 285     _make_signed_int_deserializer(tags.INT8, 8),
 286     _make_signed_int_deserializer(tags.INT16, 16),
 287     _make_signed_int_deserializer(tags.INT32, 32),
 288     _make_signed_int_deserializer(tags.INT64, 64),
 289     _deserialize_binary,
 290     _make_string_deserializer(tags.UTF8, 'utf8'),
 291     _make_string_deserializer(tags.UTF16, 'utf16'),
 292     _make_string_deserializer(tags.UTF32, 'utf32'),
 293     _deserialize_list,
 294     _deserialize_object,
 295 ]
 296
 297 def _deserialize_one(s):
 298     s = s.lstrip()
 299
 300     for deserializer in _DESERIALIZERS:
 301         succeeded, result, remaining = deserializer(s)
 302
 303         if succeeded:
 304             return succeeded, result, remaining
 305
 306     return False, None, None
 307
 308 def deserialize(s):
 309     succeeded, result, remaining = _deserialize_one(s)
 310
 311     assert succeeded
 312     assert remaining == ''
 313
 314     return result