code.kerkeslager.com Git - ton/blob - don/string.py

   1 import binascii
   2 import collections
   3 import functools
   4 import re
   5
   6 from don import tags, _shared
   7
   8 def _integer_size_to_string_serializer(integer_size):
   9     minimum = -(2 ** (integer_size - 1))
  10     maximum = 2 ** (integer_size - 1) - 1
  11
  12     def serializer(integer):
  13         assert minimum <= integer and integer <= maximum
  14         return '{}i{}'.format(integer, integer_size)
  15
  16     return serializer
  17
  18 def _serialize_binary(b):
  19     return '"{}"b'.format(binascii.hexlify(b).decode('ascii'))
  20
  21 def _utf_encoding_to_serializer(utf_encoding):
  22     def serializer(s):
  23         return '"{}"{}'.format(s, utf_encoding)
  24
  25     return serializer
  26
  27 def _string_serialize_list(l):
  28     return '[{}]'.format(', '.join(map(serialize, l)))
  29
  30 def _string_serialize_dictionary(d):
  31     def serialize_kvp(kvp):
  32         return serialize(kvp[0]) + ': ' + serialize(kvp[1])
  33     return '{ ' + ', '.join(map(serialize_kvp, d.items())) + ' }'
  34
  35 _STRING_SERIALIZERS = {
  36     tags.VOID: lambda o: 'null',
  37     tags.TRUE: lambda o: 'true',
  38     tags.FALSE: lambda o: 'false',
  39     tags.INT8: _integer_size_to_string_serializer(8),
  40     tags.INT16: _integer_size_to_string_serializer(16),
  41     tags.INT32: _integer_size_to_string_serializer(32),
  42     tags.INT64: _integer_size_to_string_serializer(64),
  43     tags.BINARY: _serialize_binary,
  44     tags.UTF8: _utf_encoding_to_serializer('utf8'),
  45     tags.UTF16: _utf_encoding_to_serializer('utf16'),
  46     tags.UTF32: _utf_encoding_to_serializer('utf32'),
  47     tags.LIST: _string_serialize_list,
  48     tags.DICTIONARY: _string_serialize_dictionary,
  49 }
  50
  51 def serialize(o):
  52     o = tags.autotag(o)
  53
  54     return _STRING_SERIALIZERS[o.tag](o.value)
  55
  56 def _consume_leading_whitespace(wrapped_parser):
  57     @functools.wraps(wrapped_parser)
  58     def parser(s):
  59         s = s.lstrip()
  60         return wrapped_parser(s)
  61
  62     return parser
  63
  64 def _make_constant_parser(constant, value):
  65     @_consume_leading_whitespace
  66     def constant_parser(s):
  67         if s.startswith(constant):
  68             result = _shared.ParseResult(
  69                 success = True,
  70                 value = value,
  71                 remaining = s[len(constant):],
  72             )
  73             return result
  74
  75         return _shared._FAILED_PARSE_RESULT
  76
  77     return constant_parser
  78
  79 def _make_integer_parser(width):
  80     matcher = re.compile(r'(-?\d+)i' + str(width))
  81
  82     @_consume_leading_whitespace
  83     def integer_parser(s):
  84         match = matcher.match(s)
  85
  86         if match:
  87             # TODO Validate that the integer is in range
  88             return _shared.ParseResult(
  89                 success = True,
  90                 value = int(match.group(1)),
  91                 remaining = s[match.end():],
  92             )
  93
  94         return _shared._FAILED_PARSE_RESULT
  95
  96     return integer_parser
  97
  98 _BINARY_MATCHER = re.compile(r'"([\da-f]*)"b')
  99
 100 @_consume_leading_whitespace
 101 def _binary_parser(s):
 102     match = _BINARY_MATCHER.match(s)
 103
 104     if match:
 105         return _shared.ParseResult(
 106             success = True,
 107             value = binascii.unhexlify(match.group(1)),
 108             remaining = s[match.end():],
 109         )
 110
 111     return _shared._FAILED_PARSE_RESULT
 112
 113 def _make_utf_parser(encoding):
 114     matcher = re.compile(r'"(.*?)"' + encoding)
 115
 116     @_consume_leading_whitespace
 117     def utf_parser(s):
 118         match = matcher.match(s)
 119
 120         if match:
 121             return _shared.ParseResult(
 122                 success = True,
 123                 value = match.group(1),
 124                 remaining = s[match.end():],
 125             )
 126
 127         return _shared._FAILED_PARSE_RESULT
 128
 129     return utf_parser
 130
 131 def _make_consume_constant_parser(constant):
 132     @_consume_leading_whitespace
 133     def consume_character_parser(s):
 134         if s.startswith(constant):
 135             return _shared.ParseResult(
 136                 success = True,
 137                 value = None,
 138                 remaining = s[len(constant):],
 139             )
 140         return _shared._FAILED_PARSE_RESULT
 141
 142     return consume_character_parser
 143
 144 _consume_comma_parser = _make_consume_constant_parser(',')
 145
 146 def _prefix_with_comma(parser):
 147     def wrapped(s):
 148         result = _consume_comma_parser(s)
 149         if result.success:
 150             s = result.remaining
 151         else:
 152             return _shared._FAILED_PARSE_RESULT
 153
 154         result = parser(s)
 155         if not result.success:
 156             raise Exception('Trailing comma before "{}"'.format(s))
 157
 158         return result
 159
 160     return wrapped
 161
 162 def _comma_separate_and_wrap(wrapped_parser, start_wrap, end_wrap, typecaster):
 163     parser_prefixed_with_comma = _prefix_with_comma(wrapped_parser)
 164     start_wrap_parser = _make_consume_constant_parser(start_wrap)
 165     end_wrap_parser = _make_consume_constant_parser(end_wrap)
 166
 167     def parser(s):
 168         result = start_wrap_parser(s)
 169         if result.success:
 170             s = result.remaining
 171         else:
 172             return _shared._FAILED_PARSE_RESULT
 173
 174         value = []
 175         first = True
 176
 177         parse_result = wrapped_parser(s)
 178
 179         while parse_result.success:
 180             value.append(parse_result.value)
 181             s = parse_result.remaining
 182             parse_result = parser_prefixed_with_comma(s)
 183
 184         result = end_wrap_parser(s)
 185         if result.success:
 186             s = result.remaining
 187         else:
 188             return _shared._FAILED_PARSE_RESULT
 189
 190         return _shared.ParseResult(
 191             success = True,
 192             value = typecaster(value),
 193             remaining = s,
 194         )
 195
 196     return parser
 197
 198 # This uses _PARSERS which has not been defined yet, but is defined here so it can be used in
 199 # the definition of _list_parser
 200 def _object_parser(source):
 201     for parser in _PARSERS:
 202         result = parser(source)
 203
 204         if result.success:
 205             return result
 206
 207     return _shared._FAILED_PARSE_RESULT
 208
 209 _list_parser = _comma_separate_and_wrap(_object_parser, '[', ']', list)
 210
 211 _consume_colon_parser = _make_consume_constant_parser(':')
 212
 213 def _kvp_parser(s):
 214     key_parse_result = _object_parser(s)
 215     if key_parse_result.success:
 216         s = key_parse_result.remaining
 217     else:
 218         return _shared._FAILED_PARSE_RESULT
 219
 220     result = _consume_colon_parser(s)
 221     if result.success:
 222         s = result.remaining
 223     else:
 224         return _shared._FAILED_PARSE_RESULT
 225
 226     value_parse_result = _object_parser(s)
 227     if value_parse_result.success:
 228         s = value_parse_result.remaining
 229     else:
 230         return _shared._FAILED_PARSE_RESULT
 231
 232     return _shared.ParseResult(
 233         success = True,
 234         value = (key_parse_result.value, value_parse_result.value),
 235         remaining = s,
 236     )
 237
 238 _dictionary_parser = _comma_separate_and_wrap(_kvp_parser, '{', '}', collections.OrderedDict)
 239
 240
 241 _PARSERS = [
 242     _make_constant_parser('null', None),
 243     _make_constant_parser('true', True),
 244     _make_constant_parser('false', False),
 245     _make_integer_parser(8),
 246     _make_integer_parser(16),
 247     _make_integer_parser(32),
 248     _make_integer_parser(64),
 249     _binary_parser,
 250     _make_utf_parser('utf8'),
 251     _make_utf_parser('utf16'),
 252     _make_utf_parser('utf32'),
 253     _list_parser,
 254     _dictionary_parser,
 255 ]
 256
 257 def _parse(parser, source):
 258     result = parser(source)
 259
 260     if result.success:
 261         if result.remaining.strip() == '':
 262             return result.value
 263
 264         raise Exception('Unparsed trailing characters: "{}"'.format(result.remaining))
 265
 266     raise Exception('Unable to parse: "{}"'.format(source))
 267
 268 def deserialize(s):
 269     return _parse(_object_parser, s)