feat: adds alarm support to Yale home

README.md

@@ -52,33 +52,35 @@ pip install yalexs
 ## Usage
 
 ```python
-from yalexs.api import Api
-from yalexs.authenticator import Authenticator, AuthenticationState
+import asyncio
+from aiohttp import ClientSession
 
-api = Api(timeout=20)
-authenticator = Authenticator(api, "phone", "YOUR_USERNAME", "YOUR_PASSWORD",
-                              access_token_cache_file="PATH_TO_ACCESS_TOKEN_CACHE_FILE")
+from yalexs.api_async import ApiAsync
+from yalexs.authenticator_async import AuthenticatorAsync
+from yalexs.const import Brand
+from yalexs.alarm import ArmState
 
-authentication = authenticator.authenticate()
 
-# State can be either REQUIRES_VALIDATION, BAD_PASSWORD or AUTHENTICATED
-# You'll need to call different methods to finish authentication process, see below
-state = authentication.state
+async def main():
+    api = ApiAsync(ClientSession(), timeout=20, brand=Brand.YALE_HOME)
+    authenticator = AuthenticatorAsync(api, "email", "EMAIL_ADDRESS", "PASSWORD}",
+        access_token_cache_file="auth.txt",install_id="UUID")
+    await authenticator.async_setup_authentication()
+    authentication = await authenticator.async_authenticate()
+    access_token = authentication.access_token
 
-# If AuthenticationState is BAD_PASSWORD, that means your login_method, username and password do not match
+    # if(authentication.state == AuthenticationState.REQUIRES_VALIDATION) :
+    #   await authenticator.async_send_verification_code()
+    # await authenticator.async_validate_verification_code("12345")
 
-# If AuthenticationState is AUTHENTICATED, that means you're authenticated already. If you specify "access_token_cache_file", the authentication is cached in a file. Every time you try to authenticate again, it'll read from that file and if you're authenticated already, Authenticator won't call Yale Access again as you have a valid access_token
+    # DO STUFF HERE LIKE GET THE ALARMS, LOCS, ETC....
+    alarms = await api.async_get_alarms(access_token)
+    locks = api.get_locks(access_token)
 
+    # OR ARM YOUR ALARM
+    await api.async_arm_alarm(access_token, alarms[0], ArmState.Away)
 
-# If AuthenticationState is REQUIRES_VALIDATION, then you'll need to go through verification process
-# send_verification_code() will send a code to either your phone or email depending on login_method
-authenticator.send_verification_code()
-# Wait for your code and pass it in to validate_verification_code()
-validation_result = authenticator.validate_verification_code(123456)
-# If ValidationResult is INVALID_VERIFICATION_CODE, then you'll need to either enter correct one or resend by calling send_verification_code() again
-# If ValidationResult is VALIDATED, then you'll need to call authenticate() again to finish authentication process
-authentication = authenticator.authenticate()
 
-# Once you have authenticated and validated you can use the access token to make API calls
-locks = api.get_locks(authentication.access_token)
+
+asyncio.run(main())
 ```

yalexs/alarm.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from ._compat import cached_property
+from .backports.enum import StrEnum
+from .device import Device, DeviceDetail
+
+
+class ArmState(StrEnum):
+    Away = "FULL_ARM"
+    Home = "PARTIAL_ARM"
+    Disarm = "DISARM"
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Alarm(Device):
+    """Class to hold details about an alarm."""
+
+    def __init__(self, device_id: str, data: dict[str, Any]) -> None:
+        _LOGGER.info("Alarm init - %s", data["location"])
+        super().__init__(device_id, data["location"], data["houseID"])
+        self._pubsub_channel = data["pubsubChannel"]
+        self._serial_number = data["serialNumber"]
+        self._status = data["status"]
+        self._areaIDs = data["areaIDs"]
+
+    @cached_property
+    def pubsub_channel(self):
+        return self._pubsub_channel
+
+    @cached_property
+    def serial_number(self):
+        return self._serial_number
+
+    @cached_property
+    def status(self):
+        return self._status
+
+    @cached_property
+    def areaIDs(self):
+        return self._areaIDs
+
+    def __repr__(self):
+        return f"Alarm(id={self.device_id}, name={self.device_name}, house_id={self.house_id})"
+
+
+class AlarmDevice(DeviceDetail):
+    """Class to hold details about a device attached to the alarm."""
+
+    def __init__(self, data: dict[str, Any]) -> None:
+        _LOGGER.info("Alarm init - %s (%s)", data["name"], data["type"])
+        super().__init__(
+            data["_id"],
+            data["name"],
+            data["alarmID"],
+            data["serialNumber"],
+            data["status"]["firmwareVersion"],
+            data.get("pubsubChannel"),
+            data,
+        )
+
+        self._status: str = data["status"]
+        self._model = data["type"]
+
+        self._battery_level = 100
+        if self._status.get("lowBattery", False):
+            self._battery_level = 10
+
+    @cached_property
+    def status(self) -> str:
+        return self._status
+
+    @cached_property
+    def model(self) -> str:
+        return self._model
+
+    @cached_property
+    def is_online(self) -> bool:
+        return self.status.get("online", False)
+
+    @cached_property
+    def contact_open(self) -> bool:
+        return self.status.get("contactOpen", False)
+
+    @cached_property
+    def fault(self) -> bool:
+        return self.status.get("fault", False)
+
+    @cached_property
+    def tamperOpen(self) -> bool:
+        return self.status.get("tamperOpen", False)
+
+    @cached_property
+    def battery_level(self) -> int | None:
+        """Return an approximation of the battery percentage."""
+        return self._battery_level
+
+    def __repr__(self):
+        return f"AlarmDevice(id={self.device_id}, name={self.device_name}, type={self.model}, alarm_id={self.house_id})"

