从 python 子进程获取输出并向其发出命令

Posted 2023-02-23

【中文标题】从 python 子进程获取输出并向其发出命令

【英文标题】：Getting output from and giving commands to a python subprocess

【发布时间】：2012-01-07 06:34:31

【问题描述】：

我正在尝试从子进程获取输出，然后根据前面的输出向该进程发出命令。当程序需要进一步输入时，我需要多次执行此操作。 （如果可能，我还需要能够隐藏子进程命令提示符）。

我认为这将是一项简单的任务，因为我已经看到在 2003 年的帖子中讨论过这个问题，现在已经快 2012 年了，这似乎是一个非常普遍的需求，而且看起来它应该是任何问题的基本组成部分编程语言。显然我错了，不知何故，将近 9 年后，仍然没有以稳定、非破坏性、独立于平台的方式完成这项任务的标准方法！

我不太了解文件 i/o 和缓冲或线程，因此我更喜欢尽可能简单的解决方案。如果有一个与 python 3.x 兼容的模块，我会非常愿意下载它。我意识到有多个问题的问题基本相同，但我还没有找到解决我想要完成的简单任务的答案。

这是我目前基于各种来源的代码；但是我完全不知道下一步该做什么。我所有的尝试都以失败告终，有些人设法使用了我 100% 的 CPU（基本上什么都不做）并且不会退出。

import subprocess
from subprocess import Popen, PIPE
p = Popen(r'C:\postgis_testing\shellcomm.bat',stdin=PIPE,stdout=PIPE,stderr=subprocess.STDOUT shell=True)
stdout,stdin = p.communicate(b'command string')

如果我的问题不清楚，我将发布示例批处理文件的文本，我演示了需要向子进程发送多个命令的情况（如果您键入不正确的命令字符串，程序将循环）。

@echo off
:looper
set INPUT=
set /P INPUT=Type the correct command string:
if "%INPUT%" == "command string" (echo you are correct) else (goto looper)

如果有人可以帮助我，我将不胜感激，我相信许多其他人也会如此！

这里编辑的是使用 eryksun 的代码的功能代码（下一篇）：

import subprocess
import threading
import time
import sys

try: 
    import queue
except ImportError:
    import Queue as queue

def read_stdout(stdout, q, p):
    it = iter(lambda: stdout.read(1), b'')
    for c in it:
        q.put(c)
        if stdout.closed:
            break

_encoding = getattr(sys.stdout, 'encoding', 'latin-1')
def get_stdout(q, encoding=_encoding):
    out = []
    while 1:
        try:
            out.append(q.get(timeout=0.2))
        except queue.Empty:
            break
    return b''.join(out).rstrip().decode(encoding)

def printout(q):
    outdata = get_stdout(q)
    if outdata:
        print('Output: %s' % outdata)

if __name__ == '__main__':
    #setup
    p = subprocess.Popen(['shellcomm.bat'], stdin=subprocess.PIPE, 
                     stdout=subprocess.PIPE, stderr=subprocess.PIPE, 
                     bufsize=0, shell=True) # I put shell=True to hide prompt
    q = queue.Queue()
    encoding = getattr(sys.stdin, 'encoding', 'utf-8')

    #for reading stdout
    t = threading.Thread(target=read_stdout, args=(p.stdout, q, p))
    t.daemon = True
    t.start()

    #command loop
    while p.poll() is None:
        printout(q)
        cmd = input('Input: ')
        cmd = (cmd + '\n').encode(encoding)
        p.stdin.write(cmd)
        time.sleep(0.1) # I added this to give some time to check for closure (otherwise it doesn't work)

    #tear down
    for n in range(4):
        rc = p.poll()
        if rc is not None:
            break
        time.sleep(0.25)
    else:
        p.terminate()
        rc = p.poll()
        if rc is None:
            rc = 1

    printout(q)
    print('Return Code: %d' % rc)

但是，当从命令提示符运行脚本时，会发生以下情况：

C:\Users\username>python C:\postgis_testing\shellcomm7.py
Input: sth
Traceback (most recent call last):
File "C:\postgis_testing\shellcomm7.py", line 51, in <module>
    p.stdin.write(cmd)
IOError: [Errno 22] Invalid argument

从命令提示符运行时，程序似乎关闭了。有什么想法吗？

【参考方案1】：

此演示使用专用线程从标准输出读取。如果您四处搜索，我相信您可以找到一个用面向对象接口编写的更完整的实现。至少我可以说这对您在 Python 2.7.2 和 3.2.2 中提供的批处理文件有用。

shellcomm.bat：

