如何从 Python 函数调用捕获 stdout 输出？

我正在使用一个 Python 库对一个对象进行处理

do_something(my_object)

然后改变它。在这样做的同时，它将一些统计数据打印到 stdout，我希望能够掌握这些信息。正确的解决方案是更改 do_something()以返回相关信息,

out = do_something(my_object)

但 do_something()的开发人员还需要一段时间才能解决这个问题。作为一种变通方法，我考虑解析 do_something()写到 stdout 的任何内容。

如何捕获代码中两点之间的 stdout 输出，例如,

start_capturing()
do_something(my_object)
out = end_capturing()

？

95882 次浏览
小开
最佳答案

试试这个上下文管理器:

from io import StringIO
import sys


class Capturing(list):
def __enter__(self):
self._stdout = sys.stdout
sys.stdout = self._stringio = StringIO()
return self
def __exit__(self, *args):
self.extend(self._stringio.getvalue().splitlines())
del self._stringio    # free up some memory
sys.stdout = self._stdout

用法:

with Capturing() as output:
do_something(my_object)

output现在是一个包含函数调用打印的行的列表。

高级用法:

可能不太明显的是，这种做法可以不止一次，其结果是:

with Capturing() as output:
print('hello world')


print('displays on screen')


with Capturing(output) as output:  # note the constructor argument
print('hello world2')


print('done')
print('output:', output)

产出:

displays on screen
done
output: ['hello world', 'hello world2']

更新 : 他们在 Python 3.4中将 redirect_stdout()添加到 contextlib(与 redirect_stderr()一起)。因此，您可以使用 io.StringIO来实现类似的结果(尽管 Capturing作为列表和上下文管理器可以说更方便)。

小开

在 python > = 3.4中，contextlib 包含一个 redirect_stdout装饰符:

import io
from contextlib import redirect_stdout


f = io.StringIO()
with redirect_stdout(f):
do_something(my_object)
out = f.getvalue()

来自 那些文件:

上下文管理器，用于临时将 sys.stdout 重定向到另一个文件 或类似文件的对象。

此工具增加了现有函数或类的灵活性 输出硬连接到标准输出。

例如，help ()的输出通常发送到 sys.stdout 可以通过将输出重定向到 对象:

  f = io.StringIO()
with redirect_stdout(f):
help(pow)
s = f.getvalue()

要将 help ()的输出发送到磁盘上的文件，请将输出重定向到 普通档案:

 with open('help.txt', 'w') as f:
with redirect_stdout(f):
help(pow)

要将 help ()的输出发送到 sys.stderr:

with redirect_stdout(sys.stderr):
help(pow)

注意，sys.stdout 的全局副作用意味着这个上下文 管理器不适合在库代码和大多数线程中使用 它对子进程的输出也没有影响。 然而，对于许多实用程序脚本来说，它仍然是一种有用的方法。

这个上下文管理器是可重入的。

小开

下面是一个使用文件管道的异步解决方案。

import threading
import sys
import os


class Capturing():
def __init__(self):
self._stdout = None
self._stderr = None
self._r = None
self._w = None
self._thread = None
self._on_readline_cb = None


def _handler(self):
while not self._w.closed:
try:
while True:
line = self._r.readline()
if len(line) == 0: break
if self._on_readline_cb: self._on_readline_cb(line)
except:
break


def print(self, s, end=""):
print(s, file=self._stdout, end=end)


def on_readline(self, callback):
self._on_readline_cb = callback


def start(self):
self._stdout = sys.stdout
self._stderr = sys.stderr
r, w = os.pipe()
r, w = os.fdopen(r, 'r'), os.fdopen(w, 'w', 1)
self._r = r
self._w = w
sys.stdout = self._w
sys.stderr = self._w
self._thread = threading.Thread(target=self._handler)
self._thread.start()


def stop(self):
self._w.close()
if self._thread: self._thread.join()
self._r.close()
sys.stdout = self._stdout
sys.stderr = self._stderr

示例用法:

from Capturing import *
import time


capturing = Capturing()


def on_read(line):
# do something with the line
capturing.print("got line: "+line)


capturing.on_readline(on_read)
capturing.start()
print("hello 1")
time.sleep(1)
print("hello 2")
time.sleep(1)
print("hello 3")
capturing.stop()
小开

根据 kindall和 ForeverWintr的答案。

我为 Python<3.4创建 redirect_stdout函数:

import io
from contextlib import contextmanager


@contextmanager
def redirect_stdout(f):
try:
_stdout = sys.stdout
sys.stdout = f
yield
finally:
sys.stdout = _stdout




f = io.StringIO()
with redirect_stdout(f):
do_something()
out = f.getvalue()
小开

同样利用@kindall 和@ForeveWintr 的答案，这里有一门课可以完成这个任务。与以前的答案主要不同的是，这个答案捕获它的 就像一根绳子，而不是作为一个 StringIO对象，这是更方便的工作！

import io
from collections import UserString
from contextlib import redirect_stdout


class capture(UserString, str, redirect_stdout):
'''
Captures stdout (e.g., from ``print()``) as a variable.


Based on ``contextlib.redirect_stdout``, but saves the user the trouble of
defining and reading from an IO stream. Useful for testing the output of functions
that are supposed to print certain output.
'''


def __init__(self, seq='', *args, **kwargs):
self._io = io.StringIO()
UserString.__init__(self, seq=seq, *args, **kwargs)
redirect_stdout.__init__(self, self._io)
return


def __enter__(self, *args, **kwargs):
redirect_stdout.__enter__(self, *args, **kwargs)
return self


def __exit__(self, *args, **kwargs):
self.data += self._io.getvalue()
redirect_stdout.__exit__(self, *args, **kwargs)
return


def start(self):
self.__enter__()
return self


def stop(self):
self.__exit__(None, None, None)
return

例子:

# Using with...as
with capture() as txt1:
print('Assign these lines')
print('to a variable')


# Using start()...stop()
txt2 = capture().start()
print('This works')
print('the same way')
txt2.stop()


print('Saved in txt1:')
print(txt1)
print('Saved in txt2:')
print(txt2)

这在 Sciris中实现为 Sc.catch ()。


提交答案