Bagaimana cara menangkap output stdout dari pemanggilan fungsi Python?

Saya menggunakan pustaka Python yang melakukan sesuatu pada suatu objek

do_something(my_object)

dan mengubahnya. Saat melakukannya, ia mencetak beberapa statistik ke stdout, dan saya ingin memahami informasi ini. Solusi yang tepat adalah mengubah do_something()untuk mengembalikan informasi yang relevan,

out = do_something(my_object)

tetapi akan memakan waktu lama sebelum developer membahas do_something()masalah ini. Sebagai solusinya, saya berpikir untuk mengurai apa pun yang do_something()ditulis ke stdout.

Bagaimana cara menangkap output stdout antara dua titik dalam kode, misalnya,

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

?

Coba pengelola konteks ini:

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

Pemakaian:

with Capturing() as output:
    do_something(my_object)

output sekarang menjadi daftar yang berisi garis-garis yang dicetak oleh pemanggilan fungsi.

Penggunaan lanjutan:

Apa yang mungkin tidak jelas adalah bahwa ini dapat dilakukan lebih dari sekali dan hasilnya digabungkan:

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)

Keluaran:

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

Pembaruan : Mereka ditambahkan redirect_stdout()ke contextlibdalam Python 3.4 (bersama dengan redirect_stderr()). Jadi Anda bisa menggunakannya io.StringIOuntuk mencapai hasil yang serupa (meskipun Capturingmenjadi daftar serta pengelola konteks bisa dibilang lebih nyaman).

Dalam python> = 3.4, contextlib berisi redirect_stdoutdekorator. Ini dapat digunakan untuk menjawab pertanyaan Anda seperti:

import io
from contextlib import redirect_stdout

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

Dari dokumen :

Pengelola konteks untuk mengalihkan sementara sys.stdout ke file lain atau objek mirip file.

Alat ini menambahkan fleksibilitas ke fungsi atau kelas yang sudah ada yang keluarannya sudah terpasang ke stdout.

Misalnya, output help () biasanya dikirim ke sys.stdout. Anda dapat menangkap output itu dalam sebuah string dengan mengarahkan output ke objek io.StringIO:

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

Untuk mengirim output help () ke file di disk, alihkan output ke file biasa:

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

Untuk mengirim keluaran help () ke sys.stderr:

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

Perhatikan bahwa efek samping global pada sys.stdout berarti bahwa pengelola konteks ini tidak cocok untuk digunakan dalam kode perpustakaan dan sebagian besar aplikasi berulir. Ini juga tidak berpengaruh pada keluaran subproses. Namun, ini masih merupakan pendekatan yang berguna untuk banyak skrip utilitas.

Manajer konteks ini adalah reentrant.

Berikut adalah solusi async menggunakan pipa file.

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

Contoh penggunaan:

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