# -*- coding: utf-8 -*- """ Dekoratoren für das Event-System. Ermöglicht die einfache Registrierung von Event-Handlern über Dekoratoren auf Methoden. """ from typing import Callable, Any, Type from functools import wraps from trixy_core.events.enums import EventPriority from trixy_core.events.event_data.base import EventData # Attribut-Name für registrierte Event-Handler _EVENT_HANDLERS_ATTR = "_trixy_event_handlers" def TrixyEvent( events: list[str], priority: EventPriority = EventPriority.NORMAL ) -> Callable[[Callable], Callable]: """ Dekorator zur Registrierung einer Methode als Event-Handler. Verwendung: @TrixyEvent(["wakeword_received", "satellite_connected"]) def on_events(self, event_name: str, event_data: EventData): pass Args: events: Liste der Event-Namen, auf die reagiert werden soll priority: Handler-Priorität Returns: Dekorator-Funktion """ def decorator(func: Callable) -> Callable: # Speichere Handler-Metadaten direkt auf der Funktion # Kein Wrapper nötig - bewahrt async-Erkennung via iscoroutinefunction() if not hasattr(func, _EVENT_HANDLERS_ATTR): setattr(func, _EVENT_HANDLERS_ATTR, []) handlers = getattr(func, _EVENT_HANDLERS_ATTR) for event_name in events: handlers.append({ "event_name": event_name, "priority": priority, "handler": func, }) return func return decorator def event_handler( event_class: Type[EventData], priority: EventPriority = EventPriority.NORMAL ) -> Callable[[Callable], Callable]: """ Dekorator zur Registrierung einer Funktion als typisierter Event-Handler. Verwendung: @event_handler(WakewordReceived) async def on_wakeword(self, event: WakewordReceived): pass Args: event_class: Die Event-Datenklasse, auf die reagiert werden soll priority: Handler-Priorität Returns: Dekorator-Funktion """ event_name = event_class.event_name() def decorator(func: Callable) -> Callable: if not hasattr(func, _EVENT_HANDLERS_ATTR): setattr(func, _EVENT_HANDLERS_ATTR, []) handlers = getattr(func, _EVENT_HANDLERS_ATTR) handlers.append({ "event_name": event_name, "event_class": event_class, "priority": priority, "handler": func, }) return func return decorator def get_event_handlers(obj: Any) -> list[dict[str, Any]]: """ Extrahiert alle Event-Handler aus einem Objekt. Durchsucht alle Methoden des Objekts nach dem Event-Handler-Attribut. Args: obj: Das zu durchsuchende Objekt Returns: Liste der Handler-Metadaten """ handlers = [] for attr_name in dir(obj): if attr_name.startswith("_"): continue try: attr = getattr(obj, attr_name) if callable(attr) and hasattr(attr, _EVENT_HANDLERS_ATTR): for handler_info in getattr(attr, _EVENT_HANDLERS_ATTR): handlers.append({ **handler_info, "instance": obj, "method": attr, }) except Exception: continue return handlers