Improve async_track_template_result callback typing (#97135)

This commit is contained in:
Marc Mueller 2023-07-24 12:42:17 +02:00 committed by GitHub
parent c0da6b822e
commit 582499a260
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 228 additions and 72 deletions

View File

@ -27,7 +27,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE, STATE_UNAVAILABLE,
STATE_UNKNOWN, STATE_UNKNOWN,
) )
from homeassistant.core import Event, HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConditionError, TemplateError from homeassistant.exceptions import ConditionError, TemplateError
from homeassistant.helpers import condition from homeassistant.helpers import condition
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
@ -259,14 +259,13 @@ class BayesianBinarySensor(BinarySensorEntity):
@callback @callback
def _async_template_result_changed( def _async_template_result_changed(
event: Event | None, updates: list[TrackTemplateResult] event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None: ) -> None:
track_template_result = updates.pop() track_template_result = updates.pop()
template = track_template_result.template template = track_template_result.template
result = track_template_result.result result = track_template_result.result
entity: str | None = ( entity_id = None if event is None else event.data["entity_id"]
None if event is None else event.data.get(CONF_ENTITY_ID)
)
if isinstance(result, TemplateError): if isinstance(result, TemplateError):
_LOGGER.error( _LOGGER.error(
"TemplateError('%s') while processing template '%s' in entity '%s'", "TemplateError('%s') while processing template '%s' in entity '%s'",
@ -283,8 +282,8 @@ class BayesianBinarySensor(BinarySensorEntity):
observation.observed = observed observation.observed = observed
# in some cases a template may update because of the absence of an entity # in some cases a template may update because of the absence of an entity
if entity is not None: if entity_id is not None:
observation.entity_id = entity observation.entity_id = entity_id
self.current_observations[observation.id] = observation self.current_observations[observation.id] = observation

View File

@ -1,5 +1,7 @@
"""Offer template automation rules.""" """Offer template automation rules."""
from datetime import timedelta
import logging import logging
from typing import Any
import voluptuous as vol import voluptuous as vol
@ -8,13 +10,15 @@ from homeassistant.const import CONF_FOR, CONF_PLATFORM, CONF_VALUE_TEMPLATE
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, template from homeassistant.helpers import config_validation as cv, template
from homeassistant.helpers.event import ( from homeassistant.helpers.event import (
EventStateChangedData,
TrackTemplate, TrackTemplate,
TrackTemplateResult,
async_call_later, async_call_later,
async_track_template_result, async_track_template_result,
) )
from homeassistant.helpers.template import Template, result_as_boolean from homeassistant.helpers.template import Template, result_as_boolean
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType, EventType
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -59,7 +63,10 @@ async def async_attach_trigger(
) )
@callback @callback
def template_listener(event, updates): def template_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
"""Listen for state changes and calls action.""" """Listen for state changes and calls action."""
nonlocal delay_cancel, armed nonlocal delay_cancel, armed
result = updates.pop().result result = updates.pop().result
@ -88,9 +95,9 @@ async def async_attach_trigger(
# Fire! # Fire!
armed = False armed = False
entity_id = event and event.data.get("entity_id") entity_id = event and event.data["entity_id"]
from_s = event and event.data.get("old_state") from_s = event and event.data["old_state"]
to_s = event and event.data.get("new_state") to_s = event and event.data["new_state"]
if entity_id is not None: if entity_id is not None:
description = f"{entity_id} via template" description = f"{entity_id} via template"
@ -110,7 +117,7 @@ async def async_attach_trigger(
} }
@callback @callback
def call_action(*_): def call_action(*_: Any) -> None:
"""Call action with right context.""" """Call action with right context."""
nonlocal trigger_variables nonlocal trigger_variables
hass.async_run_hass_job( hass.async_run_hass_job(
@ -124,7 +131,7 @@ async def async_attach_trigger(
return return
try: try:
period = cv.positive_time_period( period: timedelta = cv.positive_time_period(
template.render_complex(time_delta, {"trigger": template_variables}) template.render_complex(time_delta, {"trigger": template_variables})
) )
except (exceptions.TemplateError, vol.Invalid) as ex: except (exceptions.TemplateError, vol.Invalid) as ex:

View File

@ -87,6 +87,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import ( from homeassistant.helpers.event import (
EventStateChangedData, EventStateChangedData,
TrackTemplate, TrackTemplate,
TrackTemplateResult,
async_track_state_change_event, async_track_state_change_event,
async_track_template_result, async_track_template_result,
) )
@ -192,7 +193,10 @@ class UniversalMediaPlayer(MediaPlayerEntity):
self.async_schedule_update_ha_state(True) self.async_schedule_update_ha_state(True)
@callback @callback
def _async_on_template_update(event, updates): def _async_on_template_update(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
"""Update state when template state changes.""" """Update state when template state changes."""
for data in updates: for data in updates:
template = data.template template = data.template

View File

@ -26,6 +26,7 @@ from homeassistant.exceptions import (
from homeassistant.helpers import config_validation as cv, entity, template from homeassistant.helpers import config_validation as cv, entity, template
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.event import ( from homeassistant.helpers.event import (
EventStateChangedData,
TrackTemplate, TrackTemplate,
TrackTemplateResult, TrackTemplateResult,
async_track_template_result, async_track_template_result,
@ -37,6 +38,7 @@ from homeassistant.helpers.json import (
json_dumps, json_dumps,
) )
from homeassistant.helpers.service import async_get_all_descriptions from homeassistant.helpers.service import async_get_all_descriptions
from homeassistant.helpers.typing import EventType
from homeassistant.loader import ( from homeassistant.loader import (
Integration, Integration,
IntegrationNotFound, IntegrationNotFound,
@ -535,7 +537,8 @@ async def handle_render_template(
@callback @callback
def _template_listener( def _template_listener(
event: Event | None, updates: list[TrackTemplateResult] event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None: ) -> None:
nonlocal info nonlocal info
track_template_result = updates.pop() track_template_result = updates.pop()

View File

@ -888,9 +888,7 @@ class TrackTemplateResultInfo:
self, self,
hass: HomeAssistant, hass: HomeAssistant,
track_templates: Sequence[TrackTemplate], track_templates: Sequence[TrackTemplate],
action: Callable[ action: TrackTemplateResultListener,
[EventType[EventStateChangedData] | None, list[TrackTemplateResult]], None
],
has_super_template: bool = False, has_super_template: bool = False,
) -> None: ) -> None:
"""Handle removal / refresh of tracker init.""" """Handle removal / refresh of tracker init."""
@ -1209,7 +1207,7 @@ TrackTemplateResultListener = Callable[
EventType[EventStateChangedData] | None, EventType[EventStateChangedData] | None,
list[TrackTemplateResult], list[TrackTemplateResult],
], ],
None, Coroutine[Any, Any, None] | None,
] ]
"""Type for the listener for template results. """Type for the listener for template results.

View File

@ -15,7 +15,6 @@ from homeassistant.components.sensor import (
SensorEntity, SensorEntity,
) )
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_ENTITY_PICTURE, ATTR_ENTITY_PICTURE,
ATTR_FRIENDLY_NAME, ATTR_FRIENDLY_NAME,
ATTR_ICON, ATTR_ICON,
@ -33,7 +32,12 @@ from homeassistant.util.json import JSON_DECODE_EXCEPTIONS, json_loads
from . import config_validation as cv from . import config_validation as cv
from .entity import Entity from .entity import Entity
from .event import TrackTemplate, TrackTemplateResult, async_track_template_result from .event import (
EventStateChangedData,
TrackTemplate,
TrackTemplateResult,
async_track_template_result,
)
from .script import Script, _VarsType from .script import Script, _VarsType
from .template import ( from .template import (
Template, Template,
@ -42,7 +46,7 @@ from .template import (
render_complex, render_complex,
result_as_boolean, result_as_boolean,
) )
from .typing import ConfigType from .typing import ConfigType, EventType
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -327,14 +331,14 @@ class TemplateEntity(Entity):
@callback @callback
def _handle_results( def _handle_results(
self, self,
event: Event | None, event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult], updates: list[TrackTemplateResult],
) -> None: ) -> None:
"""Call back the results to the attributes.""" """Call back the results to the attributes."""
if event: if event:
self.async_set_context(event.context) self.async_set_context(event.context)
entity_id = event and event.data.get(ATTR_ENTITY_ID) entity_id = event and event.data["entity_id"]
if entity_id and entity_id == self.entity_id: if entity_id and entity_id == self.entity_id:
self._self_ref_update_count += 1 self._self_ref_update_count += 1

View File

@ -965,7 +965,10 @@ async def test_track_template_result(hass: HomeAssistant) -> None:
"{{(states.sensor.test.state|int) + test }}", hass "{{(states.sensor.test.state|int) + test }}", hass
) )
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
specific_runs.append(int(track_result.result)) specific_runs.append(int(track_result.result))
@ -974,7 +977,10 @@ async def test_track_template_result(hass: HomeAssistant) -> None:
) )
@ha.callback @ha.callback
def wildcard_run_callback(event, updates): def wildcard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
wildcard_runs.append( wildcard_runs.append(
(int(track_result.last_result or 0), int(track_result.result)) (int(track_result.last_result or 0), int(track_result.result))
@ -984,7 +990,10 @@ async def test_track_template_result(hass: HomeAssistant) -> None:
hass, [TrackTemplate(template_condition, None)], wildcard_run_callback hass, [TrackTemplate(template_condition, None)], wildcard_run_callback
) )
async def wildercard_run_callback(event, updates): async def wildercard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
wildercard_runs.append( wildercard_runs.append(
(int(track_result.last_result or 0), int(track_result.result)) (int(track_result.last_result or 0), int(track_result.result))
@ -1051,7 +1060,10 @@ async def test_track_template_result_none(hass: HomeAssistant) -> None:
"{{(state_attr('sensor.test', 'battery')|int(default=0)) + test }}", hass "{{(state_attr('sensor.test', 'battery')|int(default=0)) + test }}", hass
) )
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
result = int(track_result.result) if track_result.result is not None else None result = int(track_result.result) if track_result.result is not None else None
specific_runs.append(result) specific_runs.append(result)
@ -1061,7 +1073,10 @@ async def test_track_template_result_none(hass: HomeAssistant) -> None:
) )
@ha.callback @ha.callback
def wildcard_run_callback(event, updates): def wildcard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
last_result = ( last_result = (
int(track_result.last_result) int(track_result.last_result)
@ -1075,7 +1090,10 @@ async def test_track_template_result_none(hass: HomeAssistant) -> None:
hass, [TrackTemplate(template_condition, None)], wildcard_run_callback hass, [TrackTemplate(template_condition, None)], wildcard_run_callback
) )
async def wildercard_run_callback(event, updates): async def wildercard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
last_result = ( last_result = (
int(track_result.last_result) int(track_result.last_result)
@ -1122,7 +1140,10 @@ async def test_track_template_result_super_template(hass: HomeAssistant) -> None
"{{(states.sensor.test.state|int) + test }}", hass "{{(states.sensor.test.state|int) + test }}", hass
) )
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
specific_runs.append(int(track_result.result)) specific_runs.append(int(track_result.result))
@ -1140,7 +1161,10 @@ async def test_track_template_result_super_template(hass: HomeAssistant) -> None
) )
@ha.callback @ha.callback
def wildcard_run_callback(event, updates): def wildcard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
wildcard_runs.append( wildcard_runs.append(
@ -1159,7 +1183,10 @@ async def test_track_template_result_super_template(hass: HomeAssistant) -> None
has_super_template=True, has_super_template=True,
) )
async def wildercard_run_callback(event, updates): async def wildercard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition_var: if track_result.template is template_condition_var:
wildercard_runs.append( wildercard_runs.append(
@ -1272,7 +1299,10 @@ async def test_track_template_result_super_template_initially_false(
hass.states.async_set("sensor.test", "unavailable") hass.states.async_set("sensor.test", "unavailable")
await hass.async_block_till_done() await hass.async_block_till_done()
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
specific_runs.append(int(track_result.result)) specific_runs.append(int(track_result.result))
@ -1290,7 +1320,10 @@ async def test_track_template_result_super_template_initially_false(
) )
@ha.callback @ha.callback
def wildcard_run_callback(event, updates): def wildcard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
wildcard_runs.append( wildcard_runs.append(
@ -1309,7 +1342,10 @@ async def test_track_template_result_super_template_initially_false(
has_super_template=True, has_super_template=True,
) )
async def wildercard_run_callback(event, updates): async def wildercard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition_var: if track_result.template is template_condition_var:
wildercard_runs.append( wildercard_runs.append(
@ -1434,7 +1470,10 @@ async def test_track_template_result_super_template_2(
return result_as_boolean(result) return result_as_boolean(result)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
specific_runs.append(int(track_result.result)) specific_runs.append(int(track_result.result))
@ -1454,7 +1493,10 @@ async def test_track_template_result_super_template_2(
) )
@ha.callback @ha.callback
def wildcard_run_callback(event, updates): def wildcard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
wildcard_runs.append( wildcard_runs.append(
@ -1475,7 +1517,10 @@ async def test_track_template_result_super_template_2(
has_super_template=True, has_super_template=True,
) )
async def wildercard_run_callback(event, updates): async def wildercard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition_var: if track_result.template is template_condition_var:
wildercard_runs.append( wildercard_runs.append(
@ -1580,7 +1625,10 @@ async def test_track_template_result_super_template_2_initially_false(
return result_as_boolean(result) return result_as_boolean(result)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
specific_runs.append(int(track_result.result)) specific_runs.append(int(track_result.result))
@ -1600,7 +1648,10 @@ async def test_track_template_result_super_template_2_initially_false(
) )
@ha.callback @ha.callback
def wildcard_run_callback(event, updates): def wildcard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition: if track_result.template is template_condition:
wildcard_runs.append( wildcard_runs.append(
@ -1621,7 +1672,10 @@ async def test_track_template_result_super_template_2_initially_false(
has_super_template=True, has_super_template=True,
) )
async def wildercard_run_callback(event, updates): async def wildercard_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_condition_var: if track_result.template is template_condition_var:
wildercard_runs.append( wildercard_runs.append(
@ -1701,7 +1755,10 @@ async def test_track_template_result_complex(hass: HomeAssistant) -> None:
""" """
template_complex = Template(template_complex_str, hass) template_complex = Template(template_complex_str, hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
hass.states.async_set("light.one", "on") hass.states.async_set("light.one", "on")
@ -1854,7 +1911,10 @@ async def test_track_template_result_with_wildcard(hass: HomeAssistant) -> None:
""" """
template_complex = Template(template_complex_str, hass) template_complex = Template(template_complex_str, hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
hass.states.async_set("cover.office_drapes", "closed") hass.states.async_set("cover.office_drapes", "closed")
@ -1906,7 +1966,10 @@ async def test_track_template_result_with_group(hass: HomeAssistant) -> None:
""" """
template_complex = Template(template_complex_str, hass) template_complex = Template(template_complex_str, hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -1963,7 +2026,10 @@ async def test_track_template_result_and_conditional(hass: HomeAssistant) -> Non
template = Template(template_str, hass) template = Template(template_str, hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2028,7 +2094,10 @@ async def test_track_template_result_and_conditional_upper_case(
template = Template(template_str, hass) template = Template(template_str, hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2087,7 +2156,10 @@ async def test_track_template_result_iterator(hass: HomeAssistant) -> None:
iterator_runs = [] iterator_runs = []
@ha.callback @ha.callback
def iterator_callback(event, updates): def iterator_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
iterator_runs.append(updates.pop().result) iterator_runs.append(updates.pop().result)
async_track_template_result( async_track_template_result(
@ -2120,7 +2192,10 @@ async def test_track_template_result_iterator(hass: HomeAssistant) -> None:
filter_runs = [] filter_runs = []
@ha.callback @ha.callback
def filter_callback(event, updates): def filter_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
filter_runs.append(updates.pop().result) filter_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2170,7 +2245,10 @@ async def test_track_template_result_errors(
not_exist_runs = [] not_exist_runs = []
@ha.callback @ha.callback
def syntax_error_listener(event, updates): def syntax_error_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
syntax_error_runs.append( syntax_error_runs.append(
( (
@ -2190,7 +2268,10 @@ async def test_track_template_result_errors(
assert "TemplateSyntaxError" in caplog.text assert "TemplateSyntaxError" in caplog.text
@ha.callback @ha.callback
def not_exist_runs_error_listener(event, updates): def not_exist_runs_error_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
template_track = updates.pop() template_track = updates.pop()
not_exist_runs.append( not_exist_runs.append(
( (
@ -2255,7 +2336,10 @@ async def test_track_template_result_transient_errors(
sometimes_error_runs = [] sometimes_error_runs = []
@ha.callback @ha.callback
def sometimes_error_listener(event, updates): def sometimes_error_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
track_result = updates.pop() track_result = updates.pop()
sometimes_error_runs.append( sometimes_error_runs.append(
( (
@ -2300,7 +2384,10 @@ async def test_static_string(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2320,7 +2407,10 @@ async def test_track_template_rate_limit(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2379,7 +2469,10 @@ async def test_track_template_rate_limit_super(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_refresh: if track_result.template is template_refresh:
refresh_runs.append(track_result.result) refresh_runs.append(track_result.result)
@ -2452,7 +2545,10 @@ async def test_track_template_rate_limit_super_2(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_refresh: if track_result.template is template_refresh:
refresh_runs.append(track_result.result) refresh_runs.append(track_result.result)
@ -2521,7 +2617,10 @@ async def test_track_template_rate_limit_super_3(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for track_result in updates: for track_result in updates:
if track_result.template is template_refresh: if track_result.template is template_refresh:
refresh_runs.append(track_result.result) refresh_runs.append(track_result.result)
@ -2592,7 +2691,10 @@ async def test_track_template_rate_limit_suppress_listener(hass: HomeAssistant)
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2689,7 +2791,10 @@ async def test_track_template_rate_limit_five(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2725,7 +2830,10 @@ async def test_track_template_has_default_rate_limit(hass: HomeAssistant) -> Non
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2766,7 +2874,10 @@ async def test_track_template_unavailable_states_has_default_rate_limit(
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2807,7 +2918,10 @@ async def test_specifically_referenced_entity_is_not_rate_limited(
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2850,7 +2964,10 @@ async def test_track_two_templates_with_different_rate_limits(
} }
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
for update in updates: for update in updates:
refresh_runs[update.template].append(update.result) refresh_runs[update.template].append(update.result)
@ -2911,7 +3028,10 @@ async def test_string(hass: HomeAssistant) -> None:
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2931,7 +3051,10 @@ async def test_track_template_result_refresh_cancel(hass: HomeAssistant) -> None
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates.pop().result) refresh_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -2993,7 +3116,10 @@ async def test_async_track_template_result_multiple_templates(
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates) refresh_runs.append(updates)
async_track_template_result( async_track_template_result(
@ -3054,7 +3180,10 @@ async def test_async_track_template_result_multiple_templates_mixing_domain(
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates) refresh_runs.append(updates)
async_track_template_result( async_track_template_result(
@ -3139,7 +3268,10 @@ async def test_track_template_with_time(hass: HomeAssistant) -> None:
specific_runs = [] specific_runs = []
template_complex = Template("{{ states.switch.test.state and now() }}", hass) template_complex = Template("{{ states.switch.test.state and now() }}", hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -3169,7 +3301,10 @@ async def test_track_template_with_time_default(hass: HomeAssistant) -> None:
specific_runs = [] specific_runs = []
template_complex = Template("{{ now() }}", hass) template_complex = Template("{{ now() }}", hass)
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -3218,7 +3353,10 @@ async def test_track_template_with_time_that_leaves_scope(hass: HomeAssistant) -
hass, hass,
) )
def specific_run_callback(event, updates): def specific_run_callback(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
specific_runs.append(updates.pop().result) specific_runs.append(updates.pop().result)
info = async_track_template_result( info = async_track_template_result(
@ -3283,7 +3421,10 @@ async def test_async_track_template_result_multiple_templates_mixing_listeners(
refresh_runs = [] refresh_runs = []
@ha.callback @ha.callback
def refresh_listener(event, updates): def refresh_listener(
event: EventType[EventStateChangedData] | None,
updates: list[TrackTemplateResult],
) -> None:
refresh_runs.append(updates) refresh_runs.append(updates)
now = dt_util.utcnow() now = dt_util.utcnow()