Как получить вывод stdout из вызова функции Python?

Я использую библиотеку Python, которая что-то делает с объектом

do_something(my_object)

и меняет это. При этом он выводит некоторую статистику на стандартный вывод, и я хотел бы получить эту информацию. Правильным решением было бы изменить do_something(), чтобы он возвращал соответствующую информацию,

out = do_something(my_object)

но пройдет некоторое время, прежде чем разработчики do_something() займутся этой проблемой. В качестве обходного пути я подумал о разборе всего, что do_something() записывает в стандартный вывод.

Как я могу захватить вывод stdout между двумя точками кода, например,

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

?


person Nico Schlömer    schedule 15.05.2013    source источник
comment
возможный дубликат захвата результатов dis.dis   -  person Martijn Pieters    schedule 15.05.2013
comment
Мой ответ в связанном вопросе применим и здесь.   -  person Martijn Pieters    schedule 15.05.2013
comment
Я попытался сделать это один раз, и лучший ответ, который я нашел, был: stackoverflow.com/a/3113913/1330293   -  person elyase    schedule 15.05.2013
comment
Связанный ответ @elyase - элегантное решение   -  person Pykler    schedule 15.05.2013
comment
См. Этот ответ.   -  person martineau    schedule 15.05.2013


Ответы (3)


Попробуйте этот диспетчер контекста:

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

Использование:

with Capturing() as output:
    do_something(my_object)

output теперь является списком, содержащим строки, напечатанные вызовом функции.

Расширенное использование:

Что может быть неочевидным, так это то, что это можно делать более одного раза, а результаты объединяются:

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)

Вывод:

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

Обновление: они добавили redirect_stdout() в _7 _ в Python 3.4 (вместе с redirect_stderr()). Таким образом, вы можете использовать io.StringIO с этим для достижения аналогичного результата (хотя Capturing, являющийся списком, а также диспетчером контекста, возможно, более удобен).

person kindall    schedule 15.05.2013
comment
Спасибо! И спасибо за добавление расширенного раздела ... Изначально я использовал назначение срезов, чтобы вставить захваченный текст в список, затем я ударил себя по голове и вместо этого использовал .extend(), чтобы его можно было использовать конкатентивно, как вы заметили. :-) - person kindall; 16.05.2013
comment
P.S. Если он будет использоваться повторно, я бы предложил добавить self._stringio.truncate(0) после вызова self.extend() в метод __exit__(), чтобы освободить часть памяти, удерживаемой членом _stringio. - person martineau; 16.05.2013
comment
@martineau Новый экземпляр StringIO будет создаваться каждый раз, когда менеджер контекста все равно будет использоваться, а старый в это время уйдет, но вы можете установить self._stringio на None (или del его) в __exit__(), если временное использование памяти является проблема. - person kindall; 16.05.2013
comment
Ах да, верно ... моя ошибка. Предыдущее содержимое мешало чему-то экспериментальному, что я делал с первым экземпляром до того, как был создан второй, что (неправильно) навело меня на мысль, что результаты будут накапливаться. Тот факт, что у них обоих одинаковые имена, усугублял иллюзию. ;-) - person martineau; 16.05.2013
comment
@kindall Очень умное решение! Работает как шарм, даже если в моем случае мне также пришлось перенаправить stderr, поскольку я хотел зафиксировать сообщение об ошибке argsparse. - person Joël; 26.03.2014
comment
Отличный ответ, спасибо. Для Python 3 используйте from io import StringIO вместо первой строки в диспетчере контекста. - person Wtower; 02.11.2015
comment
Это потокобезопасно? Что произойдет, если какой-то другой поток / вызов использует print () во время выполнения do_something? - person Derorrist; 17.11.2015
comment
Отвечаю сам .. Не потокобезопасный. Если другой поток вызывает печать, он попадет в «вывод». - person Derorrist; 17.11.2015
comment
Да, не потокобезопасный, хотя печать в стандартный вывод в любом случае может быть немного шаткой с потоками. - person kindall; 17.11.2015
comment
К сожалению, это не работает с модулями C (такими как fontforge) - person Clément; 05.12.2015
comment
@kindall, это отличная идея. Спасибо тебе за это! - person kennes; 16.06.2016
comment
Требуется ли инструкция del? Я бы подумал, что после того, как __exit__ завершится, все выйдет за рамки и будет сразу же собрано мусором. - person user1071847; 18.10.2017
comment
Почему это выходит за рамки? - person kindall; 18.10.2017
comment
Этот ответ не будет работать для вывода из общих библиотек C, см. этот ответ. - person craymichael; 14.08.2018
comment
Фантастика, именно то, что мне нужно для моего случая использования. - person Btibert3; 18.10.2020
comment
можно ли отправить output другому объекту в реальном времени? - person Pablo; 28.04.2021

В python> = 3.4 contextlib содержит декоратор redirect_stdout. Его можно использовать, чтобы ответить на ваш вопрос так:

import io
from contextlib import redirect_stdout

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

Из документов:

Диспетчер контекста для временного перенаправления sys.stdout в другой файл или файловый объект.

Этот инструмент добавляет гибкости существующим функциям или классам, вывод которых жестко привязан к stdout.

Например, вывод help () обычно отправляется в sys.stdout. Вы можете записать этот вывод в строку, перенаправив вывод на объект io.StringIO:

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

Чтобы отправить вывод help () в файл на диске, перенаправьте вывод в обычный файл:

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

Чтобы отправить вывод help () в sys.stderr:

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

Обратите внимание, что глобальный побочный эффект для sys.stdout означает, что этот диспетчер контекста не подходит для использования в коде библиотеки и в большинстве многопоточных приложений. Это также не влияет на вывод подпроцессов. Тем не менее, это все еще полезный подход для многих служебных скриптов.

Этот диспетчер контекста является реентерабельным.

person ForeverWintr    schedule 05.12.2016
comment
когда пробовал 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'). Выдает ошибку: AssertionError: '' != 'Test debug message' - person Eziz Durdyyev; 12.12.2019
comment
Это означает, что я сделал что-то не так, или он не смог поймать журнал stdout. - person Eziz Durdyyev; 12.12.2019
comment
@EzizDurdyyev, logger.debug по умолчанию не пишет в stdout. Если вы замените вызов журнала на print(), вы должны увидеть это сообщение. - person ForeverWintr; 12.12.2019
comment
Да, я знаю, но я заставляю его писать в stdout вот так: stream_handler = logging.StreamHandler(sys.stdout). И добавьте этот обработчик в мой регистратор. так что он должен писать в stdout и redirect_stdout должен его поймать, верно? - person Eziz Durdyyev; 13.12.2019
comment
Я подозреваю, что проблема в том, как вы настроили регистратор. Я бы проверил, что он выводится на стандартный вывод без redirect_stdout. Если это так, возможно, буфер не очищается до выхода из диспетчера контекста. - person ForeverWintr; 13.12.2019
comment
Когда ваш код не перенаправляет ни std_out, ни std_err (я имею в виду: у меня все еще есть вывод на консоль, и на диск ничего не сохраняется), что это означает? - person Antonio Sesto; 14.05.2021

Вот асинхронное решение с использованием файловых каналов.

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

Пример использования:

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()
person miXo    schedule 15.06.2020