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

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

?

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).

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.

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