如何捕获子进程的输入和输出？

Question

我正在尝试制作一个以可执行文件名称作为参数的程序，运行可执行文件并报告该运行的输入和输出。例如，考虑一个名为“circle”的子程序。我的程序需要运行以下内容：

$ python3 capture_io.py ./circle 输入圆半径：10 区域：314.158997 [('output', '输入圆的半径:'), ('input', '10\n'), ('output', '面积: 314.158997\n')]

我决定使用pexpect 模块来完成这项工作。它有一个名为interact 的方法，可以让用户与子程序进行交互，如上所示。它还需要 2 个可选参数：output_filter 和 input_filter。来自文档：

output_filter 将传递子进程的所有输出。 input_filter 将传递来自用户的所有键盘输入。

这是我写的代码：

capture_io.py

import sys
import pexpect

_stdios = []


def read(data):
    _stdios.append(("output", data.decode("utf8")))
    return data


def write(data):
    _stdios.append(("input", data.decode("utf8")))
    return data


def capture_io(argv):
    _stdios.clear()
    child = pexpect.spawn(argv)
    child.interact(input_filter=write, output_filter=read)
    child.wait()
    return _stdios


if __name__ == '__main__':
    stdios_of_child = capture_io(sys.argv[1:])
    print(stdios_of_child)

circle.c

#include <stdio.h>
#include <stdlib.h>

int main(int argc, char* argv[]) {
    float radius, area;

    printf("Enter radius of circle: ");
    scanf("%f", &radius);

    if (radius < 0) {
        fprintf(stderr, "Negative radius values are not allowed.\n");
        exit(1);
    }

    area = 3.14159 * radius * radius;
    printf("Area: %f\n", area);
    return 0;
}

产生以下输出：

$ python3 capture_io.py ./circle 输入圆半径：10 区域：314.158997 [('output', '输入圆的半径:'), ('input', '1'), ('output', '1'), ('input', '0'), ('output', '0'), ('input', '\r'), ('output', '\r\n'), ('output', '区域: 314.158997\r\n')]

您可以从输出中观察到，输入是逐个字符处理的，并且还会作为输出回显，这会造成如此混乱。是否可以更改此行为，以便我的 input_filter 仅在按下 Enter 时运行？

或者更一般地说，实现我的目标的最佳方式是什么（有或没有pexpect）？

Answer 1

当我开始写一个helper的时候，我意识到主要的问题是输入应该是记录行缓冲的，所以退格和其他编辑在输入到达程序之前完成，但输出应该是无缓冲的记录未被新行终止的提示。

为了记录日志而捕获输出，需要一个管道，但这会自动打开行缓冲。众所周知，伪终端可以解决问题（expect 模块是围绕伪终端构建的），但是终端同时具有输入和输出，我们只想取消缓冲输出。

幸运的是，有 stdbuf 实用程序。在 Linux 上，它改变了动态链接可执行文件的 C 库函数。不能普遍使用。

我修改了一个 Python 双向复制程序来记录它复制的数据。结合stdbuf，它会产生所需的输出。

import select
import os

STDIN = 0
STDOUT = 1

BUFSIZE = 4096

def main(cmd):
    ipipe_r, ipipe_w = os.pipe()
    opipe_r, opipe_w = os.pipe()
    if os.fork():
        # parent
        os.close(ipipe_r)
        os.close(opipe_w)
        fdlist_r = [STDIN, opipe_r]
        while True:
            ready_r, _, _ = select.select(fdlist_r, [], []) 
            if STDIN in ready_r:
                # STDIN -> program
                data = os.read(STDIN, BUFSIZE)
                if data:
                    yield('in', data)   # optional: convert to str
                    os.write(ipipe_w, data)
                else:
                    # send EOF
                    fdlist_r.remove(STDIN)
                    os.close(ipipe_w)
            if opipe_r in ready_r:
                # program -> STDOUT
                data = os.read(opipe_r, BUFSIZE)
                if not data:
                    # got EOF
                    break
                yield('out', data)
                os.write(STDOUT, data)
        os.wait()
    else:
        # child
        os.close(ipipe_w)
        os.close(opipe_r)
        os.dup2(ipipe_r, STDIN)
        os.dup2(opipe_w, STDOUT)
        os.execlp(*cmd)
        # not reached
        os._exit(127)

if __name__ == '__main__':
    log = list(main(['stdbuf', 'stdbuf', '-o0', './circle']))
    print(log)

打印出来：

