Index

HassEvent: TypeAlias = Event[HassPayload[Any]] module-attribute

Alias for Home Assistant events.

Event dataclass

Bases: Generic[PayloadT]

Base event with strongly typed payload.

Source code in src/hassette/events/base.py

@dataclass(frozen=True, slots=True, repr=False)
class Event(Generic[PayloadT]):
    """Base event with strongly typed payload."""

    topic: "Topic | str"
    """Topic of the event."""

    payload: PayloadT
    """The event payload."""

    def __repr__(self) -> str:
        return f"Event({self.payload})"

topic: Topic | str instance-attribute

Topic of the event.

payload: PayloadT instance-attribute

The event payload.

EventPayload dataclass

Bases: Generic[DataT]

Base payload with typed data.

Source code in src/hassette/events/base.py

@dataclass(frozen=True, slots=True)
class EventPayload(Generic[DataT]):
    """Base payload with typed data."""

    data: DataT
    """The actual event data."""

    event_id: str = field(default="", kw_only=True)
    """Unique identifier for this event. Subclasses override with a more specific default."""

    origin: str = field(default="UNKNOWN", kw_only=True)
    """Origin of the event. Subclasses override with a concrete value."""

data: DataT instance-attribute

The actual event data.

event_id: str = field(default='', kw_only=True) class-attribute instance-attribute

Unique identifier for this event. Subclasses override with a more specific default.

origin: str = field(default='UNKNOWN', kw_only=True) class-attribute instance-attribute

Origin of the event. Subclasses override with a concrete value.

HassContext dataclass

Structure for the context of a Home Assistant event.

Source code in src/hassette/events/base.py

@dataclass(frozen=True, slots=True)
class HassContext:
    """Structure for the context of a Home Assistant event."""

    id: str
    parent_id: str | None
    user_id: str | None

HassettePayload dataclass

Bases: EventPayload[DataT]

Hassette event payload with additional metadata.

Source code in src/hassette/events/base.py

@dataclass(frozen=True, slots=True, repr=False)
class HassettePayload(EventPayload[DataT]):
    """Hassette event payload with additional metadata."""

    data: DataT
    """The actual event data."""

    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    """Unique identifier for this payload instance (UUID4), generated at construction time.

    The bus shares one payload instance with all matched listeners, so all handlers
    for the same event see the same ``event_id``.
    """

    time_fired: ZonedDateTime = field(default_factory=lambda: ZonedDateTime.now("UTC"))
    """The time the event was fired, defaulting to the current UTC time at construction."""

    origin: str = field(default="HASSETTE", init=False)
    """Origin of the event, always 'HASSETTE' for framework-generated events."""

    def __repr__(self) -> str:
        return f"HassettePayload(event_id={self.event_id}, time_fired={self.time_fired})"

data: DataT instance-attribute

The actual event data.

event_id: str = field(default_factory=(lambda: str(uuid.uuid4()))) class-attribute instance-attribute

Unique identifier for this payload instance (UUID4), generated at construction time.

The bus shares one payload instance with all matched listeners, so all handlers for the same event see the same event_id.

time_fired: ZonedDateTime = field(default_factory=(lambda: ZonedDateTime.now('UTC'))) class-attribute instance-attribute

The time the event was fired, defaulting to the current UTC time at construction.

origin: str = field(default='HASSETTE', init=False) class-attribute instance-attribute

Origin of the event, always 'HASSETTE' for framework-generated events.

HassPayload dataclass

Bases: EventPayload[DataT]

Home Assistant event payload with additional metadata.

Source code in src/hassette/events/base.py

