A python equivalent for the :code:`Archive` serialization protocol.
| 21 | |
| 22 | |
| 23 | class OutArchive(object): |
| 24 | """A python equivalent for the :code:`Archive` serialization protocol.""" |
| 25 | |
| 26 | def __init__(self, buffer): |
| 27 | self._buffer = memoryview(buffer) |
| 28 | self._buffer = self._buffer.cast("b", shape=(self._buffer.nbytes,)) |
| 29 | self._head = 0 |
| 30 | # assume all size is int64_t |
| 31 | self._size_of_int64 = struct.calcsize("q") |
| 32 | |
| 33 | @property |
| 34 | def buffer(self): |
| 35 | return self._buffer |
| 36 | |
| 37 | @property |
| 38 | def size(self): |
| 39 | return self.__len__() |
| 40 | |
| 41 | def __len__(self): |
| 42 | return len(self._buffer) |
| 43 | |
| 44 | @property |
| 45 | def isempty(self): |
| 46 | return len(self._buffer) == 0 |
| 47 | |
| 48 | def get_block(self, size): |
| 49 | """Peek a block of given size.""" |
| 50 | block = self._buffer[self._head : self._head + size] |
| 51 | self._head += size |
| 52 | return block |
| 53 | |
| 54 | def get_size(self): |
| 55 | size = struct.unpack("q", self.get_block(self._size_of_int64))[0] |
| 56 | return size |
| 57 | |
| 58 | def get_sized_block(self): |
| 59 | """Peek a block with the size as the prefix bytes.""" |
| 60 | size = self.get_size() |
| 61 | block = self.get_block(size) |
| 62 | return block |
| 63 | |
| 64 | def get_bytes(self): |
| 65 | """Peek bytes""" |
| 66 | size = self.get_size() |
| 67 | byt = self.get_block(size).tobytes() |
| 68 | return byt |
| 69 | |
| 70 | def get_string(self): |
| 71 | """Peek a string. |
| 72 | Get string's length first (stored as a size_t), then get number of bytes |
| 73 | equivalents to the length. |
| 74 | Default using utf-8 to decode, if there is any bytes cannot be converted, |
| 75 | 'errors="backslashreplace' inserts a escape sequence.So these special |
| 76 | bytes can be reserved. |
| 77 | Ref:https://docs.python.org/3/howto/unicode.html?highlight=bytes%20decode |
| 78 | """ |
| 79 | size = self.get_size() |
| 80 | string = ( |
no outgoing calls
no test coverage detected