标签归档:capture

如何从Python函数调用中捕获标准输出?

问题:如何从Python函数调用中捕获标准输出?

我正在使用对对象执行某些操作的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()

I’m using a Python library that does something to an object

do_something(my_object)

and changes it. While doing so, it prints some statistics to stdout, and I’d like to get a grip on this information. The proper solution would be to change do_something() to return the relevant information,

out = do_something(my_object)

but it will be a while before the devs of do_something() get to this issue. As a workaround, I thought about parsing whatever do_something() writes to stdout.

How can I capture stdout output between two points in the code, e.g.,

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

?


回答 0

试试这个上下文管理器:

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']

更新:它们已添加redirect_stdout()contextlibPython 3.4中(以及redirect_stderr())。因此,您可以使用io.StringIO它来达到类似的结果(尽管Capturing列表和上下文管理器可能更方便)。

Try this context manager:

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

Usage:

with Capturing() as output:
    do_something(my_object)

output is now a list containing the lines printed by the function call.

Advanced usage:

What may not be obvious is that this can be done more than once and the results concatenated:

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)

Output:

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

Update: They added redirect_stdout() to contextlib in Python 3.4 (along with redirect_stderr()). So you could use io.StringIO with that to achieve a similar result (though Capturing being a list as well as a context manager is arguably more convenient).


回答 1

在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重定向到另一个文件或类似文件的对象。

该工具为现有功能或类的输出增加了灵活性,这些功能或类的输出被硬连线到stdout。

例如,通常将help()的输出发送到sys.stdout。您可以通过将输出重定向到io.StringIO对象来捕获该输出的字符串:

  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的全局副作用意味着此上下文管理器不适合在库代码和大多数线程应用程序中使用。它还对子流程的输出没有影响。但是,对于许多实用程序脚本,它仍然是一种有用的方法。

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

In python >= 3.4, contextlib contains a redirect_stdout decorator. It can be used to answer your question like so:

import io
from contextlib import redirect_stdout

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

From the docs:

Context manager for temporarily redirecting sys.stdout to another file or file-like object.

This tool adds flexibility to existing functions or classes whose output is hardwired to stdout.

For example, the output of help() normally is sent to sys.stdout. You can capture that output in a string by redirecting the output to an io.StringIO object:

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

To send the output of help() to a file on disk, redirect the output to a regular file:

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

To send the output of help() to sys.stderr:

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

Note that the global side effect on sys.stdout means that this context manager is not suitable for use in library code and most threaded applications. It also has no effect on the output of subprocesses. However, it is still a useful approach for many utility scripts.

This context manager is reentrant.


回答 2

这是使用文件管道的异步解决方案。

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()

Here is an async solution using file pipes.

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

Example usage:

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()