@dataclass(frozen=True, slots=True, repr=False)
class HassPayload(EventPayload[DataT]):
    """Home Assistant event payload with additional metadata."""

    event_type: str
    """Type of the event, e.g., 'state_changed', 'call_service', etc."""

    data: DataT
    """The actual event data from Home Assistant."""

    origin: Literal["LOCAL", "REMOTE"]  # pyright: ignore[reportGeneralTypeIssues]
    """Origin of the event, either 'LOCAL' or 'REMOTE'."""

    time_fired: ZonedDateTime  # pyright: ignore[reportGeneralTypeIssues]
    """The time the event was fired."""

    context: HassContext  # pyright: ignore[reportGeneralTypeIssues]
    """The context of the event."""

    def __post_init__(self) -> None:
        object.__setattr__(self, "event_id", self.context.id)

    @property
    def entity_id(self) -> str | None:
        """Return the entity_id if present in the data."""
        return getattr(self.data, "entity_id", None)

    @property
    def domain(self) -> str | None:
        """Return the domain if present in the data."""
        domain = getattr(self.data, "domain", None)
        if domain is not None:
            return domain

        entity_id = self.entity_id
        if entity_id:
            return entity_id.split(".")[0]
        return None

    @property
    def service(self) -> str | None:
        """Return the service if present in the data."""
        return getattr(self.data, "service", None)

    def __repr__(self) -> str:
        if self.entity_id:
            return f"HassPayload(event_type={self.event_type}, entity_id={self.entity_id}, event_id={self.event_id})"

        return f"HassPayload(event_type={self.event_type}, event_id={self.event_id})"

event_type: str instance-attribute

Type of the event, e.g., 'state_changed', 'call_service', etc.

data: DataT instance-attribute

The actual event data from Home Assistant.

origin: Literal['LOCAL', 'REMOTE'] instance-attribute

Origin of the event, either 'LOCAL' or 'REMOTE'.

time_fired: ZonedDateTime instance-attribute

The time the event was fired.

context: HassContext instance-attribute

The context of the event.

entity_id: str | None property

Return the entity_id if present in the data.

domain: str | None property

Return the domain if present in the data.

service: str | None property

Return the service if present in the data.

AutomationTriggeredEvent dataclass

Bases: Event[HassPayload[AutomationTriggeredPayload]]

Event representing an automation triggered in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class AutomationTriggeredEvent(Event[HassPayload[AutomationTriggeredPayload]]):
    """Event representing an automation triggered in Home Assistant."""

CallServiceEvent dataclass

Bases: Event[HassPayload[CallServicePayload]]

Event representing a call service in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class CallServiceEvent(Event[HassPayload[CallServicePayload]]):
    """Event representing a call service in Home Assistant."""

ComponentLoadedEvent dataclass

Bases: Event[HassPayload[ComponentLoadedPayload]]

Event representing a component loaded in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class ComponentLoadedEvent(Event[HassPayload[ComponentLoadedPayload]]):
    """Event representing a component loaded in Home Assistant."""

LogbookEntryEvent dataclass

Bases: Event[HassPayload[LogbookEntryPayload]]

Event representing a logbook entry in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class LogbookEntryEvent(Event[HassPayload[LogbookEntryPayload]]):
    """Event representing a logbook entry in Home Assistant."""

RawStateChangeEvent dataclass

Bases: Event[HassPayload[RawStateChangePayload]]

Event representing a state change in Home Assistant, with raw state data.

Source code in src/hassette/events/hass/hass.py

class RawStateChangeEvent(Event[HassPayload[RawStateChangePayload]]):
    """Event representing a state change in Home Assistant, with raw state data."""

ScriptStartedEvent dataclass

Bases: Event[HassPayload[ScriptStartedPayload]]

Event representing a script started in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class ScriptStartedEvent(Event[HassPayload[ScriptStartedPayload]]):
    """Event representing a script started in Home Assistant."""

ServiceRegisteredEvent dataclass

Bases: Event[HassPayload[ServiceRegisteredPayload]]

Event representing a service registered in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class ServiceRegisteredEvent(Event[HassPayload[ServiceRegisteredPayload]]):
    """Event representing a service registered in Home Assistant."""

ServiceRemovedEvent dataclass

Bases: Event[HassPayload[ServiceRemovedPayload]]

Event representing a service removed in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class ServiceRemovedEvent(Event[HassPayload[ServiceRemovedPayload]]):
    """Event representing a service removed in Home Assistant."""

UserAddedEvent dataclass

Bases: Event[HassPayload[UserAddedPayload]]

Event representing a user added in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class UserAddedEvent(Event[HassPayload[UserAddedPayload]]):
    """Event representing a user added in Home Assistant."""

UserRemovedEvent dataclass

Bases: Event[HassPayload[UserRemovedPayload]]

