Reputation: 3932
I am working with a shared library that is being called through the ctypes module. I would like to redirect the stdout associated with this module to a variable or a file that I can access in my program. However ctypes uses a separate stdout from sys.stdout.
I'll demonstrate the problem I am having with libc. If anyone is copying and pasting the code they might have to change the filename on line 2.
import ctypes
libc = ctypes.CDLL('libc.so.6')
from cStringIO import StringIO
import sys
oldStdOut = sys.stdout
sys.stdout = myStdOut = StringIO()
print 'This text gets captured by myStdOut'
libc.printf('This text fails to be captured by myStdOut\n')
sys.stdout = oldStdOut
myStdOut.getvalue()
Is there any way I can capture the stdout that is associated with the ctypes loaded shared library?
Upvotes: 10
Views: 5566
Reputation: 804
Late to the party but here is my approach, basically using the flag --capture=fd
will use a different file descriptor value than 1
which is the value for c-level stdout, that being said I came up with this code, the commented section is how I came up with the answer
import os
import sys
import pytest
import ctypes
libc = ctypes.CDLL(None)
c_stdout = ctypes.c_void_p.in_dll(libc, 'stdout')
c_stdout_fd = libc.fileno(c_stdout)
def test_one():
c_fd = os.memfd_create('c_fd')
# py_fd = sys.stdout.fileno()
# print (c_fd, py_fd, c_stdout_fd)
# assert py_fd == c_stdout_fd
try:
os.dup2(c_fd, c_stdout_fd)
libc.printf(b"Hello world")
libc.fflush(None)
print("hello world")
os.lseek(c_fd, 0, os.SEEK_SET)
output = os.read(c_fd, 1024)
assert output == "Hello world".encode("utf-8")
finally:
os.dup2(c_stdout_fd, 1)
os.close(c_fd)
I don't recommend using this code as it does not safely handle errors and it uses platform specifc approach os.memfd_create
.
Note you could replace os.memfd_create
for tempfile
Upvotes: 1
Reputation: 785
There is a Python project called Wurlitzer that very elegantly solves this problem. It's a work of art and deserves to be one of the top answers to this question.
https://github.com/minrk/wurlitzer
https://pypi.org/project/wurlitzer/
pip install wurlitzer
from wurlitzer import pipes
with pipes() as (out, err):
call_some_c_function()
stdout = out.read()
from io import StringIO
from wurlitzer import pipes, STDOUT
out = StringIO()
with pipes(stdout=out, stderr=STDOUT):
call_some_c_function()
stdout = out.getvalue()
from wurlitzer import sys_pipes
with sys_pipes():
call_some_c_function()
And the most magical part: it supports Jupyter:
%load_ext wurlitzer
Upvotes: 0
Reputation: 3083
If the data the native process writes are large (larger than pipe buffer), the native program would block until you make some space in the pipe by reading it.
The solution from lunixbochs, however, needs the native process to finish before it starts reading the pipe. I improved the solution so that it reads the pipe in parallel from a separate thread. This way you can capture output of any size.
This solution is also inspired by https://stackoverflow.com/a/16571630/1076564 and captures both stdout and stderr:
class CtypesStdoutCapture(object):
def __enter__(self):
self._pipe_out, self._pipe_in = os.pipe()
self._err_pipe_out, self._err_pipe_in = os.pipe()
self._stdout = os.dup(1)
self._stderr = os.dup(2)
self.text = ""
self.err = ""
# replace stdout with our write pipe
os.dup2(self._pipe_in, 1)
os.dup2(self._err_pipe_in, 2)
self._stop = False
self._read_thread = threading.Thread(target=self._read, args=["text", self._pipe_out])
self._read_err_thread = threading.Thread(target=self._read, args=["err", self._err_pipe_out])
self._read_thread.start()
self._read_err_thread.start()
return self
def __exit__(self, *args):
self._stop = True
self._read_thread.join()
self._read_err_thread.join()
# put stdout back in place
os.dup2(self._stdout, 1)
os.dup2(self._stderr, 2)
self.text += self.read_pipe(self._pipe_out)
self.err += self.read_pipe(self._err_pipe_out)
# check if we have more to read from the pipe
def more_data(self, pipe):
r, _, _ = select.select([pipe], [], [], 0)
return bool(r)
# read the whole pipe
def read_pipe(self, pipe):
out = ''
while self.more_data(pipe):
out += os.read(pipe, 1024)
return out
def _read(self, type, pipe):
while not self._stop:
setattr(self, type, getattr(self, type) + self.read_pipe(pipe))
sleep(0.001)
def __str__(self):
return self.text
# Usage:
with CtypesStdoutCapture as capture:
lib.native_fn()
print(capture.text)
print(capture.err)
Upvotes: 0
Reputation: 466
Simplest example, because this question in google top.
import os
from ctypes import CDLL
libc = CDLL(None)
stdout = os.dup(1)
silent = os.open(os.devnull, os.O_WRONLY)
os.dup2(silent, 1)
libc.printf(b"Hate this text")
os.dup2(stdout, 1)
Upvotes: 2
Reputation: 22425
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()
Upvotes: 7