evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add missing type hints to homekit_controller (#65368)

Browse source
This commit is contained in:
Jc2k 2022-02-01 19:30:37 +00:00 • committed by GitHub
parent aef6f49eff
commit 9f5d77e0df
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 389 additions and 312 deletions
Download patch file Download diff file Expand all files Collapse all files

102
homeassistant/components/homekit_controller/connection.py
View file

@ -1,7 +1,11 @@
"""Helpers for managing a pairing with a HomeKit accessory or bridge."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
import datetime
import logging
from typing import Any
from aiohomekit.exceptions import (
AccessoryDisconnectedError,
@ -9,11 +13,11 @@ from aiohomekit.exceptions import (
EncryptionError,
)
from aiohomekit.model import Accessories, Accessory
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from aiohomekit.model.characteristics import Characteristic, CharacteristicsTypes
from aiohomekit.model.services import Service, ServicesTypes
from homeassistant.const import ATTR_VIA_DEVICE
from homeassistant.core import callback
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.event import async_track_time_interval
@ -37,6 +41,10 @@ MAX_POLL_FAILURES_TO_DECLARE_UNAVAILABLE = 3
_LOGGER = logging.getLogger(__name__)
AddAccessoryCb = Callable[[Accessory], bool]
AddServiceCb = Callable[[Service], bool]
AddCharacteristicCb = Callable[[Characteristic], bool]
def valid_serial_number(serial):
"""Return if the serial number appears to be valid."""
@ -51,7 +59,7 @@ def valid_serial_number(serial):
class HKDevice:
"""HomeKit device."""
def __init__(self, hass, config_entry, pairing_data):
def __init__(self, hass, config_entry, pairing_data) -> None:
"""Initialise a generic HomeKit device."""
self.hass = hass
@ -71,28 +79,28 @@ class HKDevice:
self.entity_map = Accessories()
# A list of callbacks that turn HK accessories into entities
self.accessory_factories = []
self.accessory_factories: list[AddAccessoryCb] = []
# A list of callbacks that turn HK service metadata into entities
self.listeners = []
self.listeners: list[AddServiceCb] = []
# A list of callbacks that turn HK characteristics into entities
self.char_factories = []
self.char_factories: list[AddCharacteristicCb] = []
# The platorms we have forwarded the config entry so far. If a new
# accessory is added to a bridge we may have to load additional
# platforms. We don't want to load all platforms up front if its just
# a lightbulb. And we don't want to forward a config entry twice
# (triggers a Config entry already set up error)
self.platforms = set()
self.platforms: set[str] = set()
# This just tracks aid/iid pairs so we know if a HK service has been
# mapped to a HA entity.
self.entities = []
self.entities: list[tuple[int, int | None, int | None]] = []
# A map of aid -> device_id
# Useful when routing events to triggers
self.devices = {}
self.devices: dict[int, str] = {}
self.available = False
@ -100,13 +108,13 @@ class HKDevice:
# Current values of all characteristics homekit_controller is tracking.
# Key is a (accessory_id, characteristic_id) tuple.
self.current_state = {}
self.current_state: dict[tuple[int, int], Any] = {}
self.pollable_characteristics = []
self.pollable_characteristics: list[tuple[int, int]] = []
# If this is set polling is active and can be disabled by calling
# this method.
self._polling_interval_remover = None
self._polling_interval_remover: CALLBACK_TYPE | None = None
# Never allow concurrent polling of the same accessory or bridge
self._polling_lock = asyncio.Lock()
@ -116,33 +124,37 @@ class HKDevice:
# This is set to True if we can't rely on serial numbers to be unique
self.unreliable_serial_numbers = False
self.watchable_characteristics = []
self.watchable_characteristics: list[tuple[int, int]] = []
self.pairing.dispatcher_connect(self.process_new_events)
def add_pollable_characteristics(self, characteristics):
def add_pollable_characteristics(
self, characteristics: list[tuple[int, int]]
) -> None:
"""Add (aid, iid) pairs that we need to poll."""
self.pollable_characteristics.extend(characteristics)
def remove_pollable_characteristics(self, accessory_id):
def remove_pollable_characteristics(self, accessory_id: int) -> None:
"""Remove all pollable characteristics by accessory id."""
self.pollable_characteristics = [
char for char in self.pollable_characteristics if char[0] != accessory_id
]
def add_watchable_characteristics(self, characteristics):
def add_watchable_characteristics(
self, characteristics: list[tuple[int, int]]
) -> None:
"""Add (aid, iid) pairs that we need to poll."""
self.watchable_characteristics.extend(characteristics)
self.hass.async_create_task(self.pairing.subscribe(characteristics))
def remove_watchable_characteristics(self, accessory_id):
def remove_watchable_characteristics(self, accessory_id: int) -> None:
"""Remove all pollable characteristics by accessory id."""
self.watchable_characteristics = [
char for char in self.watchable_characteristics if char[0] != accessory_id
]
@callback
def async_set_available_state(self, available):
def async_set_available_state(self, available: bool) -> None:
"""Mark state of all entities on this connection when it becomes available or unavailable."""
_LOGGER.debug(
"Called async_set_available_state with %s for %s", available, self.unique_id
@ -152,7 +164,7 @@ class HKDevice:
self.available = available
self.hass.helpers.dispatcher.async_dispatcher_send(self.signal_state_updated)
async def async_setup(self):
async def async_setup(self) -> bool:
"""Prepare to use a paired HomeKit device in Home Assistant."""
cache = self.hass.data[ENTITY_MAP].get_map(self.unique_id)
if not cache:
@ -214,7 +226,7 @@ class HKDevice:
return device_info
@callback
def async_migrate_devices(self):
def async_migrate_devices(self) -> None:
"""Migrate legacy device entries from 3-tuples to 2-tuples."""
_LOGGER.debug(
"Migrating device registry entries for pairing %s", self.unique_id
@ -246,7 +258,7 @@ class HKDevice:
(DOMAIN, IDENTIFIER_LEGACY_SERIAL_NUMBER, serial_number)
)
device = device_registry.async_get_device(identifiers=identifiers)
device = device_registry.async_get_device(identifiers=identifiers) # type: ignore[arg-type]
if not device:
continue
@ -286,7 +298,7 @@ class HKDevice:
)
@callback
def async_create_devices(self):
def async_create_devices(self) -> None:
"""
Build device registry entries for all accessories paired with the bridge.
@ -315,7 +327,7 @@ class HKDevice:
self.devices = devices
@callback
def async_detect_workarounds(self):
def async_detect_workarounds(self) -> None:
"""Detect any workarounds that are needed for this pairing."""
unreliable_serial_numbers = False
@ -354,7 +366,7 @@ class HKDevice:
self.unreliable_serial_numbers = unreliable_serial_numbers
async def async_process_entity_map(self):
async def async_process_entity_map(self) -> None:
"""
Process the entity map and load any platforms or entities that need adding.
@ -388,18 +400,18 @@ class HKDevice:
await self.async_update()
async def async_unload(self):
async def async_unload(self) -> None:
"""Stop interacting with device and prepare for removal from hass."""
if self._polling_interval_remover:
self._polling_interval_remover()
await self.pairing.close()
return await self.hass.config_entries.async_unload_platforms(
await self.hass.config_entries.async_unload_platforms(
self.config_entry, self.platforms
)
async def async_refresh_entity_map(self, config_num):
async def async_refresh_entity_map(self, config_num: int) -> bool:
"""Handle setup of a HomeKit accessory."""
try:
self.accessories = await self.pairing.list_accessories_and_characteristics()
@ -419,26 +431,26 @@ class HKDevice:
return True
def add_accessory_factory(self, add_entities_cb):
def add_accessory_factory(self, add_entities_cb) -> None:
"""Add a callback to run when discovering new entities for accessories."""
self.accessory_factories.append(add_entities_cb)
self._add_new_entities_for_accessory([add_entities_cb])
def _add_new_entities_for_accessory(self, handlers):
def _add_new_entities_for_accessory(self, handlers) -> None:
for accessory in self.entity_map.accessories:
for handler in handlers:
if (accessory.aid, None) in self.entities:
if (accessory.aid, None, None) in self.entities:
continue
if handler(accessory):
self.entities.append((accessory.aid, None))
self.entities.append((accessory.aid, None, None))
break
def add_char_factory(self, add_entities_cb):
def add_char_factory(self, add_entities_cb) -> None:
"""Add a callback to run when discovering new entities for accessories."""
self.char_factories.append(add_entities_cb)
self._add_new_entities_for_char([add_entities_cb])
def _add_new_entities_for_char(self, handlers):
def _add_new_entities_for_char(self, handlers) -> None:
for accessory in self.entity_map.accessories:
for service in accessory.services:
for char in service.characteristics:
@ -449,33 +461,33 @@ class HKDevice:
self.entities.append((accessory.aid, service.iid, char.iid))
break
def add_listener(self, add_entities_cb):
def add_listener(self, add_entities_cb) -> None:
"""Add a callback to run when discovering new entities for services."""
self.listeners.append(add_entities_cb)
self._add_new_entities([add_entities_cb])
def add_entities(self):
def add_entities(self) -> None:
"""Process the entity map and create HA entities."""
self._add_new_entities(self.listeners)
self._add_new_entities_for_accessory(self.accessory_factories)
self._add_new_entities_for_char(self.char_factories)
def _add_new_entities(self, callbacks):
def _add_new_entities(self, callbacks) -> None:
for accessory in self.entity_map.accessories:
aid = accessory.aid
for service in accessory.services:
iid = service.iid
if (aid, iid) in self.entities:
if (aid, None, iid) in self.entities:
# Don't add the same entity again
continue
for listener in callbacks:
if listener(service):
self.entities.append((aid, iid))
self.entities.append((aid, None, iid))
break
async def async_load_platform(self, platform):
async def async_load_platform(self, platform: str) -> None:
"""Load a single platform idempotently."""
if platform in self.platforms:
return
@ -489,7 +501,7 @@ class HKDevice:
self.platforms.remove(platform)
raise
async def async_load_platforms(self):
async def async_load_platforms(self) -> None:
"""Load any platforms needed by this HomeKit device."""
tasks = []
for accessory in self.entity_map.accessories:
@ -558,7 +570,7 @@ class HKDevice:
_LOGGER.debug("Finished HomeKit controller update: %s", self.unique_id)
def process_new_events(self, new_values_dict):
def process_new_events(self, new_values_dict) -> None:
"""Process events from accessory into HA state."""
self.async_set_available_state(True)
@ -575,11 +587,11 @@ class HKDevice:
self.hass.helpers.dispatcher.async_dispatcher_send(self.signal_state_updated)
async def get_characteristics(self, *args, **kwargs):
async def get_characteristics(self, *args, **kwargs) -> dict[str, Any]:
"""Read latest state from homekit accessory."""
return await self.pairing.get_characteristics(*args, **kwargs)
async def put_characteristics(self, characteristics):
async def put_characteristics(self, characteristics) -> None:
"""Control a HomeKit device state from Home Assistant."""
results = await self.pairing.put_characteristics(characteristics)
@ -604,7 +616,7 @@ class HKDevice:
self.process_new_events(new_entity_state)
@property
def unique_id(self):
def unique_id(self) -> str:
"""
Return a unique id for this accessory or bridge.