Add typing to Alert integration (#68365)

This commit is contained in:
Franck Nijhof 2022-03-22 14:48:36 +01:00 committed by GitHub
parent a8df10bb2c
commit 1c2b8ee606
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 72 additions and 42 deletions

View File

@ -28,6 +28,7 @@ homeassistant.util.unit_system
# --- Add components below this line --- # --- Add components below this line ---
homeassistant.components homeassistant.components
homeassistant.components.alert.*
homeassistant.components.abode.* homeassistant.components.abode.*
homeassistant.components.acer_projector.* homeassistant.components.acer_projector.*
homeassistant.components.accuweather.* homeassistant.components.accuweather.*

View File

@ -1,6 +1,10 @@
"""Support for repeating alerts when conditions are met.""" """Support for repeating alerts when conditions are met."""
from __future__ import annotations
from collections.abc import Callable
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any, final
import voluptuous as vol import voluptuous as vol
@ -23,10 +27,15 @@ from homeassistant.const import (
STATE_OFF, STATE_OFF,
STATE_ON, STATE_ON,
) )
from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.core import Event, HomeAssistant, ServiceCall
from homeassistant.helpers import event, service from homeassistant.helpers import service
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import ToggleEntity 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.helpers.typing import ConfigType
from homeassistant.util.dt import now from homeassistant.util.dt import now
@ -62,7 +71,7 @@ ALERT_SCHEMA = vol.Schema(
vol.Optional(CONF_DONE_MESSAGE): cv.template, vol.Optional(CONF_DONE_MESSAGE): cv.template,
vol.Optional(CONF_TITLE): cv.template, vol.Optional(CONF_TITLE): cv.template,
vol.Optional(CONF_DATA): dict, 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}) 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 if the alert is firing and not acknowledged."""
return hass.states.is_state(entity_id, STATE_ON) return hass.states.is_state(entity_id, STATE_ON)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Alert component.""" """Set up the Alert component."""
entities = [] entities: list[Alert] = []
for object_id, cfg in config[DOMAIN].items(): for object_id, cfg in config[DOMAIN].items():
if not cfg: if not cfg:
@ -144,10 +153,16 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
schema=ALERT_SERVICE_SCHEMA, schema=ALERT_SERVICE_SCHEMA,
) )
hass.services.async_register( 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( 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: for alert in entities:
@ -163,20 +178,20 @@ class Alert(ToggleEntity):
def __init__( def __init__(
self, self,
hass, hass: HomeAssistant,
entity_id, entity_id: str,
name, name: str,
watched_entity_id, watched_entity_id: str,
state, state: str,
repeat, repeat: list[float],
skip_first, skip_first: bool,
message_template, message_template: Template | None,
done_message_template, done_message_template: Template | None,
notifiers, notifiers: list[str],
can_ack, can_ack: bool,
title_template, title_template: Template | None,
data, data: dict[Any, Any],
): ) -> None:
"""Initialize the alert.""" """Initialize the alert."""
self.hass = hass self.hass = hass
self._attr_name = name self._attr_name = name
@ -204,16 +219,18 @@ class Alert(ToggleEntity):
self._firing = False self._firing = False
self._ack = False self._ack = False
self._cancel = None self._cancel: Callable[[], None] | None = None
self._send_done_message = False self._send_done_message = False
self.entity_id = f"{DOMAIN}.{entity_id}" 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 hass, [watched_entity_id], self.watched_entity_change
) )
@final # type: ignore[misc]
@property @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.""" """Return the alert status."""
if self._firing: if self._firing:
if self._ack: if self._ack:
@ -221,17 +238,17 @@ class Alert(ToggleEntity):
return STATE_ON return STATE_ON
return STATE_IDLE 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.""" """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 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: if to_state.state == self._alert_state and not self._firing:
await self.begin_alerting() await self.begin_alerting()
if to_state.state != self._alert_state and self._firing: if to_state.state != self._alert_state and self._firing:
await self.end_alerting() await self.end_alerting()
async def begin_alerting(self): async def begin_alerting(self) -> None:
"""Begin the alert procedures.""" """Begin the alert procedures."""
_LOGGER.debug("Beginning Alert: %s", self._attr_name) _LOGGER.debug("Beginning Alert: %s", self._attr_name)
self._ack = False self._ack = False
@ -245,26 +262,27 @@ class Alert(ToggleEntity):
self.async_write_ha_state() self.async_write_ha_state()
async def end_alerting(self): async def end_alerting(self) -> None:
"""End the alert procedures.""" """End the alert procedures."""
_LOGGER.debug("Ending Alert: %s", self._attr_name) _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._ack = False
self._firing = False self._firing = False
if self._send_done_message: if self._send_done_message:
await self._notify_done_message() await self._notify_done_message()
self.async_write_ha_state() self.async_write_ha_state()
async def _schedule_notify(self): async def _schedule_notify(self) -> None:
"""Schedule a notification.""" """Schedule a notification."""
delay = self._delay[self._next_delay] delay = self._delay[self._next_delay]
next_msg = now() + delay next_msg = now() + delay
self._cancel = event.async_track_point_in_time( self._cancel = async_track_point_in_time(self.hass, self._notify, next_msg)
self.hass, self._notify, next_msg
)
self._next_delay = min(self._next_delay + 1, len(self._delay) - 1) 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.""" """Send the alert notification."""
if not self._firing: if not self._firing:
return return
@ -281,7 +299,7 @@ class Alert(ToggleEntity):
await self._send_notification_message(message) await self._send_notification_message(message)
await self._schedule_notify() await self._schedule_notify()
async def _notify_done_message(self, *args): async def _notify_done_message(self) -> None:
"""Send notification of complete alert.""" """Send notification of complete alert."""
_LOGGER.info("Alerting: %s", self._done_message_template) _LOGGER.info("Alerting: %s", self._done_message_template)
self._send_done_message = False self._send_done_message = False
@ -293,15 +311,15 @@ class Alert(ToggleEntity):
await self._send_notification_message(message) 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} msg_payload = {ATTR_MESSAGE: message}
if self._title_template is not None: if self._title_template is not None:
title = self._title_template.async_render(parse_result=False) title = self._title_template.async_render(parse_result=False)
msg_payload.update({ATTR_TITLE: title}) msg_payload[ATTR_TITLE] = title
if self._data: if self._data:
msg_payload.update({ATTR_DATA: self._data}) msg_payload[ATTR_DATA] = self._data
_LOGGER.debug(msg_payload) _LOGGER.debug(msg_payload)
@ -310,19 +328,19 @@ class Alert(ToggleEntity):
DOMAIN_NOTIFY, target, msg_payload, context=self._context 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.""" """Async Unacknowledge alert."""
_LOGGER.debug("Reset Alert: %s", self._attr_name) _LOGGER.debug("Reset Alert: %s", self._attr_name)
self._ack = False self._ack = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Async Acknowledge alert.""" """Async Acknowledge alert."""
_LOGGER.debug("Acknowledged Alert: %s", self._attr_name) _LOGGER.debug("Acknowledged Alert: %s", self._attr_name)
self._ack = True self._ack = True
self.async_write_ha_state() self.async_write_ha_state()
async def async_toggle(self, **kwargs): async def async_toggle(self, **kwargs: Any) -> None:
"""Async toggle alert.""" """Async toggle alert."""
if self._ack: if self._ack:
return await self.async_turn_on() return await self.async_turn_on()

View File

@ -110,6 +110,17 @@ warn_return_any = true
warn_unreachable = true warn_unreachable = true
no_implicit_reexport = 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.*] [mypy-homeassistant.components.abode.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true