|
44 | 44 |
|
45 | 45 | @runtime_checkable |
46 | 46 | class BufferFactory(Protocol): # pragma: no cover |
47 | | - "BufferFactory Protocol definition" |
| 47 | + """BufferFactory Protocol definition. |
| 48 | + """ |
48 | 49 | def create(self, init_or_size: Union[int, bytes], size: int=None) -> Any: |
49 | 50 | """This function must create and return a mutable character buffer. |
50 | 51 |
|
@@ -91,7 +92,8 @@ def create(self, init_or_size: Union[int, bytes], size: int=None) -> bytearray: |
91 | 92 | buffer[:limit] = init_or_size[:limit] |
92 | 93 | return buffer |
93 | 94 | def clear(self, buffer: bytearray) -> None: |
94 | | - "Fills the buffer with zero." |
| 95 | + """Fills the buffer with zero. |
| 96 | + """ |
95 | 97 | buffer[:] = b'\x00' * len(buffer) |
96 | 98 |
|
97 | 99 | class CTypesBufferFactory: |
@@ -125,7 +127,8 @@ def create(self, init_or_size: Union[int, bytes], size: int=None) -> bytearray: |
125 | 127 | buffer[:limit] = init_or_size[:limit] |
126 | 128 | return buffer |
127 | 129 | def clear(self, buffer: bytearray, init: int=0) -> None: |
128 | | - "Fills the buffer with specified value (default)." |
| 130 | + """Fills the buffer with specified value (default). |
| 131 | + """ |
129 | 132 | memset(buffer, init, len(buffer)) |
130 | 133 |
|
131 | 134 | def safe_ord(byte: Union[bytes, int]) -> int: |
@@ -169,108 +172,133 @@ def _check_space(self, size: int): |
169 | 172 | if len(self.raw) < self.pos + size: |
170 | 173 | raise IOError("Insufficient buffer size") |
171 | 174 | def clear(self) -> None: |
172 | | - "Fills the buffer with zeros and resets the position in buffer to zero." |
| 175 | + """Fills the buffer with zeros and resets the position in buffer to zero. |
| 176 | + """ |
173 | 177 | self.factory.clear(self.raw) |
174 | 178 | self.pos = 0 |
175 | 179 | def resize(self, size: int) -> None: |
176 | | - "Resize buffer to specified length." |
| 180 | + """Resize buffer to specified length. |
| 181 | + """ |
177 | 182 | if self.max_size is not UNLIMITED and self.max_size < size: |
178 | 183 | raise IOError(f"Cannot resize buffer past max. size {self.max_size} bytes") |
179 | 184 | self.raw = self.factory.create(self.raw, size) |
180 | 185 | def is_eof(self) -> bool: |
181 | | - "Return True when positioned past the end of buffer or on `.eof_marker` (if defined)." |
| 186 | + """Return True when positioned past the end of buffer or on `.eof_marker` |
| 187 | + (if defined). |
| 188 | + """ |
182 | 189 | if self.pos >= len(self.raw): |
183 | 190 | return True |
184 | 191 | if self.eof_marker is not None and safe_ord(self.raw[self.pos]) == self.eof_marker: |
185 | 192 | return True |
186 | 193 | return False |
187 | 194 | def write(self, data: bytes) -> None: |
188 | | - "Write bytes." |
| 195 | + """Write bytes. |
| 196 | + """ |
189 | 197 | size = len(data) |
190 | 198 | self._ensure_space(size) |
191 | 199 | self.raw[self.pos:self.pos + size] = data |
192 | 200 | self.pos += size |
193 | 201 | def write_byte(self, byte: int) -> None: |
194 | | - "Write one byte." |
| 202 | + """Write one byte. |
| 203 | + """ |
195 | 204 | self._ensure_space(1) |
196 | 205 | self.raw[self.pos] = byte |
197 | 206 | self.pos += 1 |
198 | 207 | def write_number(self, value: int, size: int, *, signed: bool=False) -> None: |
199 | | - "Write number with specified size (in bytes)." |
| 208 | + """Write number with specified size (in bytes). |
| 209 | + """ |
200 | 210 | self.write(value.to_bytes(size, self.byteorder.value, signed=signed)) |
201 | 211 | def write_short(self, value: int) -> None: |
202 | | - "Write 2 byte number (c_ushort)." |
| 212 | + """Write 2 byte number (c_ushort). |
| 213 | + """ |
203 | 214 | self.write_number(value, 2) |
204 | 215 | def write_int(self, value: int) -> None: |
205 | | - "Write 4 byte number (c_uint)." |
| 216 | + """Write 4 byte number (c_uint). |
| 217 | + """ |
206 | 218 | self.write_number(value, 4) |
207 | 219 | def write_bigint(self, value: int) -> None: |
208 | | - "Write tagged 8 byte number (c_ulonglong)." |
| 220 | + """Write tagged 8 byte number (c_ulonglong). |
| 221 | + """ |
209 | 222 | self.write_number(value, 8) |
210 | 223 | def write_string(self, value: str, *, encoding='ascii') -> None: |
211 | | - "Write zero-terminated string." |
| 224 | + """Write zero-terminated string. |
| 225 | + """ |
212 | 226 | self.write(value.encode(encoding)) |
213 | 227 | self.write_byte(0) |
214 | 228 | def write_pascal_string(self, value: str, *, encoding='ascii') -> None: |
215 | | - "Write tagged Pascal string (2 byte length followed by data)." |
| 229 | + """Write tagged Pascal string (2 byte length followed by data). |
| 230 | + """ |
216 | 231 | value = value.encode(encoding) |
217 | 232 | size = len(value) |
218 | 233 | self.write_byte(size) |
219 | 234 | self.write(value) |
220 | 235 | def read(self, size: int=-1) -> bytes: |
221 | | - "Reads specified number of bytes, or all remaining data." |
| 236 | + """Reads specified number of bytes, or all remaining data. |
| 237 | + """ |
222 | 238 | if size < 0: |
223 | 239 | size = self.buffer_size - self.pos |
224 | 240 | self._check_space(size) |
225 | 241 | result = self.raw[self.pos: self.pos + size] |
226 | 242 | self.pos += size |
227 | 243 | return result |
228 | 244 | def read_number(self, size: int, *, signed=False) -> int: |
229 | | - "Read number with specified size in bytes." |
| 245 | + """Read number with specified size in bytes. |
| 246 | + """ |
230 | 247 | self._check_space(size) |
231 | 248 | result = (0).from_bytes(self.raw[self.pos: self.pos + size], self.byteorder.value, signed=signed) |
232 | 249 | self.pos += size |
233 | 250 | return result |
234 | 251 | def read_byte(self, *, signed=False) -> int: |
235 | | - "Read 1 byte number (c_ubyte)." |
| 252 | + """Read 1 byte number (c_ubyte). |
| 253 | + """ |
236 | 254 | return self.read_number(1, signed=signed) |
237 | 255 | def read_short(self, *, signed=False) -> int: |
238 | | - "Read 2 byte number (c_ushort)." |
| 256 | + """Read 2 byte number (c_ushort). |
| 257 | + """ |
239 | 258 | return self.read_number(2, signed=signed) |
240 | 259 | def read_int(self, *, signed=False) -> int: |
241 | | - "Read 4 byte number (c_uint)." |
| 260 | + """Read 4 byte number (c_uint). |
| 261 | + """ |
242 | 262 | return self.read_number(4, signed=signed) |
243 | 263 | def read_bigint(self, *, signed=False) -> int: |
244 | | - "Read 8 byte number (c_ulonglong)." |
| 264 | + """Read 8 byte number (c_ulonglong). |
| 265 | + """ |
245 | 266 | return self.read_number(8, signed=signed) |
246 | 267 | def read_sized_int(self, *, signed=False) -> int: |
247 | | - "Read number cluster (2 byte length followed by data)." |
| 268 | + """Read number cluster (2 byte length followed by data). |
| 269 | + """ |
248 | 270 | return self.read_number(self.read_short(), signed=signed) |
249 | 271 | def read_string(self, *, encoding='ascii') -> str: |
250 | | - "Read null-terminated string." |
| 272 | + """Read null-terminated string. |
| 273 | + """ |
251 | 274 | i = self.pos |
252 | 275 | while i < self.buffer_size and safe_ord(self.raw[i]) != 0: |
253 | 276 | i += 1 |
254 | 277 | result = self.read(i - self.pos).decode(encoding) |
255 | 278 | self.pos += 1 |
256 | 279 | return result |
257 | 280 | def read_pascal_string(self, *, encoding='ascii') -> str: |
258 | | - "Read Pascal string (1 byte length followed by string data)." |
| 281 | + """Read Pascal string (1 byte length followed by string data). |
| 282 | + """ |
259 | 283 | return self.read(self.read_byte()).decode(encoding) |
260 | 284 | def read_sized_string(self, *, encoding='ascii') -> str: |
261 | | - "Read string (2 byte length followed by data)." |
| 285 | + """Read string (2 byte length followed by data). |
| 286 | + """ |
262 | 287 | return self.read(self.read_short()).decode(encoding) |
263 | 288 | def read_bytes(self) -> bytes: |
264 | | - "Read content of binary cluster (2 bytes data length followed by data)." |
| 289 | + """Read content of binary cluster (2 bytes data length followed by data). |
| 290 | + """ |
265 | 291 | return self.read(self.read_short()) |
266 | 292 | # Properties |
267 | 293 | @property |
268 | 294 | def buffer_size(self) -> int: |
269 | | - "Current buffer size in bytes." |
| 295 | + """Current buffer size in bytes. |
| 296 | + """ |
270 | 297 | return len(self.raw) |
271 | 298 | @property |
272 | 299 | def last_data(self) -> int: |
273 | | - "Index of first non-zero byte when searched from the end of buffer." |
| 300 | + """Index of first non-zero byte when searched from the end of buffer. |
| 301 | + """ |
274 | 302 | i = len(self.raw) - 1 |
275 | 303 | while i >= 0: |
276 | 304 | if safe_ord(self.raw[i]) != 0: |
|
0 commit comments