Is there a way in python to "emulate" running a python file?

Question

I want to be able to programmatically run a file, arbitrarily give it inputs (stdin) at any time, and periodically poll for any stdout. I also want to be able to kill the process whenever I want.

Here's what I have tried:

import subprocess
from threading import Thread


class Runner:
    def __init__(self):
        self.process = subprocess.Popen(
            ["python", "x.py"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

    def run(self):
        self.process.wait()

    def poll(self):
        print("got stdout:", self.process.stdout.readline().decode(), end="")

    def give_input(self, text=""):
        return self.process.stdin.write(bytes(text))

    def kill(self):
        self.process.kill()


r = Runner()
t = Thread(target=r.run)
t.start()
r.poll() # should be "hi"
r.poll() # should be "your name:"
r.give_input("hi\n")
r.kill()
t.join()

Here's the code in x.py

print("hi")
input("Your name: ")

So in my demo, I start a thread that will call run which will run the process. Then, I poll for stdout twice. It should print

got stdout: hi
got stdout: Your name:

Afterwards, I give an stdin to the process -- hi\n.

At this point, the program should terminate, and I make sure of it by doing r.kill().

However, the program does not work as expected.

Instead, the program freezes at the second r.poll(), and I'm not sure why. I suspect that since there is no newline, the program will keep reading, but I'm not sure how to prevent it from doing this.

Any ideas?

Answer 1

Here's what I finally came up with, (with code from a non blocking read on subprocess.PIPE:

import subprocess
from queue import Queue, Empty
from threading import Thread
from typing import IO
import io


class Runner:
    def __init__(self, stdin_input: str):
        self.process = subprocess.Popen(
            "py x.py",
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=2,
            close_fds=False,
        )

        self.process.stdin.write(stdin_input.encode() + b"\n")

    def reader(self, out: IO[str], queue: Queue[bytes]):
        def enqueue_output(out: IO[str], queue: Queue[bytes]):
            stream = io.open(out.fileno(), "rb", closefd=False)
            while True:
                n = stream.read1()
                if len(n) > 0:
                    queue.put(n)
                else:
                    return

        t = Thread(target=enqueue_output, args=(out, queue))
        # end thread when program ends
        t.daemon = True
        t.start()

    def run(self, timeout=1):
        stdout_queue: Queue[bytes] = Queue()
        stderr_queue: Queue[bytes] = Queue()

        self.reader(self.process.stdout, stdout_queue)
        self.reader(self.process.stderr, stderr_queue)

        try:
            self.process.communicate(timeout=timeout)
        except:
            print("ERROR: TIMED OUT")
        finally:
            print("=== STDOUT ===")
            try:
                while True:
                    print(stdout_queue.get_nowait().decode(), end="")
            except Empty:
                print("=== STDOUT ===\n\n")

            print("=== STDERR ===")
            try:
                while True:
                    print(stderr_queue.get_nowait().decode(), end="")
            except Empty:
                print("=== STDERR ===")


r = Runner("5\n6")
r.run()

Unfortunately, I was not able to figure out how to interactively give stdin to the process, nor was I able to periodically poll for stdout, but I was able to figure out a decent solution for my use case. Instead of getting output every so often, I instead just ran it for a few seconds before terminating the program.