From 1c2b8ee606ec5a645d676c83ca896fd6291006d0 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Tue, 22 Mar 2022 14:48:36 +0100 Subject: [PATCH] Add typing to Alert integration (#68365) --- .strict-typing | 1 + homeassistant/components/alert/__init__.py | 102 ++++++++++++--------- mypy.ini | 11 +++ 3 files changed, 72 insertions(+), 42 deletions(-) diff --git a/.strict-typing b/.strict-typing index cf77894f77f..a7a157c54bf 100644 --- a/.strict-typing +++ b/.strict-typing @@ -28,6 +28,7 @@ homeassistant.util.unit_system # --- Add components below this line --- homeassistant.components +homeassistant.components.alert.* homeassistant.components.abode.* homeassistant.components.acer_projector.* homeassistant.components.accuweather.* diff --git a/homeassistant/components/alert/__init__.py b/homeassistant/components/alert/__init__.py index 252bba54b1c..20f3eaf30ca 100644 --- a/homeassistant/components/alert/__init__.py +++ b/homeassistant/components/alert/__init__.py @@ -1,6 +1,10 @@ """Support for repeating alerts when conditions are met.""" +from __future__ import annotations + +from collections.abc import Callable from datetime import timedelta import logging +from typing import Any, final import voluptuous as vol @@ -23,10 +27,15 @@ from homeassistant.const import ( STATE_OFF, STATE_ON, ) -from homeassistant.core import HomeAssistant, ServiceCall -from homeassistant.helpers import event, service +from homeassistant.core import Event, HomeAssistant, ServiceCall +from homeassistant.helpers import service import homeassistant.helpers.config_validation as cv from homeassistant.helpers.entity import ToggleEntity +from homeassistant.helpers.event import ( + async_track_point_in_time, + async_track_state_change_event, +) +from homeassistant.helpers.template import Template from homeassistant.helpers.typing import ConfigType from homeassistant.util.dt import now @@ -62,7 +71,7 @@ ALERT_SCHEMA = vol.Schema( vol.Optional(CONF_DONE_MESSAGE): cv.template, vol.Optional(CONF_TITLE): cv.template, vol.Optional(CONF_DATA): dict, - vol.Required(CONF_NOTIFIERS): cv.ensure_list, + vol.Required(CONF_NOTIFIERS): vol.All(cv.ensure_list, [cv.string]), } ) @@ -73,14 +82,14 @@ CONFIG_SCHEMA = vol.Schema( ALERT_SERVICE_SCHEMA = vol.Schema({vol.Required(ATTR_ENTITY_ID): cv.entity_ids}) -def is_on(hass, entity_id): +def is_on(hass: HomeAssistant, entity_id: str) -> bool: """Return if the alert is firing and not acknowledged.""" return hass.states.is_state(entity_id, STATE_ON) async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the Alert component.""" - entities = [] + entities: list[Alert] = [] for object_id, cfg in config[DOMAIN].items(): if not cfg: @@ -144,10 +153,16 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: schema=ALERT_SERVICE_SCHEMA, ) hass.services.async_register( - DOMAIN, SERVICE_TURN_ON, async_handle_alert_service, schema=ALERT_SERVICE_SCHEMA + DOMAIN, + SERVICE_TURN_ON, + async_handle_alert_service, + schema=ALERT_SERVICE_SCHEMA, ) hass.services.async_register( - DOMAIN, SERVICE_TOGGLE, async_handle_alert_service, schema=ALERT_SERVICE_SCHEMA + DOMAIN, + SERVICE_TOGGLE, + async_handle_alert_service, + schema=ALERT_SERVICE_SCHEMA, ) for alert in entities: @@ -163,20 +178,20 @@ class Alert(ToggleEntity): def __init__( self, - hass, - entity_id, - name, - watched_entity_id, - state, - repeat, - skip_first, - message_template, - done_message_template, - notifiers, - can_ack, - title_template, - data, - ): + hass: HomeAssistant, + entity_id: str, + name: str, + watched_entity_id: str, + state: str, + repeat: list[float], + skip_first: bool, + message_template: Template | None, + done_message_template: Template | None, + notifiers: list[str], + can_ack: bool, + title_template: Template | None, + data: dict[Any, Any], + ) -> None: """Initialize the alert.""" self.hass = hass self._attr_name = name @@ -204,16 +219,18 @@ class Alert(ToggleEntity): self._firing = False self._ack = False - self._cancel = None + self._cancel: Callable[[], None] | None = None self._send_done_message = False self.entity_id = f"{DOMAIN}.{entity_id}" - event.async_track_state_change_event( + async_track_state_change_event( hass, [watched_entity_id], self.watched_entity_change ) + @final # type: ignore[misc] @property - def state(self): # pylint: disable=overridden-final-method + # pylint: disable=overridden-final-method + def state(self) -> str: # type: ignore[override] """Return the alert status.""" if self._firing: if self._ack: @@ -221,17 +238,17 @@ class Alert(ToggleEntity): return STATE_ON return STATE_IDLE - async def watched_entity_change(self, ev): + async def watched_entity_change(self, event: Event) -> None: """Determine if the alert should start or stop.""" - if (to_state := ev.data.get("new_state")) is None: + if (to_state := event.data.get("new_state")) is None: return - _LOGGER.debug("Watched entity (%s) has changed", ev.data.get("entity_id")) + _LOGGER.debug("Watched entity (%s) has changed", event.data.get("entity_id")) if to_state.state == self._alert_state and not self._firing: await self.begin_alerting() if to_state.state != self._alert_state and self._firing: await self.end_alerting() - async def begin_alerting(self): + async def begin_alerting(self) -> None: """Begin the alert procedures.""" _LOGGER.debug("Beginning Alert: %s", self._attr_name) self._ack = False @@ -245,26 +262,27 @@ class Alert(ToggleEntity): self.async_write_ha_state() - async def end_alerting(self): + async def end_alerting(self) -> None: """End the alert procedures.""" _LOGGER.debug("Ending Alert: %s", self._attr_name) - self._cancel() + if self._cancel is not None: + self._cancel() + self._cancel = None + self._ack = False self._firing = False if self._send_done_message: await self._notify_done_message() self.async_write_ha_state() - async def _schedule_notify(self): + async def _schedule_notify(self) -> None: """Schedule a notification.""" delay = self._delay[self._next_delay] next_msg = now() + delay - self._cancel = event.async_track_point_in_time( - self.hass, self._notify, next_msg - ) + self._cancel = async_track_point_in_time(self.hass, self._notify, next_msg) self._next_delay = min(self._next_delay + 1, len(self._delay) - 1) - async def _notify(self, *args): + async def _notify(self, *args: Any) -> None: """Send the alert notification.""" if not self._firing: return @@ -281,7 +299,7 @@ class Alert(ToggleEntity): await self._send_notification_message(message) await self._schedule_notify() - async def _notify_done_message(self, *args): + async def _notify_done_message(self) -> None: """Send notification of complete alert.""" _LOGGER.info("Alerting: %s", self._done_message_template) self._send_done_message = False @@ -293,15 +311,15 @@ class Alert(ToggleEntity): await self._send_notification_message(message) - async def _send_notification_message(self, message): + async def _send_notification_message(self, message: Any) -> None: msg_payload = {ATTR_MESSAGE: message} if self._title_template is not None: title = self._title_template.async_render(parse_result=False) - msg_payload.update({ATTR_TITLE: title}) + msg_payload[ATTR_TITLE] = title if self._data: - msg_payload.update({ATTR_DATA: self._data}) + msg_payload[ATTR_DATA] = self._data _LOGGER.debug(msg_payload) @@ -310,19 +328,19 @@ class Alert(ToggleEntity): DOMAIN_NOTIFY, target, msg_payload, context=self._context ) - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Async Unacknowledge alert.""" _LOGGER.debug("Reset Alert: %s", self._attr_name) self._ack = False self.async_write_ha_state() - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Async Acknowledge alert.""" _LOGGER.debug("Acknowledged Alert: %s", self._attr_name) self._ack = True self.async_write_ha_state() - async def async_toggle(self, **kwargs): + async def async_toggle(self, **kwargs: Any) -> None: """Async toggle alert.""" if self._ack: return await self.async_turn_on() diff --git a/mypy.ini b/mypy.ini index 2f1792438a5..293b8a62c37 100644 --- a/mypy.ini +++ b/mypy.ini @@ -110,6 +110,17 @@ warn_return_any = true warn_unreachable = true no_implicit_reexport = true +[mypy-homeassistant.components.alert.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +no_implicit_optional = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.abode.*] check_untyped_defs = true disallow_incomplete_defs = true