Trixy7/docs/api/events.md

trixy_core.events

Zentrales Event-System mit Pub/Sub, Middleware, Throttling und Persistenz.

Übersicht

from trixy_core.events import (
    # Core
    EventManager, EventPriority,
    TrixyEvent, event_handler,
    # Event Data
    EventData, SatelliteConnected, WakewordReceived,
    # Middleware
    EventMiddleware, MiddlewarePipeline,
    LoggingMiddleware, TimingMiddleware,
    # Throttling
    EventThrottler, EventDebouncer,
    # Persistence
    EventStore, InMemoryEventStore, FileEventStore,
    EventReplayer,
    # DLQ
    DeadLetterQueue, RetryStrategy,
    # Async
    AsyncEventQueue, WorkerPool,
)

---

EventManager

Zentrale Event-Verteilung.

from trixy_core.events import EventManager, EventPriority

manager = EventManager()

# Handler registrieren
def on_wakeword(event_name: str, data: dict) -> None:
    print(f"Wakeword: {data}")

manager.register("wakeword_received", on_wakeword)

# Mit Priorität
manager.register(
    "speech_recognized",
    handler,
    priority=EventPriority.HIGH
)

# Event auslösen
manager.emit("wakeword_received", {
    "satellite": "living_room",
    "confidence": 0.95,
})

# Async Event
await manager.async_emit("speech_recognized", {
    "text": "Schalte das Licht an",
})

# Handler entfernen
manager.unregister("wakeword_received", on_wakeword)

EventPriority

Priorität	Wert	Verwendung
SYSTEM	0	System-interne Handler
HIGHEST	10	Kritische Handler
HIGH	25	Wichtige Handler
NORMAL	50	Standard
LOW	75	Nachgelagerte Handler
LOWEST	90	Letzte Handler
MONITOR	100	Nur Logging/Monitoring

---

Decorators

@TrixyEvent

Registriert Event-Handler in Klassen.

from trixy_core.events import TrixyEvent, EventPriority

class MyPlugin:
    @TrixyEvent(["wakeword_received"])
    def on_wakeword(self, event_name: str, data: dict) -> None:
        satellite = data.get("satellite")
        self.process_wakeword(satellite)

    @TrixyEvent(
        ["speech_recognized", "text_input"],
        priority=EventPriority.HIGH
    )
    async def on_text(self, event_name: str, data: dict) -> None:
        text = data.get("text")
        await self.process_text(text)

    @TrixyEvent(["*"])  # Alle Events
    def on_any(self, event_name: str, data: dict) -> None:
        self.logger.debug(f"Event: {event_name}")

@event_handler

Typisierte Event-Handler.

from trixy_core.events import event_handler
from trixy_core.events.event_data import WakewordReceived

class MyService:
    @event_handler(WakewordReceived)
    async def handle_wakeword(self, event: WakewordReceived) -> None:
        print(f"Wakeword von {event.satellite_id}")
        print(f"Confidence: {event.confidence}")

---

Event Data

Strukturierte Event-Daten.

from trixy_core.events.event_data import (
    EventData,
    SatelliteConnected,
    SatelliteDisconnected,
    WakewordReceived,
    RawAudioInputReceived,
    SpeechRecognized,
    IntentReceived,
)

# Satellite-Events
connected = SatelliteConnected(
    satellite_id="living_room",
    mac_address="AA:BB:CC:DD:EE:FF",
    alias="Wohnzimmer",
)

# Wakeword-Event
wakeword = WakewordReceived(
    satellite_id="living_room",
    wakeword="trixy",
    confidence=0.95,
    audio_level=0.7,
)

# Speech-Event
speech = SpeechRecognized(
    satellite_id="living_room",
    text="Schalte das Licht an",
    confidence=0.88,
    language="de",
)

# Intent-Event
intent = IntentReceived(
    text="Schalte das Licht an",
    intent="smart_home.light.on",
    confidence=0.92,
    entities={
        "device": "light",
        "action": "on",
    },
)

