Sometimes you need to debug how a tool interacts with another command. This script runs the command, passing through stderr and stdout while also logging it to seperate files.

ProcessPassthroughLogger.py

import sys
import subprocess
import threading
import codecs
import selectors

from typing import BinaryIO, TextIO

UTF8Reader = codecs.getreader("utf-8")

class ProcessPassthroughLogger:
    class PassthroughConfig:
        def __init__(self, name, decoder, buffer, logfile, output) -> None:
            self.name = name
            self.decoder:codecs.IncrementalDecoder = decoder
            self.buffer = buffer
            self.logfile = logfile
            self.output = output

    def __init__(self, stdout:BinaryIO, stderr:BinaryIO, stdout_file:TextIO, stderr_file:TextIO, timeout_seconds:float=0.01 ) -> None:
        assert stdout
        assert stderr
        self.stdout = stdout
        self.stderr = stderr
        self.stdout_file = stdout_file
        self.stderr_file = stderr_file
        self.timeout_seconds = timeout_seconds

        self.stdout_decoder = codecs.getincrementaldecoder("utf-8")("replace")
        self.stderr_decoder = codecs.getincrementaldecoder("utf-8")("replace")
        self.stdout_buffer = []
        self.stderr_buffer = []

        self.selector  = selectors.DefaultSelector()
        stdout_key = self.selector.register(self.stdout, selectors.EVENT_READ, self.PassthroughConfig("stdout", self.stdout_decoder, self.stdout_buffer, self.stdout_file, sys.stdout ))
        stderr_key = self.selector.register(self.stderr, selectors.EVENT_READ, self.PassthroughConfig("stderr", self.stderr_decoder, self.stderr_buffer, self.stderr_file, sys.stderr ))

        self.selector_data = {
            stdout_key: stdout_key.data,
            stderr_key: stderr_key.data
        }

        self.capture_thread =  threading.Thread(target=self._capture_output)
        self.capture_thread.start()

    def finish(self):
        self.capture_thread.join()


    def _capture_output(self):
        while len(self.selector.get_map().keys()) > 0:
            try:
                for (selector_key, _) in self.selector.select(self.timeout_seconds):
                    config:self.PassthroughConfig = selector_key.data
                    current_byte = selector_key.fileobj.read(1)

                    if current_byte is None:
                        continue # the selector should prevent this

                    is_eof = False
                    if len(current_byte) == 0: # EOF
                        is_eof = True
                        self.selector.unregister(selector_key.fileobj)
                    else:
                        current_char = config.decoder.decode(current_byte)
                        if current_char:
                            config.buffer.append(current_char)

                    if current_char == "\n" or is_eof:
                        line = "".join(config.buffer)
                        config.buffer.clear()
                        config.output.write(line)
                        config.logfile.write(line)
                        config.output.flush()

            except TimeoutError:
                pass


def main():
    cmd = ["python", "executable.py"] # put name of executable to intercept here, maybe wrap in stdbuf call
    cmd.extend(sys.argv[1:])  # append original arguments


    with open("stdout.log", "w") as stdout_file, open("stderr.log", "w") as stderr_file:
        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env={"PYTHONUNBUFFERED": "True"})
        passthrough_logger = ProcessPassthroughLogger(process.stdout, process.stderr, stdout_file, stderr_file)
        process.wait()
        passthrough_logger.finish()


if __name__  == "__main__":
    main()