Enforce strict typing for SimpliSafe (#53417)

This commit is contained in:
Aaron Bach 2021-07-27 14:11:54 -06:00 committed by GitHub
parent f71980a634
commit f92ba75791
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 211 additions and 97 deletions

View File

@ -84,6 +84,7 @@ homeassistant.components.scene.*
homeassistant.components.select.* homeassistant.components.select.*
homeassistant.components.sensor.* homeassistant.components.sensor.*
homeassistant.components.shelly.* homeassistant.components.shelly.*
homeassistant.components.simplisafe.*
homeassistant.components.slack.* homeassistant.components.slack.*
homeassistant.components.sonos.media_player homeassistant.components.sonos.media_player
homeassistant.components.ssdp.* homeassistant.components.ssdp.*

View File

@ -1,17 +1,28 @@
"""Support for SimpliSafe alarm systems.""" """Support for SimpliSafe alarm systems."""
from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable
from typing import Callable, cast
from uuid import UUID from uuid import UUID
from simplipy import get_api from simplipy import get_api
from simplipy.api import API
from simplipy.errors import ( from simplipy.errors import (
EndpointUnavailableError, EndpointUnavailableError,
InvalidCredentialsError, InvalidCredentialsError,
SimplipyError, SimplipyError,
) )
from simplipy.sensor.v2 import SensorV2
from simplipy.sensor.v3 import SensorV3
from simplipy.system import SystemNotification
from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_CODE, CONF_CODE, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import ATTR_CODE, CONF_CODE, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import CoreState, callback from homeassistant.core import CoreState, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import ( from homeassistant.helpers import (
aiohttp_client, aiohttp_client,
@ -109,7 +120,7 @@ SERVICE_SET_SYSTEM_PROPERTIES_SCHEMA = SERVICE_BASE_SCHEMA.extend(
CONFIG_SCHEMA = cv.deprecated(DOMAIN) CONFIG_SCHEMA = cv.deprecated(DOMAIN)
async def async_get_client_id(hass): async def async_get_client_id(hass: HomeAssistant) -> str:
"""Get a client ID (based on the HASS unique ID) for the SimpliSafe API. """Get a client ID (based on the HASS unique ID) for the SimpliSafe API.
Note that SimpliSafe requires full, "dashed" versions of UUIDs. Note that SimpliSafe requires full, "dashed" versions of UUIDs.
@ -118,7 +129,9 @@ async def async_get_client_id(hass):
return str(UUID(hass_id)) return str(UUID(hass_id))
async def async_register_base_station(hass, system, config_entry_id): async def async_register_base_station(
hass: HomeAssistant, system: SystemV2 | SystemV3, config_entry_id: str
) -> None:
"""Register a new bridge.""" """Register a new bridge."""
device_registry = await dr.async_get_registry(hass) device_registry = await dr.async_get_registry(hass)
device_registry.async_get_or_create( device_registry.async_get_or_create(
@ -130,11 +143,11 @@ async def async_register_base_station(hass, system, config_entry_id):
) )
async def async_setup_entry(hass, config_entry): # noqa: C901 @callback
"""Set up SimpliSafe as config entry.""" def _async_standardize_config_entry(
hass.data.setdefault(DOMAIN, {DATA_CLIENT: {}}) hass: HomeAssistant, config_entry: ConfigEntry
hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = [] ) -> None:
"""Bring a config entry up to current standards."""
if CONF_PASSWORD not in config_entry.data: if CONF_PASSWORD not in config_entry.data:
raise ConfigEntryAuthFailed("Config schema change requires re-authentication") raise ConfigEntryAuthFailed("Config schema change requires re-authentication")
@ -154,6 +167,14 @@ async def async_setup_entry(hass, config_entry): # noqa: C901
if entry_updates: if entry_updates:
hass.config_entries.async_update_entry(config_entry, **entry_updates) hass.config_entries.async_update_entry(config_entry, **entry_updates)
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up SimpliSafe as config entry."""
hass.data.setdefault(DOMAIN, {DATA_CLIENT: {}})
hass.data[DOMAIN][DATA_CLIENT][config_entry.entry_id] = []
_async_standardize_config_entry(hass, config_entry)
_verify_domain_control = verify_domain_control(hass, DOMAIN) _verify_domain_control = verify_domain_control(hass, DOMAIN)
client_id = await async_get_client_id(hass) client_id = await async_get_client_id(hass)
@ -183,10 +204,12 @@ async def async_setup_entry(hass, config_entry): # noqa: C901
hass.config_entries.async_setup_platforms(config_entry, PLATFORMS) hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
@callback @callback
def verify_system_exists(coro): def verify_system_exists(
coro: Callable[..., Awaitable]
) -> Callable[..., Awaitable]:
"""Log an error if a service call uses an invalid system ID.""" """Log an error if a service call uses an invalid system ID."""
async def decorator(call): async def decorator(call: ServiceCall) -> None:
"""Decorate.""" """Decorate."""
system_id = int(call.data[ATTR_SYSTEM_ID]) system_id = int(call.data[ATTR_SYSTEM_ID])
if system_id not in simplisafe.systems: if system_id not in simplisafe.systems:
@ -197,10 +220,10 @@ async def async_setup_entry(hass, config_entry): # noqa: C901
return decorator return decorator
@callback @callback
def v3_only(coro): def v3_only(coro: Callable[..., Awaitable]) -> Callable[..., Awaitable]:
"""Log an error if the decorated coroutine is called with a v2 system.""" """Log an error if the decorated coroutine is called with a v2 system."""
async def decorator(call): async def decorator(call: ServiceCall) -> None:
"""Decorate.""" """Decorate."""
system = simplisafe.systems[int(call.data[ATTR_SYSTEM_ID])] system = simplisafe.systems[int(call.data[ATTR_SYSTEM_ID])]
if system.version != 3: if system.version != 3:
@ -212,43 +235,40 @@ async def async_setup_entry(hass, config_entry): # noqa: C901
@verify_system_exists @verify_system_exists
@_verify_domain_control @_verify_domain_control
async def clear_notifications(call): async def clear_notifications(call: ServiceCall) -> None:
"""Clear all active notifications.""" """Clear all active notifications."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try: try:
await system.clear_notifications() await system.clear_notifications()
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
return
@verify_system_exists @verify_system_exists
@_verify_domain_control @_verify_domain_control
async def remove_pin(call): async def remove_pin(call: ServiceCall) -> None:
"""Remove a PIN.""" """Remove a PIN."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try: try:
await system.remove_pin(call.data[ATTR_PIN_LABEL_OR_VALUE]) await system.remove_pin(call.data[ATTR_PIN_LABEL_OR_VALUE])
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
return
@verify_system_exists @verify_system_exists
@_verify_domain_control @_verify_domain_control
async def set_pin(call): async def set_pin(call: ServiceCall) -> None:
"""Set a PIN.""" """Set a PIN."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try: try:
await system.set_pin(call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE]) await system.set_pin(call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE])
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
return
@verify_system_exists @verify_system_exists
@v3_only @v3_only
@_verify_domain_control @_verify_domain_control
async def set_system_properties(call): async def set_system_properties(call: ServiceCall) -> None:
"""Set one or more system parameters.""" """Set one or more system parameters."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = cast(SystemV3, simplisafe.systems[call.data[ATTR_SYSTEM_ID]])
try: try:
await system.set_properties( await system.set_properties(
{ {
@ -259,7 +279,6 @@ async def async_setup_entry(hass, config_entry): # noqa: C901
) )
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
return
for service, method, schema in ( for service, method, schema in (
("clear_notifications", clear_notifications, None), ("clear_notifications", clear_notifications, None),
@ -278,7 +297,7 @@ async def async_setup_entry(hass, config_entry): # noqa: C901
return True return True
async def async_unload_entry(hass, entry): async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a SimpliSafe config entry.""" """Unload a SimpliSafe config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok: if unload_ok:
@ -287,7 +306,7 @@ async def async_unload_entry(hass, entry):
return unload_ok return unload_ok
async def async_reload_entry(hass, config_entry): async def async_reload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Handle an options update.""" """Handle an options update."""
await hass.config_entries.async_reload(config_entry.entry_id) await hass.config_entries.async_reload(config_entry.entry_id)
@ -295,17 +314,19 @@ async def async_reload_entry(hass, config_entry):
class SimpliSafe: class SimpliSafe:
"""Define a SimpliSafe data object.""" """Define a SimpliSafe data object."""
def __init__(self, hass, config_entry, api): def __init__(
self, hass: HomeAssistant, config_entry: ConfigEntry, api: API
) -> None:
"""Initialize.""" """Initialize."""
self._api = api self._api = api
self._hass = hass self._hass = hass
self._system_notifications = {} self._system_notifications: dict[int, set[SystemNotification]] = {}
self.config_entry = config_entry self.config_entry = config_entry
self.coordinator = None self.coordinator: DataUpdateCoordinator | None = None
self.systems = {} self.systems: dict[int, SystemV2 | SystemV3] = {}
@callback @callback
def _async_process_new_notifications(self, system): def _async_process_new_notifications(self, system: SystemV2 | SystemV3) -> None:
"""Act on any new system notifications.""" """Act on any new system notifications."""
if self._hass.state != CoreState.running: if self._hass.state != CoreState.running:
# If HASS isn't fully running yet, it may cause the SIMPLISAFE_NOTIFICATION # If HASS isn't fully running yet, it may cause the SIMPLISAFE_NOTIFICATION
@ -324,8 +345,6 @@ class SimpliSafe:
LOGGER.debug("New system notifications: %s", to_add) LOGGER.debug("New system notifications: %s", to_add)
self._system_notifications[system.system_id].update(to_add)
for notification in to_add: for notification in to_add:
text = notification.text text = notification.text
if notification.link: if notification.link:
@ -341,7 +360,9 @@ class SimpliSafe:
}, },
) )
async def async_init(self): self._system_notifications[system.system_id] = latest_notifications
async def async_init(self) -> None:
"""Initialize the data class.""" """Initialize the data class."""
self.systems = await self._api.get_systems() self.systems = await self._api.get_systems()
for system in self.systems.values(): for system in self.systems.values():
@ -361,10 +382,10 @@ class SimpliSafe:
update_method=self.async_update, update_method=self.async_update,
) )
async def async_update(self): async def async_update(self) -> None:
"""Get updated data from SimpliSafe.""" """Get updated data from SimpliSafe."""
async def async_update_system(system): async def async_update_system(system: SystemV2 | SystemV3) -> None:
"""Update a system.""" """Update a system."""
await system.update(cached=system.version != 3) await system.update(cached=system.version != 3)
self._async_process_new_notifications(system) self._async_process_new_notifications(system)
@ -389,8 +410,16 @@ class SimpliSafe:
class SimpliSafeEntity(CoordinatorEntity): class SimpliSafeEntity(CoordinatorEntity):
"""Define a base SimpliSafe entity.""" """Define a base SimpliSafe entity."""
def __init__(self, simplisafe, system, name, *, serial=None): def __init__(
self,
simplisafe: SimpliSafe,
system: SystemV2 | SystemV3,
name: str,
*,
serial: str | None = None,
) -> None:
"""Initialize.""" """Initialize."""
assert simplisafe.coordinator
super().__init__(simplisafe.coordinator) super().__init__(simplisafe.coordinator)
if serial: if serial:
@ -413,32 +442,33 @@ class SimpliSafeEntity(CoordinatorEntity):
self._system = system self._system = system
@property @property
def available(self): def available(self) -> bool:
"""Return whether the entity is available.""" """Return whether the entity is available."""
# We can easily detect if the V3 system is offline, but no simple check exists # We can easily detect if the V3 system is offline, but no simple check exists
# for the V2 system. Therefore, assuming the coordinator hasn't failed, we mark # for the V2 system. Therefore, assuming the coordinator hasn't failed, we mark
# the entity as available if: # the entity as available if:
# 1. We can verify that the system is online (assuming True if we can't) # 1. We can verify that the system is online (assuming True if we can't)
# 2. We can verify that the entity is online # 2. We can verify that the entity is online
return ( if isinstance(self._system, SystemV3):
super().available system_offline = self._system.offline
and self._online else:
and not (self._system.version == 3 and self._system.offline) system_offline = False
)
return super().available and self._online and not system_offline
@callback @callback
def _handle_coordinator_update(self): def _handle_coordinator_update(self) -> None:
"""Update the entity with new REST API data.""" """Update the entity with new REST API data."""
self.async_update_from_rest_api() self.async_update_from_rest_api()
self.async_write_ha_state() self.async_write_ha_state()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
await super().async_added_to_hass() await super().async_added_to_hass()
self.async_update_from_rest_api() self.async_update_from_rest_api()
@callback @callback
def async_update_from_rest_api(self): def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
raise NotImplementedError() raise NotImplementedError()
@ -446,13 +476,22 @@ class SimpliSafeEntity(CoordinatorEntity):
class SimpliSafeBaseSensor(SimpliSafeEntity): class SimpliSafeBaseSensor(SimpliSafeEntity):
"""Define a SimpliSafe base (binary) sensor.""" """Define a SimpliSafe base (binary) sensor."""
def __init__(self, simplisafe, system, sensor): def __init__(
self,
simplisafe: SimpliSafe,
system: SystemV2 | SystemV3,
sensor: SensorV2 | SensorV3,
) -> None:
"""Initialize.""" """Initialize."""
super().__init__(simplisafe, system, sensor.name, serial=sensor.serial) super().__init__(simplisafe, system, sensor.name, serial=sensor.serial)
self._attr_device_info["identifiers"] = {(DOMAIN, sensor.serial)} self._attr_device_info = {
self._attr_device_info["model"] = sensor.type.name "identifiers": {(DOMAIN, sensor.serial)},
self._attr_device_info["name"] = sensor.name "manufacturer": "SimpliSafe",
"model": sensor.type.name,
"name": sensor.name,
"via_device": (DOMAIN, system.serial),
}
human_friendly_name = " ".join([w.title() for w in sensor.type.name.split("_")]) human_friendly_name = " ".join([w.title() for w in sensor.type.name.split("_")])
self._attr_name = f"{super().name} {human_friendly_name}" self._attr_name = f"{super().name} {human_friendly_name}"

