Skip to content

Bus & Events

This page covers how to migrate AppDaemon event listeners and state change listeners to Hassette's event bus (self.bus).

Overview

AppDaemon exposes event subscriptions as methods directly on self: self.listen_state(...), self.listen_event(...). You cancel subscriptions using a handle returned by the listen call.

Hassette centralizes event subscriptions on self.bus. Each subscription method returns a Subscription object. You cancel it by calling .cancel() on that object.

Handler constraints

Handlers cannot use positional-only parameters (parameters before /) or variadic positional arguments (*args). This applies to all self.bus subscription methods.

Event payload values are untyped

Event objects are typed, but the values inside payload dicts (such as service_data) are dict[str, Any]. Use dependency injection or convert data manually to work with typed objects.

State Change Listeners

AppDaemon

In AppDaemon, self.listen_state() listens for state changes on an entity. Callback signatures must follow a fixed pattern:

from appdaemon.plugins.hass import Hass


class ButtonPressed(Hass):
    def initialize(self):
        self.listen_state(self.button_pressed, "input_button.test_button", arg1=123)

    def button_pressed(self, entity, attribute, old, new, arg1, **kwargs):
        self.log(f"{entity=} {attribute=} {old=} {new=} {arg1=}")

In Hassette, self.bus.on_state_change() is an async method — it must be awaited. Handler signatures are flexible — use type annotations and Hassette extracts the data for you:

from hassette import App, AppConfig, D, states


class MyConfig(AppConfig):
    button_entity: str = "input_button.test_button"


class MyApp(App[MyConfig]):
    async def on_initialize(self):
        sub = await self.bus.on_state_change(
            entity_id=self.app_config.button_entity,
            handler=self.button_pressed,
            name="button_pressed",
        )
        self.logger.info("Subscribed: %s", sub)

    async def button_pressed(self, new_state: D.StateNew[states.InputButtonState], entity_id: D.EntityId) -> None:
        friendly_name = new_state.attributes.friendly_name or entity_id
        self.logger.info("Button %s pressed at %s", friendly_name, new_state.last_changed)

Hassette: with the full event object

If you prefer to receive the raw event and inspect it yourself:

from hassette import App, AppConfig, D, states


class MyConfig(AppConfig):
    button_entity: str = "input_button.test_button"


class MyApp(App[MyConfig]):
    async def on_initialize(self):
        sub = await self.bus.on_state_change(
            entity_id=self.app_config.button_entity,
            handler=self.button_pressed,
            name="button_pressed",
        )
        self.logger.info("Subscribed: %s", sub)

    async def button_pressed(self, event: D.TypedStateChangeEvent[states.InputButtonState]) -> None:
        self.logger.info("Button pressed: %s", event)

Filter options

on_state_change() supports built-in filter arguments:

AppDaemon argument Hassette equivalent
new="on" changed_to="on"
old="off" changed_from="off"
attribute="battery" Use on_attribute_change() instead

For more complex filtering, pass a predicate via the where parameter (where=P.StateTo('on') for example). See the Bus filtering docs for the full reference.

Service Call Listeners

AppDaemon

In AppDaemon, you use self.listen_event("call_service", ...) to monitor service calls:

from datetime import datetime
from typing import Any

from appdaemon.adapi import ADAPI


class ButtonHandler(ADAPI):
    def initialize(self):
        # Listen for a button press event with a specific entity_id
        self.listen_event(
            self.minimal_callback,
            "call_service",
            service="press",
            entity_id="input_button.test_button",
        )

    def minimal_callback(self, event_name: str, event_data: dict[str, Any], **kwargs: Any) -> None:
        self.log(f"{event_name=}, {event_data=}, {kwargs=}")

The callback signature must follow (self, event_name, event_data, **kwargs). Extra keyword arguments you passed when subscribing arrive in **kwargs.

Use self.bus.on_call_service() and annotate your handler to extract exactly the fields you need:

from typing import Annotated, Any

from hassette import A, App, AppConfig, D


class MyConfig(AppConfig):
    button_entity: str = "input_button.test_button"


class MyApp(App[MyConfig]):
    async def on_initialize(self):
        # Handler with dependency injection
        sub = await self.bus.on_call_service(
            service="press",
            handler=self.minimal_callback,
            where={"entity_id": self.app_config.button_entity},
            name="button_press_di",
        )
        self.logger.info("Subscribed: %s", sub)

    # Extract only what you need from the event
    async def minimal_callback(
        self,
        domain: D.Domain,
        service: Annotated[str, A.get_service],
        service_data: Annotated[Any, A.get_service_data],
    ) -> None:
        entity_id = service_data.get("entity_id", "unknown")
        self.logger.info("Button %s pressed (domain=%s, service=%s)", entity_id, domain, service)
        self.logger.info("Service data: %s", service_data)

Available dependency markers for service call handlers include:

  • D.Domain — the service domain (e.g., "light")
  • D.EntityId / D.MaybeEntityId — entity ID from the service data
  • D.EventContext — the HA event context object
  • Annotated[str, A.get_service] — the service name
  • Annotated[Any, A.get_service_data] — the full service data dict

Hassette: with the full event object

from hassette import App, AppConfig
from hassette.events import CallServiceEvent


class MyConfig(AppConfig):
    button_entity: str = "input_button.test_button"


class MyApp(App[MyConfig]):
    async def on_initialize(self):
        sub = await self.bus.on_call_service(
            service="press",
            handler=self.minimal_callback,
            where={"entity_id": self.app_config.button_entity},
            name="button_press_event",
        )
        self.logger.info("Subscribed: %s", sub)

    def minimal_callback(self, event: CallServiceEvent) -> None:
        self.logger.info("Button pressed: %s", event.payload.data.service_data)

Canceling Subscriptions

handle = self.listen_state(...)
self.cancel_listen_state(handle)

from hassette import App


class MyApp(App):
    async def on_initialize(self):
        # Subscribe and save the subscription object
        subscription = await self.bus.on_state_change("light.kitchen", handler=self.on_change, name="kitchen_light")

        # Cancel when no longer needed
        subscription.cancel()

    async def on_change(self):
        pass

In Hassette, the subscription object returned by on_state_change(), on_call_service(), and on() all support .cancel(). All three methods are async and must be awaited.

Common Migration Patterns

State changes with a filter

def initialize(self):
    self.listen_state(self.on_motion, "binary_sensor.motion", new="on")

def on_motion(self, entity, attribute, old, new, **kwargs):
    self.log(f"Motion detected on {entity}")

from hassette import App, AppConfig, D, states


class MyConfig(AppConfig):
    motion_entity: str = "binary_sensor.motion"


class MyApp(App[MyConfig]):
    async def on_initialize(self):
        await self.bus.on_state_change(
            "binary_sensor.motion",
            handler=self.on_motion,
            changed_to="on",
            name="motion_on",
        )

    async def on_motion(self, new_state: D.StateNew[states.BinarySensorState]):
        self.logger.info("Motion detected on %s", new_state.entity_id)

Service call subscriptions

def initialize(self):
    self.listen_event(
        self.on_service,
        "call_service",
        domain="light",
        service="turn_on",
    )

from hassette import App


class MyApp(App):
    async def on_initialize(self):
        await self.bus.on_call_service(
            domain="light",
            service="turn_on",
            handler=self.on_service,
            name="light_turn_on",
        )

    async def on_service(self):
        self.logger.info("Light turned on")

See Also