c++ - 如何在 python3 中使用 ctypes 导入 o​​stringstream？

Question

我正在为自然语言编写分析器，我有一个用 swig 创建的 python 3 中的 c++ 代码包装器。我想使用一个函数，它是某种流编写器，它以 std::ostream & os 作为参数。所以我想如果我以某种方式导入 ostringstream(读作我应该在我的 ctypes.CDLL 中使用的 lib.so)它会起作用python代码然后将它传递给这个函数，以免调用它create_stream_writer(stream)，然后使用stream.str()来获取字符串。是否可以使用 ctypes 或任何其他库来执行此操作？我正在使用运行 Ubuntu 18.04、python3.6 的 docker 容器

我猜代码应该是这样的:

def analyse(text, config):
    reader = PlainTextReader.create_string_reader(text, config)
    stream = ctypes.ostringstream() # some magic hear
    writer = TokenWriter.create_stream_writer('plain', stream, reader.tagset())

    for sentence in sentences(reader):
        writer.write_sentence(sentence)
    return stream.str()

Answer 1

您可以做到这一点(并且对 Python 开发人员也有好处)。这个答案本质上是我旧答案的 Python 3 版本 on wrapping iostreams .

为了简化这里的事情，我使用了 boost 的 iostreams 库。如果你不能/不使用 boost，那么你可以从标准 C++ 库组件中编写所有这些，只是要冗长得多。

我的目标不仅仅是将 io.StringIO 映射到 std::stringstream，而是将任何“类似文件”的 Python 对象映射到任何 iostream。也就是说，我们使用目标是在 Python 对象上使用鸭子类型，以便在我们的 C++ 流对象需要时明智地调用 read() 和 write()。

%module test

%{
#include <boost/iostreams/stream.hpp>
#include <boost/iostreams/categories.hpp> 

// This is just a helper that we can use with boost iostreams to proxy everything to a Python object    
struct python_stream_device {
  typedef char char_type;
  typedef boost::iostreams::bidirectional_device_tag category;

  std::streamsize read(char* s, std::streamsize n) {
    PyObject *p = PyObject_CallMethod(o, "read", "l", static_cast<long int>(n));
    if (PyErr_Occurred()) {
      // TODO: throw a C++ exception to back out of wherever we are and then re-throw the Python one...
      assert(false);
    }
    assert(p);
    char *ptr = nullptr;
    Py_ssize_t len = 0;
    PyObject *str = PyUnicode_AsUTF8String(p);
    PyBytes_AsStringAndSize(str, &ptr, &len);
    if (PyErr_Occurred()) {
      assert(false); // Let's just pretend this is error handlng...
    }

    memcpy(s, ptr, len);
    Py_DECREF(str);
    Py_DECREF(p);
    return len;
  }

  std::streamsize write(const char* s, std::streamsize n) {
    PyObject *ret = PyObject_CallMethod(o, "write", "s#", s, static_cast<Py_ssize_t>(n));
    if (PyErr_Occurred()) {
      // See above
      assert(false);
    }
    std::streamsize r = PyLong_AsSsize_t(ret);
    Py_DECREF(ret);
    return r;
  }

  // Using this means we can rely on the default synthesised operator= + copy ctor etc. and saves us some code.
  swig::SwigPtr_PyObject o;

  python_stream_device(PyObject *o) : o(o) {}
};

typedef boost::iostreams::stream<python_stream_device> python_stream;

%}

// Here is the stuff that wraps it neatly
%typemap(in) std::iostream& (python_stream tmp) {
  // Writing the typemap this way lets us get RAII semantics despite the goto in the SWIG macros in the simplest way
  tmp.open(python_stream_device($input));  
  $1 = &tmp;
}

// We can just use the same typemaps for other cases too:
%apply std::iostream& { std::istream&, std::ostream& }; 


// Below is just for testing:    
%{
#include <iostream>
%}

%inline %{
  // This is the function you want to call
  void fun1(std::ostream& out) {
    assert(out.good());
    out << "Hello world, from C++";
    assert(out.good());
  }

  // This one is here for completeness because once you've got this far you may as well support this too.
  void fun2(std::istream& in) {
    std::string tmp;
    //in >> tmp;
    std::getline(in, tmp);
    assert(in.good());
    std::cout << "fun2 got: " << tmp << std::endl;
  }
%}

这就足够了，您可以像这样使用一些 Python:

import io
import test

i=io.StringIO()
test.fun1(i)
print('After fun1: %s' % i.getvalue())

i=io.StringIO('hello world, from Python!\n')
test.fun2(i)