In Python, How to Capture the Stdout from a C++ Shared Library to a Variable

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

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

Capturing print output from shared library called from python with ctypes module

We can use os.dup2() and os.pipe() to replace the entire stdout file descriptor (fd 1) with a pipe we can read from ourselves. You can do the same thing with stderr (fd 2).

This example uses select.select() to see if the pipe (our fake stdout) has data waiting to be written, so we can print it safely without blocking execution of our script.

As we are completely replacing the stdout file descriptor for this process and any subprocesses, this example can even capture output from child processes.

import os, sys, select

# the pipe would fail for some reason if I didn't write to stdout at some point
# so I write a space, then backspace (will show as empty in a normal terminal)
sys.stdout.write(' \b')
pipe_out, pipe_in = os.pipe()
# save a copy of stdout
stdout = os.dup(1)
# replace stdout with our write pipe
os.dup2(pipe_in, 1)

# check if we have more to read from the pipe
def more_data():
        r, _, _ = select.select([pipe_out], [], [], 0)
        return bool(r)

# read the whole pipe
def read_pipe():
        out = ''
        while more_data():
                out += os.read(pipe_out, 1024)

        return out

# testing print methods
import ctypes
libc = ctypes.CDLL('libc.so.6')

print 'This text gets captured by myStdOut'
libc.printf('This text fails to be captured by myStdOut\n')

# put stdout back in place 
os.dup2(stdout, 1)
print 'Contents of our stdout pipe:'
print read_pipe()

How do I prevent a C shared library to print on stdout in python?

Based on @Yinon Ehrlich's answer. This variant tries to avoid leaking file descriptors:

import os
import sys
from contextlib import contextmanager

@contextmanager
def stdout_redirected(to=os.devnull):
    '''
    import os

    with stdout_redirected(to=filename):
        print("from Python")
        os.system("echo non-Python applications are also supported")
    '''
    fd = sys.stdout.fileno()

    ##### assert that Python and C stdio write using the same file descriptor
    ####assert libc.fileno(ctypes.c_void_p.in_dll(libc, "stdout")) == fd == 1

    def _redirect_stdout(to):
        sys.stdout.close() # + implicit flush()
        os.dup2(to.fileno(), fd) # fd writes to 'to' file
        sys.stdout = os.fdopen(fd, 'w') # Python writes to fd

    with os.fdopen(os.dup(fd), 'w') as old_stdout:
        with open(to, 'w') as file:
            _redirect_stdout(to=file)
        try:
            yield # allow code to be run with the redirected stdout
        finally:
            _redirect_stdout(to=old_stdout) # restore stdout.
                                            # buffering and flags such as
                                            # CLOEXEC may be different

Capture print output from a c-module in python

The subprocess solution found here https://stackoverflow.com/a/5136686/4270148, actually works!

import subprocess
proc = subprocess.Popen(["python", "-c",
    "cnf = [[1, -5, 4], [-1, 5, 3, 4], [-3, -4]];\
    import pycosat;\
    pycosat.solve(cnf,verbose=5);"],
    stdout=subprocess.PIPE)
out = proc.communicate()[0]

I don't like that the way the program is passed (as an eval-string), but at least it works.

How to capture stdout output from a Python function call?

Try this context manager:

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

Usage:

with Capturing() as output:
    do_something(my_object)

output is now a list containing the lines printed by the function call.

Advanced usage:

What may not be obvious is that this can be done more than once and the results concatenated:

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)

Output:

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

Update: They added redirect_stdout() to contextlib in Python 3.4 (along with redirect_stderr()). So you could use io.StringIO with that to achieve a similar result (though Capturing being a list as well as a context manager is arguably more convenient).

Pass Python's stdout to a C function to which write

Hope it'd help others.

import os
import sys
from tempfile import TemporaryFile
from io import TextIOWrapper, SEEK_SET
from contextlib import contextmanager
from ctypes import c_void_p

@contextmanager
def capture(stream, libc):
    osfd = sys.stdout.fileno()
    fd = os.dup(osfd)
    try:
        tfile = TemporaryFile(mode='w+b')
        redirect_stdout(tfile.fileno(), osfd, libc)
        yield
        redirect_stdout(fd, osfd, libc)
        tfile.flush()
        tfile.seek(0, SEEK_SET)
        stream.write(tfile.read())
    finally:
        tfile.close()
        os.close(fd)

def redirect_stdout(fd, osfd, libc):
    libc.fflush(c_void_p.in_dll(libc, 'stdout'))
    sys.stdout.close()
    os.dup2(fd, osfd)
    sys.stdout = TextIOWrapper(os.fdopen(osfd, 'wb'))

How I used it to test the output of the timer funciton:

from io import BytesIO
from ctypes import CDLL
from unittest import TestCase
from helpers import capture