Event representing a user removed in Home Assistant.

Source code in src/hassette/events/hass/hass.py

class UserRemovedEvent(Event[HassPayload[UserRemovedPayload]]):
    """Event representing a user removed in Home Assistant."""

HassContextDict

Bases: TypedDict

Structure for the context of a state change event.

Source code in src/hassette/events/hass/raw.py

class HassContextDict(TypedDict):
    """Structure for the context of a state change event."""

    id: str
    parent_id: str | None
    user_id: str | None

HassEventDict

Bases: TypedDict

Structure for the state change event data.

Source code in src/hassette/events/hass/raw.py

class HassEventDict(TypedDict):
    """Structure for the state change event data."""

    event_type: str
    data: dict[str, Any] | None
    origin: Literal["LOCAL", "REMOTE"]
    time_fired: str
    context: HassContextDict

HassEventEnvelopeDict

Bases: TypedDict

The structure of what comes from Home Assistant's websocket API for state change events.

When turned into an Event, the event attribute is popped and used to create the event, with type and id being discarded.

Source code in src/hassette/events/hass/raw.py

class HassEventEnvelopeDict(TypedDict):
    """The structure of what comes from Home Assistant's websocket API for state change events.

    When turned into an Event, the `event` attribute is popped and used to create the event,
    with `type` and `id` being discarded.
    """

    event: HassEventDict
    type: Literal["event"]
    id: int  # from the websocket message, not the event's ID

HassStateDict

Bases: TypedDict

Structure for the state of an entity.

This structure is seen both in a state change event or by calling the HA API to get the state of an entity.

Source code in src/hassette/events/hass/raw.py

class HassStateDict(TypedDict, total=False):
    """Structure for the state of an entity.

    This structure is seen both in a state change event or by calling the HA API to get the state of an entity.
    """

    domain: str
    entity_id: Required[str]
    last_changed: str | None
    last_reported: str | None
    last_updated: str | None
    context: Required[HassContextDict]

    state: Required[Any]
    attributes: Required[dict[str, Any]]

AppStateChangePayload dataclass

Payload for app instance state change events.

Source code in src/hassette/events/hassette.py

@dataclass(slots=True, frozen=True)
class AppStateChangePayload:
    """Payload for app instance state change events."""

    app_key: str
    index: int
    status: ResourceStatus
    previous_status: ResourceStatus | None = None
    instance_name: str | None = None
    class_name: str | None = None
    exception: str | None = None
    exception_type: str | None = None
    exception_traceback: str | None = None

ExecutionCompletedPayload dataclass

Payload for a completed execution — handler invocation or scheduled job.

kind distinguishes the two: listener_id is set when kind == "handler", job_id when kind == "job".

Source code in src/hassette/events/hassette.py

@dataclass(slots=True, frozen=True)
class ExecutionCompletedPayload:
    """Payload for a completed execution — handler invocation or scheduled job.

    ``kind`` distinguishes the two: ``listener_id`` is set when ``kind == "handler"``,
    ``job_id`` when ``kind == "job"``.
    """

    kind: Literal["handler", "job"]
    status: str
    duration_ms: float
    listener_id: int | None = None
    job_id: int | None = None
    app_key: str = ""
    instance_index: int = 0
    error_type: str | None = None

FileWatcherEventPayload dataclass

Payload for file watcher events.

Source code in src/hassette/events/hassette.py

@dataclass(slots=True, frozen=True)
class FileWatcherEventPayload:
    """Payload for file watcher events."""

    changed_file_paths: frozenset[Path]

HassetteAppStateEvent dataclass

Bases: Event[HassettePayload[AppStateChangePayload]]

Event emitted when an app instance changes state.

Source code in src/hassette/events/hassette.py

