lock.py

Go to the documentation of this file.

"""Support for SimpliSafe locks."""
 
from __future__ import annotations
 
from typing import Any
 
from simplipy.device.lock import Lock, LockStates
from simplipy.errors import SimplipyError
from simplipy.system.v3 import SystemV3
from simplipy.websocket import EVENT_LOCK_LOCKED, EVENT_LOCK_UNLOCKED, WebsocketEvent
 
from homeassistant.components.lock import LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
from . import SimpliSafe
from .const import DOMAIN, LOGGER
from .entity import SimpliSafeEntity
 
ATTR_LOCK_LOW_BATTERY = "lock_low_battery"
ATTR_PIN_PAD_LOW_BATTERY = "pin_pad_low_battery"
 
STATE_MAP_FROM_WEBSOCKET_EVENT = {
    EVENT_LOCK_LOCKED: True,
    EVENT_LOCK_UNLOCKED: False,
}
 
WEBSOCKET_EVENTS_TO_LISTEN_FOR = (EVENT_LOCK_LOCKED, EVENT_LOCK_UNLOCKED)
 
 
async def async_setup_entry(
    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up SimpliSafe locks based on a config entry."""
    simplisafe = hass.data[DOMAIN][entry.entry_id]
    locks: list[SimpliSafeLock] = []
 
    for system in simplisafe.systems.values():
        if system.version == 2:
            LOGGER.warning("Skipping lock setup for V2 system: %s", system.system_id)
            continue
 
        locks.extend(
            SimpliSafeLock(simplisafe, system, lock) for lock in system.locks.values()
        )
 
    async_add_entities(locks)
 
 
class SimpliSafeLock(SimpliSafeEntity, LockEntity):
    """Define a SimpliSafe lock."""
 
    _attr_name = None
    _device: Lock
 
    def __init__(self, simplisafe: SimpliSafe, system: SystemV3, lock: Lock) -> None:
        """Initialize."""
        super().__init__(
            simplisafe,
            system,
            device=lock,
            additional_websocket_events=WEBSOCKET_EVENTS_TO_LISTEN_FOR,
        )
 
    async def async_lock(self, **kwargs: Any) -> None:
        """Lock the lock."""
        try:
            await self._device_device.async_lock()
        except SimplipyError as err:
            raise HomeAssistantError(
                f'Error while locking "{self._device.name}": {err}'
            ) from err
 
        self._attr_is_locked_attr_is_locked = True
        self.async_write_ha_stateasync_write_ha_state()
 
    async def async_unlock(self, **kwargs: Any) -> None:
        """Unlock the lock."""
        try:
            await self._device_device.async_unlock()
        except SimplipyError as err:
            raise HomeAssistantError(
                f'Error while unlocking "{self._device.name}": {err}'
            ) from err
 
        self._attr_is_locked_attr_is_locked = False
        self.async_write_ha_stateasync_write_ha_state()
 
    @callback
    def async_update_from_rest_api(self) -> None:
        """Update the entity with the provided REST API data."""
        self._attr_is_jammed_attr_is_jammed = self._device_device.state == LockStates.JAMMED
        self._attr_is_locked_attr_is_locked = self._device_device.state == LockStates.LOCKED
 
        self._attr_extra_state_attributes_attr_extra_state_attributes.update(
            {
                ATTR_LOCK_LOW_BATTERY: self._device_device.lock_low_battery,
                ATTR_PIN_PAD_LOW_BATTERY: self._device_device.pin_pad_low_battery,
            }
        )
 
    @callback
    def async_update_from_websocket_event(self, event: WebsocketEvent) -> None:
        """Update the entity when new data comes from the websocket."""
        assert event.event_type
 
        if state := STATE_MAP_FROM_WEBSOCKET_EVENT.get(event.event_type) is not None:
            self._attr_is_locked_attr_is_locked = state
            self.async_reset_error_countasync_reset_error_count()
        else:
            LOGGER.error("Unknown lock websocket event: %s", event.event_type)
            self.async_increment_error_countasync_increment_error_count()