包装子进程的stdout/stderr

Question

我想捕获并显示通过 Python 子进程调用的进程的输出。

我想我可以将我的类文件对象作为命名参数 stdout 和 stderr 传递

我可以看到它访问

fileno

属性 - 所以它正在对对象执行某些操作。 但是，永远不会调用

write()

方法。我的方法完全不对劲还是我只是错过了什么？

class Process(object):
    class StreamWrapper(object):
        def __init__(self, stream):
            self._stream = stream
            self._buffer = []
        def _print(self, msg):
            print repr(self), msg
        def __getattr__(self, name):
            if not name in ['fileno']:
                self._print("# Redirecting: %s" % name)
            return getattr(self._stream, name)
        def write(self, data):
            print "###########"
            self._buffer.append(data)
            self._stream.write(data)
            self._stream.flush()
        def getBuffer(self):
            return self._buffer[:]
    def __init__(self, *args, **kwargs):
        print ">> Running `%s`" % " ".join(args[0])
        self._stdout = self.StreamWrapper(sys.stdout)
        self._stderr = self.StreamWrapper(sys.stderr)
        kwargs.setdefault('stdout', self._stdout)
        kwargs.setdefault('stderr', self._stderr)
        self._process = subprocess.Popen(*args, **kwargs)
        self._process.communicate()

更新：

我还想使用 ANSI 控制字符来移动光标并覆盖以前的输出内容。我不知道这是否是正确的术语，但这里有一个例子来说明我的意思：我正在尝试自动化一些 GIT 的东西，并且它们的进度可以自我更新，而无需每次都写入新行。

更新2

对我来说很重要的是，子流程的输出立即显示。我尝试使用 subprocess.PIPE 捕获输出并手动显示它，但我只能在进程完成后让它显示输出。但是，我想实时查看输出。

Answer 1

进程的stdin、stdout和stderr需要是真实的文件描述符。 （这实际上不是 Python 施加的限制，而是管道在操作系统级别上的工作方式。）因此您将需要不同的解决方案。

如果您想实时跟踪

stdout

和

stderr

，您将需要异步 I/O 或线程。

• 异步 I/O： 使用标准同步（=阻塞）I/O，对其中一个流的读取可能会阻塞，从而不允许实时访问另一个流。如果您使用的是 Unix，则可以使用非阻塞 I/O，如此答案中所述。然而，在 Windows 上，您将无法采用这种方法。有关 Python 中的异步 I/O 的更多信息以及一些替代方案，请参阅此视频。

• 线程：处理此问题的另一种常见方法是为要实时读取的每个文件描述符创建一个线程。线程只处理它们所分配的文件描述符，因此阻塞 I/O 不会造成损害。

Answer 2

看看这里。

p = subprocess.Popen(cmd,
                 shell=True,
                 bufsize=64,
                 stdin=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 stdout=subprocess.PIPE)

Answer 3

类文件还不够接近。它必须是具有实际文件描述符的实际文件。使用

subprocess

对管道的支持并根据需要从中读取。

Answer 4

这是来自 Chromium CI 产品 Catapult（BSD-3-Clause 许可证）的函数。

我也会在这里研究

os.set_blocking()

而不是

fcntl

。

os.set_blocking()

在Python 3.5中引入，从Python 3.12开始支持Windows管道。因此，从 3.12 开始，代码也可以移植到 Windows。

def _IterProcessStdoutFcntl(process,
                            iter_timeout=None,
                            timeout=None,
                            buffer_size=4096,
                            poll_interval=1):
  """An fcntl-based implementation of _IterProcessStdout."""
  # pylint: disable=too-many-nested-blocks
  import fcntl
  try:
    # Enable non-blocking reads from the child's stdout.
    child_fd = process.stdout.fileno()
    fl = fcntl.fcntl(child_fd, fcntl.F_GETFL)
    fcntl.fcntl(child_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)

    end_time = (time.time() + timeout) if timeout else None
    iter_end_time = (time.time() + iter_timeout) if iter_timeout else None

    while True:
      if end_time and time.time() > end_time:
        raise TimeoutError()
      if iter_end_time and time.time() > iter_end_time:
        yield None
        iter_end_time = time.time() + iter_timeout

      if iter_end_time:
        iter_aware_poll_interval = min(poll_interval,
                                       max(0, iter_end_time - time.time()))
      else:
        iter_aware_poll_interval = poll_interval

      read_fds, _, _ = select.select([child_fd], [], [],
                                     iter_aware_poll_interval)
      if child_fd in read_fds:
        data = _read_and_decode(child_fd, buffer_size)
        if not data:
          break
        yield data

      if process.poll() is not None:
        # If process is closed, keep checking for output data (because of timing
        # issues).
        while True:
          read_fds, _, _ = select.select([child_fd], [], [],
                                         iter_aware_poll_interval)
          if child_fd in read_fds:
            data = _read_and_decode(child_fd, buffer_size)
            if data:
              yield data
              continue
          break
        break
  finally:
    try:
      if process.returncode is None:
        # Make sure the process doesn't stick around if we fail with an
        # exception.
        process.kill()
    except OSError:
      pass
    process.wait()

Answer 5

您在课堂上上课有什么原因吗？ stdout 和 stderr 可以采用任何文件，例如 line 。因此，只需传递一个打开的文件类型或 stringIO 就足以修改流

import sys
sys.stdout = open('test.txt','w')
print "Testing!"
sys.stdout.write('\nhehehe')
sys.stdout = sys.__stdout__
sys.exit(0)