feat(sdk): add event bus for inter-addon communication (#382) #398

Merged
forbes merged 1 commits from feat/sdk-event-bus into main 2026-03-04 15:48:19 +00:00
2 changed files with 85 additions and 0 deletions

View File

@@ -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",

View 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"
)