Improve august typing (5) (#108332)

This commit is contained in:
Marc Mueller 2024-01-19 01:09:52 +01:00 committed by GitHub
parent 0206833cfd
commit 94c8c71ffb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 38 additions and 28 deletions

View file

@ -9,6 +9,7 @@ import logging
from typing import Any, ParamSpec, TypeVar
from aiohttp import ClientError, ClientResponseError
from yalexs.activity import ActivityTypes
from yalexs.const import DEFAULT_BRAND
from yalexs.doorbell import Doorbell, DoorbellDetail
from yalexs.exceptions import AugustApiAIOHTTPError
@ -274,7 +275,7 @@ class AugustData(AugustSubscriberMixin):
"""Return the py-august LockDetail or DoorbellDetail object for a device."""
return self._device_detail_by_id[device_id]
async def _async_refresh(self, time):
async def _async_refresh(self, time: datetime) -> None:
await self._async_refresh_device_detail_by_ids(self._subscriptions.keys())
async def _async_refresh_device_detail_by_ids(
@ -329,7 +330,13 @@ class AugustData(AugustSubscriberMixin):
)
self.async_signal_device_id_update(device_id)
async def _async_update_device_detail(self, device, api_call):
async def _async_update_device_detail(
self,
device: Doorbell | Lock,
api_call: Callable[
[str, str], Coroutine[Any, Any, DoorbellDetail | LockDetail]
],
) -> None:
_LOGGER.debug(
"Started retrieving detail for %s (%s)",
device.device_name,
@ -363,7 +370,7 @@ class AugustData(AugustSubscriberMixin):
return device.device_name
return None
async def async_lock(self, device_id: str):
async def async_lock(self, device_id: str) -> list[ActivityTypes]:
"""Lock the device."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -372,9 +379,7 @@ class AugustData(AugustSubscriberMixin):
device_id,
)
async def async_status_async(
self, device_id: str, hyper_bridge: bool
) -> str | None:
async def async_status_async(self, device_id: str, hyper_bridge: bool) -> str:
"""Request status of the device but do not wait for a response since it will come via pubnub."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -384,7 +389,7 @@ class AugustData(AugustSubscriberMixin):
hyper_bridge,
)
async def async_lock_async(self, device_id: str, hyper_bridge: bool) -> str | None:
async def async_lock_async(self, device_id: str, hyper_bridge: bool) -> str:
"""Lock the device but do not wait for a response since it will come via pubnub."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -394,7 +399,7 @@ class AugustData(AugustSubscriberMixin):
hyper_bridge,
)
async def async_unlock(self, device_id: str):
async def async_unlock(self, device_id: str) -> list[ActivityTypes]:
"""Unlock the device."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -403,9 +408,7 @@ class AugustData(AugustSubscriberMixin):
device_id,
)
async def async_unlock_async(
self, device_id: str, hyper_bridge: bool
) -> str | None:
async def async_unlock_async(self, device_id: str, hyper_bridge: bool) -> str:
"""Unlock the device but do not wait for a response since it will come via pubnub."""
return await self._async_call_api_op_requires_bridge(
device_id,
@ -421,9 +424,8 @@ class AugustData(AugustSubscriberMixin):
func: Callable[_P, Coroutine[Any, Any, _R]],
*args: _P.args,
**kwargs: _P.kwargs,
) -> _R | None:
) -> _R:
"""Call an API that requires the bridge to be online and will change the device state."""
ret = None
try:
ret = await func(*args, **kwargs)
except AugustApiAIOHTTPError as err:
@ -479,7 +481,7 @@ class AugustData(AugustSubscriberMixin):
del self._locks_by_id[device_id]
def _save_live_attrs(lock_detail):
def _save_live_attrs(lock_detail: DoorbellDetail | LockDetail) -> dict[str, Any]:
"""Store the attributes that the lock detail api may have an invalid cache for.
Since we are connected to pubnub we may have more current data
@ -489,7 +491,9 @@ def _save_live_attrs(lock_detail):
return {attr: getattr(lock_detail, attr) for attr in API_CACHED_ATTRS}
def _restore_live_attrs(lock_detail, attrs):
def _restore_live_attrs(
lock_detail: DoorbellDetail | LockDetail, attrs: dict[str, Any]
) -> None:
"""Restore the non-cache attributes after a cached update."""
for attr, value in attrs.items():
setattr(lock_detail, attr, value)

View file

@ -58,7 +58,7 @@ class AugustCamera(AugustEntityMixin, Camera):
return self._device.has_subscription
@property
def model(self):
def model(self) -> str | None:
"""Return the camera model."""
return self._detail.model

View file

@ -1,7 +1,7 @@
"""Base class for August entity."""
from abc import abstractmethod
from yalexs.doorbell import Doorbell
from yalexs.doorbell import Doorbell, DoorbellDetail
from yalexs.lock import Lock, LockDetail
from yalexs.util import get_configuration_url
@ -46,7 +46,7 @@ class AugustEntityMixin(Entity):
return self._device.device_id
@property
def _detail(self):
def _detail(self) -> DoorbellDetail | LockDetail:
return self._data.get_device_detail(self._device.device_id)
@property

View file

@ -1,9 +1,12 @@
"""Support for August lock."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
import logging
from typing import Any
from aiohttp import ClientResponseError
from yalexs.activity import SOURCE_PUBNUB, ActivityType
from yalexs.activity import SOURCE_PUBNUB, ActivityType, ActivityTypes
from yalexs.lock import Lock, LockStatus
from yalexs.util import get_latest_activity, update_lock_detail_from_activity
@ -62,7 +65,9 @@ class AugustLock(AugustEntityMixin, RestoreEntity, LockEntity):
return
await self._call_lock_operation(self._data.async_unlock)
async def _call_lock_operation(self, lock_operation):
async def _call_lock_operation(
self, lock_operation: Callable[[str], Coroutine[Any, Any, list[ActivityTypes]]]
) -> None:
try:
activities = await lock_operation(self._device_id)
except ClientResponseError as err:

