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

Question

我正在尝试编写一个程序，该程序以可执行文件名作为参数，运行可执行文件并报告该运行的输入和输出。例如，考虑一个名为“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的方法，允许用户与子程序进行交互，就像上面看到的那样。它还有两个可选参数：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')]

从输出结果可以看出，输入字符逐个处理，并作为输出回显，这会导致混乱。有没有可能更改此行为，使得只有在按下“Enter”键时才运行 input_filter ？

或者更一般地说，无论是否使用 pexpect ，实现我的目标的最佳方法是什么？

- Asocia

---

Linux有相关的工具script（检查--log-in和--log-out选项）和tee。 - VPfB

请参考这个问题。 - igrinis

@VPfB 我将在我无法控制的机器上运行此代码。因此，需要另一个程序并不适合我。即使在我的电脑上，我也找不到 --log-in 和 --log-out 选项。（来自 util-linux 2.31.1 的脚本） - Asocia

@igrinis 我认为它没有做我想要的事情（至少当我阅读它时感觉是这样），而且比它应该的复杂。 - Asocia

@Asocia 好的，我不确定哪种解决方案适合您的需求。许多人更喜欢现有的工具。关于--log-in，您是正确的，它是最近在2.35中添加的。 - VPfB

@VPfB 我对bash编程不熟悉，不确定该如何实现。最终，唯一的要求是简单。如果安装tee命令可以以自然的方式解决问题，那么这可能不是什么大问题。当我输入man tee时，它说将标准输入复制到每个文件，并同时复制到标准输出中。所以看起来好像它分离了输入和输出。是否有可能在保留顺序的情况下（即哪个输入在哪个输出之后，反之亦然），将这两者结合起来呢？ - Asocia

Answer 1

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

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

- VPfB

Answer 2

这个行为可以更改吗，使得我的input_filter只在按下Enter时运行？

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

正如VPfB在他们的答案中指出的那样，您不能使用管道，我认为值得一提的是，这个问题也在pexpect的文档中得到了解决。

您说过：

...输入是逐个字符处理并作为输出回显...

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

tty.setraw(self.STDIN_FILENO)

这将使您的终端进入原始模式：

逐个字符输入可用，...，并且禁用了终端输入和输出字符的所有特殊处理。

这就是为什么您的input_filter函数会在每次按键时运行，并且它会看到退格键或其他特殊字符。如果您注释掉此行代码，当您运行程序时，您会看到像这样的结果：

$ python3 test.py ./circle
Enter radius of circle: 10
10
Area: 314.158997
[('output', 'Enter radius of circle: '), ('input', '10\n'), ('output', '10\r\n'), ('output', 'Area: 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)

- Asocia

Answer 3

我不认为你能轻易地做到这一点，但是我认为这应该适合你：

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

- Cargo23

---

感谢您的建议。不幸的是，正如我所说，input_filter会在每次按键时运行。因此，当用户输入像1[Backspace]5这样的内容时，它也会在退格键上运行。而我只想要5这种情况。因此，我正在寻找一种改变子进程底层pty的方法。 - Asocia