Wie erfasst man die Standardausgabe eines Python-Funktionsaufrufs?

112

Ich verwende eine Python-Bibliothek, die etwas mit einem Objekt macht

do_something(my_object)

und ändert es. Dabei werden einige Statistiken an stdout gedruckt, und ich möchte diese Informationen in den Griff bekommen. Die richtige Lösung wäre, zu ändern do_something(), um die relevanten Informationen zurückzugeben.

out = do_something(my_object)

Aber es wird eine Weile dauern, bis die Entwickler do_something()zu diesem Thema kommen. Um dieses Problem do_something()zu umgehen , habe ich darüber nachgedacht, alles zu analysieren, was in stdout geschrieben wird.

Wie kann ich die Standardausgabe zwischen zwei Punkten im Code erfassen, z.

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

?

Nico Schlömer
quelle
2
Mögliches Duplikat der Erfassung von dis.dis-Ergebnissen
Martijn Pieters
Meine Antwort in der verknüpften Frage gilt auch hier.
Martijn Pieters
Ich habe das einmal versucht und die beste Antwort, die ich gefunden habe, war: stackoverflow.com/a/3113913/1330293
elyase
@elyase verknüpfte Antwort ist eine elegante Lösung
Pykler
Siehe diese Antwort .
Martineau

Antworten:

183

Versuchen Sie diesen Kontextmanager:

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

Verwendung:

with Capturing() as output:
    do_something(my_object)

output ist jetzt eine Liste mit den vom Funktionsaufruf gedruckten Zeilen.

Erweiterte Verwendung:

Was möglicherweise nicht offensichtlich ist, ist, dass dies mehr als einmal durchgeführt und die Ergebnisse verkettet werden können:

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)

Ausgabe:

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

Update : Sie fügten hinzu , redirect_stdout()um contextlibin Python 3.4 (zusammen mit redirect_stderr()). Sie könnten damit io.StringIOein ähnliches Ergebnis erzielen (obwohl Capturinges wohl bequemer ist, sowohl eine Liste als auch ein Kontextmanager zu sein).

irgendwie
quelle
Vielen Dank! Und danke, dass Sie den erweiterten Abschnitt hinzugefügt haben ... Ich habe ursprünglich eine Slice-Zuweisung verwendet, um den erfassten Text in die Liste aufzunehmen, dann habe ich mich in den Kopf gestoßen und .extend()stattdessen verwendet, damit er verkettet verwendet werden kann, genau wie Sie es bemerkt haben. :-)
bitte alle
PS Wenn es wiederholt verwendet werden soll, würde ich vorschlagen, self._stringio.truncate(0)nach dem self.extend()Aufruf der __exit__()Methode ein Element hinzuzufügen, um einen Teil des vom _stringioMitglied gespeicherten Speichers freizugeben .
Martineau
25
Tolle Antwort, danke. Verwenden Sie für Python 3 from io import StringIOanstelle der ersten Zeile im Kontextmanager.
Wtower
1
Ist das threadsicher? Was passiert, wenn ein anderer Thread / Aufruf print () verwendet, während do_something ausgeführt wird?
Derorrist
1
Diese Antwort funktioniert nicht für die Ausgabe von gemeinsam genutzten C-Bibliotheken. Sehen Sie stattdessen diese Antwort .
Craymichael
81

In Python> = 3.4 enthält contextlib einen redirect_stdoutDekorator. Es kann verwendet werden, um Ihre Frage wie folgt zu beantworten:

import io
from contextlib import redirect_stdout

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

Aus den Dokumenten :

Kontextmanager zum vorübergehenden Umleiten von sys.stdout in eine andere Datei oder ein dateiähnliches Objekt.

Dieses Tool bietet Flexibilität für vorhandene Funktionen oder Klassen, deren Ausgabe fest mit stdout verbunden ist.

Beispielsweise wird die Ausgabe von help () normalerweise an sys.stdout gesendet. Sie können diese Ausgabe in einer Zeichenfolge erfassen, indem Sie die Ausgabe auf ein io.StringIO-Objekt umleiten:

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

Um die Ausgabe von help () an eine Datei auf der Festplatte zu senden, leiten Sie die Ausgabe in eine reguläre Datei um:

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

So senden Sie die Ausgabe von help () an sys.stderr:

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

Beachten Sie, dass der globale Nebeneffekt auf sys.stdout bedeutet, dass dieser Kontextmanager nicht für die Verwendung in Bibliothekscode und den meisten Thread-Anwendungen geeignet ist. Es hat auch keine Auswirkung auf die Ausgabe von Unterprozessen. Es ist jedoch immer noch ein nützlicher Ansatz für viele Dienstprogramm-Skripte.

Dieser Kontextmanager ist wiedereintrittsfähig.

ForeverWintr
quelle
wenn versucht 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'). Es gibt mir einen Fehler:AssertionError: '' != 'Test debug message'
Eziz Durdyyev
was bedeutet, dass ich etwas falsch gemacht habe oder es konnte nicht stdout log fangen.
Eziz Durdyyev
@EzizDurdyyev, logger.debugschreibt standardmäßig nicht in stdout. Wenn Sie Ihren Protokollanruf durch ersetzen print(), sollte die Meldung angezeigt werden.
ForeverWintr
Ja, ich weiß, aber ich lasse es so schreiben : stream_handler = logging.StreamHandler(sys.stdout). Und fügen Sie diesen Handler meinem Logger hinzu. also sollte es an stdout schreiben und redirect_stdoutes fangen, oder?
Eziz Durdyyev
Ich vermute, das Problem liegt in der Art und Weise, wie Sie Ihren Logger konfiguriert haben. Ich würde überprüfen, ob es ohne redirect_stdout auf stdout gedruckt wird. Wenn dies der Fall ist, wird der Puffer möglicherweise erst geleert, wenn der Kontextmanager beendet wird.
ForeverWintr
0

Hier ist eine asynchrone Lösung mit 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

Anwendungsbeispiel:

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
quelle