Simple bit packer to handle ints with a non standard width, e.g. 10 bits. Note that for some bandwidth (1.5, 3), the codebook representation will not cover an integer number of bytes. Args: bits (int): number of bits per value that will be pushed. fo (IO[bytes]): file-ob
| 52 | |
| 53 | |
| 54 | class BitPacker: |
| 55 | """Simple bit packer to handle ints with a non standard width, e.g. 10 bits. |
| 56 | Note that for some bandwidth (1.5, 3), the codebook representation |
| 57 | will not cover an integer number of bytes. |
| 58 | |
| 59 | Args: |
| 60 | bits (int): number of bits per value that will be pushed. |
| 61 | fo (IO[bytes]): file-object to push the bytes to. |
| 62 | """ |
| 63 | |
| 64 | def __init__(self, bits: int, fo: tp.IO[bytes]): |
| 65 | self._current_value = 0 |
| 66 | self._current_bits = 0 |
| 67 | self.bits = bits |
| 68 | self.fo = fo |
| 69 | |
| 70 | def push(self, value: int): |
| 71 | """Push a new value to the stream. This will immediately |
| 72 | write as many uint8 as possible to the underlying file-object.""" |
| 73 | self._current_value += (value << self._current_bits) |
| 74 | self._current_bits += self.bits |
| 75 | while self._current_bits >= 8: |
| 76 | lower_8bits = self._current_value & 0xff |
| 77 | self._current_bits -= 8 |
| 78 | self._current_value >>= 8 |
| 79 | self.fo.write(bytes([lower_8bits])) |
| 80 | |
| 81 | def flush(self): |
| 82 | """Flushes the remaining partial uint8, call this at the end |
| 83 | of the stream to encode.""" |
| 84 | if self._current_bits: |
| 85 | self.fo.write(bytes([self._current_value])) |
| 86 | self._current_value = 0 |
| 87 | self._current_bits = 0 |
| 88 | self.fo.flush() |
| 89 | |
| 90 | |
| 91 | class BitUnpacker: |