BitUnpacker does the opposite of `BitPacker`. Args: bits (int): number of bits of the values to decode. fo (IO[bytes]): file-object to push the bytes to.
| 89 | |
| 90 | |
| 91 | class BitUnpacker: |
| 92 | """BitUnpacker does the opposite of `BitPacker`. |
| 93 | |
| 94 | Args: |
| 95 | bits (int): number of bits of the values to decode. |
| 96 | fo (IO[bytes]): file-object to push the bytes to. |
| 97 | """ |
| 98 | |
| 99 | def __init__(self, bits: int, fo: tp.IO[bytes]): |
| 100 | self.bits = bits |
| 101 | self.fo = fo |
| 102 | self._mask = (1 << bits) - 1 |
| 103 | self._current_value = 0 |
| 104 | self._current_bits = 0 |
| 105 | |
| 106 | def pull(self) -> tp.Optional[int]: |
| 107 | """ |
| 108 | Pull a single value from the stream, potentially reading some |
| 109 | extra bytes from the underlying file-object. |
| 110 | Returns `None` when reaching the end of the stream. |
| 111 | """ |
| 112 | while self._current_bits < self.bits: |
| 113 | buf = self.fo.read(1) |
| 114 | if not buf: |
| 115 | return None |
| 116 | character = buf[0] |
| 117 | self._current_value += character << self._current_bits |
| 118 | self._current_bits += 8 |
| 119 | |
| 120 | out = self._current_value & self._mask |
| 121 | self._current_value >>= self.bits |
| 122 | self._current_bits -= self.bits |
| 123 | return out |
| 124 | |
| 125 | |
| 126 | def test(): |