class HassetteAppStateEvent(Event[HassettePayload[AppStateChangePayload]]):
    """Event emitted when an app instance changes state."""

    @classmethod
    def from_data(
        cls,
        app: "App",
        status: ResourceStatus,
        previous_status: ResourceStatus | None = None,
        exception: Exception | BaseException | None = None,
    ) -> "HassetteAppStateEvent":
        exc_str, exc_type, exc_tb = _extract_exception_fields(exception)
        payload = AppStateChangePayload(
            app_key=app.app_manifest.app_key,
            index=app.index,
            status=status,
            previous_status=previous_status,
            instance_name=app.instance_name,
            class_name=type(app).__name__,
            exception=exc_str,
            exception_type=exc_type,
            exception_traceback=exc_tb,
        )
        return cls(
            topic=Topic.HASSETTE_EVENT_APP_STATE_CHANGED,
            payload=HassettePayload(data=payload),
        )

HassetteExecutionCompletedEvent dataclass

Bases: Event[HassettePayload[ExecutionCompletedPayload]]

Event emitted after a scheduled job execution is persisted to telemetry.

Source code in src/hassette/events/hassette.py

class HassetteExecutionCompletedEvent(Event[HassettePayload[ExecutionCompletedPayload]]):
    """Event emitted after a scheduled job execution is persisted to telemetry."""

    @classmethod
    def from_record(
        cls,
        kind: Literal["handler", "job"],
        status: str,
        duration_ms: float,
        listener_id: int | None = None,
        job_id: int | None = None,
        app_key: str = "",
        instance_index: int = 0,
        error_type: str | None = None,
    ) -> "HassetteExecutionCompletedEvent":
        payload = ExecutionCompletedPayload(
            kind=kind,
            status=status,
            duration_ms=duration_ms,
            listener_id=listener_id,
            job_id=job_id,
            app_key=app_key,
            instance_index=instance_index,
            error_type=error_type,
        )
        return cls(
            topic=Topic.HASSETTE_EVENT_EXECUTION_COMPLETED,
            payload=HassettePayload(data=payload),
        )

HassetteFileWatcherEvent dataclass

Bases: Event[HassettePayload[FileWatcherEventPayload]]

Alias for file watcher events.

Source code in src/hassette/events/hassette.py

class HassetteFileWatcherEvent(Event[HassettePayload[FileWatcherEventPayload]]):
    """Alias for file watcher events."""

    @classmethod
    def create_event(cls, *, changed_file_paths: set[Path]) -> "HassetteFileWatcherEvent":
        payload = FileWatcherEventPayload(changed_file_paths=frozenset(changed_file_paths))
        return cls(
            topic=Topic.HASSETTE_EVENT_FILE_WATCHER,
            payload=HassettePayload(data=payload),
        )

HassetteServiceEvent dataclass

Bases: Event[HassettePayload[ServiceStatusPayload]]

Alias for service status events.

Source code in src/hassette/events/hassette.py

class HassetteServiceEvent(Event[HassettePayload[ServiceStatusPayload]]):
    """Alias for service status events."""

    @classmethod
    def from_data(
        cls,
        resource_name: str,
        role: ResourceRole,
        status: ResourceStatus,
        previous_status: ResourceStatus | None = None,
        exception: Exception | BaseException | None = None,
        ready: bool = False,
        ready_phase: str | None = None,
    ) -> "HassetteServiceEvent":
        exc_str, exc_type, exc_tb = _extract_exception_fields(exception)
        payload = ServiceStatusPayload(
            resource_name=resource_name,
            role=role,
            status=status,
            previous_status=previous_status,
            exception=exc_str,
            exception_type=exc_type,
            exception_traceback=exc_tb,
            ready=ready,
            ready_phase=ready_phase,
        )
        return cls(
            topic=Topic.HASSETTE_EVENT_SERVICE_STATUS,
            payload=HassettePayload(data=payload),
        )

HassetteSimpleEvent dataclass

Bases: Event[HassettePayload[HassetteEmptyPayload]]

Alias for simple events with empty payload.

Source code in src/hassette/events/hassette.py

class HassetteSimpleEvent(Event[HassettePayload[HassetteEmptyPayload]]):
    """Alias for simple events with empty payload."""

    @classmethod
    def create_event(cls, topic: Topic) -> "HassetteSimpleEvent":
        payload = HassetteEmptyPayload()
        return cls(
            topic=topic,
            payload=HassettePayload(data=payload),
        )

ServiceStatusPayload dataclass

Payload for service events.

Source code in src/hassette/events/hassette.py

