【问题标题】:Capture output including control characters of subprocess捕获输出,包括子进程的控制字符
【发布时间】:2021-11-29 09:04:39
【问题描述】:

我有以下简单的程序来运行一个子进程并将tee 输出到stdout 和一些缓冲区

import subprocess
import sys
import time

import unicodedata

p = subprocess.Popen(
    "top",
    shell=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

stdout_parts = []
while p.poll() is None:
    for bytes in iter(p.stdout.readline, b''):
        stdout_parts.append(bytes)
        str = bytes.decode("utf-8")
        sys.stdout.write(str)
        for ch in str:
            if unicodedata.category(ch)[0]=="C" and ord(ch) != 10:
                raise Exception(f"control character! {ord(ch)}")
    time.sleep(0.01)

当运行一些终端更新程序时,例如topdocker pull,我也希望能够捕获它的整个输出,即使它不能立即读取。

例如阅读How do commands like top update output without appending in the console?,它似乎是通过控制字符实现的。 但是,从进程输出流(stdout/stderr)中读取行时,我没有收到任何消息。还是他们使用的技术不同,我无法从子流程中捕捉到它?

【问题讨论】:

  • 我回滚了你最近的编辑(文本仍然可以从revision history获得);非常欢迎您将其发布为答案,但您的问题应严格保留。

标签: python subprocess tty control-characters python-module-unicodedata


【解决方案1】:

许多工具会根据它们是否连接到终端来调整其输出。如果您想准确接收在终端中以交互方式运行该工具时看到的输出,请使用 pexpect 之类的包装器来模拟此行为。 (还有一个低级的pty 库,但使用起来很棘手,尤其是如果您不熟悉问题空间。)

一些工具还允许您为脚本指定批处理操作模式;也许看看top -b(尽管这在 MacOS 上不可用)。

据记录,许多屏幕控制序列并不完全或主要由控制字符组成;例如,将光标移动到curses 中特定位置的控制序列以转义字符(0x1B)开头,否则由常规可打印字符组成。如果您真的想处理这些序列,可能会考虑使用 curses / ANSI 控制代码解析库。但对于大多数目的,更好的方法是使用机器可读的 API 并完全禁用屏幕更新。在 Linux 上,/proc 伪文件系统提供了大量机器可读信息。

【讨论】:

  • 你知道是否有一种方法可以设置子进程的伪终端大小吗?它似乎知道它们并相应地打印控制序列
  • stackoverflow.com/questions/263890/… 是一个 PHP 问题,但有几个有用资源的链接。这有点像鸡和蛋的问题。 curses 确实会尝试计算屏幕的大小,但有一些方法可以覆盖 IIRC。
【解决方案2】:

从恢复编辑到问题中挽救的内容:

一些解决方案可以很好地打印top,并带有来自答案的提示:

import os
import pty
import subprocess
import sys
import time

import select

stdout_master_fd, stdout_slave_fd = pty.openpty()
stderr_master_fd, stderr_slave_fd = pty.openpty()

p = subprocess.Popen(
    "top",
    shell=True,
    stdout=stdout_slave_fd,
    stderr=stderr_slave_fd,
    close_fds=True
)

stdout_parts = []
while p.poll() is None:
    rlist, _, _ = select.select([stdout_master_fd, stderr_master_fd], [], [])
    for f in rlist:
        output = os.read(f, 1000)  # This is used because it doesn't block
        sys.stdout.write(output.decode("utf-8"))
        sys.stdout.flush()
    time.sleep(0.01)

【讨论】:

猜你喜欢
  • 1970-01-01
  • 2021-05-06
  • 2016-07-21
  • 2011-02-01
  • 1970-01-01
  • 2018-05-28
  • 2010-10-29
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多