---

Middleware

Event-Verarbeitung vor/nach Handlern.

MiddlewarePipeline

from trixy_core.events.middleware import (
    MiddlewarePipeline,
    LoggingMiddleware,
    TimingMiddleware,
    ValidationMiddleware,
)

pipeline = MiddlewarePipeline()

# Middleware hinzufügen
pipeline.add(LoggingMiddleware(level="DEBUG"))
pipeline.add(TimingMiddleware(threshold=0.1))
pipeline.add(ValidationMiddleware(schema=event_schema))

# Mit EventManager verwenden
manager.set_middleware(pipeline)

Eigene Middleware

from trixy_core.events.middleware import EventMiddleware, MiddlewareContext

class AuthenticationMiddleware(EventMiddleware):
    async def before(self, ctx: MiddlewareContext) -> MiddlewareContext:
        if not ctx.metadata.get("authenticated"):
            ctx.skip = True
            ctx.skip_reason = "Nicht authentifiziert"
        return ctx

    async def after(self, ctx: MiddlewareContext) -> MiddlewareContext:
        ctx.metadata["processed_at"] = datetime.now()
        return ctx

class MetricsMiddleware(EventMiddleware):
    def __init__(self, collector):
        self.collector = collector

    async def before(self, ctx: MiddlewareContext) -> MiddlewareContext:
        ctx.metadata["start_time"] = time.time()
        return ctx

    async def after(self, ctx: MiddlewareContext) -> MiddlewareContext:
        duration = time.time() - ctx.metadata["start_time"]
        self.collector.record(ctx.event_name, duration)
        return ctx

---

Throttling

EventThrottler

Rate-Limiting für Events.

from trixy_core.events.throttling import (
    EventThrottler,
    ThrottleConfig,
    ThrottleStrategy,
)

config = ThrottleConfig(
    rate=10,                          # Events pro Sekunde
    burst=20,                         # Burst-Kapazität
    strategy=ThrottleStrategy.DROP,   # Überschüssige droppen
)

throttler = EventThrottler(config)

# Prüfen ob Event durchgelassen wird
result = throttler.check("wakeword_received")
if result.allowed:
    process_event()
else:
    print(f"Gedrosselt, Retry nach {result.retry_after}s")

# Pro Event-Typ konfigurieren
throttler.configure("audio_data", ThrottleConfig(rate=100))
throttler.configure("status_update", ThrottleConfig(rate=1))

EventDebouncer

Verzögert Events und führt nur das letzte aus.

from trixy_core.events.throttling import EventDebouncer, DebounceConfig

config = DebounceConfig(
    delay=0.5,       # Wartezeit in Sekunden
    max_wait=5.0,    # Maximale Verzögerung
    leading=False,   # Erstes Event sofort?
    trailing=True,   # Letztes Event ausführen?
)

debouncer = EventDebouncer(config)

# Events werden gesammelt
debouncer.emit("search_query", {"query": "t"})
debouncer.emit("search_query", {"query": "te"})
debouncer.emit("search_query", {"query": "tes"})
debouncer.emit("search_query", {"query": "test"})

# Nach 0.5s wird nur das letzte ausgeführt
# -> {"query": "test"}

---

Persistence

Event-Speicherung für Replay und Audit.

InMemoryEventStore

from trixy_core.events.persistence import InMemoryEventStore

store = InMemoryEventStore(max_events=10000)

# Event speichern
event_id = store.store("user_action", {
    "user": "admin",
    "action": "login",
})

# Events abfragen
events = store.query(
    event_names=["user_action"],
    start_time=datetime.now() - timedelta(hours=1),
    limit=100,
)

# Event abrufen
event = store.get(event_id)

FileEventStore

from trixy_core.events.persistence import FileEventStore

store = FileEventStore(
    directory=Path("./events"),
    rotation="daily",           # Tägliche Rotation
    max_files=30,               # Aufbewahrung
    compression=True,           # Komprimierung
)

