lib/bup/vint.py

   1 """Binary encodings for bup."""
   2
   3 # Copyright (C) 2010 Rob Browning
   4 #
   5 # This code is covered under the terms of the GNU Library General
   6 # Public License as described in the bup LICENSE file.
   7
   8 from __future__ import absolute_import
   9 from io import BytesIO
  10
  11 # Variable length integers are encoded as vints -- see jakarta lucene.
  12
  13 def write_vuint(port, x):
  14     if x < 0:
  15         raise Exception("vuints must not be negative")
  16     elif x == 0:
  17         port.write('\0')
  18     else:
  19         while True:
  20             seven_bits = x & 0x7f
  21             x >>= 7
  22             if x:
  23                 port.write(chr(0x80 | seven_bits))
  24             else:
  25                 port.write(chr(seven_bits))
  26                 break
  27
  28
  29 def read_vuint(port):
  30     c = port.read(1)
  31     if not c:
  32         raise EOFError('encountered EOF while reading vuint')
  33     if ord(c) == 0:
  34         return 0
  35     result = 0
  36     offset = 0
  37     while True:
  38         b = ord(c)
  39         if b & 0x80:
  40             result |= ((b & 0x7f) << offset)
  41             offset += 7
  42             c = port.read(1)
  43             if not c:
  44                 raise EOFError('encountered EOF while reading vuint')
  45         else:
  46             result |= (b << offset)
  47             break
  48     return result
  49
  50
  51 def write_vint(port, x):
  52     # Sign is handled with the second bit of the first byte.  All else
  53     # matches vuint.
  54     if x == 0:
  55         port.write('\0')
  56     else:
  57         if x < 0:
  58             x = -x
  59             sign_and_six_bits = (x & 0x3f) | 0x40
  60         else:
  61             sign_and_six_bits = x & 0x3f
  62         x >>= 6
  63         if x:
  64             port.write(chr(0x80 | sign_and_six_bits))
  65             write_vuint(port, x)
  66         else:
  67             port.write(chr(sign_and_six_bits))
  68
  69
  70 def read_vint(port):
  71     c = port.read(1)
  72     if not c:
  73         raise EOFError('encountered EOF while reading vint')
  74     negative = False
  75     result = 0
  76     offset = 0
  77     # Handle first byte with sign bit specially.
  78     if c:
  79         b = ord(c)
  80         if b & 0x40:
  81             negative = True
  82         result |= (b & 0x3f)
  83         if b & 0x80:
  84             offset += 6
  85             c = port.read(1)
  86         elif negative:
  87             return -result
  88         else:
  89             return result
  90     while True:
  91         b = ord(c)
  92         if b & 0x80:
  93             result |= ((b & 0x7f) << offset)
  94             offset += 7
  95             c = port.read(1)
  96             if not c:
  97                 raise EOFError('encountered EOF while reading vint')
  98         else:
  99             result |= (b << offset)
 100             break
 101     if negative:
 102         return -result
 103     else:
 104         return result
 105
 106
 107 def write_bvec(port, x):
 108     write_vuint(port, len(x))
 109     port.write(x)
 110
 111
 112 def read_bvec(port):
 113     n = read_vuint(port)
 114     return port.read(n)
 115
 116
 117 def skip_bvec(port):
 118     port.read(read_vuint(port))
 119
 120 def send(port, types, *args):
 121     if len(types) != len(args):
 122         raise Exception('number of arguments does not match format string')
 123     for (type, value) in zip(types, args):
 124         if type == 'V':
 125             write_vuint(port, value)
 126         elif type == 'v':
 127             write_vint(port, value)
 128         elif type == 's':
 129             write_bvec(port, value)
 130         else:
 131             raise Exception('unknown xpack format string item "' + type + '"')
 132
 133 def recv(port, types):
 134     result = []
 135     for type in types:
 136         if type == 'V':
 137             result.append(read_vuint(port))
 138         elif type == 'v':
 139             result.append(read_vint(port))
 140         elif type == 's':
 141             result.append(read_bvec(port))
 142         else:
 143             raise Exception('unknown xunpack format string item "' + type + '"')
 144     return result
 145
 146 def pack(types, *args):
 147     port = BytesIO()
 148     send(port, types, *args)
 149     return port.getvalue()
 150
 151 def unpack(types, data):
 152     port = BytesIO(data)
 153     return recv(port, types)