Improve august typing (1) (#108325)

This commit is contained in:
Marc Mueller 2024-01-18 23:13:08 +01:00 committed by GitHub
parent edd7feaf10
commit 5f08e2a2d1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 42 additions and 33 deletions

View file

@ -2,7 +2,7 @@
from __future__ import annotations
import asyncio
from collections.abc import ValuesView
from collections.abc import Iterable, ValuesView
from datetime import datetime
from itertools import chain
import logging
@ -104,7 +104,7 @@ async def async_setup_august(
@callback
def _async_trigger_ble_lock_discovery(
hass: HomeAssistant, locks_with_offline_keys: list[LockDetail]
):
) -> None:
"""Update keys for the yalexs-ble integration if available."""
for lock_detail in locks_with_offline_keys:
discovery_flow.async_create_flow(
@ -213,7 +213,7 @@ class AugustData(AugustSubscriberMixin):
self._hass, self._async_initial_sync(), "august-initial-sync"
)
async def _async_initial_sync(self):
async def _async_initial_sync(self) -> None:
"""Attempt to request an initial sync."""
# We don't care if this fails because we only want to wake
# locks that are actually online anyways and they will be
@ -274,7 +274,9 @@ class AugustData(AugustSubscriberMixin):
async def _async_refresh(self, time):
await self._async_refresh_device_detail_by_ids(self._subscriptions.keys())
async def _async_refresh_device_detail_by_ids(self, device_ids_list):
async def _async_refresh_device_detail_by_ids(
self, device_ids_list: Iterable[str]
) -> None:
"""Refresh each device in sequence.
This used to be a gather but it was less reliable with august's
@ -421,7 +423,7 @@ class AugustData(AugustSubscriberMixin):
return ret
def _remove_inoperative_doorbells(self):
def _remove_inoperative_doorbells(self) -> None:
for doorbell in list(self.doorbells):
device_id = doorbell.device_id
if self._device_detail_by_id.get(device_id):
@ -435,7 +437,7 @@ class AugustData(AugustSubscriberMixin):
)
del self._doorbells_by_id[device_id]
def _remove_inoperative_locks(self):
def _remove_inoperative_locks(self) -> None:
# Remove non-operative locks as there must
# be a bridge (August Connect) for them to
# be usable

View file

@ -1,4 +1,6 @@
"""Consume the august activity stream."""
from __future__ import annotations
import asyncio
from datetime import datetime
from functools import partial

View file

@ -228,7 +228,7 @@ class AugustDoorBinarySensor(AugustEntityMixin, BinarySensorEntity):
self._attr_unique_id = f"{self._device_id}_{description.key}"
@callback
def _update_from_data(self):
def _update_from_data(self) -> None:
"""Get the latest state of the sensor and update activity."""
assert self._data.activity_stream is not None
door_activity = self._data.activity_stream.get_latest_device_activity(
@ -270,12 +270,12 @@ class AugustDoorbellBinarySensor(AugustEntityMixin, BinarySensorEntity):
"""Initialize the sensor."""
super().__init__(data, device)
self.entity_description = description
self._check_for_off_update_listener = None
self._check_for_off_update_listener: Callable[[], None] | None = None
self._data = data
self._attr_unique_id = f"{self._device_id}_{description.key}"
@callback
def _update_from_data(self):
def _update_from_data(self) -> None:
"""Get the latest state of the sensor."""
self._cancel_any_pending_updates()
self._attr_is_on = self.entity_description.value_fn(self._data, self._detail)
@ -286,14 +286,14 @@ class AugustDoorbellBinarySensor(AugustEntityMixin, BinarySensorEntity):
else:
self._attr_available = True
def _schedule_update_to_recheck_turn_off_sensor(self):
def _schedule_update_to_recheck_turn_off_sensor(self) -> None:
"""Schedule an update to recheck the sensor to see if it is ready to turn off."""
# If the sensor is already off there is nothing to do
if not self.is_on:
return
@callback
def _scheduled_update(now):
def _scheduled_update(now: datetime) -> None:
"""Timer callback for sensor update."""
self._check_for_off_update_listener = None
self._update_from_data()
@ -304,7 +304,7 @@ class AugustDoorbellBinarySensor(AugustEntityMixin, BinarySensorEntity):
self.hass, TIME_TO_RECHECK_DETECTION.total_seconds(), _scheduled_update
)
def _cancel_any_pending_updates(self):
def _cancel_any_pending_updates(self) -> None:
"""Cancel any updates to recheck a sensor to see if it is ready to turn off."""
if not self._check_for_off_update_listener:
return

View file

@ -36,5 +36,5 @@ class AugustWakeLockButton(AugustEntityMixin, ButtonEntity):
await self._data.async_status_async(self._device_id, self._hyper_bridge)
@callback
def _update_from_data(self):
def _update_from_data(self) -> None:
"""Nothing to update as buttons are stateless."""

View file

@ -1,7 +1,9 @@
"""Support for August doorbell camera."""
from __future__ import annotations
from aiohttp import ClientSession
from yalexs.activity import ActivityType
from yalexs.doorbell import Doorbell
from yalexs.util import update_doorbell_image_from_activity
from homeassistant.components.camera import Camera
@ -37,7 +39,9 @@ class AugustCamera(AugustEntityMixin, Camera):
_attr_translation_key = "camera"
def __init__(self, data, device, session, timeout):
def __init__(
self, data: AugustData, device: Doorbell, session: ClientSession, timeout: int
) -> None:
"""Initialize an August security camera."""
super().__init__(data, device)
self._timeout = timeout

View file

@ -42,7 +42,7 @@ class AugustEntityMixin(Entity):
self._attr_device_info[ATTR_CONNECTIONS] = {(dr.CONNECTION_BLUETOOTH, mac)}
@property
def _device_id(self):
def _device_id(self) -> str:
return self._device.device_id
@property
@ -50,17 +50,17 @@ class AugustEntityMixin(Entity):
return self._data.get_device_detail(self._device.device_id)
@property
def _hyper_bridge(self):
def _hyper_bridge(self) -> bool:
"""Check if the lock has a paired hyper bridge."""
return bool(self._detail.bridge and self._detail.bridge.hyper_bridge)
@callback
def _update_from_data_and_write_state(self):
def _update_from_data_and_write_state(self) -> None:
self._update_from_data()
self.async_write_ha_state()
@abstractmethod
def _update_from_data(self):
def _update_from_data(self) -> None:
"""Update the entity state from the data object."""
async def async_added_to_hass(self):
@ -77,7 +77,7 @@ class AugustEntityMixin(Entity):
)
def _remove_device_types(name, device_types):
def _remove_device_types(name: str, device_types: list[str]) -> str:
"""Strip device types from a string.
August stores the name as Master Bed Lock

View file

@ -132,7 +132,7 @@ class AugustGateway:
return self.authentication
async def async_reset_authentication(self):
async def async_reset_authentication(self) -> None:
"""Remove the cache file."""
await self._hass.async_add_executor_job(self._reset_authentication)

View file

@ -4,7 +4,7 @@ from typing import Any
from aiohttp import ClientResponseError
from yalexs.activity import SOURCE_PUBNUB, ActivityType
from yalexs.lock import LockStatus
from yalexs.lock import Lock, LockStatus
from yalexs.util import get_latest_activity, update_lock_detail_from_activity
from homeassistant.components.lock import ATTR_CHANGED_BY, LockEntity
@ -39,7 +39,7 @@ class AugustLock(AugustEntityMixin, RestoreEntity, LockEntity):
_attr_name = None
def __init__(self, data, device):
def __init__(self, data: AugustData, device: Lock) -> None:
"""Initialize the lock."""
super().__init__(data, device)
self._lock_status = None
@ -82,7 +82,7 @@ class AugustLock(AugustEntityMixin, RestoreEntity, LockEntity):
)
self._data.async_signal_device_id_update(self._device_id)
def _update_lock_status_from_detail(self):
def _update_lock_status_from_detail(self) -> bool:
self._attr_available = self._detail.bridge_is_online
if self._lock_status != self._detail.lock_status:

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import Generic, TypeVar
from typing import Any, Generic, TypeVar
from yalexs.activity import ActivityType
from yalexs.doorbell import Doorbell
@ -179,7 +179,7 @@ class AugustOperatorSensor(AugustEntityMixin, RestoreSensor):
_attr_translation_key = "operator"
def __init__(self, data, device):
def __init__(self, data: AugustData, device) -> None:
"""Initialize the sensor."""
super().__init__(data, device)
self._data = data
@ -211,9 +211,9 @@ class AugustOperatorSensor(AugustEntityMixin, RestoreSensor):
self._attr_entity_picture = lock_activity.operator_thumbnail_url
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the device specific state attributes."""
attributes = {}
attributes: dict[str, Any] = {}
if self._operated_remote is not None:
attributes[ATTR_OPERATION_REMOTE] = self._operated_remote
@ -291,7 +291,7 @@ class AugustBatterySensor(AugustEntityMixin, SensorEntity, Generic[_T]):
self._update_from_data()
@callback
def _update_from_data(self):
def _update_from_data(self) -> None:
"""Get the latest state of the sensor."""
self._attr_native_value = self.entity_description.value_fn(self._detail)
self._attr_available = self._attr_native_value is not None

View file

@ -1,11 +1,11 @@
"""Base class for August entity."""
from __future__ import annotations
from abc import abstractmethod
from datetime import datetime, timedelta
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
from homeassistant.helpers.event import async_track_time_interval
@ -34,7 +34,7 @@ class AugustSubscriberMixin:
self._subscriptions.setdefault(device_id, []).append(update_callback)
def _unsubscribe():
def _unsubscribe() -> None:
self.async_unsubscribe_device_id(device_id, update_callback)
return _unsubscribe
@ -54,9 +54,10 @@ class AugustSubscriberMixin:
)
@callback
def _async_cancel_update_interval(_):
def _async_cancel_update_interval(_: Event) -> None:
self._stop_interval = None
self._unsub_interval()
if self._unsub_interval:
self._unsub_interval()
self._stop_interval = self._hass.bus.async_listen(
EVENT_HOMEASSISTANT_STOP, _async_cancel_update_interval