class TomatTestCase(TestCase):
    def setUp(self):
        self.libc = CDLL('./tomat.so')
        self.maxDiff = None

    def test_prints_every_second(self):
        seconds = [
            '01:00', '01:01', '01:02', '01:03', '01:04', '01:05', '01:06',
            '01:07', '01:08', '01:09', '01:10', '01:11', '01:12', '01:13',
            '01:14', '01:15', '01:16', '01:17', '01:18', '01:19', '01:20',
            '01:21', '01:22', '01:23', '01:24', '01:25', '01:26', '01:27',
            '01:28', '01:29', '01:30', '01:31', '01:32', '01:33', '01:34',
            '01:35', '01:36', '01:37', '01:38', '01:39', '01:40', '01:41',
            '01:42', '01:43', '01:44', '01:45', '01:46', '01:47', '01:48',
            '01:49', '01:50', '01:51', '01:52', '01:53', '01:54', '01:55',
            '01:56', '01:57', '01:58', '01:59', '01:60']

        stream = BytesIO()
        with capture(stream, self.libc):
            self.libc.tomat(2)
        stream = stream.getvalue().split()
        output = [byte.decode() for byte in stream]

        self.assertListEqual(output, seconds)

for more information about how it works you can take a look at the Eli Bendersky's post: http://eli.thegreenplace.net/2015/redirecting-all-kinds-of-stdout-in-python/

Assign to variable exported from shared library via Python CTypes?

Your c_int does point at the value of test.c. You will want to write to the valueattribute of the c_int to write to the underlying memory that the c_int wraps. Otherwise you would simply be replacing one Python object with another in Python-land and not having any effect in C land. For example:

>>> import ctypes
>>> i = ctypes.c_int()
>>> i
c_int(0)
>>> i.value = 10
>>> i
c_int(10)

Proof of Concept

Below, I'll give the C file, how the C file should be compiled, and the python code to show that the above works. This is specifically for a Unix-like environment (including WSL). If you're using Windows directly, then compilation instructions and how to load the library will differ.

mylib.c

int value = 10;

int
get_value() {
        return value;
}

Compilation Instructions

gcc -Wall -Werror -fPIC -c mylib.c
gcc -Wall -Werror -fPIC -shared -o mylib.so mylib.o

main.py

import ctypes

mylib = ctypes.cdll.LoadLibrary('./mylib.so')

get_value = mylib.get_value
get_value.argtypes = []
get_value.returntype = int

library_value = ctypes.c_int.in_dll(mylib, 'value')

initial_value = get_value()

print(f'{initial_value=}')
print(f'{library_value=}')
print()

library_value.value = 123
assigned_value = get_value()

print(f'{initial_value=}')
print(f'{assigned_value=}')
print(f'{library_value=}')

output

initial_value=10
library_value=c_int(10)

initial_value=10
assigned_value=123
library_value=c_int(123)

Capture stdout from a script?

Setting stdout is a reasonable way to do it. Another is to run it as another process:

import subprocess

proc = subprocess.Popen(["python", "-c", "import writer; writer.write()"], stdout=subprocess.PIPE)
out = proc.communicate()[0]
print out.upper()

grep library output from within Python

I'm a little confused about exactly what your program is doing, but it sounds like you have a C library that writes to the C stdout (not the Python sys.stdout) and you want to capture this output and postprocess it, and you already have a Python binding for the C library, which you would prefer to use rather than a separate C program.

First off, you must use a child process to do this; nothing else will work reliably. This is because stdout is process-global, so there's no reliable way to capture only one thread's writes to stdout.

Second off, you can use subprocess.Popen, because you can re-invoke the current script using it! This is what the Python multiprocessing module does under the hood, and it's not terribly hard to do yourself. I would use a special, hidden command line argument to distinguish the child, like this:

import argparse
import subprocess
import sys

def subprocess_call_c_lib():
    import c_lib
    c_lib.do_stuff()

def invoke_c_lib():
    proc = subprocess.Popen([sys.executable, __file__,
                             "--internal-subprocess-call-c-lib"
                             # , ...
                             ],
                            stdin=subprocess.DEVNULL,
                            stdout=subprocess.PIPE)
    for line in proc.stdout:
        # filter output from the library here
        # to display to "screen", write to sys.stdout as usual

    if proc.wait():
        raise subprocess.CalledProcessError(proc.returncode, "c_lib")

def main():
    ap = argparse.Parser(...)
    ap.add_argument("--internal-subprocess-call-c-lib", action="store_true",
                    help=argparse.SUPPRESS)
    # ... more arguments ...

    args = ap.parse_args()
    if args.internal_subprocess_call_c_lib:
        subprocess_call_c_lib()
        sys.exit(0)

    # otherwise, proceed as before ...

main()

---

Related Topics