View file

@ -4,9 +4,9 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import Any, Generic, TypeVar
from typing import Any, Generic, TypeVar, cast
from yalexs.activity import ActivityType
from yalexs.activity import ActivityType, LockOperationActivity
from yalexs.doorbell import Doorbell
from yalexs.keypad import KeypadDetail
from yalexs.lock import Lock, LockDetail
@ -158,7 +158,7 @@ async def async_setup_entry(
async_add_entities(entities)
async def _async_migrate_old_unique_ids(hass, devices):
async def _async_migrate_old_unique_ids(hass: HomeAssistant, devices) -> None:
"""Keypads now have their own serial number."""
registry = er.async_get(hass)
for device in devices:
@ -184,11 +184,11 @@ class AugustOperatorSensor(AugustEntityMixin, RestoreSensor):
super().__init__(data, device)
self._data = data
self._device = device
self._operated_remote = None
self._operated_keypad = None
self._operated_manual = None
self._operated_tag = None
self._operated_autorelock = None
self._operated_remote: bool | None = None
self._operated_keypad: bool | None = None
self._operated_manual: bool | None = None
self._operated_tag: bool | None = None
self._operated_autorelock: bool | None = None
self._operated_time = None
self._attr_unique_id = f"{self._device_id}_lock_operator"
self._update_from_data()
@ -202,6 +202,7 @@ class AugustOperatorSensor(AugustEntityMixin, RestoreSensor):
self._attr_available = True
if lock_activity is not None:
lock_activity = cast(LockOperationActivity, lock_activity)
self._attr_native_value = lock_activity.operated_by
self._operated_remote = lock_activity.operated_remote
self._operated_keypad = lock_activity.operated_keypad