Как зафиксировать входы и выходы дочернего процесса?

Когда я начал писать хелпер, я понял, что основная проблема заключается в том, что входные данные должны логироваться буферизованными строками, поэтому возврат и другое редактирование выполняются до того, как входные данные достигнут программы, а выходные данные должны быть небуферизованными, чтобы логировать приглашение, которое не завершается новой строкой.

Для захвата вывода с целью ведения журнала необходим канал, но он автоматически включает буферизацию строк. Известно, что псевдотерминал решает проблему (модуль 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')]

Можно ли изменить это поведение, чтобы мой 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
Enter radius of circle: 10
Area: 314.158997
[('output', 'Enter radius of circle: '), ('input', '10\n'), ('output', 'Area: 314.158997\r\n')]

Бинго! Теперь вы можете наследоваться от pexpect.spawn и переопределить метод interact с этими изменениями или реализовать то же самое, используя встроенный модуль pty Python:

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)

Я не думаю, что вы сможете сделать это легко, однако я думаю, что это должно сработать для вас:

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