feat: add async_get_doorbell_image method to the YaleXSData object (#120

)

yalexs/doorbell.py

@@ -19,6 +19,8 @@
 
 
 class Doorbell(Device):
+    """Class to hold details about a doorbell."""
+
     def __init__(self, device_id: str, data: dict[str, Any]) -> None:
         _LOGGER.info("Doorbell init - %s", data["name"])
         super().__init__(device_id, data["name"], data["HouseID"])
@@ -64,7 +66,9 @@ def __repr__(self):
 
 
 class DoorbellDetail(DeviceDetail):
-    def __init__(self, data):
+    """Class to hold details about a doorbell."""
+
+    def __init__(self, data: dict[str, Any]) -> None:
         super().__init__(
             data["doorbellID"],
             data["name"],
@@ -75,21 +79,23 @@ def __init__(self, data):
             data,
         )
 
-        self._status = data["status"]
-        recent_image = data.get("recentImage", {})
-        self._image_url = recent_image.get("secure_url", None)
-        self._has_subscription = data.get("dvrSubscriptionSetupDone", False)
-        self._image_created_at_datetime = None
-        self._model = None
-        self._content_token = data.get("contentToken", "")
+        self._status: str = data["status"]
+        recent_image: dict[str, Any] = data.get("recentImage", {})
+        self._image_url: str | None = recent_image.get("secure_url")
+        self._has_subscription: bool = data.get("dvrSubscriptionSetupDone", False)
+        self._image_created_at_datetime: datetime.datetime | None = None
+        self._model: str | int | None = None
+        self._content_token: str = data.get("contentToken", "")
 
         if "type" in data:
             self._model = data["type"]
 
         if "created_at" in recent_image:
-            self._image_created_at_datetime = parse_datetime(recent_image["created_at"])
+            self._image_created_at_datetime: datetime.datetime = parse_datetime(
+                recent_image["created_at"]
+            )
 
-        self._battery_level = None
+        self._battery_level: int | None = None
         if "telemetry" in data:
             telemetry = data["telemetry"]
             if "battery_soc" in telemetry:
@@ -108,62 +114,62 @@ def __init__(self, data):
                     self._battery_level = 25
 
     @cached_property
-    def status(self):
+    def status(self) -> str:
         return self._status
 
     @cached_property
-    def model(self):
+    def model(self) -> str | int | None:
         return self._model
 
     @cached_property
-    def is_online(self):
+    def is_online(self) -> bool:
         return self.status == "doorbell_call_status_online"
 
     @cached_property
-    def is_standby(self):
+    def is_standby(self) -> bool:
         return self.status == "standby"
 
     @property
-    def image_created_at_datetime(self):
+    def image_created_at_datetime(self) -> datetime.datetime | datetime.date | None:
         return self._image_created_at_datetime
 
     @image_created_at_datetime.setter
-    def image_created_at_datetime(self, var):
+    def image_created_at_datetime(self, var: datetime.date):
         """Update the doorbell image created_at datetime (usually form the activity log)."""
         if not isinstance(var, datetime.date):
             raise ValueError
         self._image_created_at_datetime = var
 
     @property
-    def image_url(self):
+    def image_url(self) -> str | None:
         return self._image_url
 
     @property
-    def content_token(self):
+    def content_token(self) -> str:
         return self._content_token
 
     @image_url.setter
-    def image_url(self, var):
+    def image_url(self, var: str | None) -> None:
         """Update the doorbell image url (usually form the activity log)."""
         _LOGGER.debug("image_url updated for %s", self.device_name)
         self._image_url = var
 
     @content_token.setter
-    def content_token(self, var):
+    def content_token(self, var: str) -> None:
         _LOGGER.debug("content_token updated for %s", self.device_name)
         self._content_token = var or ""
 
     @cached_property
-    def battery_level(self):
+    def battery_level(self) -> int | None:
         """Return an approximation of the battery percentage."""
         return self._battery_level
 
     @cached_property
-    def has_subscription(self):
+    def has_subscription(self) -> bool:
         return self._has_subscription
 
     async def async_get_doorbell_image(
-        self, aiohttp_session: ClientSession, timeout=10
+        self, aiohttp_session: ClientSession, timeout: float = 10.0
     ) -> bytes:
         _LOGGER.debug("async_get_doorbell_image %s", self.device_name)
         response = await aiohttp_session.request(
@@ -179,7 +185,7 @@ async def async_get_doorbell_image(
             raise ContentTokenExpired
         return await response.read()
 
-    def get_doorbell_image(self, timeout=10) -> bytes:
+    def get_doorbell_image(self, timeout: float = 10.0) -> bytes:
         _LOGGER.debug("get_doorbell_image sync %s", self.device_name)
         return requests.get(
             self._image_url,

yalexs/manager/data.py

@@ -11,11 +11,12 @@
 from itertools import chain
 from typing import Any, ParamSpec, TypeVar
 
-from aiohttp import ClientError, ClientResponseError
+from aiohttp import ClientError, ClientResponseError, ClientSession
 
 from ..activity import ActivityTypes
 from ..backports.tasks import create_eager_task
-from ..doorbell import Doorbell, DoorbellDetail
+from ..const import Brand
+from ..doorbell import ContentTokenExpired, Doorbell, DoorbellDetail
 from ..exceptions import AugustApiAIOHTTPError
 from ..lock import Lock, LockDetail
 from ..pubnub_activity import activities_from_pubnub_message
@@ -120,7 +121,7 @@ async def async_setup_activity_stream(self) -> None:
         await self.activity_stream.async_setup()
         pubnub.subscribe(self.async_pubnub_message)
         self._pubnub_unsub = async_create_pubnub(
-            user_data["UserID"], pubnub, self.brand
+            user_data["UserID"], pubnub, self._gateway.api.brand
         )
 
     async def _async_initial_sync(self) -> None:
@@ -384,6 +385,26 @@ async def _async_call_api_op_requires_bridge(
 
         return ret
 
+    async def async_get_doorbell_image(
+        self,
+        device_id: str,
+        aiohttp_session: ClientSession,
+        timeout: float = 10.0,
+    ) -> bytes:
+        """Get the latest image from the doorbell."""
+        doorbell = self.get_device_detail(device_id)
+        try:
+            return await doorbell.async_get_doorbell_image(aiohttp_session, timeout)
+        except ContentTokenExpired:
+            if self._gateway.api.brand != Brand.YALE_HOME:
+                raise
+            _LOGGER.debug(
+                "Error fetching camera image, updating content-token from api to retry"
+            )
+            await self.refresh_camera_by_id(device_id)
+            doorbell = self.get_device_detail(device_id)
+            return await doorbell.async_get_doorbell_image(aiohttp_session, timeout)
+
     def _remove_inoperative_doorbells(self) -> None:
         for doorbell in list(self.doorbells):
             device_id = doorbell.device_id

yalexs/manager/gateway.py

@@ -10,12 +10,11 @@
 
 from aiohttp import ClientError, ClientResponseError, ClientSession
 
-from yalexs.api_async import ApiAsync
-from yalexs.authenticator_async import AuthenticationState, AuthenticatorAsync
-from yalexs.authenticator_common import Authentication
-from yalexs.const import DEFAULT_BRAND
-from yalexs.exceptions import AugustApiAIOHTTPError
-
+from ..api_async import ApiAsync
+from ..authenticator_async import AuthenticationState, AuthenticatorAsync
+from ..authenticator_common import Authentication
+from ..const import DEFAULT_BRAND
+from ..exceptions import AugustApiAIOHTTPError
 from .const import (
     CONF_ACCESS_TOKEN_CACHE_FILE,
     CONF_BRAND,