Jak przechwycić wyjście standardowego wyjścia z wywołania funkcji Pythona?

112

Używam biblioteki Python, która robi coś z obiektem

do_something(my_object)

i zmienia to. Robiąc to, wypisuje statystyki na standardowe wyjście i chciałbym mieć kontrolę nad tymi informacjami. Właściwym rozwiązaniem byłaby zmiana do_something()i zwrócenie odpowiednich informacji,

out = do_something(my_object)

ale minie trochę czasu, zanim twórcy zabiorą do_something()się do tego wydania. Aby obejść ten problem, pomyślałem o parsowaniu wszystkiego, co do_something()zapisuje na standardowe wyjście.

Jak mogę przechwycić wyjście standardowego wyjścia między dwoma punktami w kodzie, np.

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

?

Nico Schlömer
źródło
2
możliwy duplikat przechwytywania wyników dis.dis
Martijn Pieters
Moja odpowiedź w powiązanym pytaniu ma również zastosowanie tutaj.
Martijn Pieters
Próbowałem to zrobić raz i najlepsza odpowiedź, jaką znalazłem, brzmiała: stackoverflow.com/a/3113913/1330293
elyase
@elyase linked answer to eleganckie rozwiązanie
Pykler
Zobacz tę odpowiedź .
martineau

Odpowiedzi:

185

Wypróbuj tego menedżera kontekstu:

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

Stosowanie:

with Capturing() as output:
    do_something(my_object)

output jest teraz listą zawierającą wiersze wypisane przez wywołanie funkcji.

Zaawansowane użycie:

Co może nie być oczywiste, to fakt, że można to zrobić więcej niż raz, a wyniki są połączone:

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)

Wynik:

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

Aktualizacja : Dodali redirect_stdout()do contextlibw Pythonie 3.4 (wraz z redirect_stderr()). Możesz więc użyć io.StringIOtego do osiągnięcia podobnego wyniku (chociaż Capturingbycie listą i menedżerem kontekstu jest prawdopodobnie wygodniejsze).

kindall
źródło
Dzięki! Dziękuję za dodanie sekcji zaawansowanej ... Pierwotnie użyłem przypisania do wycinka, aby wkleić przechwycony tekst do listy, a następnie uderzyłem się w głowę i .extend()zamiast tego użyłem, aby można było go używać łącznie, tak jak zauważyłeś. :-)
kindall
PS Jeśli ma być używany wielokrotnie, sugerowałbym dodanie self._stringio.truncate(0)po self.extend()wywołaniu __exit__()metody, aby zwolnić część pamięci posiadanej przez _stringioczłonka.
martineau
25
Świetna odpowiedź, dzięki. W przypadku Pythona 3 użyj from io import StringIOzamiast pierwszego wiersza w menedżerze kontekstu.
Wtower
1
Czy to jest bezpieczne dla wątków? Co się stanie, jeśli jakiś inny wątek / wywołanie użyje print () podczas działania do_something?
Derorrist,
1
Ta odpowiedź nie będzie działać dla danych wyjściowych z bibliotek współdzielonych C. Zamiast tego zobacz tę odpowiedź .
craymichael
82

W pythonie> = 3.4, contextlib zawiera redirect_stdoutdekorator. Można go użyć, aby odpowiedzieć na twoje pytanie w następujący sposób:

import io
from contextlib import redirect_stdout

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

Z dokumentów :

Menedżer kontekstu do tymczasowego przekierowania sys.stdout do innego pliku lub obiektu podobnego do pliku.

To narzędzie zwiększa elastyczność istniejących funkcji lub klas, których dane wyjściowe są podłączone na stałe do standardowego wyjścia.

Na przykład, wyjście help () jest normalnie wysyłane do sys.stdout. Możesz przechwycić te dane wyjściowe w postaci ciągu, przekierowując dane wyjściowe do obiektu io.StringIO:

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

Aby wysłać wynik help () do pliku na dysku, przekieruj wynik do zwykłego pliku:

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

Aby wysłać wyjście help () do sys.stderr:

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

Zauważ, że globalny efekt uboczny na sys.stdout oznacza, że ​​ten menedżer kontekstu nie jest odpowiedni do użycia w kodzie biblioteki i większości aplikacji wielowątkowych. Nie ma również wpływu na wynik podprocesów. Jednak jest to nadal przydatne podejście w przypadku wielu skryptów narzędziowych.

Ten menedżer kontekstu jest ponownie wprowadzany.

ForeverWintr
źródło
kiedy próbowałem 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'). Daje mi błąd:AssertionError: '' != 'Test debug message'
Eziz Durdyyev
co oznacza, że ​​zrobiłem coś źle lub nie mogę przechwycić dziennika standardowego.
Eziz Durdyyev
@EzizDurdyyev, domyślnie logger.debugnie zapisuje na standardowe wyjście. Jeśli zamienisz wywołanie dziennika na print(), powinieneś zobaczyć komunikat.
ForeverWintr
Tak, wiem, ale robią to pisać na standardowe wyjście tak: stream_handler = logging.StreamHandler(sys.stdout). I dodaj tę obsługę do mojego rejestratora. więc powinien napisać na stdout i redirect_stdoutpowinien to złapać, prawda?
Eziz Durdyyev
Podejrzewam, że problem dotyczy sposobu, w jaki skonfigurowałeś swój rejestrator. Chciałbym sprawdzić, czy drukuje na stdout bez redirect_stdout. Jeśli tak, być może bufor nie jest opróżniany do momentu zakończenia pracy menedżera kontekstu.
ForeverWintr,
0

Oto rozwiązanie asynchroniczne wykorzystujące potoki plików.

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

Przykładowe użycie:

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
źródło