do_something()
out = do_something(my_object)
但是要在
do_something()
devs到这个问题之前需要一段时间。作为解决方法,我考虑过解析do_something()
撰写的任何内容。
如何在代码中的两个点之间捕获stdout输出,例如
start_capturing()
do_something(my_object)
out = end_capturing()
?
步调此上下文经理:
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)
update
:他们将displays on screen
done
output: ['hello world', 'hello world2']
redirect_stdout()
)中。因此,您可以使用contextlib
在python> = 3.4,Contextlib包含一个redirect_stderr()
io.StringIO
theDocs:
ContextManager暂时将SYS.Stdout重定向到另一个文件 或类似文件的对象该工具为现有功能或类 输出与Stdout有联系。例如,help()的输出通常发送到sys.stdout。你 可以通过将输出重定向到一个字符串中的输出 io.stringio对象:
Capturing
要将help()的输出发送到磁盘上的文件,请将输出重定向到 常规文件:
redirect_stdout
将帮助()的输出发送到sys.stderr:
import io
from contextlib import redirect_stdout
f = io.StringIO()
with redirect_stdout(f):
do_something(my_object)
out = f.getvalue()
注意,全球副作用对sys.stdout意味着此上下文
管理器不适合在库代码和大多数线程中使用
申请。它也对子过程的产出没有影响。
但是,对于许多实用程序脚本而言,它仍然是一种有用的方法。
here是一种使用文件管道的异步解决方案。
f = io.StringIO() with redirect_stdout(f): help(pow) s = f.getvalue()
示例用法:
with open('help.txt', 'w') as f:
with redirect_stdout(f):
help(pow)
基于和
with redirect_stdout(sys.stderr): help(pow)
的答案。
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
I创建
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
,而不是作为
redirect_stdout
对象,它更方便地使用!
Python<3.4
示例: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()
在Sciris
中实现为Sc.capture().
。
您的功能可能会引起例外。如果您还要捕获错误消息,则应将函数调用包含在StringIO
中,如下(基于@foreverwintr的答案)
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)
还将将异常从do_something(my_object)
重定向到
try...except
,以便它也将被import io
from contextlib import redirect_stdout
f = io.StringIO()
with redirect_stdout(f):
try:
do_something(my_object)
except Exception as e:
print(f"ERROR: {str(e)}") # will redirect the exception to stdout
out = f.getvalue()
捕获。