python - 如何在 Python 中使用 ctypes 创建回调函数？

Question

我为 Corsair Utility Engine SDK 编写了一个包装器，但有一个功能我无法包装。这是一个接受回调函数的异步函数，但我似乎无法弄清楚如何给它回调。

函数如下所示:

bool CorsairSetLedsColorsAsync(int size, CorsairLedColor* ledsColors, void (*CallbackType)(void* context, bool result, CorsairError error), void *context)

这些是我到目前为止尝试过的实现方式:

def SetLedsColorsAsync(self, size, led_color, callback, context):
    c_func = CFUNCTYPE(c_void_p, c_void_p, c_bool, c_int)
    c_callback = c_func(callback)
    self._libcue.CorsairSetLedsColorsAsync.restype = c_bool
    self._libcue.CorsairSetLedsColorsAsync.argtypes = [c_int, POINTER(CorsairLedColor), c_void_p, c_void_p]
    return self._libcue.CorsairSetLedsColorsAsync(size, led_color, c_callback, context)

还有

def SetLedsColorsAsync(self, size, led_color, callback, context):
    c_func = CFUNCTYPE(None, c_void_p, c_bool, c_int)
    c_callback = c_func(callback)
    self._libcue.CorsairSetLedsColorsAsync.restype = c_bool
    self._libcue.CorsairSetLedsColorsAsync.argtypes = [c_int, POINTER(CorsairLedColor), c_func, c_void_p]
    return self._libcue.CorsairSetLedsColorsAsync(size, led_color, c_callback, context)

我正在测试的代码是

from cue_sdk import *
import time


def test(context, result, error):
    print context, result, error
    return 0

Corsair = CUE("CUESDK.x64_2013.dll")
Corsair.RequestControl(CAM_ExclusiveLightingControl)
Corsair.SetLedsColorsAsync(1, CorsairLedColor(CLK_H, 255, 255, 255), test, 1)

while True:
time.sleep(1)

time.sleep() 只是为了让程序保持活跃。

运行它时，它在 Windows 上崩溃并显示错误代码 3221225477 (STATUS_ACCESS_VIOLATION)。

If you need to see the actual wrapper, you can find it here .

Answer 1

直到eryksun提醒我，我完全忘记了垃圾回收的问题。他建议我创建一个永久回调处理程序，用于存储所有回调并在必要时调用 + 弹出它们。这就是我所做的。

函数原型(prototype)如下所示:

self._callback_type = CFUNCTYPE(None, c_void_p, c_bool, c_int)
self._callback = self._callback_type(self._callback_handler)
self._libcue.CorsairSetLedsColorsAsync.restype = c_bool
self._libcue.CorsairSetLedsColorsAsync.argtypes = [c_int, POINTER(CorsairLedColor), self._callback_type, c_void_p]

_callback_handler 函数如下所示:

def _callback_handler(self, context, result, error):
    if context is not None and context in self._callbacks:
        self._callbacks.pop(context)(context, result, error)

实际的功能是这样的。

def SetLedsColorsAsync(self, size, led_color, callback=None, context=None):
    if callback:
        if context is None:
            context = id(callback)
        self._callbacks[context] = callback
    return self._libcue.CorsairSetLedsColorsAsync(size, led_color, self._callback, context)

_callback_type 是包装永久回调 (_callback_handler) 的实际 CFUNCTYPE，它是原型(prototype)的 argtype 之一。当 SetLedsColorsAsync 被调用时，callback 参数被放入一个字典中(以函数的上下文或 ID 为键)。不是将回调提供给函数，而是传递永久回调。一旦调用永久回调，它将调用适当的函数并将其从字典中删除。

我用的测试:

#!python3
import time

from cue_sdk import *


def test(context, result, error):
    print(context, result, error)
    assert context == id(test)


Corsair = CUE("CUESDK.x64_2013.dll")
Corsair.RequestControl(CAM_ExclusiveLightingControl)
Corsair.SetLedsColorsAsync(1, CorsairLedColor(CLK_H, 255, 255, 255), test)

while True:
    time.sleep(1)

示例输出:

2969710418936 True 0

If I'm not making sense, the commit is here.