[('out', b'Enter radius of circle: '), ('in', b'12\n'), ('out', b'Area: 452.388947\n')]

Answer 2

是否可以更改此行为，以便我的 input_filter 仅在按下 Enter 时运行？

是的，您可以通过继承pexpect.spawn 并覆盖interact 方法来实现。我很快就会谈到这一点。

正如 VPfB 在 their answer 中指出的那样，您不能使用管道，我认为值得一提的是，pexpect's documentation 中也解决了这个问题。

你说过：

...输入被逐个字符处理，也作为输出回显...

如果您检查interact 的源代码，您可以看到这一行：

tty.setraw(self.STDIN_FILENO)

这会将您的终端设置为raw mode:

输入是逐字符可用的，...，终端输入和输出字符的所有特殊处理都被禁用。

这就是为什么您的input_filter 函数在每次按键时都会运行并且它会看到退格或其他特殊字符。如果你可以注释掉这一行，你会在运行程序时看到如下内容：

$ python3 test.py ./circle 输入圆半径：10 10 区域：314.158997 [('output', '输入圆的半径:'), ('input', '10\n'), ('output', '10\r\n'), ('output', '面积: 314.158997 \r\n')]

这也可以让你编辑输入（即12[Backspace]0 会给你同样的结果）。但正如您所看到的，它仍然与输入相呼应。这可以通过为孩子的终端设置一个简单的标志来禁用：

mode = tty.tcgetattr(self.child_fd)
mode[3] &= ~termios.ECHO
tty.tcsetattr(self.child_fd, termios.TCSANOW, mode)

使用最新更改运行：

$ python3 test.py ./circle 输入圆半径：10 区域：314.158997 [('output', '输入圆半径:'), ('input', '10\n'), ('output', '面积: 314.158997\r\n')]

宾果！现在您可以继承 pexpect.spawn 并使用这些更改覆盖 interact 方法，或者使用 Python 的内置 pty 模块实现相同的功能：

pty:

import os
import pty
import sys
import termios
import tty

_stdios = []

def _read(fd):
    data = os.read(fd, 1024)
    _stdios.append(("output", data.decode("utf8")))
    return data


def _stdin_read(fd):
    data = os.read(fd, 1024)
    _stdios.append(("input", data.decode("utf8")))
    return data


def _spawn(argv):
    pid, master_fd = pty.fork()
    if pid == pty.CHILD:
        os.execlp(argv[0], *argv)

    mode = tty.tcgetattr(master_fd)
    mode[3] &= ~termios.ECHO
    tty.tcsetattr(master_fd, termios.TCSANOW, mode)

    try:
        pty._copy(master_fd, _read, _stdin_read)
    except OSError:
        pass

    os.close(master_fd)
    return os.waitpid(pid, 0)[1]


def capture_io_and_return_code(argv):
    _stdios.clear()
    return_code = _spawn(argv)
    return _stdios, return_code >> 8


if __name__ == '__main__':
    stdios, ret = capture_io_and_return_code(sys.argv[1:])
    print(stdios)

pexpect:

import sys
import termios
import tty
import pexpect

_stdios = []


def read(data):
    _stdios.append(("output", data.decode("utf8")))
    return data


def write(data):
    _stdios.append(("input", data.decode("utf8")))
    return data


class CustomSpawn(pexpect.spawn):
    def interact(self, escape_character=chr(29),
                 input_filter=None, output_filter=None):
        self.write_to_stdout(self.buffer)
        self.stdout.flush()
        self._buffer = self.buffer_type()
        mode = tty.tcgetattr(self.child_fd)
        mode[3] &= ~termios.ECHO
        tty.tcsetattr(self.child_fd, termios.TCSANOW, mode)
        if escape_character is not None and pexpect.PY3:
            escape_character = escape_character.encode('latin-1')
        self._spawn__interact_copy(escape_character, input_filter, output_filter)


def capture_io_and_return_code(argv):
    _stdios.clear()
    child = CustomSpawn(argv)
    child.interact(input_filter=write, output_filter=read)
    child.wait()
    return _stdios, child.status >> 8


if __name__ == '__main__':
    stdios, ret = capture_io_and_return_code(sys.argv[1:])
    print(stdios)

Answer 3

我认为您不会轻易做到这一点，但是，我认为这应该适合您：

output_buffer=''
def read(data):
    output_buffer+=data
    if data == '\r':
         _stdios.append(("output", output_buffer.decode("utf8")))
         output_buffer = ''
    return data