feat(sdk): add event bus for inter-addon communication (#382) #398
@@ -12,6 +12,7 @@ from kindred_sdk.context import (
|
||||
unregister_overlay,
|
||||
)
|
||||
from kindred_sdk.dock import register_dock_panel
|
||||
from kindred_sdk.events import emit, off, on
|
||||
from kindred_sdk.menu import register_menu
|
||||
from kindred_sdk.origin import (
|
||||
active_origin,
|
||||
@@ -33,6 +34,7 @@ __all__ = [
|
||||
"addon_version",
|
||||
"create_version",
|
||||
"current_context",
|
||||
"emit",
|
||||
"freecad_version",
|
||||
"get_origin",
|
||||
"get_theme_tokens",
|
||||
@@ -41,6 +43,8 @@ __all__ = [
|
||||
"list_origins",
|
||||
"load_palette",
|
||||
"loaded_addons",
|
||||
"off",
|
||||
"on",
|
||||
"refresh_context",
|
||||
"register_command",
|
||||
"register_context",
|
||||
|
||||
81
mods/sdk/kindred_sdk/events.py
Normal file
81
mods/sdk/kindred_sdk/events.py
Normal file
@@ -0,0 +1,81 @@
|
||||
# kindred_sdk.events — lightweight event bus for inter-addon communication
|
||||
#
|
||||
# Pure Python pub/sub so addons can signal each other without direct imports.
|
||||
# Synchronous dispatch on the Qt main thread. No C++ dependency.
|
||||
#
|
||||
# Usage:
|
||||
# import kindred_sdk as sdk
|
||||
# sdk.on("silo.document_locked", handler)
|
||||
# sdk.emit("silo.document_locked", {"doc_id": "abc"})
|
||||
# sdk.off("silo.document_locked", handler)
|
||||
|
||||
import FreeCAD
|
||||
|
||||
_listeners: dict[str, list] = {}
|
||||
|
||||
|
||||
def on(event, handler):
|
||||
"""Subscribe *handler* to *event*.
|
||||
|
||||
>>> import kindred_sdk as sdk
|
||||
>>> sdk.on("silo.document_locked", my_handler)
|
||||
"""
|
||||
if not isinstance(event, str):
|
||||
raise TypeError(f"event must be str, got {type(event).__name__}")
|
||||
if not callable(handler):
|
||||
raise TypeError(f"handler must be callable, got {type(handler).__name__}")
|
||||
|
||||
subs = _listeners.setdefault(event, [])
|
||||
if handler not in subs:
|
||||
subs.append(handler)
|
||||
FreeCAD.Console.PrintLog(f"kindred_sdk: subscribed to '{event}'\n")
|
||||
|
||||
|
||||
def off(event, handler):
|
||||
"""Unsubscribe *handler* from *event*. No-op if not subscribed.
|
||||
|
||||
>>> import kindred_sdk as sdk
|
||||
>>> sdk.off("silo.document_locked", my_handler)
|
||||
"""
|
||||
if not isinstance(event, str):
|
||||
raise TypeError(f"event must be str, got {type(event).__name__}")
|
||||
if not callable(handler):
|
||||
raise TypeError(f"handler must be callable, got {type(handler).__name__}")
|
||||
|
||||
subs = _listeners.get(event)
|
||||
if subs is None:
|
||||
return
|
||||
try:
|
||||
subs.remove(handler)
|
||||
except ValueError:
|
||||
return
|
||||
if not subs:
|
||||
del _listeners[event]
|
||||
|
||||
|
||||
def emit(event, data=None):
|
||||
"""Publish *event* with an optional *data* dict to all subscribers.
|
||||
|
||||
Handlers that raise are logged and skipped — remaining handlers still fire.
|
||||
|
||||
>>> import kindred_sdk as sdk
|
||||
>>> sdk.emit("silo.document_locked", {"doc_id": "abc", "user": "forbes"})
|
||||
"""
|
||||
if not isinstance(event, str):
|
||||
raise TypeError(f"event must be str, got {type(event).__name__}")
|
||||
if data is None:
|
||||
data = {}
|
||||
if not isinstance(data, dict):
|
||||
raise TypeError(f"data must be dict, got {type(data).__name__}")
|
||||
|
||||
subs = _listeners.get(event)
|
||||
if not subs:
|
||||
return
|
||||
|
||||
for handler in list(subs): # snapshot — safe if handlers call on/off/emit
|
||||
try:
|
||||
handler(data)
|
||||
except Exception as e:
|
||||
FreeCAD.Console.PrintWarning(
|
||||
f"kindred_sdk: event handler for '{event}' raised {type(e).__name__}: {e}\n"
|
||||
)
|
||||
Reference in New Issue
Block a user