使用python进行交互式输入/输出
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/19880190/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
Interactive input/output using python
提问by Talor Abramovich
I have a program that interacts with the user (acts like a shell), and I want to run it using python subprocess module interactively. That means, I want the possibility to write to stdin and immediately get the output from stdout. I tried many solutions offered here, but none of them seems to work for my needs.
我有一个与用户交互的程序（就像一个 shell），我想以交互方式使用 python 子进程模块运行它。这意味着，我希望能够写入标准输入并立即从标准输出获得输出。我尝试了这里提供的许多解决方案，但似乎没有一个适合我的需求。
The code I've written based on Running an interactive command from within 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)
The problem is that the queue remains empty, as if there is no output. Only if I write to stdin all the input that the program needs to execute and terminate, then I get the output (which is not what I want). For example if I do something like:
问题是队列一直是空的，好像没有输出一样。只有当我将程序需要执行和终止的所有输入写入 stdin 时，我才会得到输出（这不是我想要的）。例如，如果我执行以下操作：
p.stdin.write("1\n5\n")
errors = getOutput(errQueue)
output = getOutput(outQueue)
Is there any way to do what I want to do?
有什么办法可以做我想做的事吗？
EDIT:The script will run on a Linux machine. I changed my script and deleted the universal_newlines=True + set the bufsize to 1 and flushed stdin immediately after wrtie. Still I get no output.
编辑：该脚本将在 Linux 机器上运行。我更改了我的脚本并删除了universal_newlines=True + 将bufsize 设置为1 并在wrtie 之后立即刷新stdin。我仍然没有输出。
Second try:I tried this solution and it works for me:
第二次尝试：我尝试了这个解决方案，它对我有用：
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()
采纳答案by Talor Abramovich
Two solutions for this issue on Linux:
Linux 上此问题的两种解决方案：
First one is to use a file to write the output to, and read from it simultaneously:
第一个是使用文件将输出写入并同时读取：
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()
Second, as J.F. Sebastian offered, is to make p.stdout and p.stderr pipes non-blocking using fnctl module:
其次，正如 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
回答by gorcajo
None of the current answers worked for me. At the end, I've got this working:
目前的答案都不适合我。最后，我有这个工作：
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)
Tested with this dummy.pyscript:
用这个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])
The caveats are: input/output always lines with newline, flushing child's stdin after every write, and use readline()from child's stdout (all of that managed in the functions).
警告是：输入/输出总是与换行符，每次写入后刷新孩子的标准输入，并使用readline()孩子的标准输出（所有这些都在函数中管理）。
It's a pretty simple solution, in my opinion (not mine, I found it here: https://eli.thegreenplace.net/2017/interacting-with-a-long-running-child-process-in-python/). I was using Python 3.6.
在我看来，这是一个非常简单的解决方案（不是我的，我在这里找到了它：https: //eli.thegreenplace.net/2017/interacting-with-a-long-running-child-process-in-python/）。我使用的是 Python 3.6。
回答by Shawn
Here is an interactive shell. You have to run read() on a separate thread, otherwise it will block the write()
这是一个交互式外壳。你必须在一个单独的线程上运行 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()