@dataclass(slots=True, frozen=True)
class ServiceStatusPayload:
    """Payload for service events."""

    resource_name: str
    role: ResourceRole
    status: ResourceStatus
    previous_status: ResourceStatus | None = None
    exception: str | None = None
    exception_type: str | None = None
    exception_traceback: str | None = None
    retry_at: float | None = None
    """Unix timestamp when the next restart will be attempted.
    Populated for EXHAUSTED_COOLING. None for EXHAUSTED_DEAD and all other statuses."""
    ready: bool = False
    """Whether the service has signalled readiness at the time of this status event."""
    ready_phase: str | None = None
    """Human-readable description of the current readiness phase, or None if not available."""

retry_at: float | None = None class-attribute instance-attribute

Unix timestamp when the next restart will be attempted. Populated for EXHAUSTED_COOLING. None for EXHAUSTED_DEAD and all other statuses.

ready: bool = False class-attribute instance-attribute

Whether the service has signalled readiness at the time of this status event.

ready_phase: str | None = None class-attribute instance-attribute

Human-readable description of the current readiness phase, or None if not available.

create_event_from_hass(data: HassEventEnvelopeDict) -> Event

Create an Event from a dictionary.

Source code in src/hassette/events/hass/hass.py

def create_event_from_hass(data: HassEventEnvelopeDict) -> Event:
    """Create an Event from a dictionary."""

    event = data.get("event", {})
    event_type = event.get("event_type")
    if not event_type:
        raise ValueError("Event data must contain 'event_type' key")

    event_data = event.get("data", {}) or {}
    event_payload = {
        "event_type": event_type,
        "origin": event["origin"],
        "context": HassContext(**event["context"]),
        "time_fired": convert_datetime_str_to_system_tz(event["time_fired"]),
    }

    match event_type:
        case "state_changed":
            return RawStateChangeEvent(
                topic=Topic.HASS_EVENT_STATE_CHANGED,
                payload=HassPayload(**event_payload, data=RawStateChangePayload(**event_data)),
            )

        case "call_service":
            return CallServiceEvent(
                topic=Topic.HASS_EVENT_CALL_SERVICE,
                payload=HassPayload(**event_payload, data=CallServicePayload(**event_data)),
            )
        case "component_loaded":
            return ComponentLoadedEvent(
                topic=Topic.HASS_EVENT_COMPONENT_LOADED,
                payload=HassPayload(**event_payload, data=ComponentLoadedPayload(**event_data)),
            )
        case "service_registered":
            return ServiceRegisteredEvent(
                topic=Topic.HASS_EVENT_SERVICE_REGISTERED,
                payload=HassPayload(**event_payload, data=ServiceRegisteredPayload(**event_data)),
            )
        case "service_removed":
            return ServiceRemovedEvent(
                topic=Topic.HASS_EVENT_SERVICE_REMOVED,
                payload=HassPayload(**event_payload, data=ServiceRemovedPayload(**event_data)),
            )
        case "logbook_entry":
            return LogbookEntryEvent(
                topic=Topic.HASS_EVENT_LOGBOOK_ENTRY,
                payload=HassPayload(**event_payload, data=LogbookEntryPayload(**event_data)),
            )
        case "user_added":
            return UserAddedEvent(
                topic=Topic.HASS_EVENT_USER_ADDED,
                payload=HassPayload(**event_payload, data=UserAddedPayload(**event_data)),
            )
        case "user_removed":
            return UserRemovedEvent(
                topic=Topic.HASS_EVENT_USER_REMOVED,
                payload=HassPayload(**event_payload, data=UserRemovedPayload(**event_data)),
            )
        case "automation_triggered":
            return AutomationTriggeredEvent(
                topic=Topic.HASS_EVENT_AUTOMATION_TRIGGERED,
                payload=HassPayload(**event_payload, data=AutomationTriggeredPayload(**event_data)),
            )
        case "script_started":
            return ScriptStartedEvent(
                topic=Topic.HASS_EVENT_SCRIPT_STARTED,
                payload=HassPayload(**event_payload, data=ScriptStartedPayload(**event_data)),
            )
        case _:
            pass

    # fallback to generic event
    return Event(topic=f"hass.event.{event_type}", payload=HassPayload(**event_payload, data=event_data))