In python, how to capture the stdout from a c++ shared library to a variable
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).
Thank you Devan!
Your code helped me a lot, but I had some problems using it I want to share here:
For any reason the line where you want to force the capture to stop
self.origstream.write(self.escape_char)
does not work. I commented it out and made sure that my stdout captured string contains the escape character otherwise the line
data = os.read(self.pipe_out, 1) # Read One Byte Only
in the while loop waits forever.
One other thing is the usage. Make sure the object of the OutputGrabber class is a local variable. If you use a global object or class attribute (such as self.out = OutputGrabber()) you will run into trouble when recreating it.
That's all. Again thank you!
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
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.