evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reorganize SimpliSafe services (#58722)

Browse source
This commit is contained in:
Aaron Bach 2021-12-02 12:07:14 -07:00 • committed by GitHub
parent d9c567e205
commit e641214c60
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 164 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

226
homeassistant/components/simplisafe/__init__.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable, Iterable
from collections.abc import Callable, Iterable
from datetime import timedelta
from typing import TYPE_CHECKING, Any, cast
@ -14,7 +14,6 @@ from simplipy.errors import (
SimplipyError,
)
from simplipy.system import SystemNotification
from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import (
MAX_ALARM_DURATION,
MAX_ENTRY_DELAY_AWAY,
@ -45,9 +44,10 @@ from simplipy.websocket import (
)
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import (
ATTR_CODE,
ATTR_DEVICE_ID,
CONF_CODE,
CONF_TOKEN,
EVENT_HOMEASSISTANT_STOP,
@ -88,6 +88,7 @@ from .const import (
DOMAIN,
LOGGER,
)
from .typing import SystemType
ATTR_CATEGORY = "category"
ATTR_LAST_EVENT_CHANGED_BY = "last_event_changed_by"
@ -131,18 +132,37 @@ VOLUME_MAP = {
"off": Volume.OFF,
}
SERVICE_BASE_SCHEMA = vol.Schema({vol.Required(ATTR_SYSTEM_ID): cv.positive_int})
SERVICE_NAME_CLEAR_NOTIFICATIONS = "clear_notifications"
SERVICE_NAME_REMOVE_PIN = "remove_pin"
SERVICE_NAME_SET_PIN = "set_pin"
SERVICE_NAME_SET_SYSTEM_PROPERTIES = "set_system_properties"
SERVICE_REMOVE_PIN_SCHEMA = SERVICE_BASE_SCHEMA.extend(
{vol.Required(ATTR_PIN_LABEL_OR_VALUE): cv.string}
SERVICES = (
SERVICE_NAME_CLEAR_NOTIFICATIONS,
SERVICE_NAME_REMOVE_PIN,
SERVICE_NAME_SET_PIN,
SERVICE_NAME_SET_SYSTEM_PROPERTIES,
)
SERVICE_SET_PIN_SCHEMA = SERVICE_BASE_SCHEMA.extend(
{vol.Required(ATTR_PIN_LABEL): cv.string, vol.Required(ATTR_PIN_VALUE): cv.string}
)
SERVICE_SET_SYSTEM_PROPERTIES_SCHEMA = SERVICE_BASE_SCHEMA.extend(
SERVICE_REMOVE_PIN_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.string,
vol.Required(ATTR_PIN_LABEL_OR_VALUE): cv.string,
}
)
SERVICE_SET_PIN_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.string,
vol.Required(ATTR_PIN_LABEL): cv.string,
vol.Required(ATTR_PIN_VALUE): cv.string,
}
)
SERVICE_SET_SYSTEM_PROPERTIES_SCHEMA = vol.Schema(
{
vol.Required(ATTR_DEVICE_ID): cv.string,
vol.Optional(ATTR_ALARM_DURATION): vol.All(
cv.time_period,
lambda value: value.total_seconds(),
@ -191,6 +211,58 @@ WEBSOCKET_EVENTS_TO_FIRE_HASS_EVENT = [
CONFIG_SCHEMA = cv.deprecated(DOMAIN)
@callback
def _async_get_system_for_service_call(
hass: HomeAssistant, call: ServiceCall
) -> SystemType:
"""Get the SimpliSafe system related to a service call (by device ID)."""
device_id = call.data[ATTR_DEVICE_ID]
device_registry = dr.async_get(hass)
if (
alarm_control_panel_device_entry := device_registry.async_get(device_id)
) is None:
raise vol.Invalid("Invalid device ID specified")
if TYPE_CHECKING:
assert alarm_control_panel_device_entry.via_device_id
if (
base_station_device_entry := device_registry.async_get(
alarm_control_panel_device_entry.via_device_id
)
) is None:
raise ValueError("No base station registered for alarm control panel")
[system_id] = [
identity[1]
for identity in base_station_device_entry.identifiers
if identity[0] == DOMAIN
]
for entry_id in base_station_device_entry.config_entries:
if (simplisafe := hass.data[DOMAIN].get(entry_id)) is None:
continue
return cast(SystemType, simplisafe.systems[system_id])
raise ValueError(f"No system for device ID: {device_id}")
@callback
def _async_register_base_station(
hass: HomeAssistant, entry: ConfigEntry, system: SystemType
) -> None:
"""Register a new bridge."""
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, system.system_id)},
manufacturer="SimpliSafe",
model=system.version,
name=system.address,
)
@callback
def _async_standardize_config_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Bring a config entry up to current standards."""
@ -216,24 +288,7 @@ def _async_standardize_config_entry(hass: HomeAssistant, entry: ConfigEntry) ->
hass.config_entries.async_update_entry(entry, **entry_updates)
@callback
def _async_register_base_station(
hass: HomeAssistant, entry: ConfigEntry, system: SystemV2 | SystemV3
) -> None:
"""Register a new bridge."""
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, system.system_id)},
manufacturer="SimpliSafe",
model=system.version,
name=system.address,
)
async def async_setup_entry( # noqa: C901
hass: HomeAssistant, entry: ConfigEntry
) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up SimpliSafe as config entry."""
_async_standardize_config_entry(hass, entry)
@ -263,94 +318,64 @@ async def async_setup_entry( # noqa: C901
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
@callback
def verify_system_exists(
coro: Callable[..., Awaitable]
) -> Callable[..., Awaitable]:
"""Log an error if a service call uses an invalid system ID."""
def extract_system(func: Callable) -> Callable:
"""Define a decorator to get the correct system for a service call."""
async def decorator(call: ServiceCall) -> None:
"""Decorate."""
system_id = int(call.data[ATTR_SYSTEM_ID])
if system_id not in simplisafe.systems:
LOGGER.error("Unknown system ID in service call: %s", system_id)
return
await coro(call)
async def wrapper(call: ServiceCall) -> None:
"""Wrap the service function."""
system = _async_get_system_for_service_call(hass, call)
return decorator
try:
await func(call, system)
except SimplipyError as err:
LOGGER.error("Error while executing %s: %s", func.__name__, err)
@callback
def v3_only(coro: Callable[..., Awaitable]) -> Callable[..., Awaitable]:
"""Log an error if the decorated coroutine is called with a v2 system."""
return wrapper
async def decorator(call: ServiceCall) -> None:
"""Decorate."""
system = simplisafe.systems[int(call.data[ATTR_SYSTEM_ID])]
if system.version != 3:
LOGGER.error("Service only available on V3 systems")
return
await coro(call)
return decorator
@verify_system_exists
@_verify_domain_control
async def clear_notifications(call: ServiceCall) -> None:
@extract_system
async def async_clear_notifications(call: ServiceCall, system: SystemType) -> None:
"""Clear all active notifications."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try:
await system.async_clear_notifications()
except SimplipyError as err:
LOGGER.error("Error during service call: %s", err)
@verify_system_exists
@_verify_domain_control
async def remove_pin(call: ServiceCall) -> None:
@extract_system
async def async_remove_pin(call: ServiceCall, system: SystemType) -> None:
"""Remove a PIN."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try:
await system.async_remove_pin(call.data[ATTR_PIN_LABEL_OR_VALUE])
except SimplipyError as err:
LOGGER.error("Error during service call: %s", err)
@verify_system_exists
@_verify_domain_control
async def set_pin(call: ServiceCall) -> None:
@extract_system
async def async_set_pin(call: ServiceCall, system: SystemType) -> None:
"""Set a PIN."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try:
await system.async_set_pin(
call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE]
)
except SimplipyError as err:
LOGGER.error("Error during service call: %s", err)
await system.async_set_pin(call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE])
@verify_system_exists
@v3_only
@_verify_domain_control
async def set_system_properties(call: ServiceCall) -> None:
@extract_system
async def async_set_system_properties(
call: ServiceCall, system: SystemType
) -> None:
"""Set one or more system parameters."""
system = cast(SystemV3, simplisafe.systems[call.data[ATTR_SYSTEM_ID]])
try:
if not isinstance(system, SystemV3):
LOGGER.error("Can only set system properties on V3 systems")
return
await system.async_set_properties(
{
prop: value
for prop, value in call.data.items()
if prop != ATTR_SYSTEM_ID
}
{prop: value for prop, value in call.data.items() if prop != ATTR_DEVICE_ID}
)
except SimplipyError as err:
LOGGER.error("Error during service call: %s", err)
for service, method, schema in (
("clear_notifications", clear_notifications, None),
("remove_pin", remove_pin, SERVICE_REMOVE_PIN_SCHEMA),
("set_pin", set_pin, SERVICE_SET_PIN_SCHEMA),
(SERVICE_NAME_CLEAR_NOTIFICATIONS, async_clear_notifications, None),
(SERVICE_NAME_REMOVE_PIN, async_remove_pin, SERVICE_REMOVE_PIN_SCHEMA),
(SERVICE_NAME_SET_PIN, async_set_pin, SERVICE_SET_PIN_SCHEMA),
(
"set_system_properties",
set_system_properties,
SERVICE_NAME_SET_SYSTEM_PROPERTIES,
async_set_system_properties,
SERVICE_SET_SYSTEM_PROPERTIES_SCHEMA,
),
):
if hass.services.has_service(DOMAIN, service):
continue
async_register_admin_service(hass, DOMAIN, service, method, schema=schema)
current_options = {**entry.options}
@ -383,6 +408,17 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
loaded_entries = [
entry
for entry in hass.config_entries.async_entries(DOMAIN)
if entry.state == ConfigEntryState.LOADED
]
if len(loaded_entries) == 1:
# If this is the last loaded instance of SimpliSafe, deregister any services
# defined during integration setup:
for service_name in SERVICES:
hass.services.async_remove(DOMAIN, service_name)
return unload_ok
@ -396,13 +432,13 @@ class SimpliSafe:
self._system_notifications: dict[int, set[SystemNotification]] = {}
self.entry = entry
self.initial_event_to_use: dict[int, dict[str, Any]] = {}
self.systems: dict[int, SystemV2 | SystemV3] = {}
self.systems: dict[int, SystemType] = {}
# This will get filled in by async_init:
self.coordinator: DataUpdateCoordinator | None = None
@callback
def _async_process_new_notifications(self, system: SystemV2 | SystemV3) -> None:
def _async_process_new_notifications(self, system: SystemType) -> None:
"""Act on any new system notifications."""
if self._hass.state != CoreState.running:
# If HASS isn't fully running yet, it may cause the SIMPLISAFE_NOTIFICATION
@ -543,7 +579,7 @@ class SimpliSafe:
async def async_update(self) -> None:
"""Get updated data from SimpliSafe."""
async def async_update_system(system: SystemV2 | SystemV3) -> None:
async def async_update_system(system: SystemType) -> None:
"""Update a system."""
await system.async_update(cached=system.version != 3)
self._async_process_new_notifications(system)
@ -571,7 +607,7 @@ class SimpliSafeEntity(CoordinatorEntity):
def __init__(
self,
simplisafe: SimpliSafe,
system: SystemV2 | SystemV3,
system: SystemType,
*,
device: Device | None = None,
additional_websocket_events: Iterable[str] | None = None,

4
homeassistant/components/simplisafe/alarm_control_panel.py
View file

@ -5,7 +5,6 @@ from typing import TYPE_CHECKING
from simplipy.errors import SimplipyError
from simplipy.system import SystemStates
from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3
from simplipy.websocket import (
EVENT_ALARM_CANCELED,
@ -58,6 +57,7 @@ from .const import (
DOMAIN,
LOGGER,
)
from .typing import SystemType
ATTR_BATTERY_BACKUP_POWER_LEVEL = "battery_backup_power_level"
ATTR_GSM_STRENGTH = "gsm_strength"
@ -119,7 +119,7 @@ async def async_setup_entry(
class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
"""Representation of a SimpliSafe alarm."""
def __init__(self, simplisafe: SimpliSafe, system: SystemV2 | SystemV3) -> None:
def __init__(self, simplisafe: SimpliSafe, system: SystemType) -> None:
"""Initialize the SimpliSafe alarm."""
super().__init__(
simplisafe,

30
homeassistant/components/simplisafe/services.yaml
View file

@ -3,13 +3,14 @@ remove_pin:
name: Remove PIN
description: Remove a PIN by its label or value.
fields:
system_id:
name: System ID
description: The SimpliSafe system ID to affect.
device_id:
name: System
description: The system to remove the PIN from
required: true
example: 123987
selector:
text:
device:
integration: simplisafe
model: alarm_control_panel
label_or_pin:
name: Label/PIN
description: The label/value to remove.
@ -21,13 +22,14 @@ set_pin:
name: Set PIN
description: Set/update a PIN
fields:
system_id:
name: System ID
description: The SimpliSafe system ID to affect
device_id:
name: System
description: The system to set the PIN on
required: true
example: 123987
selector:
text:
device:
integration: simplisafe
model: alarm_control_panel
label:
name: Label
description: The label of the PIN
@ -46,6 +48,14 @@ set_system_properties:
name: Set system properties
description: Set one or more system properties
fields:
device_id:
name: System
description: The system whose properties should be set
required: true
selector:
device:
integration: simplisafe
model: alarm_control_panel
alarm_duration:
name: Alarm duration
description: The length of a triggered alarm

7
homeassistant/components/simplisafe/typing.py Normal file
View file

@ -0,0 +1,7 @@
"""Define typing helpers for SimpliSafe."""
from typing import Union
from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3
SystemType = Union[SystemV2, SystemV3]