@echo off
echo Command Loop Test
echo.
:looper
set INPUT=
set /P INPUT=Type the correct command string:
if "%INPUT%" == "command string" (echo you are correct) else (goto looper)

这是我根据“错误”、“仍然错误”和“命令字符串”的命令序列得到的输出：

Output:
Command Loop Test

Type the correct command string:
Input: wrong
Output:
Type the correct command string:
Input: still wrong
Output:
Type the correct command string:
Input: command string
Output:
you are correct

Return Code: 0

对于读取管道输出，readline 有时可能会起作用，但批处理文件中的set /P INPUT 自然不会写行结尾。因此，我使用lambda: stdout.read(1) 一次读取一个字节（效率不高，但它有效）。读取函数将数据放入队列中。主线程在写入命令后从队列中获取输出。在此处对get 调用使用超时使其等待一小段时间以确保程序正在等待输入。相反，您可以检查输出以了解程序何时需要输入。

话虽如此，您不能指望这样的设置可以通用，因为您尝试与之交互的控制台程序可能会在管道传输时缓冲其输出。在 Unix 系统中，您可以将一些实用命令插入到管道中，以将缓冲修改为非缓冲、行缓冲或给定大小——例如 stdbuf。还有一些方法可以让程序认为它连接到了一个 pty（参见pexpect）。但是，如果您无权访问程序的源代码以使用 setvbuf 显式设置缓冲，我不知道如何在 Windows 上解决此问题。

import subprocess
import threading
import time
import sys

if sys.version_info.major >= 3:
    import queue
else:
    import Queue as queue
    input = raw_input

def read_stdout(stdout, q):
    it = iter(lambda: stdout.read(1), b'')
    for c in it:
        q.put(c)
        if stdout.closed:
            break

_encoding = getattr(sys.stdout, 'encoding', 'latin-1')
def get_stdout(q, encoding=_encoding):
    out = []
    while 1:
        try:
            out.append(q.get(timeout=0.2))
        except queue.Empty:
            break
    return b''.join(out).rstrip().decode(encoding)

def printout(q):
    outdata = get_stdout(q)
    if outdata:
        print('Output:\n%s' % outdata)

if __name__ == '__main__':

    ARGS = ["shellcomm.bat"]   ### Modify this

    #setup
    p = subprocess.Popen(ARGS, bufsize=0, stdin=subprocess.PIPE, 
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    q = queue.Queue()
    encoding = getattr(sys.stdin, 'encoding', 'utf-8')

    #for reading stdout
    t = threading.Thread(target=read_stdout, args=(p.stdout, q))
    t.daemon = True
    t.start()

    #command loop
    while 1:
        printout(q)
        if p.poll() is not None or p.stdin.closed:
            break
        cmd = input('Input: ') 
        cmd = (cmd + '\n').encode(encoding)
        p.stdin.write(cmd)

    #tear down
    for n in range(4):
        rc = p.poll()
        if rc is not None:
            break
        time.sleep(0.25)
    else:
        p.terminate()
        rc = p.poll()
        if rc is None:
            rc = 1

    printout(q)
    print('\nReturn Code: %d' % rc)

【讨论】：

非常感谢它在 IDLE 中完美运行，但是当我运行修改后的代码时（请参阅我在原始帖子中的编辑）它给了我以下错误 C:\Users\username>python C:\postgis_testing\shellcomm7.py 输入：sth Traceback（最近一次调用最后）：文件“C:\postgis_testing\shellcomm7.py”，第 51 行，在 p.stdin.write(cmd) IOError: [Errno 22] Invalid argument 从命令提示符运行时，程序似乎关闭了。有什么想法吗？

@THX1138：我修改了循环以轮询进程并在要求输入之前检查标准输入。如果进程退出，它会中断。我还从read_stdout args 中删除了进程对象；它是在我的初稿中错误地留下的。

当我从命令提示符运行新代码时，我收到错误：C:\Users\username>python C:\postgis_testing\shellcomm8.py Traceback（最近一次调用最后一次）：文件“C :\postgis_testing\shellcomm8.py"，第 38 行，在 bufsize=0) 文件 "C:\Python32\lib\subprocess.py"，第 741 行，在 init restore_signals, start_new_session) File "C:\Python32\lib\subprocess.py", line 960, in _execute_child startupinfo) WindowsError: [Error 2] The system cannot find the file specified

@THX1138：我在当前工作目录中有批处理文件shellcomm.bat。您应该使用批处理文件的完全限定路径调用Popen。

感谢它的工作，我以为我输入了一个完全限定的路径名​​，但实际上我忘记了。