Event Flow Management

Adds new events to Bob The Skull's event-driven architecture, ensuring all publishers and subscribers are properly connected.

When to Use

• Creating new event types
• Adding event publishers (components that emit events)
• Adding event subscribers (components that handle events)
• Documenting event flows
• Troubleshooting missing event handlers

Event-Driven Architecture Overview

Bob The Skull uses a central EventBus for all inter-component communication:

Component A → publish(Event) → EventBus → deliver → Component B.on_event()

No direct method calls between components! Everything goes through events.

Event Definition Pattern (events.py)

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, Optional

@dataclass
class YourEvent:
    """Published when [describe condition that triggers event]"""
    param1: type  # Description
    param2: type  # Description
    optional_data: Optional[Dict[str, Any]] = None
    timestamp: datetime = field(default_factory=datetime.now)

Common Event Types

State Change Events:

@dataclass
class ComponentStateChangedEvent:
    """Component changed state"""
    component: str
    old_state: str
    new_state: str
    timestamp: datetime = field(default_factory=datetime.now)

Detection Events:

@dataclass
class ThingDetectedEvent:
    """Thing was detected"""
    confidence: float
    location: tuple
    detector_id: str
    data: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)

Request/Response Events:

@dataclass
class ProcessRequestEvent:
    """Request to process something"""
    request_id: str
    input_data: Any
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class ProcessResponseEvent:
    """Response to process request"""
    request_id: str
    result: Any
    success: bool
    timestamp: datetime = field(default_factory=datetime.now)

Publishing Events

In component code:

from events import YourEvent

class YourComponent:
    def __init__(self, event_bus):
        self.event_bus = event_bus

    def do_something(self):
        # When condition occurs, publish event
        event = YourEvent(
            param1=value1,
            param2=value2
        )
        self.event_bus.publish(event)
        logger.info(f"Published YourEvent: {event}")

Subscribing to Events

In component code:

from events import YourEvent

class YourComponent:
    def __init__(self, event_bus):
        self.event_bus = event_bus
        # Subscribe to event during initialization
        self.event_bus.subscribe(YourEvent, self.on_your_event)

    def on_your_event(self, event: YourEvent):
        """Handle YourEvent"""
        logger.info(f"Received YourEvent: {event}")
        # Process event
        self.do_something_with(event.param1, event.param2)

State Machine Integration

State machine is a special subscriber that coordinates system behavior:

# In state_machine/state_machine.py

def _setup_event_handlers(self):
    """Subscribe to events"""
    self.event_bus.subscribe(YourEvent, self.on_your_event)

def on_your_event(self, event: YourEvent):
    """Handle YourEvent in state machine"""
    logger.info(f"State machine received: {event}")

    # Check current state
    if self.current_state == State.WAITING:
        # Transition to new state
        self._transition_to(State.PROCESSING)
        # Trigger actions
        self._handle_processing(event)

Event Flow Checklist

When adding a new event:

• Define event dataclass in

  events.py

• Find all components that should publish this event
• Find all components that should subscribe to this event
• Add publishers:

  event_bus.publish(YourEvent(...))

• Add subscribers:

  event_bus.subscribe(YourEvent, self.on_event)

• Add handlers:

  def on_your_event(self, event: YourEvent):

• Update state machine if needed
• Test event flow end-to-end
• Document event flow in comments/docstrings

Finding Event Usage

Find where event is published:

grep -r "YourEvent(" .

Find where event is subscribed:

grep -r "subscribe.*YourEvent" .

Find event handlers:

grep -r "def on_your_event" .

Common Event Patterns

Vision Detection Events

• Published by: Detector plugins
• Subscribed by: State machine, vision processor
• Pattern:

  *DetectedEvent

  ,

  *ClearedEvent

Speech Events

• Published by: STT, wake word detector
• Subscribed by: State machine
• Pattern:

  *RecognizedEvent

  ,

  *DetectedEvent

LLM Events

• Published by: LLM processor
• Subscribed by: State machine, TTS
• Pattern:

  *ResponseEvent

  ,

  *ToolCallEvent

Hardware Events

• Published by: Hardware controllers
• Subscribed by: State machine
• Pattern:

  *StatusEvent

  ,

  *ErrorEvent

Event Bus Implementation

Bob uses two event bus implementations:

LocalEventBus - In-process pub/sub

from local_event_bus import LocalEventBus
event_bus = LocalEventBus()

MQTTEventBus - Distributed pub/sub

from mqtt_event_bus import MQTTEventBus
event_bus = MQTTEventBus(broker="192.168.1.44")

Both implement the same interface:

• publish(event)

  - Send event

• subscribe(event_type, handler)

  - Register handler

• unsubscribe(event_type, handler)

  - Remove handler

Existing Events Reference

Core events in

events.py

:

• WakeWordDetectedEvent

  - Wake word heard

• SpeechRecognizedEvent

  - Speech transcribed

• LLMResponseReadyEvent

  - LLM response available

• LLMToolCallEvent

  - LLM wants to call tool

• TTSStartedEvent

  /

  TTSCompletedEvent

  - TTS lifecycle

• FaceDetectedEvent

  /

  FaceIdentifiedEvent

  - Vision events

• MotionDetectedEvent

  /

  MotionClearedEvent

  - Motion events

• StateChangedEvent

  - State machine transitions

Event Naming Conventions

Event Names:

• Use descriptive nouns:

  ThingDetected

  , not

  DetectThing

• End with

  Event

  :

  FaceDetectedEvent

• Use past tense:

  Detected

  , not

  Detect

Handler Names:

• Prefix with

  on_

  :

  on_face_detected

• Use snake_case:

  on_speech_recognized

• Match event name:

  FaceDetectedEvent

  →

  on_face_detected

Debugging Event Flows

Enable event logging:

# In LocalEventBus or MQTTEventBus
logger.setLevel(logging.DEBUG)

Trace event flow:

# In event_bus.py
def publish(self, event):
    logger.debug(f"Publishing: {type(event).__name__} to {len(self.subscribers.get(type(event), []))} subscribers")
    # ... publish logic

Check subscriptions:

# In component
logger.info(f"Subscribed to: {[t.__name__ for t in self.event_bus.subscribers.keys()]}")

Pro Tips

1. Events are immutable - Use dataclasses with frozen=False for mutable data
2. Include context - Always add detector_id, component_id, etc.
3. Timestamp everything - Use

   field(default_factory=datetime.now)

4. Document triggers - Explain in docstring what causes the event
5. Test isolation - Mock event_bus for unit tests
6. Avoid event storms - Debounce high-frequency events
7. Use type hints - Makes IDE autocomplete work perfectly