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

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

$ python3 capture_io.py ./circle
Enter radius of circle: 10
Area: 314.158997
[('output', 'Enter radius of circle: '), ('input',  '10\n'), ('output', 'Area: 314.158997\n')]
  

我决定使用pexpect这项工作的模块。它有一个方法叫做interact它允许用户与子程序交互，如上所示。它还需要 2 个可选参数：output_filter and input_filter。从文档中：

The 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
Enter radius of circle: 10
Area: 314.158997
[('output', 'Enter radius of circle: '), ('input', '1'), ('output', '1'), ('input', '0'), ('output', '0'), ('input', '\r'), ('output', '\r\n'), ('output', 'Area: 314.158997\r\n')]
  

正如您从输出中可以观察到的那样，输入是逐字符处理的，并且还作为输出回显，这造成了如此混乱。是否有可能改变这种行为，以便我的input_filter仅当以下情况时才会运行Enter被压？

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

当我开始编写助手时，我意识到主要问题是输入应该被记录为行缓冲，因此退格键和其他编辑是在输入到达程序之前完成的，但输出应该是非缓冲的以便记录提示符不以新行结束。

为了捕获用于日志记录的输出，需要一个管道，但这会自动打开行缓冲。众所周知，伪终端可以解决这个问题（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')]