Skip to content

Bus & Events

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

Coming from synchronous AppDaemon?

All registration methods (on_state_change, on_attribute_change, on_call_service, on) are async and must be awaited — see Async Basics if that shift is new to you.

The name= Requirement

Every self.bus.on_*() call requires a name= argument. Omitting it raises ListenerNameRequiredError at call time. Hassette uses this name in log output and the monitoring UI, and to avoid registering the same listener twice after a reload.

# Raises ListenerNameRequiredError immediately
await self.bus.on_state_change("light.kitchen", handler=self.on_change)

await self.bus.on_state_change("light.kitchen", handler=self.on_change, name="kitchen_light")

This is the most common cause of breakage when porting AppDaemon apps. Add name= to every subscription call before running the app.

State Change Listeners

AppDaemon uses self.listen_state() with a fixed four-argument callback signature. Hassette uses self.bus.on_state_change(), which is async and must be awaited. Handler signatures are flexible: instead of AppDaemon's fixed (entity, attribute, old, new, kwargs), declare only the parameters the handler needs and give them type hints — Hassette reads the hints and passes the matching values in. This pattern is called dependency injection.

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=}")

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)

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)

The dependency injection form is preferred. D.StateNew[states.InputButtonState] tells Hassette to extract the new state and convert it to a typed model — D is hassette.event_handling.dependencies, states is the module of typed state classes. AppConfig in the example replaces AppDaemon's self.args; fields declared on it are set in hassette.toml (see Configuration). If your editor runs a type checker, it knows the state's type and catches typos.

Filter argument mapping

on_state_change() supports built-in filter arguments that replace AppDaemon's new= and old= kwargs:

AppDaemon Hassette
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 where= — a function that receives the event and returns True or False. See Bus filtering for the full reference.

Attribute Change Listeners

AppDaemon uses self.listen_state(..., attribute="battery") to watch a specific attribute. Hassette has a dedicated method for this: on_attribute_change().

await self.bus.on_attribute_change(
    "sensor.phone",
    "battery_level",
    handler=self.on_battery,
    name="phone_battery",
)

The method signature is on_attribute_change(entity_id, attr, *, handler, name, ...). The attribute= argument on listen_state() maps directly to the second positional argument here.

Service Call Listeners

AppDaemon uses self.listen_event("call_service", ...) with a callback that receives raw dicts. Hassette uses self.bus.on_call_service(), which is async and must be awaited.

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=}")

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)

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)

These are the values Hassette can inject into service-call handler parameters — declare the ones the handler needs. A is hassette.event_handling.accessors, field extractors; Annotated[str, A.get_service] means "a str, extracted by A.get_service":

  • D.Domain, the service domain (e.g., "light")
  • D.EntityId / D.MaybeEntityId, entity ID from the service data (Maybe allows calls where it's absent)
  • 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

AppDaemon passes extra kwargs from listen_event() into the callback via **kwargs. Hassette uses where= for filtering instead. Pass a dict or predicate to match on domain, service, entity ID, or arbitrary fields.

Canceling Subscriptions

AppDaemon returns an opaque handle from listen_state() and requires a separate cancel call. Hassette returns a Subscription object with a .cancel() method.

handle = self.listen_state(self.on_change, "light.kitchen")
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

All registration methods (on_state_change, on_attribute_change, on_call_service, on) are async and must be awaited. .cancel() on the returned Subscription is synchronous.

Common Migration Patterns

State change with 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 subscription

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")

Verify the Migration

Run hassette listener --app <key> to confirm each subscription registered under its name=, then trigger the entity and watch hassette log --app <key> for the handler's log line.

See Also