simplipy/system/v3.py
"""Define a V3 (new) SimpliSafe system."""
from __future__ import annotations
from datetime import datetime, timedelta
from enum import Enum
from typing import TYPE_CHECKING, Any, Final, cast
import voluptuous as vol
from simplipy.const import LOGGER
from simplipy.device import DeviceTypes, get_device_type_from_data
from simplipy.device.camera import Camera
from simplipy.device.lock import Lock
from simplipy.device.sensor.v3 import SensorV3
from simplipy.system import (
CONF_DURESS_PIN,
CONF_MASTER_PIN,
DEFAULT_MAX_USER_PINS,
System,
SystemStates,
guard_from_missing_data,
)
from simplipy.util.dt import utcnow
if TYPE_CHECKING:
from simplipy.api import API
CONF_ALARM_DURATION = "alarm_duration"
CONF_ALARM_VOLUME = "alarm_volume"
CONF_CHIME_VOLUME = "chime_volume"
CONF_ENTRY_DELAY_AWAY = "entry_delay_away"
CONF_ENTRY_DELAY_HOME = "entry_delay_home"
CONF_EXIT_DELAY_AWAY = "exit_delay_away"
CONF_EXIT_DELAY_HOME = "exit_delay_home"
CONF_LIGHT = "light"
CONF_VOICE_PROMPT_VOLUME = "voice_prompt_volume"
DEFAULT_LOCK_STATE_CHANGE_WINDOW = timedelta(seconds=15)
SYSTEM_PROPERTIES_VALUE_MAP = {
CONF_ALARM_DURATION: "alarmDuration",
CONF_ALARM_VOLUME: "alarmVolume",
CONF_CHIME_VOLUME: "doorChime",
CONF_ENTRY_DELAY_AWAY: "entryDelayAway",
CONF_ENTRY_DELAY_HOME: "entryDelayHome",
CONF_EXIT_DELAY_AWAY: "exitDelayAway",
CONF_EXIT_DELAY_HOME: "exitDelayHome",
CONF_LIGHT: "light",
CONF_VOICE_PROMPT_VOLUME: "voicePrompts",
}
MIN_ALARM_DURATION: Final = 30
MAX_ALARM_DURATION: Final = 480
MIN_ENTRY_DELAY_AWAY: Final = 30
MAX_ENTRY_DELAY_AWAY: Final = 255
MIN_ENTRY_DELAY_HOME: Final = 0
MAX_ENTRY_DELAY_HOME: Final = 255
MIN_EXIT_DELAY_AWAY: Final = 45
MAX_EXIT_DELAY_AWAY: Final = 255
MIN_EXIT_DELAY_HOME: Final = 0
MAX_EXIT_DELAY_HOME: Final = 255
class Volume(Enum):
"""Define a representation of a SimpliSafe volume level."""
OFF = 0
LOW = 1
MEDIUM = 2
HIGH = 3
SYSTEM_PROPERTIES_PAYLOAD_SCHEMA = vol.Schema(
{
vol.Optional(CONF_ALARM_DURATION): vol.All(
vol.Coerce(int), vol.Range(min=MIN_ALARM_DURATION, max=MAX_ALARM_DURATION)
),
vol.Optional(CONF_ALARM_VOLUME): vol.All(Volume, lambda volume: volume.value),
vol.Optional(CONF_CHIME_VOLUME): vol.All(Volume, lambda volume: volume.value),
vol.Optional(CONF_ENTRY_DELAY_AWAY): vol.All(
vol.Coerce(int),
vol.Range(min=MIN_ENTRY_DELAY_AWAY, max=MAX_ENTRY_DELAY_AWAY),
),
vol.Optional(CONF_ENTRY_DELAY_HOME): vol.All(
vol.Coerce(int),
vol.Range(min=MIN_ENTRY_DELAY_HOME, max=MAX_ENTRY_DELAY_HOME),
),
vol.Optional(CONF_EXIT_DELAY_AWAY): vol.All(
vol.Coerce(int), vol.Range(min=MIN_EXIT_DELAY_AWAY, max=MAX_EXIT_DELAY_AWAY)
),
vol.Optional(CONF_EXIT_DELAY_HOME): vol.All(
vol.Coerce(int), vol.Range(min=MIN_EXIT_DELAY_HOME, max=MAX_EXIT_DELAY_HOME)
),
vol.Optional(CONF_LIGHT): bool,
vol.Optional(CONF_VOICE_PROMPT_VOLUME): vol.All(
Volume, lambda volume: volume.value
),
}
)
def create_pin_payload(pins: dict) -> dict[str, dict[str, dict[str, str]]]:
"""Create the request payload to send for updating PINs.
Args:
pins: A dictionary of pins.
Returns:
A SimpliSafe V3 PIN payload.
"""
duress_pin = pins.pop(CONF_DURESS_PIN)
master_pin = pins.pop(CONF_MASTER_PIN)
payload = {
"pins": {
CONF_DURESS_PIN: {"pin": duress_pin},
CONF_MASTER_PIN: {"pin": master_pin},
}
}
user_pins = {}
for idx, (label, pin) in enumerate(pins.items()):
user_pins[str(idx)] = {"name": label, "pin": pin}
empty_user_index = len(pins)
for idx in range(DEFAULT_MAX_USER_PINS - empty_user_index):
user_pins[str(idx + empty_user_index)] = {
"name": "",
"pin": "",
}
payload["pins"]["users"] = user_pins
LOGGER.debug("PIN payload: %s", payload)
return payload
class SystemV3(System): # pylint: disable=too-many-public-methods
"""Define a V3 (new) system.
Note that this class shouldn't be instantiated directly; it will be instantiated as
appropriate via :meth:`simplipy.API.async_get_systems`.
Args:
api: A :meth:`simplipy.API` object.
sid: A subscription ID.
"""
def __init__(self, api: API, system_id: int) -> None:
"""Initialize.
Args:
api: A :meth:`simplipy.API` object.
system_id: A system ID.
"""
super().__init__(api, system_id)
self._last_state_change_dt: datetime | None = None
# This will be filled in by the appropriate data update methods:
self.camera_data: dict[str, dict] = self._generate_camera_data()
self.cameras: dict[str, Camera] = {}
self.locks: dict[str, Lock] = {}
self.settings_data: dict[str, dict] = {}
@property
@guard_from_missing_data()
def alarm_duration(self) -> int | None:
"""Return the number of seconds an activated alarm will sound for.
Returns:
The alarm duration.
"""
return cast(
int,
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["alarm_duration"]
],
)
@property
@guard_from_missing_data()
def alarm_volume(self) -> Volume:
"""Return the volume level of the alarm.
Returns:
The alarm volume.
"""
return Volume(
int(
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["alarm_volume"]
]
)
)
@property
@guard_from_missing_data()
def battery_backup_power_level(self) -> int:
"""Return the power rating of the battery backup.
Returns:
The battery backup power rating.
"""
return cast(int, self.settings_data["basestationStatus"]["backupBattery"])
@property
@guard_from_missing_data()
def chime_volume(self) -> Volume:
"""Return the volume level of the door chime.
Returns:
The door chime volume.
"""
return Volume(
int(
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["chime_volume"]
]
)
)
@property
@guard_from_missing_data()
def entry_delay_away(self) -> int:
"""Return the number of seconds to delay when returning to an "away" alarm.
Returns:
The entry delay when returning to an "away" alarm.
"""
return cast(
int,
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["entry_delay_away"]
],
)
@property
@guard_from_missing_data()
def entry_delay_home(self) -> int:
"""Return the number of seconds to delay when returning to a "home" alarm.
Returns:
The entry delay when returning to a "home" alarm.
"""
return cast(
int,
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["entry_delay_home"]
],
)
@property
@guard_from_missing_data()
def exit_delay_away(self) -> int:
"""Return the number of seconds to delay when exiting an "away" alarm.
Returns:
The exit delay when exiting an "away" alarm.
"""
return cast(
int,
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["exit_delay_away"]
],
)
@property
@guard_from_missing_data()
def exit_delay_home(self) -> int:
"""Return the number of seconds to delay when exiting an "home" alarm.
Returns:
The exit delay when exiting a "home" alarm.
"""
return cast(
int,
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["exit_delay_home"]
],
)
@property
@guard_from_missing_data()
def gsm_strength(self) -> int:
"""Return the signal strength of the cell antenna.
Returns:
The cell antenna strength.
"""
return cast(int, self.settings_data["basestationStatus"]["gsmRssi"])
@property
@guard_from_missing_data()
def light(self) -> bool:
"""Return whether the base station light is on.
Returns:
The light status.
"""
return cast(
bool,
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["light"]
],
)
@property
@guard_from_missing_data(default_value=False)
def offline(self) -> bool:
"""Return whether the system is offline.
Returns:
The offline status.
"""
return cast(
bool,
self._api.subscription_data[self._sid]["location"]["system"]["isOffline"],
)
@property
@guard_from_missing_data(default_value=False)
def power_outage(self) -> bool:
"""Return whether the system is experiencing a power outage.
Returns:
The power outage status.
"""
return cast(
bool,
self._api.subscription_data[self._sid]["location"]["system"]["powerOutage"],
)
@property
@guard_from_missing_data(default_value=False)
def rf_jamming(self) -> bool:
"""Return whether the base station is noticing RF jamming.
Returns:
The RF jamming status.
"""
return cast(bool, self.settings_data["basestationStatus"]["rfJamming"])
@property
@guard_from_missing_data()
def voice_prompt_volume(self) -> Volume:
"""Return the volume level of the voice prompt.
Returns:
The voice prompt volume.
"""
return Volume(
int(
self.settings_data["settings"]["normal"][
SYSTEM_PROPERTIES_VALUE_MAP["voice_prompt_volume"]
]
)
)
@property
@guard_from_missing_data()
def wall_power_level(self) -> int:
"""Return the power rating of the A/C outlet.
Returns:
The A/C power rating.
"""
return cast(int, self.settings_data["basestationStatus"]["wallPower"])
@property
@guard_from_missing_data()
def wifi_ssid(self) -> str:
"""Return the ssid of the base station.
Returns:
The connected SSID.
"""
return cast(str, self.settings_data["settings"]["normal"]["wifiSSID"])
@property
@guard_from_missing_data()
def wifi_strength(self) -> int:
"""Return the signal strength of the wifi antenna.
Returns:
The WiFi strength.
"""
return cast(int, self.settings_data["basestationStatus"]["wifiRssi"])
async def _async_clear_notifications(self) -> None:
"""Clear active notifications."""
await self._api.async_request(
"delete", f"ss3/subscriptions/{self.system_id}/messages"
)
async def _async_set_state(self, value: SystemStates) -> None:
"""Set the state of the system.
Args:
value: A :meth:`simplipy.system.SystemStates` object.
"""
await self._api.async_request(
"post", f"ss3/subscriptions/{self.system_id}/state/{value.name.lower()}"
)
self._state = value
self._last_state_change_dt = utcnow()
async def _async_set_updated_pins(self, pins: dict[str, Any]) -> None:
"""Post new PINs.
Args:
pins: A dictionary of PINs.
"""
self.settings_data = await self._api.async_request(
"post",
f"ss3/subscriptions/{self.system_id}/settings/pins",
json=create_pin_payload(pins),
)
async def _async_update_device_data(self, cached: bool = True) -> None:
"""Update all device data.
Args:
cached: Whether to update with cached data.
"""
sensor_resp = await self._api.async_request(
"get",
f"ss3/subscriptions/{self.system_id}/sensors",
params={"forceUpdate": str(not cached).lower()},
)
self.sensor_data = {
sensor["serial"]: sensor for sensor in sensor_resp.get("sensors", [])
}
async def _async_update_settings_data(self, cached: bool = True) -> None:
"""Update all settings data.
Args:
cached: Whether to update with cached data.
"""
settings_resp = await self._api.async_request(
"get",
f"ss3/subscriptions/{self.system_id}/settings/normal",
params={"forceUpdate": str(not cached).lower()},
)
if settings_resp:
self.settings_data = settings_resp
async def _async_update_subscription_data(self) -> None:
"""Update subscription data."""
await super()._async_update_subscription_data()
self.camera_data = self._generate_camera_data()
def _generate_camera_data(self) -> dict[str, dict]:
"""Generate usable, hashable camera data from subscription data.
This method exists because the SimpliSafe API includes camera data with the
subscription (and not with other devices); by splitting this out, we can
separate this action from updating the subscription data itself.
Returns:
A dictionary of camera UUID to camera data.
"""
return {
camera["uuid"]: camera
for camera in self._api.subscription_data[self._sid]["location"][
"system"
].get("cameras", [])
}
def as_dict(self) -> dict[str, Any]:
"""Return dictionary version of this device.
Returns:
A dict representation of this device.
"""
data = {
**super().as_dict(),
"alarm_duration": self.alarm_duration,
"battery_backup_power_level": self.battery_backup_power_level,
"cameras": [camera.as_dict() for camera in self.cameras.values()],
"entry_delay_away": self.entry_delay_away,
"entry_delay_home": self.entry_delay_home,
"exit_delay_away": self.exit_delay_away,
"exit_delay_home": self.exit_delay_home,
"gsm_strength": self.gsm_strength,
"light": self.light,
"locks": [lock.as_dict() for lock in self.locks.values()],
"offline": self.offline,
"power_outage": self.power_outage,
"rf_jamming": self.rf_jamming,
"wall_power_level": self.wall_power_level,
"wifi_ssid": self.wifi_ssid,
"wifi_strength": self.wifi_strength,
}
for key, volume_enum in (
("alarm_volume", self.alarm_volume),
("chime_volume", self.chime_volume),
("voice_prompt_volume", self.voice_prompt_volume),
):
if volume_enum:
data[key] = volume_enum.value
return data
def generate_device_objects(self) -> None:
"""Generate device objects for this system."""
for serial, sensor in self.sensor_data.items():
if (sensor_type := get_device_type_from_data(sensor)) == DeviceTypes.LOCK:
self.locks[serial] = Lock(
self._api.async_request, self, sensor_type, serial
)
else:
self.sensors[serial] = SensorV3(self, sensor_type, serial)
for serial in self.camera_data:
self.cameras[serial] = Camera(self, DeviceTypes.CAMERA, serial)
async def async_get_pins(self, cached: bool = True) -> dict[str, str]:
"""Return all of the set PINs, including master and duress.
The ``cached`` parameter determines whether the SimpliSafe Cloud uses the last
known values retrieved from the base station (``True``) or retrieves new data.
Args:
cached: Whether to update with cached data.
Returns:
A dictionary of PINs.
"""
await self._async_update_settings_data(cached)
pins = {
CONF_MASTER_PIN: self.settings_data["settings"]["pins"]["master"]["pin"],
CONF_DURESS_PIN: self.settings_data["settings"]["pins"]["duress"]["pin"],
}
for user_pin in [
p for p in self.settings_data["settings"]["pins"]["users"] if p["pin"]
]:
pins[user_pin["name"]] = user_pin["pin"]
return pins
async def async_set_properties(
self, properties: dict[str, bool | int | Volume]
) -> None:
"""Set various system properties.
Volume properties should take values from :meth:`simplipy.system.v3.Volume`.
The following properties can be set:
1. alarm_duration (in seconds): 30-480
2. alarm_volume: Volume.OFF, Volume.LOW, Volume.MEDIUM, Volume.HIGH
3. chime_volume: Volume.OFF, Volume.LOW, Volume.MEDIUM, Volume.HIGH
4. entry_delay_away (in seconds): 30-255
5. entry_delay_home (in seconds): 0-255
6. exit_delay_away (in seconds): 45-255
7. exit_delay_home (in seconds): 0-255
8. light: True or False
9. voice_prompt_volume: Volume.OFF, Volume.LOW, Volume.MEDIUM, Volume.HIGH
Args:
properties: The system properties to set.
Raises:
ValueError: Raised on invalid properties.
"""
try:
parsed_properties = SYSTEM_PROPERTIES_PAYLOAD_SCHEMA(properties)
except vol.Invalid as err:
raise ValueError(
f"Using invalid values for system properties ({properties}): {err}"
) from None
settings_resp = await self._api.async_request(
"post",
f"ss3/subscriptions/{self.system_id}/settings/normal",
json={
"normal": {
SYSTEM_PROPERTIES_VALUE_MAP[prop]: value
for prop, value in parsed_properties.items()
}
},
)
if settings_resp:
self.settings_data = settings_resp
async def async_update(
self,
*,
include_subscription: bool = True,
include_settings: bool = True,
include_devices: bool = True,
cached: bool = True,
) -> None:
"""Get the latest system data.
The ``cached`` parameter determines whether the SimpliSafe Cloud uses the last
known values retrieved from the base station (``True``) or retrieves new data.
Args:
include_subscription: Whether system state/properties should be updated.
include_settings: Whether system settings (like PINs) should be updated.
include_devices: whether sensors/locks/etc. should be updated.
cached: Whether to used cached data.
"""
if (
self.locks
and self._last_state_change_dt
and utcnow()
<= self._last_state_change_dt + DEFAULT_LOCK_STATE_CHANGE_WINDOW
):
# The SimpliSafe cloud API currently has a bug wherein systems with locks
# will audible announce that those locks aren't responding when the system
# is updated within a certain window (around 15 seconds) of the system
# changing state. Oof. So, we refuse to update inside that window:
LOGGER.info(
"Skipping system update within %s seconds from last system arm/disarm",
DEFAULT_LOCK_STATE_CHANGE_WINDOW,
)
return
await super().async_update(
include_subscription=include_subscription,
include_settings=include_settings,
include_devices=include_devices,
cached=cached,
)