Events: implement manager, shared across all of our Python packages

Author: PMeira

cffi/dss_capi_custom.h

@@ -1,8 +1,3 @@
 extern "Python" int32_t dss_python_cb_plot(void* ctx, char* params);
 extern "Python" int32_t dss_python_cb_write(void* ctx, char* messageStr, int32_t messageType, int64_t messageSize, int32_t messageSubType);
-
-extern "Python" void dss_python_cb_init_controls(void* ctx);
-extern "Python" void dss_python_cb_check_controls(void* ctx);
-extern "Python" void dss_python_cb_step_controls(void* ctx);
-
 extern "Python" void altdss_python_util_callback(void* ctx, int32_t eventCode, int32_t step, void* ptr);

dss_python_backend/events.py

@@ -0,0 +1,91 @@
+from .enums import AltDSSEvent
+from . import ffi, lib
+
+LEGACY_EVENTS = (
+    AltDSSEvent.Legacy_InitControls,
+    AltDSSEvent.Legacy_CheckControls,
+    AltDSSEvent.Legacy_StepControls,
+)
+
+class EventCallbackManager:
+    _ctx_to_manager = {}
+
+    def __init__(self, ctx):
+        if ctx in EventCallbackManager._ctx_to_manager:
+            raise ValueError('This context already has a manager. Use "get_manager_for_ctx" to get it.')
+
+        EventCallbackManager._ctx_to_manager[ctx] = self
+        self.ctx = ctx
+        for evt_type in AltDSSEvent:
+            setattr(self, evt_type.name, [])
+
+    def register_func(self, evt: AltDSSEvent, func) -> bool:
+        handlers = getattr(self, AltDSSEvent(evt).name)
+        if len(handlers) == 0:
+            if lib.ctx_DSSEvents_RegisterAlt(
+                self.ctx,
+                evt,
+                lib.altdss_python_util_callback
+            ) == 0:
+                raise RuntimeError('Could not register main callback function.')
+
+        if func in handlers:
+            return False
+
+        handlers.append(func)
+        return True
+
+    def unregister_func(self, evt: AltDSSEvent, func) -> bool:
+        handlers = getattr(self, AltDSSEvent(evt).name)
+        prev_len = len(handlers)
+        handlers[:] = [f for f in handlers if f is not func]
+        if len(handlers) == 0:
+            lib.ctx_DSSEvents_UnregisterAlt(
+                self.ctx,
+                evt,
+                lib.altdss_python_util_callback
+            )
+
+        return prev_len != len(handlers)
+
+
+    def handle_event(self, ctx, evt: AltDSSEvent, step: int, ptr):
+        evt = AltDSSEvent(evt)
+        handlers = getattr(self, evt.name)
+
+        # No arguments for legacy (classic COM impl.) OpenDSS events
+        if evt in LEGACY_EVENTS:
+            for handler in handlers:
+                handler()
+
+            return
+
+        for handler in handlers:
+            handler(ctx, evt, step, ptr)
+
+
+
+def get_manager_for_ctx(ctx) -> EventCallbackManager:
+    m = EventCallbackManager._ctx_to_manager.get(ctx)
+    if m is None:
+        m = EventCallbackManager(ctx)
+
+    return m
+
+
+@ffi.def_extern()
+def altdss_python_util_callback(ctx, eventCode: int, step: int, ptr):
+    m = EventCallbackManager._ctx_to_manager.get(ctx)
+    if m is None:
+        return
+
+    try:
+        m.handle_event(ctx, eventCode, step, ptr)
+
+    except Exception as ex:
+        err_ptr = lib.ctx_Error_Get_NumberPtr(ctx)
+        err_ptr[0] = 1
+        lib.ctx_Error_Set_Description(ctx, f"Python callback exception: {ex}".encode())
+
+
+__all__ = ['get_manager_for_ctx']