如何读取预期子进程的整个输出

Question

问题

---

我想通过 Python 脚本与交互式终端程序进行交互，这些程序可能并不总是用 Python 编写的。我已经设法用 pexpect 和下面代码 sn-p 中的类来做到这一点，但我很难找到一种方法来捕获每条指令后的全部输出.

上下文

---

我无法捕获命令的整个输出（所有行）和使程序保持活动状态以供将来输入。 假设我想这样做：

terminal.start("/path/to/executable/repl/file") # on start returns 3 lines of output
terminal.run_command("let a = fn(a) { a + 1 }") # this command return 1 line of output
terminal.run_command("var") # this command will return 2 lines of output
terminal.run_command("invalid = invalid") # this command returns 1 line of output

请注意，每个输出的行数可能会有所不同因为我希望能够运行多个交互式终端程序。

我试过的

尝试 1

我尝试使用 readlines 但正如文档所述

请记住，因为这一直读取到 EOF，这意味着子进程应该关闭其标准输出。

这意味着一旦我运行它就会关闭我的进程以获取未来的指令，这不是我预期的行为。无论如何，当我尝试它时，我得到以下信息。

def read(self):    
    return list(self.process.readlines())

由于我不知道的原因，该程序什么都不做，什么也不打印，不引发任何错误，只是保持暂停状态，没有任何输出。

尝试 2

阅读每一行，直到找到这样的空行

def read(self):
    val = self.process.readline()
    result = ""
    while val != "":
        result += val        
        val = self.process.readline()
    return result

同样的问题，程序暂停，不打印输入，几秒钟什么都不做，然后打印错误pexpect.exceptions.TIMEOUT: Timeout exceeded.

尝试 3

使用read_nonblocking方法导致我的程序只读取几个字符，所以我使用第一个参数size如下。

def read(self):
    return self.process.read_nonblocking(999999999)

只有这样我才得到预期的行为，但只有几个命令，然后它什么也读不到，此外，如果我输入更大的数字，则会引发内存溢出错误。

代码

---

这是 Terminal 类的实现。

import pexpect

class Terminal:

    process: pexpect.spawn

    def __init__(self):
        self.process = None

    def start(self, executable_file: str):
        '''
        run a command that returns an executable TUI program, returns the output, 
        (if present) of the initialization of program
        '''

        self.process = pexpect.spawn(executable_file, encoding="utf-8", maxread=1)
        return self.read()

    def read(self):
        '''return entire output of last executed command'''
        return self.process.readline() # when executed more than amoutn of output program breaks

    def write(self, message):
        '''send value to program through keyboard input'''
        self.process.sendline(message)

    def terminate(self):
        '''kill process/program and restart property value to None'''
        self.process.kill()
        self.process.wait()
        self.process = None

    def run_command(self, command: str):
        '''
        run an instruction for the executed program 
        and get the returned result as string
        '''
        self.write(command)
        return self.read()

我如何上课。这是我在上面提到的每次尝试中运行的测试

from terminal import Terminal

term = Terminal()

print(term.start("/path/to/executable/repl/file"), end="")
print(term.run_command("let a = fn(a) { a + 1 }"), end="")
print(term.run_command("a(1)"), end="")
print(term.run_command("let b = [1,2,4]"), end="")
print(term.run_command("b[0]"), end="")
print(term.run_command("b[1]"), end="")
print(term.run_command("a(2)"), end="")

如果你想知道什么样的具体的我想运行的程序，目前只有这两个1和2，但我希望将来添加更多。

Answer 1

问题的症结在于检测您发送到控制台程序的命令何时完成写入输出。

我首先创建了一个非常简单的带有输入和输出的控制台程序：echo。它只是写回你写的东西。

这里是 ：

echo.py

import sys

print("Welcome to PythonEcho Ultimate Edition 2023")

while True:
    new_line = sys.stdin.readline()
    print(new_line, file=sys.stdout, end="")  # because the line already has an 
 at the end
    print("")  # an empty line, because that's how the webshell detects the output for this command has ended

这是一个稍微修改过的版本，它在开头打印一行（因为这是你的startProgram()在做terminal.read()时所期望的）和每个回显后的空行（因为这就是你的runCommand检测到输出完成的方式） .

我在没有 Flask 的情况下这样做，因为它不需要与控制台程序进行通信，并且有助于调试（参见 Minimal Reproducible Example）。所以这是我使用的代码：

main.py

import subprocess


class Terminal:

    def __init__(self):
        self.process = None

    def start(self, executable_file):
        self.process = subprocess.Popen(
            executable_file,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    def read(self):
        return self.process.stdout.readline().decode("utf-8")
    #                                   no `.strip()` there  ^^^^^^^^

    def write(self, message):
        self.process.stdin.write(f"{message.strip()}
".encode("utf-8"))
        self.process.stdin.flush()

    def terminate(self):
        self.process.stdin.close()
        self.process.terminate()
        self.process.wait(timeout=0.2)


terminal = Terminal()


def startProgram():
    terminal.start(["python3", "echo.py"])  # using my echo program
    return terminal.read()


def runCommand(command: str):
    terminal.write(command)
    result = ""
    line = "initialized to something else than \n"
    while line != "
":  # an empty line is considered a separator ?
        line = terminal.read()
        result += line
    return result


def stopProgram():
    terminal.terminate()
    return "connection terminated"


if __name__ == "__main__":  # for simple testing
    result = startProgram()
    print(result)
    result = runCommand("cmd1")
    print(result, end="")
    result = runCommand("cmd2")
    print(result, end="")
    result = stopProgram()
    print(result)

这给了我

Welcome to PythonEcho Ultimate Edition 2023

cmd1

cmd2

connection terminated

我删除了 Terminal.read 末尾的 split() ，否则这些行稍后将无法在 runCommand 中正确连接（或者是自愿的？）。 我将 runCommand 函数更改为在遇到 时停止读取（这是一个空行，不同于表示流结束的空字符串）。你没有解释你是如何检测到你的程序输出结束的，你应该注意这一点。