使用ctypes模块从Python调用共享库并捕获打印输出

Question

我正在使用通过ctypes模块调用的共享库。我想将与该模块相关联的stdout重定向到一个变量或文件中，以便在我的程序中访问。然而，ctypes使用与sys.stdout分离的单独的stdout。

我将用libc演示我遇到的问题。如果有人复制并粘贴代码，则可能需要更改第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()

有没有办法捕获与ctypes加载的共享库相关联的stdout输出？

- ncRubert

Answer 1

我们可以使用os.dup2()和os.pipe()将整个stdout文件描述符（fd 1）替换为我们自己可以读取的管道。您也可以对stderr（fd 2）执行相同的操作。

本示例使用select.select()来查看管道（我们的虚假stdout）是否有数据等待写入，因此我们可以安全地打印它，而不会阻塞脚本的执行。

由于我们完全替换了该进程及其任何子进程的stdout文件描述符，因此该示例甚至可以捕获子进程的输出。

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

- lunixbochs

Answer 2

这是一个最简单的例子，因为这个问题在谷歌搜索排名靠前。

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)

- Atterratio

Answer 3

有一个名为Wurlitzer的Python项目非常优雅地解决了这个问题。它是一件艺术品，值得成为这个问题的最佳答案之一。

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

而最神奇的部分是：它支持Jupyter：

%load_ext wurlitzer

- Utkonos

---

这个模块在Windows上工作效果不好。 - Libin Wen

这个模块在Windows上运行效果不好。 - undefined

@LibinWen 很高兴知道这个情况。你是否已经将你在Windows上遇到的具体问题报告给了项目维护者，作为Github上的一个问题？我敢打赌他们会很乐意查看并修复你可能遇到的任何问题。 - Utkonos

@LibinWen 很高兴知道这个情况。你是否已经在Github上将你在Windows上遇到的具体问题报告给项目维护者作为一个问题？我敢打赌他们会很乐意查看并修复你可能遇到的任何问题。 - undefined

这是一个已知的问题。 - Libin Wen

Answer 4

如果本地进程写入的数据很大（大于管道缓冲区），则本地程序将阻塞，直到您通过读取管道来腾出一些空间。

然而，lunixbochs提供的解决方案需要本地进程完成后才开始读取管道。我改进了这个解决方案，使其可以从单独的线程并行读取管道。这样，您就可以捕获任意大小的输出。

此解决方案也受到https://dev59.com/Z2Qn5IYBdhLWcg3wxZri#16571630的启发，并捕获stdout和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)

- Martin Pecka