Class to simulate behavior of sys.stdout or sys.stderr. Stores contents in internal buffer and optionally echos to the inner stream it is simulating.
| 391 | |
| 392 | |
| 393 | class StdSim: |
| 394 | """Class to simulate behavior of sys.stdout or sys.stderr. |
| 395 | |
| 396 | Stores contents in internal buffer and optionally echos to the inner stream it is simulating. |
| 397 | """ |
| 398 | |
| 399 | def __init__( |
| 400 | self, |
| 401 | inner_stream: Union[TextIO, "StdSim"], |
| 402 | *, |
| 403 | echo: bool = False, |
| 404 | encoding: str = "utf-8", |
| 405 | errors: str = "replace", |
| 406 | ) -> None: |
| 407 | """StdSim Initializer. |
| 408 | |
| 409 | :param inner_stream: the wrapped stream. Should be a TextIO or StdSim instance. |
| 410 | :param echo: if True, then all input will be echoed to inner_stream |
| 411 | :param encoding: codec for encoding/decoding strings (defaults to utf-8) |
| 412 | :param errors: how to handle encoding/decoding errors (defaults to replace) |
| 413 | """ |
| 414 | self.inner_stream = inner_stream |
| 415 | self.echo = echo |
| 416 | self.encoding = encoding |
| 417 | self.errors = errors |
| 418 | self.pause_storage = False |
| 419 | self.buffer = ByteBuf(self) |
| 420 | |
| 421 | def write(self, s: str) -> None: |
| 422 | """Add str to internal bytes buffer and if echo is True, echo contents to inner stream. |
| 423 | |
| 424 | :param s: String to write to the stream |
| 425 | """ |
| 426 | if not isinstance(s, str): |
| 427 | raise TypeError(f"write() argument must be str, not {type(s)}") |
| 428 | |
| 429 | if not self.pause_storage: |
| 430 | self.buffer.byte_buf += s.encode(encoding=self.encoding, errors=self.errors) |
| 431 | if self.echo: |
| 432 | self.inner_stream.write(s) |
| 433 | |
| 434 | def getvalue(self) -> str: |
| 435 | """Get the internal contents as a str.""" |
| 436 | return self.buffer.byte_buf.decode(encoding=self.encoding, errors=self.errors) |
| 437 | |
| 438 | def getbytes(self) -> bytes: |
| 439 | """Get the internal contents as bytes.""" |
| 440 | return bytes(self.buffer.byte_buf) |
| 441 | |
| 442 | def read(self, size: int | None = -1) -> str: |
| 443 | """Read from the internal contents as a str and then clear them out. |
| 444 | |
| 445 | :param size: Number of bytes to read from the stream |
| 446 | """ |
| 447 | if size is None or size == -1: |
| 448 | result = self.getvalue() |
| 449 | self.clear() |
| 450 | else: |