qualia_core.utils.process.AsyncTeeProtocol module

class qualia_core.utils.process.AsyncTeeProtocol.AsyncTeeProtocol(done_future: Future[tuple[int, dict[int, bytearray]]], files: dict[int, BinaryIO] | None = None)[source]

Bases: SubprocessProtocol

transport: SubprocessTransport | None = None

connection_made(transport: BaseTransport) → None[source]

pipe_data_received(fd: int, data: bytes) → None[source]

process_exited() → None[source]