46

我有一个与用户交互的程序（就像一个 shell），我想使用 Python 子进程模块交互地运行它。这意味着，我希望能够写入标准输入并立即从标准输出中获取输出。我尝试了这里提供的许多解决方案，但似乎没有一个能满足我的需求。

我编写的代码基于在 Python 中运行交互式命令。

import Queue
import threading
import subprocess
def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line)
    out.close()

def getOutput(outQueue):
    outStr = ''
    try:
        while True: # Adds output from the queue until it is empty
            outStr += outQueue.get_nowait()

    except Queue.Empty:
        return outStr

p = subprocess.Popen("./a.out", stdin=subprocess.PIPE, stout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize = 1)
#p = subprocess.Popen("./a.out", stdin=subprocess.PIPE, stout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, universal_newlines=True)

outQueue = Queue()
errQueue = Queue()

outThread = Thread(target=enqueue_output, args=(p.stdout, outQueue))
errThread = Thread(target=enqueue_output, args=(p.stderr, errQueue))

outThread.daemon = True
errThread.daemon = True

outThread.start()
errThread.start()

p.stdin.write("1\n")
p.stdin.flush()
errors = getOutput(errQueue)
output = getOutput(outQueue)

p.stdin.write("5\n")
p.stdin.flush()
erros = getOutput(errQueue)
output = getOutput(outQueue)

问题是队列仍然是空的，就好像没有输出一样。只有当我将程序需要执行和终止的所有输入写入标准输入时，我才会得到输出（这不是我想要的）。例如，如果我执行以下操作：

p.stdin.write("1\n5\n")
errors = getOutput(errQueue)
output = getOutput(outQueue)

有没有办法做我想做的事？

该脚本将在 Linux 机器上运行。我更改了脚本并删除了 universal_newlines=True + 将 bufsize 设置为 1 并在写入后立即刷新标准输入。我仍然没有得到任何输出。

第二次尝试：

我尝试了这个解决方案，它对我有用：

from subprocess import Popen, PIPE

fw = open("tmpout", "wb")
fr = open("tmpout", "r")
p = Popen("./a.out", stdin = PIPE, stdout = fw, stderr = fw, bufsize = 1)
p.stdin.write("1\n")
out = fr.read()
p.stdin.write("5\n")
out = fr.read()
fw.close()
fr.close()
4

4 回答 4

24

当前的答案都不适合我。最后，我得到了这个工作：

import subprocess


def start(executable_file):
    return subprocess.Popen(
        executable_file,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)


def read(process):
    return process.stdout.readline().decode("utf-8").strip()


def write(process, message):
    process.stdin.write(f"{message.strip()}\n".encode("utf-8"))
    process.stdin.flush()


def terminate(process):
    process.stdin.close()
    process.terminate()
    process.wait(timeout=0.2)


process = start("./dummy.py")
write(process, "hello dummy")
print(read(process))
terminate(process)

用这个dummy.py脚本测试：

#!/usr/bin/env python3.6

import random
import time

while True:
    message = input()
    time.sleep(random.uniform(0.1, 1.0)) # simulates process time
    print(message[::-1])

警告是（全部在函数中管理）：

  • 输入/输出总是与换行符对齐。
  • 每次写入后刷新孩子的标准输入。
  • readline()从孩子的标准输出中使用。

在我看来，这是一个非常简单的解决方案（不是我的，我在这里找到它：https ://eli.thegreenplace.net/2017/interacting-with-a-long-running-child-process-in-python/ ）。我使用的是 Python 3.6。

于 2018-09-27T20:42:55.730 回答
20

Linux 上此问题的两种解决方案：

第一个是使用文件将输出写入并同时读取：

from subprocess import Popen, PIPE

fw = open("tmpout", "wb")
fr = open("tmpout", "r")
p = Popen("./a.out", stdin = PIPE, stdout = fw, stderr = fw, bufsize = 1)
p.stdin.write("1\n")
out = fr.read()
p.stdin.write("5\n")
out = fr.read()
fw.close()
fr.close()

其次，正如 JF Sebastian 所提供的，是使用 fnctl 模块使 p.stdout 和 p.stderr 管道非阻塞：

import os
import fcntl
from subprocess import Popen, PIPE  
def setNonBlocking(fd):
    """
    Set the file description of the given file descriptor to non-blocking.
    """
    flags = fcntl.fcntl(fd, fcntl.F_GETFL)
    flags = flags | os.O_NONBLOCK
    fcntl.fcntl(fd, fcntl.F_SETFL, flags)

p = Popen("./a.out", stdin = PIPE, stdout = PIPE, stderr = PIPE, bufsize = 1)
setNonBlocking(p.stdout)
setNonBlocking(p.stderr)

p.stdin.write("1\n")
while True:
    try:
        out1 = p.stdout.read()
    except IOError:
        continue
    else:
        break
out1 = p.stdout.read()
p.stdin.write("5\n")
while True:
    try:
        out2 = p.stdout.read()
    except IOError:
        continue
    else:
        break
于 2013-11-10T17:56:37.523 回答
13

这是一个交互式外壳。您必须在单独的线程上运行 read()，否则会阻塞 write()

import sys
import os
import subprocess
from subprocess import Popen, PIPE
import threading


class LocalShell(object):
    def __init__(self):
        pass

    def run(self):
        env = os.environ.copy()
        p = Popen('/bin/bash', stdin=PIPE, stdout=PIPE, stderr=subprocess.STDOUT, shell=True, env=env)
        sys.stdout.write("Started Local Terminal...\r\n\r\n")

        def writeall(p):
            while True:
                # print("read data: ")
                data = p.stdout.read(1).decode("utf-8")
                if not data:
                    break
                sys.stdout.write(data)
                sys.stdout.flush()

        writer = threading.Thread(target=writeall, args=(p,))
        writer.start()

        try:
            while True:
                d = sys.stdin.read(1)
                if not d:
                    break
                self._write(p, d.encode())

        except EOFError:
            pass

    def _write(self, process, message):
        process.stdin.write(message)
        process.stdin.flush()


shell = LocalShell()
shell.run()
于 2018-11-15T04:53:48.173 回答
-1

据我所知，最简单的方法是在双方都创建专用线程：两个用于父进程的 stdout/stderr，一个用于子进程的 stdin。部分此类示例请在此处找到。

于 2020-12-08T22:58:36.137 回答