Skip to content

stream_capture.py

ofrak_io.stream_capture

StreamCapture is useful utility for capturing standard output of C processes called from python.

Usage: 

with StreamCapture(sys.stdout) as stream_capture:
    sys.stdout.write("hello\n")
assert stream_capture.get_captured_stream() == "hello\n"

StreamCapture

Capture a stream from a filelike object.

start(self)

Start capturing the stream.

Source code in ofrak_io/stream_capture.py 
def start(self) -> None:
    """
    Start capturing the stream.
    """
    self.stream_file_descriptor_copy = os.dup(self.stream_file_descriptor)
    os.dup2(self.pipe_in, self.stream_file_descriptor)

stop(self)

Stop capturing the stream and read what was captured.

Source code in ofrak_io/stream_capture.py 
def stop(self) -> None:
    """
    Stop capturing the stream and read what was captured.
    """
    if self.stream_file_descriptor_copy is None:
        # Capture was not started
        return

    # Flush and read the captured stream
    self.stream.write(self.escape_char.decode())
    self.stream.flush()
    self._read_stream()

    # Reset the file
    os.close(self.pipe_in)
    os.close(self.pipe_out)
    os.dup2(self.stream_file_descriptor_copy, self.stream_file_descriptor)
    os.close(self.stream_file_descriptor_copy)

    # Write the captured stream to the file. This way, anything that may have been captured
    # by accident still gets where it needs to go
    self.stream.write(self.stream_captured)

get_captured_stream(self)

Get the captured stream.

Source code in ofrak_io/stream_capture.py 
def get_captured_stream(self) -> str:
    """
    Get the captured stream.
    """
    return self.stream_captured

_read_stream(self) private

Read the stream data (one byte at a time) and save the text in capturedtext.

Source code in ofrak_io/stream_capture.py 
def _read_stream(self) -> None:
    """
    Read the stream data (one byte at a time)
    and save the text in `capturedtext`.
    """
    while True:
        char = os.read(self.pipe_out, 1)
        if not char or self.escape_char in char:
            break
        self.stream_captured += char.decode()