In python, how to capture the stdout from a c++ shared library to a variable

Question

For some other reasons, the c++ shared library I used outputs some texts to standard output. In python, I want to capture the output and save to a variable. There are many similar questions about redirect the stdout, but not work in my code.

Example: Suppressing output of module calling outside library

1 import sys
2 import cStringIO
3 save_stdout = sys.stdout
4 sys.stdout = cStringIO.StringIO()
5 func()
6 sys.stdout = save_stdout

In line 5, func() will call the shared library, the texts generated by shared library still output to console! If change func() to print "hello", it works!

My problem is:

1. how to capture stdout of the c++ shared library to a variable?
2. Why using StringIO, can't capture the outputs from shared library?

Answer 1

Thanks to the nice answer by Adam, I was able to get this working. His solution didn't quite work for my case, since I needed to capture text, restore, and capture text again many times, so I had to make some pretty big changes. Also, I wanted to get this to work for sys.stderr as well (with the potential for other streams).

So, here is the solution I ended up using (with or without threading):

Code

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

Usage

with sys.stdout, the default:

out = OutputGrabber()
out.start()
library.method(*args) # Call your code here
out.stop()
# Compare the output to the expected value:
# comparisonMethod(out.capturedtext, expectedtext)

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

in a with block:

out = OutputGrabber()
with out:
    library.method(*args) # Call your code here
# Compare the output to the expected value:
# comparisonMethod(out.capturedtext, expectedtext)

Tested on Windows 7 with Python 2.7.6 and Ubuntu 12.04 with Python 2.7.6.

To work in Python 3, change char = os.read(self.pipe_out,1)
to char = os.read(self.pipe_out,1).decode(self.origstream.encoding).

Answer 2

Python's sys.stdout object is simply a Python wrapper on top of the usual stdout file descriptor—changing it only affects the Python process, not the underlying file descriptor. Any non-Python code, whether it be another executable which was exec'ed or a C shared library which was loaded, won't understand that and will continue using the ordinary file descriptors for I/O.

So, in order for the shared library to output to a different location, you need to change the underlying file descriptor by opening a new file descriptor and then replacing stdout using os.dup2(). You could use a temporary file for the output, but it's a better idea to use a pipe created with os.pipe(). However, this has the danger for deadlock, if nothing is reading the pipe, so in order to prevent that we can use another thread to drain the pipe.

Below is a full working example which does not use temporary files and which is not susceptible to deadlock (tested on Mac OS X).

C shared library code:

// test.c
#include <stdio.h>

void hello(void)
{
  printf("Hello, world!\n");
}

Compiled as:

$ clang test.c -shared -fPIC -o libtest.dylib

Python driver:

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

Answer 3

More simply, the Py library has a StdCaptureFD that catches streams file descriptors, which allows to catch output from C/C++ extension modules (in a similar mechanism than the other answers). Note that the library is said to be in maintenance only.

>>> import py, sys
>>> capture = py.io.StdCaptureFD(out=False, in_=False)
>>> sys.stderr.write("world")
>>> out,err = capture.reset()
>>> err
'world'

Another solution is worth noting that if you're in a pytest test fixture, you can directly use capfd, see these docs.

While the other answers may also work well, I ran into an error when using their code within PyCharm IDE (io.UnsupportedOperation: fileno), while StdCaptureFD worked fine.