-
Notifications
You must be signed in to change notification settings - Fork 8
/
varint.py
65 lines (52 loc) · 1.42 KB
/
varint.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
"""Varint encoder/decoder
varints are a common encoding for variable length integer data, used in
libraries such as sqlite, protobuf, v8, and more.
Here's a quick and dirty module to help avoid reimplementing the same thing
over and over again.
"""
# byte-oriented StringIO was moved to io.BytesIO in py3k
try:
from io import BytesIO
except ImportError:
from StringIO import StringIO as BytesIO
import sys
if sys.version > '3':
def _byte(b):
return bytes((b, ))
else:
def _byte(b):
return chr(b)
def encode(number):
"""Pack `number` into varint bytes"""
buf = b''
while True:
towrite = number & 0x7f
number >>= 7
if number:
buf += _byte(towrite | 0x80)
else:
buf += _byte(towrite)
break
return buf
def decode_stream(stream):
"""Read a varint from `stream`"""
shift = 0
result = 0
while True:
i = _read_one(stream)
result |= (i & 0x7f) << shift
shift += 7
if not (i & 0x80):
break
return result
def decode_bytes(buf):
"""Read a varint from from `buf` bytes"""
return decode_stream(BytesIO(buf))
def _read_one(stream):
"""Read a byte from the file (as an integer)
raises EOFError if the stream ends while reading bytes.
"""
c = stream.read(1)
if c == b'':
raise EOFError("Unexpected EOF while reading bytes")
return ord(c)