使用Python的交互式输入/输出

sauutmhj 于 2024-01-05 发布在 Python

关注(0)|答案(4)|浏览(170)

我有一个与用户交互的程序（像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()

型

python

来源：https://stackoverflow.com/questions/19880190/interactive-input-output-using-python

4条答案

按热度按时间

rt4zxlrg1#

目前的答案对我来说都不起作用。最后，我得到了这个工作：

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])

型
注意事项是（所有这些都在功能中管理）：

输入/输出总是带换行符的行。
每次写入后刷新子对象的标准输入。
使用child的标准输出中的readline()。

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

展开查看全部

赞(0）回复(0）举报 2024-01-05

qc6wkl3g2#

在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.塞巴斯蒂安所提供的，使用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

型

展开查看全部

赞(0）回复(0）举报 2024-01-05

jxct1oxe3#

这是一个交互式shell。你必须在单独的线程上运行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()