¿Cómo capturar la salida estándar de una llamada a una función de Python?
Estoy usando una biblioteca de Python que le hace algo a un objeto.
do_something(my_object)
y lo cambia. Mientras lo hace, imprime algunas estadísticas en la salida estándar y me gustaría controlar esta información. La solución adecuada sería cambiar do_something()para devolver la información relevante,
out = do_something(my_object)
pero pasará un tiempo antes de que los desarrolladores aborden do_something()este problema. Como solución alternativa, pensé en analizar todo lo que do_something()se escribe en la salida estándar.
¿Cómo puedo capturar la salida estándar entre dos puntos del código, por ejemplo,
start_capturing()
do_something(my_object)
out = end_capturing()
?
Pruebe este administrador de contexto:
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
Uso:
with Capturing() as output:
do_something(my_object)
outputAhora hay una lista que contiene las líneas impresas por la llamada a la función.
Uso avanzado:
Lo que puede no ser obvio es que esto se puede hacer más de una vez y los resultados se pueden concatenar:
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)
Producción:
displays on screen
done
output: ['hello world', 'hello world2']
Actualización : se agregaron redirect_stdout()en contextlibPython 3.4 (junto con redirect_stderr()). Por lo tanto, podría usarlo io.StringIOpara lograr un resultado similar (aunque Capturingpodría decirse que ser una lista y un administrador de contexto es más conveniente).
En Python >= 3.4, contextlib contiene un redirect_stdoutadministrador de contexto. Se puede utilizar para responder a su pregunta de esta manera:
import io
from contextlib import redirect_stdout
f = io.StringIO()
with redirect_stdout(f):
do_something(my_object)
out = f.getvalue()
De los documentos :
Administrador de contexto para redirigir temporalmente sys.stdout a otro archivo u objeto similar a un archivo.
Esta herramienta agrega flexibilidad a funciones o clases existentes cuya salida está cableada a la salida estándar.
Por ejemplo, la salida de help() normalmente se envía a sys.stdout. Puede capturar esa salida en una cadena redirigiendo la salida a un objeto io.StringIO:
f = io.StringIO()
with redirect_stdout(f):
help(pow)
s = f.getvalue()
Para enviar el resultado de help() a un archivo en el disco, redirija el resultado a un archivo normal:
with open('help.txt', 'w') as f: with redirect_stdout(f): help(pow)Para enviar la salida de help() a sys.stderr:
with redirect_stdout(sys.stderr): help(pow)Tenga en cuenta que el efecto secundario global en sys.stdout significa que este administrador de contexto no es adecuado para su uso en código de biblioteca y en la mayoría de aplicaciones con subprocesos. Tampoco tiene ningún efecto sobre la salida de los subprocesos. Sin embargo, sigue siendo un enfoque útil para muchos scripts de utilidades.
Este administrador de contexto es reentrante.
Aquí hay una solución asíncrona que utiliza canalizaciones de archivos.
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
Uso de ejemplo:
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()
Basado en la respuesta de kindally .ForeverWintr
Creo redirect_stdoutuna función para 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()
También basándose en las respuestas de @kindall y @ForeveWintr, aquí hay una clase que logra esto. La principal diferencia con las respuestas anteriores es que esto lo captura como una cadena , no como un StringIOobjeto, ¡lo cual es mucho más conveniente para trabajar!
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
Ejemplos:
# 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)
Esto se implementa en Sciris como sc.capture() .