# Wie InMemoryEventStore verwenden
store.store("system_event", {"message": "Startup"})

EventReplayer

from trixy_core.events.persistence import EventReplayer, ReplayConfig

config = ReplayConfig(
    speed=1.0,          # Echtzeit
    filter_events=["wakeword_received", "speech_recognized"],
    start_time=datetime(2024, 1, 1),
    end_time=datetime(2024, 1, 2),
)

replayer = EventReplayer(store, manager, config)

# Replay starten
result = await replayer.replay()

print(f"Replayed: {result.event_count} events")
print(f"Duration: {result.duration}s")

---

Dead Letter Queue

Fehlerhafte Events mit Retry.

from trixy_core.events.dlq import (
    DeadLetterQueue,
    DLQConfig,
    ExponentialBackoffStrategy,
    FixedRetryStrategy,
)

config = DLQConfig(
    max_retries=5,
    retention_days=7,
    batch_size=10,
)

dlq = DeadLetterQueue(config)

# Fehlerhaftes Event hinzufügen
dlq.add("failed_event", data, error="Connection timeout")

# Retry-Strategie
strategy = ExponentialBackoffStrategy(
    base_delay=1.0,
    max_delay=60.0,
    multiplier=2.0,
)

# Retry ausführen
results = await dlq.retry_all(strategy)

for result in results:
    if result.success:
        print(f"Event {result.event_id} erfolgreich")
    else:
        print(f"Event {result.event_id} fehlgeschlagen: {result.error}")

# Statistiken
stats = dlq.get_stats()
print(f"Pending: {stats.pending_count}")
print(f"Failed: {stats.failed_count}")
print(f"Retried: {stats.retry_count}")

---

Async Queue

Hintergrund-Verarbeitung mit Worker-Pool.

from trixy_core.events.async_queue import (
    AsyncEventQueue,
    WorkerPool,
    QueueConfig,
    WorkerConfig,
)

# Queue-Konfiguration
queue_config = QueueConfig(
    max_size=1000,
    overflow_strategy="drop_oldest",
)

queue = AsyncEventQueue(queue_config)

# Worker-Pool
worker_config = WorkerConfig(
    workers=4,
    batch_size=10,
    idle_timeout=5.0,
)

pool = WorkerPool(worker_config)

# Handler registrieren
@pool.handler("email_notification")
async def send_email(events: list) -> None:
    for event in events:
        await email_service.send(
            to=event.data["recipient"],
            subject=event.data["subject"],
            body=event.data["body"],
        )

# Queue starten
await queue.start(pool)

# Events in Queue
queue.enqueue("email_notification", {
    "recipient": "user@example.com",
    "subject": "Willkommen",
    "body": "...",
})

# Statistiken
stats = queue.get_stats()
print(f"Queued: {stats.queued}")
print(f"Processed: {stats.processed}")
print(f"Failed: {stats.failed}")

# Graceful Shutdown
await queue.stop(drain=True)

---

Filtering & Batching

EventFilter

from trixy_core.events import EventFilter, EventNameCondition

# Filter erstellen
filter = EventFilter()
filter.add(EventNameCondition(["wakeword_*", "speech_*"]))
filter.add(lambda e: e.data.get("confidence", 0) > 0.8)

# Anwenden
if filter.matches(event):
    process(event)

EventBatcher

from trixy_core.events import EventBatcher, BatchConfig

config = BatchConfig(
    batch_size=100,
    flush_interval=5.0,
    key_func=lambda e: e.data.get("satellite_id"),
)

batcher = EventBatcher(config)

@batcher.on_batch
async def process_batch(events: list) -> None:
    # Batch verarbeiten
    await bulk_insert(events)

# Events hinzufügen
batcher.add(event1)
batcher.add(event2)
# ...

# Wird automatisch geflusht bei batch_size oder flush_interval