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

Question

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

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

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

该output_filter会通过了所有从子进程的输出。该input_filter会从用户通过了所有的键盘输入。

所以这是我写的代码：

捕获_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)

圆.c

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

产生以下输出：

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

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

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

Answer 1

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

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

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

你之前这么说：

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

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

tty.setraw(self.STDIN_FILENO)

这会将您的终端设置为原始模式：

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

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

$ python3 test.py ./圆
输入圆的半径：10
10
地区：314.158997
[('输出', '输入圆半径: '), ('输入', '10\n'), ('输出', '10\r\n'), ('输出', '面积: 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 ./圆
输入圆的半径：10
地区：314.158997
[('输出','输入圆半径:'),('输入','10\n'),('输出','面积:314.158997\r\n')]

答对了！现在，您可以使用这些更改继承pexpect.spawn并覆盖interact方法，或者使用内置方法实现相同的事情pty方法，或者使用Python 的

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