在python中，如何将c++共享库的stdout捕获到变量中

Question

因为一些其他原因，我用的C++共享库会向标准输出打印一些文本。在Python中，我想把这些输出捕获并保存到一个变量里。网上有很多关于如何重定向标准输出的类似问题，但在我的代码中都不管用。

举个例子：抑制外部库调用模块的输出

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()改成打印“hello”，那就能正常工作了！

我的问题是：

1. 如何把C++共享库的标准输出捕获到一个变量里？
2. 为什么使用StringIO却无法捕获共享库的输出？

Answer 1

如果你是从谷歌过来的，想知道怎么抑制共享库（dll）的错误输出和标准输出，跟我一样，我在这里分享一个简单的上下文管理器，灵感来自于Adam的回答：

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

使用方法（根据Adam的例子进行了调整）：

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

Answer 2

谢谢你，Devan！

你的代码对我帮助很大，但我在使用时遇到了一些问题，想在这里分享一下：

首先，想要强制停止捕获的那一行代码

self.origstream.write(self.escape_char)

没有起作用。我把它注释掉了，并确保我的输出字符串里包含了转义字符，否则那行代码

data = os.read(self.pipe_out, 1)  # Read One Byte Only

在循环里会一直等待下去。

还有一点是关于使用方式。确保OutputGrabber类的对象是一个局部变量。如果你使用全局对象或者类属性（比如self.out = OutputGrabber()），在重新创建它的时候会遇到麻烦。

就这些，再次感谢你！

Answer 3

简单来说，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则没有问题。

Answer 4

感谢Adam的精彩回答，我终于搞定了这个问题。不过他的解决方案对我来说不太适用，因为我需要多次捕获文本、恢复文本，然后再捕获文本，所以我做了一些比较大的改动。此外，我还想让这个方案能在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)。

Answer 5

Python中的 sys.stdout 对象其实就是一个包装，它在普通的标准输出（stdout）文件描述符上面加了一层。改变这个对象只会影响Python程序本身，而不会影响底层的文件描述符。也就是说，其他非Python的代码，比如被 exec 执行的其他程序，或者加载的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).decode()
        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