jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-24 05:37:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve generic event typing [mqtt_statestream] (#114732)

Browse Source
This commit is contained in:
Marc Mueller 2024-04-06 14:19:39 +02:00 committed by GitHub
parent e0f5559c8f
commit bf0309a722
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/mqtt_statestream/__init__.py
View File

@ -1,16 +1,14 @@
"""Publish simple item state changes via MQTT."""
from collections.abc import Mapping
import json
import logging
from typing import Any
import voluptuous as vol
from homeassistant.components import mqtt
from homeassistant.components.mqtt import valid_publish_topic
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, EVENT_STATE_CHANGED
from homeassistant.core import Event, HomeAssistant, State, callback
from homeassistant.core import Event, EventStateChangedData, HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entityfilter import (
INCLUDE_EXCLUDE_BASE_FILTER_SCHEMA,
@ -57,9 +55,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
if not base_topic.endswith("/"):
base_topic = f"{base_topic}/"
async def _state_publisher(evt: Event) -> None:
entity_id: str = evt.data["entity_id"]
new_state: State = evt.data["new_state"]
async def _state_publisher(evt: Event[EventStateChangedData]) -> None:
entity_id = evt.data["entity_id"]
assert (new_state := evt.data["new_state"])
payload = new_state.state
@ -92,9 +90,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@callback
def _ha_started(hass: HomeAssistant) -> None:
@callback
def _event_filter(event_data: Mapping[str, Any]) -> bool:
entity_id: str = event_data["entity_id"]
new_state: State | None = event_data["new_state"]
def _event_filter(event_data: EventStateChangedData) -> bool:
entity_id = event_data["entity_id"]
new_state = event_data["new_state"]
if new_state is None:
return False
if not publish_filter(entity_id):