在python中，如何将stdout从c ++共享库捕获到变量

问题描述 投票：17回答：5

由于某些其他原因，我使用的c ++共享库将一些文本输出到标准输出。在python中，我想捕获输出并保存到变量。关于重定向标准输出有许多类似的问题，但在我的代码中不起作用。

示例：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

在第5行，func（）将调用共享库，共享库生成的文本仍然输出到控制台！如果改变func（）打印“你好”，它的工作原理！

我的问题是：

  1. 如何将c ++共享库的stdout捕获到变量？
  2. 为什么使用StringIO，无法捕获共享库的输出？
python c++
5个回答
16
投票

Python的sys.stdout对象只是通常的stdout文件描述符之上的Python包装器 - 更改它只会影响Python进程，而不会影响底层文件描述符。任何非Python代码，无论是exec的另一个可执行文件还是加载的C共享库，都不会理解这一点，并将继续使用I / O的普通文件描述符。

因此，为了使共享库输出到不同的位置，您需要通过打开新的文件描述符然后使用os.dup2()替换stdout来更改基础文件描述符。您可以使用临时文件进行输出，但最好使用使用os.pipe()创建的管道。但是，如果没有任何东西正在读取管道，这就有死锁的危险，所以为了防止我们可以使用另一个线程来排空管道。

下面是一个完整的工作示例，它不使用临时文件，并且不容易死锁（在Mac OS X上测试）。

C共享库代码：

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

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

编译为：

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

Python驱动程序：

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
10
投票

感谢nice answer的Adam，我能够让这个工作。他的解决方案并不适合我的情况，因为我需要多次捕获文本，恢复和捕获文本，所以我不得不做一些相当大的改动。此外，我想让它也适用于sys.stderr（具有其他流的潜力）。

所以，这是我最终使用的解决方案（有或没有线程）：

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

使用sys.stdout，默认值：

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

使用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)

在with街区：

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

在使用Python 2.7.6的Windows 7和使用Python 2.7.6的Ubuntu 12.04上进行了测试。

要在Python 3中工作，请更改char = os.read(self.pipe_out,1) 到char = os.read(self.pipe_out,1).decode(self.origstream.encoding)。

2
投票

谢谢Devan！

你的代码给了我很多帮助，但是我在使用它时遇到了一些问题我想在这里分享一下：

出于任何原因，您要强制捕获的行停止

self.origstream.write(self.escape_char)

不起作用。我评论了它并确保我的stdout捕获字符串包含转义字符，否则为行

data = os.read(self.pipe_out, 1)  # Read One Byte Only

在while循环中等待永远。

另一件事是用法。确保OutputGrabber类的对象是局部变量。如果使用全局对象或类属性（例如self.out = OutputGrabber（）），则在重新创建时会遇到麻烦。

就这样。再次谢谢你！

1
投票

使用管道，即os.pipe。在调用你的图书馆之前你需要os.dup2

0
投票

从库代码中捕获stdout基本上是站不住脚的，因为这取决于你运行在一个。）你在shell和b上的环境中运行的代码。）没有其他内容进入你的stdout。虽然你可以在这些约束下使某些东西工作，但如果你打算在任何意义上部署这些代码，那么就没有办法合理地保证一致的良好行为。事实上，这个库代码以无法控制的方式打印到stdout是非常值得怀疑的。

这就是你不能做的事情。你可以做的是将任何打印调用包装到你可以在子进程中执行的内容中。使用Python的subprocess.check_output，您可以在程序中从该子进程获取stdout。缓慢，凌乱，有点生气，但另一方面，你正在使用的库将有用信息打印到stdout并且不会返回它...

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.