feat(sdk): add event bus for inter-addon communication (#382)
All checks were successful
Build and Test / build (pull_request) Successful in 28m56s

New module kindred_sdk/events.py provides lightweight publish-subscribe
so addons can signal each other without direct imports.

New public API:
- sdk.on(event, handler) — subscribe to a named event
- sdk.off(event, handler) — unsubscribe
- sdk.emit(event, data) — publish event with dict payload

Pure Python, synchronous dispatch, snapshot-safe iteration.
Handlers that raise are logged and skipped without breaking the chain.
This commit is contained in:
forbes
2026-03-04 09:45:47 -06:00
parent ba1b612628
commit 54b926a925
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"
)