Example of communication between two processes with subprocess

subprocess_read_write_example.py

import subprocess

from queue import Queue
from threading import Thread
from tempfile import NamedTemporaryFile


class ClosableQueue(Queue):
    SENTINEL = object()

    def close(self):
        self.put(self.SENTINEL)

    def __iter__(self):
        while True:
            item = self.get()
            try:
                if item is self.SENTINEL:
                    return  # Cause the thread to exit
                yield item
            finally:
                self.task_done()


class StdinWriter(Thread):
    def __init__(self, proc, in_queue):
        super().__init__()
        self.in_queue = in_queue
        self.proc = proc
        self.daemon = True

    def run(self):
        for item in self.in_queue:
            self.proc.stdin.write(f"{item}\n".encode("utf8"))
            self.proc.stdin.flush()


def communicate_with_other_process(fp):
    command = ["python", fp.name]
    proc = subprocess.Popen(
        command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, bufsize=-1
    )
    write_queue = ClosableQueue()
    stdin_writer = StdinWriter(proc, write_queue)
    stdin_writer.start()

    while (line := proc.stdout.readline().decode("utf8").rstrip()) != "":
        print(line, end=" ")
        if ":" in line:
            user_input = input()
            write_queue.put(user_input)

    write_queue.join()
    proc.stdin.close()


to_control_source = """
import sys

for expected_input in ("foo", "bar", "baz"):
     question = f"enter {expected_input}:"
     print(question, flush=True)
     while (user_input := sys.stdin.readline().rstrip()) != expected_input:
        print(f"wrong input '{user_input}', try again.. \\n{question}", flush=True)
"""

with NamedTemporaryFile() as fp:
    fp.write(to_control_source.encode("utf-8"))
    fp.flush()
    communicate_with_other_process(fp)