Como capturar a saída stdout de uma chamada de função Python?

112

Estou usando uma biblioteca Python que faz algo a um objeto

do_something(my_object)

e muda isso. Enquanto faz isso, ele imprime algumas estatísticas para stdout e eu gostaria de ter uma ideia dessas informações. A solução adequada seria mudar do_something()para retornar as informações relevantes,

out = do_something(my_object)

mas vai demorar um pouco até que os desenvolvedores do_something()cheguem a esse problema. Como solução alternativa, pensei em analisar tudodo_something() grava em stdout.

Como posso capturar a saída stdout entre dois pontos no código, por exemplo,

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

?

Nico Schlömer
fonte
2
possível duplicata dos resultados
Martijn Pieters
Minha resposta na pergunta vinculada se aplica aqui também.
Martijn Pieters
Tentei fazer isso uma vez e a melhor resposta que encontrei foi: stackoverflow.com/a/3113913/1330293
elyase
A resposta vinculada a @elyase é uma solução elegante
Pykler
Veja esta resposta .
martineau de

Respostas:

183

Experimente este gerenciador 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)

output agora é uma lista contendo as linhas impressas pela chamada de função.

Uso avançado:

O que pode não ser óbvio é que isso pode ser feito mais de uma vez e os resultados concatenados:

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)

Resultado:

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

Actualizar : Acrescentaram redirect_stdout()para contextlibem Python 3,4 (juntamente com redirect_stderr()). Portanto, você pode usar io.StringIOcom isso para obter um resultado semelhante (embora Capturingser uma lista, bem como um gerenciador de contexto, seja indiscutivelmente mais conveniente).

kindall
fonte
Obrigado! E obrigado por adicionar a seção avançada ... Originalmente, usei uma atribuição de fatia para colocar o texto capturado na lista, então me bati na cabeça e usei .extend()para que pudesse ser usado concatenativamente, como você notou. :-)
kindall
PS Se for usado repetidamente, sugiro adicionar um self._stringio.truncate(0)após a self.extend()chamada no __exit__()método para liberar parte da memória mantida pelo _stringiomembro.
martineau de
25
Ótima resposta, obrigado. Para Python 3, use em from io import StringIOvez da primeira linha no gerenciador de contexto.
Wtower
1
Isso é thread-safe? O que acontece se alguma outra thread / chamada usa print () enquanto do_something é executado?
Derorrist de
1
Esta resposta não funcionará para saída de bibliotecas compartilhadas C, veja esta resposta em vez disso.
craymichael
81

Em python> = 3.4, contextlib contém um redirect_stdoutdecorador. Ele pode ser usado para responder à sua pergunta da seguinte forma:

import io
from contextlib import redirect_stdout

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

Dos documentos :

Gerenciador de contexto para redirecionar temporariamente sys.stdout para outro arquivo ou objeto semelhante a um arquivo.

Essa ferramenta adiciona flexibilidade às funções ou classes existentes cuja saída está conectada ao stdout.

Por exemplo, a saída de help () normalmente é enviada para sys.stdout. Você pode capturar essa saída em uma string redirecionando a saída para um objeto io.StringIO:

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

Para enviar a saída de help () para um arquivo no disco, redirecione a saída para um arquivo normal:

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

Para enviar a saída de help () para sys.stderr:

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

Observe que o efeito colateral global em sys.stdout significa que esse gerenciador de contexto não é adequado para uso em código de biblioteca e na maioria dos aplicativos encadeados. Também não tem efeito na saída de subprocessos. No entanto, ainda é uma abordagem útil para muitos scripts de utilitário.

Este gerenciador de contexto é reentrante.

ForeverWintr
fonte
quando tentei f = io.StringIO() with redirect_stdout(f): logger = getLogger('test_logger') logger.debug('Test debug message') out = f.getvalue() self.assertEqual(out, 'DEBUG:test_logger:Test debug message'). Isso me dá um erro:AssertionError: '' != 'Test debug message'
Eziz Durdyyev
o que significa que fiz algo errado ou não foi possível capturar o log stdout.
Eziz Durdyyev
@EzizDurdyyev, logger.debugnão grava no stdout por padrão. Se você substituir sua chamada de registro por, print()deverá ver a mensagem.
ForeverWintr
Sim, eu sei, mas eu torná-lo escrever para stdout assim: stream_handler = logging.StreamHandler(sys.stdout). E adicione esse manipulador ao meu logger. então ele deve gravar em stdout e redirect_stdoutdeve capturá-lo, certo?
Eziz Durdyyev
Suspeito que o problema esteja na maneira como você configurou seu logger. Gostaria de verificar se ele imprime em stdout sem o redirect_stdout. Em caso afirmativo, talvez o buffer não esteja sendo esvaziado até que o gerenciador de contexto saia.
ForeverWintr
0

Aqui está uma solução assíncrona usando canais de arquivo.

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

Exemplo de uso:

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