forked from Rockachopa/Timmy-time-dashboard
Kimi Agent
9a21a4b0ff
Co-authored-by: Kimi Agent <kimi@timmy.local> Co-committed-by: Kimi Agent <kimi@timmy.local>
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
"""Sensory EventBus — simple pub/sub for SensoryEvents.
|
|
|
|
Thin facade over the infrastructure EventBus that speaks in
|
|
SensoryEvent objects instead of raw infrastructure Events.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from timmy.events import SensoryEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Handler: sync or async callable that receives a SensoryEvent
|
|
SensoryHandler = Callable[[SensoryEvent], None | Awaitable[None]]
|
|
|
|
|
|
class SensoryBus:
|
|
"""Pub/sub dispatcher for SensoryEvents."""
|
|
|
|
def __init__(self, max_history: int = 500) -> None:
|
|
self._subscribers: dict[str, list[SensoryHandler]] = {}
|
|
self._history: list[SensoryEvent] = []
|
|
self._max_history = max_history
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────
|
|
|
|
async def emit(self, event: SensoryEvent) -> int:
|
|
"""Push *event* to all subscribers whose event_type filter matches.
|
|
|
|
Returns the number of handlers invoked.
|
|
"""
|
|
self._history.append(event)
|
|
if len(self._history) > self._max_history:
|
|
self._history = self._history[-self._max_history :]
|
|
|
|
handlers = self._matching_handlers(event.event_type)
|
|
for h in handlers:
|
|
try:
|
|
result = h(event)
|
|
if asyncio.iscoroutine(result):
|
|
await result
|
|
except Exception as exc:
|
|
logger.error("SensoryBus handler error for '%s': %s", event.event_type, exc)
|
|
|
|
return len(handlers)
|
|
|
|
def subscribe(self, event_type: str, callback: SensoryHandler) -> None:
|
|
"""Register *callback* for events matching *event_type*.
|
|
|
|
Use ``"*"`` to subscribe to all event types.
|
|
"""
|
|
self._subscribers.setdefault(event_type, []).append(callback)
|
|
|
|
def recent(self, n: int = 10) -> list[SensoryEvent]:
|
|
"""Return the last *n* events (most recent last)."""
|
|
return self._history[-n:]
|
|
|
|
# ── Internals ─────────────────────────────────────────────────────────
|
|
|
|
def _matching_handlers(self, event_type: str) -> list[SensoryHandler]:
|
|
handlers: list[SensoryHandler] = []
|
|
for pattern, cbs in self._subscribers.items():
|
|
if pattern == "*" or pattern == event_type:
|
|
handlers.extend(cbs)
|
|
return handlers
|
|
|
|
|
|
# ── Module-level singleton ────────────────────────────────────────────────────
|
|
_bus: SensoryBus | None = None
|
|
|
|
|
|
def get_sensory_bus() -> SensoryBus:
|
|
"""Return the module-level SensoryBus singleton."""
|
|
global _bus
|
|
if _bus is None:
|
|
_bus = SensoryBus()
|
|
return _bus
|