python - 在 python 中使用 swig 包装的 typedef 结构和枚举来转换字符串/缓冲区数据

Question

我有一些在嵌入式系统上运行的 C 代码，生成一个数据流，我的 python 代码将在蓝牙/USB 线的另一端读取该数据流。流的协议(protocol)仍在大力开发中，并且经常更改，但在单个 .h 文件中定义。我想使用 SWIG 来保持 Python 方面的最新状态，特别是通过提供对流数据布局(结构)的访问

这里是一个示例 .h 文件，描述了许多结构和一些常量(如#defines)，为了简洁起见，显然是整个协议(protocol)的一个非常小的子集。

//datalayouts.h

#ifdef SWIG
#define __attribute__(x)
#endif

#define TOKEN_TYPE_SYNC_VALUE 1
#define TOKEN_TYPE_DELTA 2

typedef struct __attribute__((packed))
{
    uint8_t token_type;
    uint32_t timestamp;
    uint32_t value;
} struct_token_type_sync_value;

typedef struct __attribute__((packed))
{
    uint8_t token_type;
    int16_t delta;
} struct_token_type_delta;

再加上这个就是基本的接口(interface)文件

%module datalayouts
%{
#include "datalayouts.h"
%}
%include "datalayouts.h"

一切都编译和导入得很好。在Python中，我可以创建一个token_type_sync_value类型的变量，但我想要做的是转换我从流中读取的数据的一部分(作为字符串)，以强加正确的结构它。

例如:

>>> from datalayouts token_type_sync_value
>>> data = stream.read() #returns 100+ bytes
>>> if ord(data[0]) == TOKEN_TYPE_SYNC_VALUE:
...     #here I want to access data[0:9] as a token_type_sync_value

这可能吗？如果可以的话怎么办？

Answer 1

您可以使用 SWIG 来完成此操作，最简单的解决方案是使用 %extend 从 Python 中提供一个额外的构造函数，该构造函数采用 PyObect 用作缓冲区:

%module test

%include <stdint.i>

%inline %{
#ifdef SWIG
#define __attribute__(x)
#endif

#define TOKEN_TYPE_SYNC_VALUE 1
#define TOKEN_TYPE_DELTA 2

typedef struct __attribute__((packed))
{
    uint8_t token_type;
    int16_t delta;
} struct_token_type_delta;
%}

%extend struct_token_type_delta {
  struct_token_type_delta(PyObject *in) {
    assert(PyObject_CheckBuffer(in));
    Py_buffer view;
    const int ret = PyObject_GetBuffer(in, &view, PyBUF_SIMPLE);
    assert(0==ret);
    assert(view.len >= sizeof(struct_token_type_delta));
    struct_token_type_delta *result = new struct_token_type_delta(*static_cast<const struct_token_type_delta*>(view.buf));
    PyBuffer_Release(&view); // Note you could/should retain view.obj for the life of this object to prevent use after free
    return result;
  }
}

您需要对要从缓冲区构造的每种类型执行此操作，但每个构造函数的实际代码保持不变，因此可以包装为宏(使用 %define) 非常简单。您还需要采取一些措施，通过保留对底层缓冲区的引用更长时间来防止释放后使用错误。

<小时/>

就我个人而言，如果是我这样做，我会寻找不同的解决方案，因为有更好的方法可以获得相同的结果，并且编写创建和维护类似 POD/bean 的对象的代码在任何语言中都是乏味且乏味的更不用说2个或更多了。假设 protbuf 太重量级，无法在嵌入式系统中使用，我希望反向解决这个问题，使用 Python 的 ctypes，然后让 Python 代码也为 C 构建工具生成 header 。所以类似:

import ctypes

class ProtocolStructure(type(ctypes.Structure)):
  def __str__(self):
    s='''
typedef struct __attribute__((packed)) {
\t%s
}'''
    return s % '\n\t'.join(('%s %s;' % (ty.__name__[2:], name) for name,ty in self._fields_))

class struct_token_type_delta(ctypes.Structure, metaclass=ProtocolStructure):
  _fields_ = (('token_type', ctypes.c_uint8),
              ('delta', ctypes.c_int16))

if __name__ == '__main__':
  # when this file is run instead of imported print the header file to stdout

  h='''
#ifndef PROTO_H
#define PROTO_H
%s
#endif
'''

  print(h % ';\n'.join('%s %s;\n' % (ty, name)  for name,ty in globals().items() if issubclass(type(ty), ProtocolStructure)))

然后你可以写:

import proto
proto.struct_token_type_delta.from_buffer(bytearray(b'\xff\x11\x22'))