yalexs/api_async.py

@@ -18,6 +18,7 @@
 )
 
 from .activity import ActivityTypes
+from .alarm import Alarm, AlarmDevice, ArmState
 from .api_common import (
     API_EXCEPTION_RETRY_TIME,
     API_LOCK_ASYNC_URL,
@@ -35,6 +36,8 @@
     _api_headers,
     _convert_lock_result_to_activities,
     _process_activity_json,
+    _process_alarm_devices_json,
+    _process_alarms_json,
     _process_doorbells_json,
     _process_locks_json,
 )
@@ -387,6 +390,36 @@ async def async_status_async(
             API_STATUS_ASYNC_URL, access_token, lock_id
         )
 
+    async def async_get_alarms(self, access_token: str) -> list[Alarm]:
+        if not self.brand_supports_alarms:
+            return []
+        response = await self._async_dict_to_api(
+            self._build_get_alarms_request(access_token)
+        )
+        return _process_alarms_json(await response.json())
+
+    async def async_get_alarm_devices(
+        self, access_token: str, alarm: Alarm
+    ) -> list[AlarmDevice]:
+        if not self.brand_supports_alarms:
+            return []
+        response = await self._async_dict_to_api(
+            self._build_get_alarm_devices_request(
+                access_token, alarm_id=alarm.device_id
+            )
+        )
+        return _process_alarm_devices_json(await response.json())
+
+    async def async_arm_alarm(
+        self, access_token: str, alarm: Alarm, arm_state: ArmState
+    ):
+        if not self.brand_supports_alarms:
+            return {}
+        response = await self._async_dict_to_api(
+            self._build_call_alarm_state_request(access_token, alarm, arm_state)
+        )
+        return await response.json()
+
     async def async_refresh_access_token(self, access_token: str) -> str:
         """Obtain a new api token."""
         response = await self._async_dict_to_api(

yalexs/api_common.py

@@ -9,6 +9,7 @@
 
 from ._compat import cached_property
 from .activity import ACTION_TO_CLASS, SOURCE_LOCK_OPERATE, SOURCE_LOG, ActivityTypes
+from .alarm import Alarm, AlarmDevice, ArmState
 from .const import BASE_URLS, BRAND_CONFIG, BRANDING, DEFAULT_BRAND, Brand, BrandConfig
 from .doorbell import Doorbell
 from .lock import Lock, LockDoorStatus, determine_door_state, door_state_to_string
@@ -63,6 +64,10 @@
 API_GET_USER_URL = "/users/me"
 API_WEBSOCKET_SUBSCRIBERS = "/websocket/subscribers"
 API_WEBSOCKET_SUBSCRIBERS_WITH_SUBSCRIBER_ID = "/websocket/subscribers/{subscriber_id}"
+API_GET_ALARMS_URL = "/users/alarms/mine"
+API_GET_ALARM_DEVICES_URL = "/alarms/{alarm_id}/devices"
+API_PUT_ALARM_URL = "/alarms/{alarm_id}/state/{arm_state}"
+
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -175,6 +180,14 @@ def _process_locks_json(json_dict: dict[str, Any]) -> list[Lock]:
     return [Lock(device_id, data) for device_id, data in json_dict.items()]
 
 
+def _process_alarms_json(json_dict: list[dict[str, Any]]) -> list[Alarm]:
+    return [Alarm(data.get("alarmID"), data) for data in json_dict]
+
+
+def _process_alarm_devices_json(json_dict: list[dict[str, Any]]) -> list[AlarmDevice]:
+    return [AlarmDevice(data) for data in json_dict]
+
+
 class ApiCommon:
     """Api dict shared between async and sync."""
 
@@ -189,6 +202,11 @@ def brand_supports_doorbells(self) -> bool:
         """Return if the brand supports doorbells."""
         return self.brand_config.supports_doorbells
 
+    @cached_property
+    def brand_supports_alarms(self) -> bool:
+        """Return if the brand supports alarms."""
+        return self.brand_config.supports_alarms
+
     def get_brand_url(self, url_str: str) -> str:
         """Get url."""
         return f"{self._base_url}{url_str}"
@@ -262,7 +280,10 @@ def _build_wakeup_doorbell_request(
         }
 
     def _build_get_houses_request(self, access_token: str) -> dict[str, Any]:
-        return self._build_base_request(access_token)
+        return {
+            **self._build_base_request(access_token),
+            "url": self.get_brand_url(API_GET_HOUSES_URL),
+        }
 
     def _build_get_house_request(self, access_token, house_id):
         return {
@@ -363,3 +384,30 @@ def _build_call_lock_operation_request(
             "url": self.get_brand_url(url_str.format(lock_id=lock_id)),
             "timeout": timeout,
         }
+
+    def _build_get_alarms_request(self, access_token: str) -> dict[str, Any]:
+        return {
+            **self._build_base_request(access_token),
+            "url": self.get_brand_url(API_GET_ALARMS_URL),
+        }
+
+    def _build_get_alarm_devices_request(
+        self, access_token: str, alarm_id: str
+    ) -> dict[str, Any]:
+        return {
+            **self._build_base_request(access_token),
+            "url": self.get_brand_url(
+                API_GET_ALARM_DEVICES_URL.format(alarm_id=alarm_id)
+            ),
+        }
+
+    def _build_call_alarm_state_request(
+        self, access_token: str, alarm: Alarm, arm_state: ArmState
+    ) -> dict[str, Any]:
+        return {
+            **self._build_base_request(access_token=access_token, method="PUT"),
+            "url": self.get_brand_url(
+                API_PUT_ALARM_URL.format(alarm_id=alarm.device_id, arm_state=arm_state)
+            ),
+            "json": {"areaIDs": alarm.areaIDs},
+        }

yalexs/const.py

@@ -28,6 +28,7 @@ class BrandConfig:
     branding_header: str
     api_key: str
     supports_doorbells: bool
+    supports_alarms: bool
     require_oauth: bool
     base_url: str
     configuration_url: str
@@ -55,6 +56,7 @@ class BrandConfig:
         branding_header=HEADER_AUGUST_BRANDING,
         api_key=HEADER_VALUE_API_KEY,
         supports_doorbells=True,
+        supports_alarms=False,
         require_oauth=False,
         base_url="https://api-production.august.com",
         configuration_url="https://account.august.com",
@@ -69,6 +71,7 @@ class BrandConfig:
         branding_header=HEADER_AUGUST_BRANDING,
         api_key=HEADER_VALUE_API_KEY,
         supports_doorbells=True,
+        supports_alarms=False,
         require_oauth=False,
         base_url="https://api-production.august.com",
         configuration_url="https://account.august.com",
@@ -81,8 +84,9 @@ class BrandConfig:
         access_token_header=HEADER_ACCESS_TOKEN,
         api_key_header=HEADER_API_KEY,
         branding_header=HEADER_BRANDING,
-        api_key=HEADER_VALUE_API_KEY,
+        api_key="6e2a2093-6118-42c5-8a41-e1fd25dce7a1",  # 🤞
         supports_doorbells=True,
+        supports_alarms=True,
         require_oauth=False,
         base_url="https://api.aaecosystem.com",
         configuration_url="https://account.aaecosystem.com",
@@ -100,6 +104,7 @@ class BrandConfig:
         # run on the user's device
         api_key="d16a1029-d823-4b55-a4ce-a769a9b56f0e",
         supports_doorbells=True,
+        supports_alarms=True,  # ??
         require_oauth=True,
         base_url="https://api.aaecosystem.com",
         configuration_url="https://account.aaecosystem.com",