使用 Python 进行交互式输入/输出
Posted
技术标签:
【中文标题】使用 Python 进行交互式输入/输出【英文标题】：Interactive input/output using Python 【发布时间】：2013-11-21 17:02:57 【问题描述】：我有一个与用户交互的程序（就像一个 shell），我想使用 Python 子进程模块交互地运行它。 这意味着，我希望能够写入标准输入并立即从标准输出中获取输出。我尝试了这里提供的许多解决方案，但似乎没有一个能满足我的需求。
我写的代码是基于Running an interactive command from within 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()
【问题讨论】：
related. 好的代码没有显示在赏金中，所以我为它做了一个要点gist.github.com/brando90/99b10cdc73dc6b604ca661712c1c7b0d我只是想在后台用一个真正的python进程来测试它。 你看了吗：docs.python.org/3/library/asyncio.html ? 相关：***.com/q/375427/240515 如果有人想在现代 Python 中做到这一点，我已经在这里发布了明确的答案：***.com/a/56051270/240515 【参考方案1】：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()
第二，正如 J.F. 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
【讨论】：
我不太明白你的意思。一开始我想做的是只读取部分输出（一行） 我不明白你的例子。这行是做什么的： p = Popen([sys.executable, "-u", '-c' 'for line in iter(input, ""): print("a"*int(line)*10**6 )'], stdin=PIPE, stdout=PIPE, bufsize=1) 是什么意思？ 你说得对，我已经在使用第二个解决方案时对其进行了编辑。问题是，当我一开始尝试该解决方案时，我已经在 python 解释器上尝试过（我没有编写脚本，只是手动测试了它）所以它起作用了（由于人为响应时间慢）。当我尝试编写脚本时，我遇到了你提到的问题，所以我添加了 while 循环，直到输入准备好。我实际上得到了整个输出或什么都没有（IOError 异常），如果我上传 a.out 源代码（它是一个 C 程序）会有帮助吗？ 不幸的是，我想与之通信的脚本使用stty -echo，这导致它在尝试使用第一种方法时抱怨'standard input': Inappropriate ioctl for device。 （第二种方法似乎永远无法正确读取输出。假设脚本正在输出）
第二个解决方案永远挂起，当调用一个简单的 C 程序时，使用 scanf("%s", buf) 请求 2 个输入【参考方案2】：
当前的答案都不适合我。最后，我得到了这个工作：
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。
【讨论】：
只想评论说我一直在尝试执行这个看似非常简单的任务大约 45 分钟，而 stdin.flush 原来是所有问题的解决方案。我在任何地方读过的任何其他答案中都没有看到这一行。【参考方案3】：这是一个交互式外壳。你必须在单独的线程上运行 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()
【讨论】：
谢谢！我通过更改 2 行使其适用于 Windows：p = Popen('/bin/bash', stdin=PIPE... 到 p = Popen(['cmd.exe'], stdin=PIPE... 和行 data = p.stdout.read(1).decode("utf-8") 到 data = p.stdout.read(1).decode("cp437")。只是如果有人想在 Windows 上运行它。
你的答案是最好的！谢谢！
除了从 bash 修改为 cmd，您还可以在 Popen 的末尾添加参数“text=True”，然后也可以删除编码/解码。很好的答案！【参考方案4】：
据我所知，最简单的方法是在两侧创建专用线程：两个用于父进程的 stdout/stderr，一个用于子进程的 stdin。部分示例请查找here。
【讨论】：
以上是关于使用 Python 进行交互式输入/输出的主要内容，如果未能解决你的问题，请参考以下文章