View File

@ -1,8 +1,12 @@
"""Support for SimpliSafe alarm control panels.""" """Support for SimpliSafe alarm control panels."""
from __future__ import annotations
import re import re
from simplipy.errors import SimplipyError from simplipy.errors import SimplipyError
from simplipy.system import SystemStates from simplipy.system import SystemStates
from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3
from homeassistant.components.alarm_control_panel import ( from homeassistant.components.alarm_control_panel import (
FORMAT_NUMBER, FORMAT_NUMBER,
@ -13,6 +17,7 @@ from homeassistant.components.alarm_control_panel.const import (
SUPPORT_ALARM_ARM_AWAY, SUPPORT_ALARM_ARM_AWAY,
SUPPORT_ALARM_ARM_HOME, SUPPORT_ALARM_ARM_HOME,
) )
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
CONF_CODE, CONF_CODE,
STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_AWAY,
@ -21,9 +26,10 @@ from homeassistant.const import (
STATE_ALARM_DISARMED, STATE_ALARM_DISARMED,
STATE_ALARM_TRIGGERED, STATE_ALARM_TRIGGERED,
) )
from homeassistant.core import callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SimpliSafeEntity from . import SimpliSafe, SimpliSafeEntity
from .const import ( from .const import (
ATTR_ALARM_DURATION, ATTR_ALARM_DURATION,
ATTR_ALARM_VOLUME, ATTR_ALARM_VOLUME,
@ -48,7 +54,9 @@ ATTR_WALL_POWER_LEVEL = "wall_power_level"
ATTR_WIFI_STRENGTH = "wifi_strength" ATTR_WIFI_STRENGTH = "wifi_strength"
async def async_setup_entry(hass, entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a SimpliSafe alarm control panel based on a config entry.""" """Set up a SimpliSafe alarm control panel based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
async_add_entities( async_add_entities(
@ -60,7 +68,7 @@ async def async_setup_entry(hass, entry, async_add_entities):
class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity): class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
"""Representation of a SimpliSafe alarm.""" """Representation of a SimpliSafe alarm."""
def __init__(self, simplisafe, system): def __init__(self, simplisafe: SimpliSafe, system: SystemV2 | SystemV3) -> None:
"""Initialize the SimpliSafe alarm.""" """Initialize the SimpliSafe alarm."""
super().__init__(simplisafe, system, "Alarm Control Panel") super().__init__(simplisafe, system, "Alarm Control Panel")
@ -91,7 +99,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
self._attr_state = None self._attr_state = None
@callback @callback
def _is_code_valid(self, code, state): def _is_code_valid(self, code: str | None, state: str) -> bool:
"""Validate that a code matches the required one.""" """Validate that a code matches the required one."""
if not self._simplisafe.config_entry.options.get(CONF_CODE): if not self._simplisafe.config_entry.options.get(CONF_CODE):
return True return True
@ -104,7 +112,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
return True return True
async def async_alarm_disarm(self, code=None): async def async_alarm_disarm(self, code: str | None = None) -> None:
"""Send disarm command.""" """Send disarm command."""
if not self._is_code_valid(code, STATE_ALARM_DISARMED): if not self._is_code_valid(code, STATE_ALARM_DISARMED):
return return
@ -118,7 +126,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
self._attr_state = STATE_ALARM_DISARMED self._attr_state = STATE_ALARM_DISARMED
self.async_write_ha_state() self.async_write_ha_state()
async def async_alarm_arm_home(self, code=None): async def async_alarm_arm_home(self, code: str | None = None) -> None:
"""Send arm home command.""" """Send arm home command."""
if not self._is_code_valid(code, STATE_ALARM_ARMED_HOME): if not self._is_code_valid(code, STATE_ALARM_ARMED_HOME):
return return
@ -134,7 +142,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
self._attr_state = STATE_ALARM_ARMED_HOME self._attr_state = STATE_ALARM_ARMED_HOME
self.async_write_ha_state() self.async_write_ha_state()
async def async_alarm_arm_away(self, code=None): async def async_alarm_arm_away(self, code: str | None = None) -> None:
"""Send arm away command.""" """Send arm away command."""
if not self._is_code_valid(code, STATE_ALARM_ARMED_AWAY): if not self._is_code_valid(code, STATE_ALARM_ARMED_AWAY):
return return
@ -151,9 +159,9 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
self.async_write_ha_state() self.async_write_ha_state()
@callback @callback
def async_update_from_rest_api(self): def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
if self._system.version == 3: if isinstance(self._system, SystemV3):
self._attr_extra_state_attributes.update( self._attr_extra_state_attributes.update(
{ {
ATTR_ALARM_DURATION: self._system.alarm_duration, ATTR_ALARM_DURATION: self._system.alarm_duration,
@ -175,9 +183,6 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
} }
) )
# Although system state updates are designed the come via the websocket, the
# SimpliSafe cloud can sporadically fail to send those updates as expected; so,
# just in case, we synchronize the state via the REST API, too:
if self._system.state == SystemStates.alarm: if self._system.state == SystemStates.alarm:
self._attr_state = STATE_ALARM_TRIGGERED self._attr_state = STATE_ALARM_TRIGGERED
elif self._system.state == SystemStates.away: elif self._system.state == SystemStates.away:

View File

@ -1,5 +1,9 @@
"""Support for SimpliSafe binary sensors.""" """Support for SimpliSafe binary sensors."""
from simplipy.entity import EntityTypes from __future__ import annotations
from simplipy.entity import Entity as SimplipyEntity, EntityTypes
from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
DEVICE_CLASS_BATTERY, DEVICE_CLASS_BATTERY,
@ -11,9 +15,11 @@ from homeassistant.components.binary_sensor import (
DEVICE_CLASS_SMOKE, DEVICE_CLASS_SMOKE,
BinarySensorEntity, BinarySensorEntity,
) )
from homeassistant.core import callback from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SimpliSafeBaseSensor from . import SimpliSafe, SimpliSafeBaseSensor
from .const import DATA_CLIENT, DOMAIN, LOGGER from .const import DATA_CLIENT, DOMAIN, LOGGER
SUPPORTED_BATTERY_SENSOR_TYPES = [ SUPPORTED_BATTERY_SENSOR_TYPES = [
@ -39,10 +45,13 @@ TRIGGERED_SENSOR_TYPES = {
} }
async def async_setup_entry(hass, entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up SimpliSafe binary sensors based on a config entry.""" """Set up SimpliSafe binary sensors based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
sensors = []
sensors: list[BatteryBinarySensor | TriggeredBinarySensor] = []
for system in simplisafe.systems.values(): for system in simplisafe.systems.values():
if system.version == 2: if system.version == 2:
@ -68,14 +77,20 @@ async def async_setup_entry(hass, entry, async_add_entities):
class TriggeredBinarySensor(SimpliSafeBaseSensor, BinarySensorEntity): class TriggeredBinarySensor(SimpliSafeBaseSensor, BinarySensorEntity):
"""Define a binary sensor related to whether an entity has been triggered.""" """Define a binary sensor related to whether an entity has been triggered."""
def __init__(self, simplisafe, system, sensor, device_class): def __init__(
self,
simplisafe: SimpliSafe,
system: SystemV2 | SystemV3,
sensor: SimplipyEntity,
device_class: str,
) -> None:
"""Initialize.""" """Initialize."""
super().__init__(simplisafe, system, sensor) super().__init__(simplisafe, system, sensor)
self._attr_device_class = device_class self._attr_device_class = device_class
@callback @callback
def async_update_from_rest_api(self): def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
self._attr_is_on = self._sensor.triggered self._attr_is_on = self._sensor.triggered
@ -85,13 +100,18 @@ class BatteryBinarySensor(SimpliSafeBaseSensor, BinarySensorEntity):
_attr_device_class = DEVICE_CLASS_BATTERY _attr_device_class = DEVICE_CLASS_BATTERY
def __init__(self, simplisafe, system, sensor): def __init__(
self,
simplisafe: SimpliSafe,
system: SystemV2 | SystemV3,
sensor: SimplipyEntity,
) -> None:
"""Initialize.""" """Initialize."""
super().__init__(simplisafe, system, sensor) super().__init__(simplisafe, system, sensor)
self._attr_unique_id = f"{super().unique_id}-battery" self._attr_unique_id = f"{super().unique_id}-battery"
@callback @callback
def async_update_from_rest_api(self): def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
self._attr_is_on = self._sensor.low_battery self._attr_is_on = self._sensor.low_battery

View File

@ -1,5 +1,10 @@
"""Config flow to configure the SimpliSafe component.""" """Config flow to configure the SimpliSafe component."""
from __future__ import annotations
from typing import Any
from simplipy import get_api from simplipy import get_api
from simplipy.api import API
from simplipy.errors import ( from simplipy.errors import (
InvalidCredentialsError, InvalidCredentialsError,
PendingAuthorizationError, PendingAuthorizationError,
@ -8,9 +13,12 @@ from simplipy.errors import (
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.typing import ConfigType
from . import async_get_client_id from . import async_get_client_id
from .const import DOMAIN, LOGGER from .const import DOMAIN, LOGGER
@ -30,20 +38,25 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize the config flow.""" """Initialize the config flow."""
self._code = None self._code: str | None = None
self._password = None self._password: str | None = None
self._username = None self._username: str | None = None
@staticmethod @staticmethod
@callback @callback
def async_get_options_flow(config_entry): def async_get_options_flow(
config_entry: ConfigEntry,
) -> SimpliSafeOptionsFlowHandler:
"""Define the config flow to handle options.""" """Define the config flow to handle options."""
return SimpliSafeOptionsFlowHandler(config_entry) return SimpliSafeOptionsFlowHandler(config_entry)
async def _async_get_simplisafe_api(self): async def _async_get_simplisafe_api(self) -> API:
"""Get an authenticated SimpliSafe API client.""" """Get an authenticated SimpliSafe API client."""
assert self._username
assert self._password
client_id = await async_get_client_id(self.hass) client_id = await async_get_client_id(self.hass)
websession = aiohttp_client.async_get_clientsession(self.hass) websession = aiohttp_client.async_get_clientsession(self.hass)
@ -54,7 +67,9 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
session=websession, session=websession,
) )
async def _async_login_during_step(self, *, step_id, form_schema): async def _async_login_during_step(
self, *, step_id: str, form_schema: vol.Schema
) -> FlowResult:
"""Attempt to log into the API from within a config flow step.""" """Attempt to log into the API from within a config flow step."""
errors = {} errors = {}
@ -84,8 +99,10 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
} }
) )
async def async_step_finish(self, user_input=None): async def async_step_finish(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle finish config entry setup.""" """Handle finish config entry setup."""
assert self._username
existing_entry = await self.async_set_unique_id(self._username) existing_entry = await self.async_set_unique_id(self._username)
if existing_entry: if existing_entry:
self.hass.config_entries.async_update_entry(existing_entry, data=user_input) self.hass.config_entries.async_update_entry(existing_entry, data=user_input)
@ -95,7 +112,9 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="reauth_successful") return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=self._username, data=user_input) return self.async_create_entry(title=self._username, data=user_input)
async def async_step_mfa(self, user_input=None): async def async_step_mfa(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle multi-factor auth confirmation.""" """Handle multi-factor auth confirmation."""
if user_input is None: if user_input is None:
return self.async_show_form(step_id="mfa") return self.async_show_form(step_id="mfa")
@ -116,14 +135,16 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
} }
) )
async def async_step_reauth(self, config): async def async_step_reauth(self, config: ConfigType) -> FlowResult:
"""Handle configuration by re-auth.""" """Handle configuration by re-auth."""
self._code = config.get(CONF_CODE) self._code = config.get(CONF_CODE)
self._username = config[CONF_USERNAME] self._username = config[CONF_USERNAME]
return await self.async_step_reauth_confirm() return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(self, user_input=None): async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle re-auth completion.""" """Handle re-auth completion."""
if not user_input: if not user_input:
return self.async_show_form( return self.async_show_form(
@ -136,7 +157,9 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
step_id="reauth_confirm", form_schema=PASSWORD_DATA_SCHEMA step_id="reauth_confirm", form_schema=PASSWORD_DATA_SCHEMA
) )
async def async_step_user(self, user_input=None): async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the start of the config flow.""" """Handle the start of the config flow."""
if not user_input: if not user_input:
return self.async_show_form(step_id="user", data_schema=FULL_DATA_SCHEMA) return self.async_show_form(step_id="user", data_schema=FULL_DATA_SCHEMA)
@ -156,11 +179,13 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
class SimpliSafeOptionsFlowHandler(config_entries.OptionsFlow): class SimpliSafeOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a SimpliSafe options flow.""" """Handle a SimpliSafe options flow."""
def __init__(self, config_entry): def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize.""" """Initialize."""
self.config_entry = config_entry self.config_entry = config_entry
async def async_step_init(self, user_input=None): async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options.""" """Manage the options."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="", data=user_input) return self.async_create_entry(title="", data=user_input)

View File

@ -1,11 +1,18 @@
"""Support for SimpliSafe locks.""" """Support for SimpliSafe locks."""
from __future__ import annotations
from typing import Any
from simplipy.errors import SimplipyError from simplipy.errors import SimplipyError
from simplipy.lock import LockStates from simplipy.lock import Lock, LockStates
from simplipy.system.v3 import SystemV3
from homeassistant.components.lock import LockEntity from homeassistant.components.lock import LockEntity
from homeassistant.core import callback from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SimpliSafeEntity from . import SimpliSafe, SimpliSafeEntity
from .const import DATA_CLIENT, DOMAIN, LOGGER from .const import DATA_CLIENT, DOMAIN, LOGGER
ATTR_LOCK_LOW_BATTERY = "lock_low_battery" ATTR_LOCK_LOW_BATTERY = "lock_low_battery"
@ -13,7 +20,9 @@ ATTR_JAMMED = "jammed"
ATTR_PIN_PAD_LOW_BATTERY = "pin_pad_low_battery" ATTR_PIN_PAD_LOW_BATTERY = "pin_pad_low_battery"
async def async_setup_entry(hass, entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up SimpliSafe locks based on a config entry.""" """Set up SimpliSafe locks based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
locks = [] locks = []
@ -32,13 +41,13 @@ async def async_setup_entry(hass, entry, async_add_entities):
class SimpliSafeLock(SimpliSafeEntity, LockEntity): class SimpliSafeLock(SimpliSafeEntity, LockEntity):
"""Define a SimpliSafe lock.""" """Define a SimpliSafe lock."""
def __init__(self, simplisafe, system, lock): def __init__(self, simplisafe: SimpliSafe, system: SystemV3, lock: Lock) -> None:
"""Initialize.""" """Initialize."""
super().__init__(simplisafe, system, lock.name, serial=lock.serial) super().__init__(simplisafe, system, lock.name, serial=lock.serial)
self._lock = lock self._lock = lock
async def async_lock(self, **kwargs): async def async_lock(self, **kwargs: dict[str, Any]) -> None:
"""Lock the lock.""" """Lock the lock."""
try: try:
await self._lock.lock() await self._lock.lock()
@ -49,7 +58,7 @@ class SimpliSafeLock(SimpliSafeEntity, LockEntity):
self._attr_is_locked = True self._attr_is_locked = True
self.async_write_ha_state() self.async_write_ha_state()
async def async_unlock(self, **kwargs): async def async_unlock(self, **kwargs: dict[str, Any]) -> None:
"""Unlock the lock.""" """Unlock the lock."""
try: try:
await self._lock.unlock() await self._lock.unlock()
@ -61,7 +70,7 @@ class SimpliSafeLock(SimpliSafeEntity, LockEntity):
self.async_write_ha_state() self.async_write_ha_state()
@callback @callback
def async_update_from_rest_api(self): def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
self._attr_extra_state_attributes.update( self._attr_extra_state_attributes.update(
{ {

View File

@ -3,7 +3,7 @@
"name": "SimpliSafe", "name": "SimpliSafe",
"config_flow": true, "config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/simplisafe", "documentation": "https://www.home-assistant.io/integrations/simplisafe",
"requirements": ["simplisafe-python==11.0.2"], "requirements": ["simplisafe-python==11.0.3"],
"codeowners": ["@bachya"], "codeowners": ["@bachya"],
"iot_class": "cloud_polling" "iot_class": "cloud_polling"
} }

View File

@ -2,14 +2,18 @@
from simplipy.entity import EntityTypes from simplipy.entity import EntityTypes
from homeassistant.components.sensor import SensorEntity from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DEVICE_CLASS_TEMPERATURE, TEMP_FAHRENHEIT from homeassistant.const import DEVICE_CLASS_TEMPERATURE, TEMP_FAHRENHEIT
from homeassistant.core import callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SimpliSafeBaseSensor from . import SimpliSafeBaseSensor
from .const import DATA_CLIENT, DOMAIN, LOGGER from .const import DATA_CLIENT, DOMAIN, LOGGER
async def async_setup_entry(hass, entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up SimpliSafe freeze sensors based on a config entry.""" """Set up SimpliSafe freeze sensors based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id]
sensors = [] sensors = []
@ -33,6 +37,6 @@ class SimplisafeFreezeSensor(SimpliSafeBaseSensor, SensorEntity):
_attr_unit_of_measurement = TEMP_FAHRENHEIT _attr_unit_of_measurement = TEMP_FAHRENHEIT
@callback @callback
def async_update_from_rest_api(self): def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
self._attr_state = self._sensor.temperature self._attr_state = self._sensor.temperature

View File

@ -935,6 +935,17 @@ no_implicit_optional = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.simplisafe.*]
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.slack.*] [mypy-homeassistant.components.slack.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true

View File

@ -2107,7 +2107,7 @@ simplehound==0.3
simplepush==1.1.4 simplepush==1.1.4
# homeassistant.components.simplisafe # homeassistant.components.simplisafe
simplisafe-python==11.0.2 simplisafe-python==11.0.3
# homeassistant.components.sisyphus # homeassistant.components.sisyphus
sisyphus-control==3.0 sisyphus-control==3.0

View File

@ -1156,7 +1156,7 @@ sharkiqpy==0.1.8
simplehound==0.3 simplehound==0.3
# homeassistant.components.simplisafe # homeassistant.components.simplisafe
simplisafe-python==11.0.2 simplisafe-python==11.0.3
# homeassistant.components.slack # homeassistant.components.slack
slackclient==2.5.0 slackclient==2.5.0