В Python, как захватить стандартный вывод из общей библиотеки C++ в переменную
По некоторым другим причинам, совместно используемая библиотека C++ выводила некоторые тексты в стандартный вывод. В Python я хочу захватить вывод и сохранить в переменную. Есть много похожих вопросов о перенаправлении stdout, но они не работают в моем коде.
Пример: подавление вывода вызова модуля вне библиотеки
1 import sys
2 import cStringIO
3 save_stdout = sys.stdout
4 sys.stdout = cStringIO.StringIO()
5 func()
6 sys.stdout = save_stdout
В строке 5 func() вызывает общую библиотеку, тексты, созданные общей библиотекой, все еще выводятся на консоль! Если изменить func(), чтобы напечатать "привет", это работает!
Моя проблема:
- как захватить стандартный вывод совместно используемой библиотеки C++ в переменную?
- Почему при использовании StringIO не удается захватить выходные данные из общей библиотеки?
5 ответов
Питона sys.stdout object - это просто оболочка Python поверх обычного дескриптора файла stdout - изменение его влияет только на процесс Python, а не на основной дескриптор файла. Любой не-Python-код, будь то другой исполняемый файл, который был exec 'ed или разделяемая библиотека C, которая была загружена, не поймет этого и продолжит использовать обычные файловые дескрипторы для ввода-вывода.
Таким образом, чтобы общая библиотека могла выводить данные в другое место, вам нужно изменить базовый файловый дескриптор, открыв новый файловый дескриптор, а затем заменив стандартный вывод с помощью os.dup2(), Вы можете использовать временный файл для вывода, но лучше использовать канал, созданный с os.pipe(), Тем не менее, это может привести к тупиковой ситуации, если труба не читает ничего, поэтому для предотвращения этого мы можем использовать другую нить для слива трубы.
Ниже приведен полный рабочий пример, в котором не используются временные файлы и который не подвержен взаимоблокировке (протестировано на Mac OS X).
C код общей библиотеки:
// test.c
#include <stdio.h>
void hello(void)
{
printf("Hello, world!\n");
}
Составлено как:
$ clang test.c -shared -fPIC -o libtest.dylib
Python драйвер:
import ctypes
import os
import sys
import threading
print 'Start'
liba = ctypes.cdll.LoadLibrary('libtest.dylib')
# Create pipe and dup2() the write end of it on top of stdout, saving a copy
# of the old stdout
stdout_fileno = sys.stdout.fileno()
stdout_save = os.dup(stdout_fileno)
stdout_pipe = os.pipe()
os.dup2(stdout_pipe[1], stdout_fileno)
os.close(stdout_pipe[1])
captured_stdout = ''
def drain_pipe():
global captured_stdout
while True:
data = os.read(stdout_pipe[0], 1024)
if not data:
break
captured_stdout += data
t = threading.Thread(target=drain_pipe)
t.start()
liba.hello() # Call into the shared library
# Close the write end of the pipe to unblock the reader thread and trigger it
# to exit
os.close(stdout_fileno)
t.join()
# Clean up the pipe and restore the original stdout
os.close(stdout_pipe[0])
os.dup2(stdout_save, stdout_fileno)
os.close(stdout_save)
print 'Captured stdout:\n%s' % captured_stdout
Благодаря хорошему ответу Adam Rosenfield, я смог заставить это работать. Его решение не вполне сработало для моего случая, так как мне нужно было многократно захватывать текст, восстанавливать и захватывать текст, поэтому мне пришлось внести довольно большие изменения. Кроме того, я хотел, чтобы это работало и для sys.stderr (с потенциалом для других потоков).
Итак, вот решение, которое я использовал (с потоками или без):
Код
import os
import sys
import threading
import time
class OutputGrabber(object):
"""
Class used to grab standard output or another stream.
"""
escape_char = "\b"
def __init__(self, stream=None, threaded=False):
self.origstream = stream
self.threaded = threaded
if self.origstream is None:
self.origstream = sys.stdout
self.origstreamfd = self.origstream.fileno()
self.capturedtext = ""
# Create a pipe so the stream can be captured:
self.pipe_out, self.pipe_in = os.pipe()
def __enter__(self):
self.start()
return self
def __exit__(self, type, value, traceback):
self.stop()
def start(self):
"""
Start capturing the stream data.
"""
self.capturedtext = ""
# Save a copy of the stream:
self.streamfd = os.dup(self.origstreamfd)
# Replace the original stream with our write pipe:
os.dup2(self.pipe_in, self.origstreamfd)
if self.threaded:
# Start thread that will read the stream:
self.workerThread = threading.Thread(target=self.readOutput)
self.workerThread.start()
# Make sure that the thread is running and os.read() has executed:
time.sleep(0.01)
def stop(self):
"""
Stop capturing the stream data and save the text in `capturedtext`.
"""
# Print the escape character to make the readOutput method stop:
self.origstream.write(self.escape_char)
# Flush the stream to make sure all our data goes in before
# the escape character:
self.origstream.flush()
if self.threaded:
# wait until the thread finishes so we are sure that
# we have until the last character:
self.workerThread.join()
else:
self.readOutput()
# Close the pipe:
os.close(self.pipe_in)
os.close(self.pipe_out)
# Restore the original stream:
os.dup2(self.streamfd, self.origstreamfd)
# Close the duplicate stream:
os.close(self.streamfd)
def readOutput(self):
"""
Read the stream data (one byte at a time)
and save the text in `capturedtext`.
"""
while True:
char = os.read(self.pipe_out, 1)
if not char or self.escape_char in char:
break
self.capturedtext += char
использование
с sys.stdout по умолчанию:
out = OutputGrabber()
out.start()
library.method(*args) # Call your code here
out.stop()
# Compare the output to the expected value:
# comparisonMethod(out.capturedtext, expectedtext)
с sys.stderr:
out = OutputGrabber(sys.stderr)
out.start()
library.method(*args) # Call your code here
out.stop()
# Compare the output to the expected value:
# comparisonMethod(out.capturedtext, expectedtext)
в with блок:
out = OutputGrabber()
with out:
library.method(*args) # Call your code here
# Compare the output to the expected value:
# comparisonMethod(out.capturedtext, expectedtext)
Протестировано на Windows 7 с Python 2.7.6 и Ubuntu 12.04 с Python 2.7.6.
Для работы в Python 3 измените char = os.read(self.pipe_out,1)
в char = os.read(self.pipe_out,1).decode(self.origstream.encoding),
Проще говоря, библиотека Py имеет StdCaptureFD который улавливает дескрипторы файлов потоков, что позволяет улавливать вывод модулей расширения C/C++ (в аналогичном механизме, чем другие ответы). Обратите внимание, что библиотека находится только на техническом обслуживании.
>>> import py, sys
>>> capture = py.io.StdCaptureFD(out=False, in_=False)
>>> sys.stderr.write("world")
>>> out,err = capture.reset()
>>> err
'world'
Еще одно решение, которое стоит отметить, если вы используете тестовую среду pytest, вы можете напрямую использоватьcapfdсм. эти документы.
Хотя другие ответы также могут работать хорошо, я столкнулся с ошибкой при использовании их кода в PyCharm IDE (io.UnsupportedOperation: fileno), пока StdCaptureFD работал нормально.
Для всех, кто пришел сюда из Google, чтобы узнать, как подавить вывод stderr/stdout из общей библиотеки (dll), как и я, я публикую следующий простой диспетчер контекста на основе ответа Адама:
class SuppressStream(object):
def __init__(self, stream=sys.stderr):
self.orig_stream_fileno = stream.fileno()
def __enter__(self):
self.orig_stream_dup = os.dup(self.orig_stream_fileno)
self.devnull = open(os.devnull, 'w')
os.dup2(self.devnull.fileno(), self.orig_stream_fileno)
def __exit__(self, type, value, traceback):
os.close(self.orig_stream_fileno)
os.dup2(self.orig_stream_dup, self.orig_stream_fileno)
os.close(self.orig_stream_dup)
self.devnull.close()
Использование (адаптированный пример Адама):
import ctypes
import sys
print('Start')
liba = ctypes.cdll.LoadLibrary('libtest.so')
with SuppressStream(sys.stdout):
liba.hello() # Call into the shared library
print('End')
Спасибо, Деван!
Ваш код мне очень помог, но у меня были некоторые проблемы с его использованием, которым я хочу поделиться здесь:
По любой причине линия, где вы хотите, чтобы захват остановить
self.origstream.write(self.escape_char)
не работает. Я закомментировал это и удостоверился, что моя захваченная строка stdout содержит escape-символ, иначе строка
data = os.read(self.pipe_out, 1) # Read One Byte Only
в то время как цикл ждет вечно.
Еще одна вещь, это использование. Убедитесь, что объект класса OutputGrabber является локальной переменной. Если вы используете глобальный объект или атрибут класса (например, self.out = OutputGrabber()), у вас возникнут проблемы при его воссоздании.
Это все. Еще раз спасибо!
В принципе невозможно захватить стандартный вывод из библиотечного кода, потому что это зависит от того, какой код выполняется в среде, в которой: а) вы находитесь в оболочке и б) нет другого контента, поступающего на ваш стандартный вывод. Хотя вы, вероятно, можете заставить что-то работать при этих ограничениях, если вы намерены развернуть этот код в каком-либо смысле, просто нет способа разумно гарантировать стабильно хорошее поведение. На самом деле, довольно сомнительно, что этот библиотечный код печатает на стандартный вывод таким образом, что его невозможно контролировать.
Так что это то, что вы не можете сделать. Что вы можете сделать, это обернуть любые вызовы печати в эту библиотеку внутри чего-то, что вы можете выполнить в подпроцессе. Использование Python subprocess.check_output затем вы можете получить стандартный вывод этого подпроцесса обратно в вашу программу. Медленно, грязно, вроде как круто, но, с другой стороны, используемая вами библиотека печатает полезную информацию в стандартный вывод и не возвращает ее так...