Event-Driven Notification System

Build a simple event-driven notification system that tracks device state changes and sends notifications when specific conditions are met.

Requirements

Your system should:

1. Listen for device state change events and maintain a count of how many times each device has changed state
2. Fire a custom notification event when any device reaches exactly 3 state changes
3. Support registering one-time listeners that automatically unregister after being triggered once
4. Fire events from both synchronous and asynchronous contexts
5. Ensure thread-safe event handling

Implementation Details

Create a NotificationManager class with the following functionality:

• Initialize with a Home Assistant event bus instance
• Track state changes for devices and count them
• Emit a "notification_ready" event (with device name in event data) when a device reaches 3 state changes
• Provide a method to register one-time listeners for notifications
• Provide a method to fire test events from synchronous contexts

Test Cases

• When a device state changes 3 times, a "notification_ready" event is fired with the device name @test
• One-time listeners are automatically removed after being triggered once @test
• Events can be fired from synchronous contexts using the appropriate method @test
• Multiple devices can be tracked independently @test

Implementation

@generates

API

from homeassistant.core import HomeAssistant, Event

class NotificationManager:
    """Manages device state change notifications using the event bus."""

    def __init__(self, hass: HomeAssistant) -> None:
        """
        Initialize the notification manager.

        Args:
            hass: Home Assistant instance with access to the event bus
        """
        pass

    async def start_tracking(self) -> None:
        """Start listening for state change events."""
        pass

    def register_one_time_listener(self, callback) -> None:
        """
        Register a listener that will be called once when notification_ready event fires.

        Args:
            callback: Async function to call when the event fires
        """
        pass

    def fire_test_event(self, event_type: str, event_data: dict) -> None:
        """
        Fire an event from a synchronous context.

        Args:
            event_type: Type of event to fire
            event_data: Data to include with the event
        """
        pass

Dependencies { .dependencies }

homeassistant { .dependency }

Provides the event bus system for event-driven communication.