jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Strict typing for homekit part 1 (#67657)

Browse Source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
J. Nick Koston 2022-03-29 23:21:07 -10:00 committed by GitHub
parent 496d90bf00
commit af6a62ca79
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 443 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.strict-typing
View File

@ -99,6 +99,14 @@ homeassistant.components.group.*
homeassistant.components.guardian.*
homeassistant.components.history.*
homeassistant.components.homeassistant.triggers.event
homeassistant.components.homekit
homeassistant.components.homekit.accessories
homeassistant.components.homekit.aidmanager
homeassistant.components.homekit.config_flow
homeassistant.components.homekit.diagnostics
homeassistant.components.homekit.logbook
homeassistant.components.homekit.type_triggers
homeassistant.components.homekit.util
homeassistant.components.homekit_controller
homeassistant.components.homekit_controller.alarm_control_panel
homeassistant.components.homekit_controller.button

184
homeassistant/components/homekit/__init__.py
View File

@ -2,14 +2,18 @@
from __future__ import annotations
import asyncio
from collections.abc import Iterable
from copy import deepcopy
import ipaddress
import logging
import os
from typing import Any, cast
from uuid import UUID
from aiohttp import web
from pyhap.const import STANDALONE_AID
import voluptuous as vol
from zeroconf.asyncio import AsyncZeroconf
from homeassistant.components import device_automation, network, zeroconf
from homeassistant.components.binary_sensor import (
@ -39,7 +43,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
SERVICE_RELOAD,
)
from homeassistant.core import CoreState, HomeAssistant, ServiceCall, callback
from homeassistant.core import CoreState, HomeAssistant, ServiceCall, State, callback
from homeassistant.exceptions import HomeAssistantError, Unauthorized
from homeassistant.helpers import device_registry, entity_registry
import homeassistant.helpers.config_validation as cv
@ -67,7 +71,7 @@ from . import ( # noqa: F401
type_switches,
type_thermostats,
)
from .accessories import HomeBridge, HomeDriver, get_accessory
from .accessories import HomeAccessory, HomeBridge, HomeDriver, get_accessory
from .aidmanager import AccessoryAidStorage
from .const import (
ATTR_INTEGRATION,
@ -114,7 +118,7 @@ from .util import (
_LOGGER = logging.getLogger(__name__)
MAX_DEVICES = 150
MAX_DEVICES = 150 # includes the bridge
# #### Driver Status ####
STATUS_READY = 0
@ -129,7 +133,9 @@ _HOMEKIT_CONFIG_UPDATE_TIME = (
)
def _has_all_unique_names_and_ports(bridges):
def _has_all_unique_names_and_ports(
bridges: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""Validate that each homekit bridge configured has a unique name."""
names = [bridge[CONF_NAME] for bridge in bridges]
ports = [bridge[CONF_PORT] for bridge in bridges]
@ -184,7 +190,9 @@ def _async_all_homekit_instances(hass: HomeAssistant) -> list[HomeKit]:
]
def _async_get_entries_by_name(current_entries):
def _async_get_entries_by_name(
current_entries: list[ConfigEntry],
) -> dict[str, ConfigEntry]:
"""Return a dict of the entries by name."""
# For backwards compat, its possible the first bridge is using the default
@ -221,7 +229,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@callback
def _async_update_config_entry_if_from_yaml(hass, entries_by_name, conf):
def _async_update_config_entry_if_from_yaml(
hass: HomeAssistant, entries_by_name: dict[str, ConfigEntry], conf: ConfigType
) -> bool:
"""Update a config entry with the latest yaml.
Returns True if a matching config entry was found
@ -346,13 +356,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Remove a config entry."""
return await hass.async_add_executor_job(
await hass.async_add_executor_job(
remove_state_files_for_entry_id, hass, entry.entry_id
)
@callback
def _async_import_options_from_data_if_missing(hass: HomeAssistant, entry: ConfigEntry):
def _async_import_options_from_data_if_missing(
hass: HomeAssistant, entry: ConfigEntry
) -> None:
options = deepcopy(dict(entry.options))
data = deepcopy(dict(entry.data))
modified = False
@ -367,7 +379,7 @@ def _async_import_options_from_data_if_missing(hass: HomeAssistant, entry: Confi
@callback
def _async_register_events_and_services(hass: HomeAssistant):
def _async_register_events_and_services(hass: HomeAssistant) -> None:
"""Register events and services for HomeKit."""
hass.http.register_view(HomeKitPairingQRView)
@ -381,7 +393,7 @@ def _async_register_events_and_services(hass: HomeAssistant):
)
continue
entity_ids = service.data.get("entity_id")
entity_ids = cast(list[str], service.data.get("entity_id"))
await homekit.async_reset_accessories(entity_ids)
hass.services.async_register(
@ -453,27 +465,29 @@ def _async_register_events_and_services(hass: HomeAssistant):
class HomeKit:
"""Class to handle all actions between HomeKit and Home Assistant."""
driver: HomeDriver
def __init__(
self,
hass,
name,
port,
ip_address,
entity_filter,
exclude_accessory_mode,
entity_config,
homekit_mode,
advertise_ip=None,
entry_id=None,
entry_title=None,
devices=None,
):
hass: HomeAssistant,
name: str,
port: int,
ip_address: str | None,
entity_filter: EntityFilter,
exclude_accessory_mode: bool,
entity_config: dict,
homekit_mode: str,
advertise_ip: str | None,
entry_id: str,
entry_title: str,
devices: Iterable[str] | None = None,
) -> None:
"""Initialize a HomeKit object."""
self.hass = hass
self._name = name
self._port = port
self._ip_address = ip_address
self._filter: EntityFilter = entity_filter
self._filter = entity_filter
self._config = entity_config
self._exclude_accessory_mode = exclude_accessory_mode
self._advertise_ip = advertise_ip
@ -481,13 +495,12 @@ class HomeKit:
self._entry_title = entry_title
self._homekit_mode = homekit_mode
self._devices = devices or []
self.aid_storage = None
self.aid_storage: AccessoryAidStorage | None = None
self.status = STATUS_READY
self.bridge = None
self.driver = None
self.bridge: HomeBridge | None = None
def setup(self, async_zeroconf_instance, uuid):
def setup(self, async_zeroconf_instance: AsyncZeroconf, uuid: UUID) -> None:
"""Set up bridge and accessory driver."""
persist_file = get_persist_fullpath_for_entry_id(self.hass, self._entry_id)
@ -510,22 +523,24 @@ class HomeKit:
if os.path.exists(persist_file):
self.driver.load()
async def async_reset_accessories(self, entity_ids):
async def async_reset_accessories(self, entity_ids: Iterable[str]) -> None:
"""Reset the accessory to load the latest configuration."""
if not self.bridge:
await self.async_reset_accessories_in_accessory_mode(entity_ids)
return
await self.async_reset_accessories_in_bridge_mode(entity_ids)
async def async_reset_accessories_in_accessory_mode(self, entity_ids):
async def async_reset_accessories_in_accessory_mode(
self, entity_ids: Iterable[str]
) -> None:
"""Reset accessories in accessory mode."""
acc = self.driver.accessory
acc = cast(HomeAccessory, self.driver.accessory)
if acc.entity_id not in entity_ids:
return
await acc.stop()
if not (state := self.hass.states.get(acc.entity_id)):
_LOGGER.warning(
"The underlying entity %s disappeared during reset", acc.entity
"The underlying entity %s disappeared during reset", acc.entity_id
)
return
if new_acc := self._async_create_single_accessory([state]):
@ -533,9 +548,14 @@ class HomeKit:
self.hass.async_add_job(new_acc.run)
await self.async_config_changed()
async def async_reset_accessories_in_bridge_mode(self, entity_ids):
async def async_reset_accessories_in_bridge_mode(
self, entity_ids: Iterable[str]
) -> None:
"""Reset accessories in bridge mode."""
assert self.aid_storage is not None
assert self.bridge is not None
new = []
acc: HomeAccessory | None
for entity_id in entity_ids:
aid = self.aid_storage.get_or_allocate_aid_for_entity_id(entity_id)
if aid not in self.bridge.accessories:
@ -545,12 +565,13 @@ class HomeKit:
self._name,
entity_id,
)
acc = await self.async_remove_bridge_accessory(aid)
if state := self.hass.states.get(acc.entity_id):
if (acc := await self.async_remove_bridge_accessory(aid)) and (
state := self.hass.states.get(acc.entity_id)
):
new.append(state)
else:
_LOGGER.warning(
"The underlying entity %s disappeared during reset", acc.entity
"The underlying entity %s disappeared during reset", entity_id
)
if not new:
@ -560,23 +581,22 @@ class HomeKit:
await self.async_config_changed()
await asyncio.sleep(_HOMEKIT_CONFIG_UPDATE_TIME)
for state in new:
acc = self.add_bridge_accessory(state)
if acc:
if acc := self.add_bridge_accessory(state):
self.hass.async_add_job(acc.run)
await self.async_config_changed()
async def async_config_changed(self):
async def async_config_changed(self) -> None:
"""Call config changed which writes out the new config to disk."""
await self.hass.async_add_executor_job(self.driver.config_changed)
def add_bridge_accessory(self, state):
def add_bridge_accessory(self, state: State) -> HomeAccessory | None:
"""Try adding accessory to bridge if configured beforehand."""
if self._would_exceed_max_devices(state.entity_id):
return
return None
if state_needs_accessory_mode(state):
if self._exclude_accessory_mode:
return
return None
_LOGGER.warning(
"The bridge %s has entity %s. For best performance, "
"and to prevent unexpected unavailability, create and "
@ -586,6 +606,8 @@ class HomeKit:
state.entity_id,
)
assert self.aid_storage is not None
assert self.bridge is not None
aid = self.aid_storage.get_or_allocate_aid_for_entity_id(state.entity_id)
conf = self._config.get(state.entity_id, {}).copy()
# If an accessory cannot be created or added due to an exception
@ -602,9 +624,10 @@ class HomeKit:
)
return None
def _would_exceed_max_devices(self, name):
def _would_exceed_max_devices(self, name: str | None) -> bool:
"""Check if adding another devices would reach the limit and log."""
# The bridge itself counts as an accessory
assert self.bridge is not None
if len(self.bridge.accessories) + 1 >= MAX_DEVICES:
_LOGGER.warning(
"Cannot add %s as this would exceed the %d device limit. Consider using the filter option",
@ -614,16 +637,20 @@ class HomeKit:
return True
return False
def add_bridge_triggers_accessory(self, device, device_triggers):
def add_bridge_triggers_accessory(
self, device: device_registry.DeviceEntry, device_triggers: list[dict[str, Any]]
) -> None:
"""Add device automation triggers to the bridge."""
if self._would_exceed_max_devices(device.name):
return
assert self.aid_storage is not None
assert self.bridge is not None
aid = self.aid_storage.get_or_allocate_aid(device.id, device.id)
# If an accessory cannot be created or added due to an exception
# of any kind (usually in pyhap) it should not prevent
# the rest of the accessories from being created
config = {}
config: dict[str, Any] = {}
self._fill_config_from_device_registry_entry(device, config)
self.bridge.add_accessory(
DeviceTriggerAccessory(
@ -638,13 +665,15 @@ class HomeKit:
)
)
async def async_remove_bridge_accessory(self, aid):
async def async_remove_bridge_accessory(self, aid: int) -> HomeAccessory | None:
"""Try adding accessory to bridge if configured beforehand."""
assert self.bridge is not None
if acc := self.bridge.accessories.pop(aid, None):
await acc.stop()
return acc
return cast(HomeAccessory, acc)
return None
async def async_configure_accessories(self):
async def async_configure_accessories(self) -> list[State]:
"""Configure accessories for the included states."""
dev_reg = device_registry.async_get(self.hass)
ent_reg = entity_registry.async_get(self.hass)
@ -680,7 +709,7 @@ class HomeKit:
return entity_states
async def async_start(self, *args):
async def async_start(self, *args: Any) -> None:
"""Load storage and start."""
if self.status != STATUS_READY:
return
@ -704,7 +733,7 @@ class HomeKit:
self._async_show_setup_message()
@callback
def _async_show_setup_message(self):
def _async_show_setup_message(self) -> None:
"""Show the pairing setup message."""
async_show_setup_message(
self.hass,
@ -715,7 +744,7 @@ class HomeKit:
)
@callback
def async_unpair(self):
def async_unpair(self) -> None:
"""Remove all pairings for an accessory so it can be repaired."""
state = self.driver.state
for client_uuid in list(state.paired_clients):
@ -730,8 +759,9 @@ class HomeKit:
self._async_show_setup_message()
@callback
def _async_register_bridge(self):
def _async_register_bridge(self) -> None:
"""Register the bridge as a device so homekit_controller and exclude it from discovery."""
assert self.driver is not None
dev_reg = device_registry.async_get(self.hass)
formatted_mac = device_registry.format_mac(self.driver.state.mac)
# Connections and identifiers are both used here.
@ -753,7 +783,9 @@ class HomeKit:
hk_mode_name = "Accessory" if is_accessory_mode else "Bridge"
dev_reg.async_get_or_create(
config_entry_id=self._entry_id,
identifiers={identifier},
identifiers={
identifier # type: ignore[arg-type]
}, # this needs to be migrated as a 2 item tuple at some point
connections={connection},
manufacturer=MANUFACTURER,
name=accessory_friendly_name(self._entry_title, self.driver.accessory),
@ -762,12 +794,17 @@ class HomeKit:
)
@callback
def _async_purge_old_bridges(self, dev_reg, identifier, connection):
def _async_purge_old_bridges(
self,
dev_reg: device_registry.DeviceRegistry,
identifier: tuple[str, str, str],
connection: tuple[str, str],
) -> None:
"""Purge bridges that exist from failed pairing or manual resets."""
devices_to_purge = []
for entry in dev_reg.devices.values():
if self._entry_id in entry.config_entries and (
identifier not in entry.identifiers
identifier not in entry.identifiers # type: ignore[comparison-overlap]
or connection not in entry.connections
):
devices_to_purge.append(entry.id)
@ -776,7 +813,9 @@ class HomeKit:
dev_reg.async_remove_device(device_id)
@callback
def _async_create_single_accessory(self, entity_states):
def _async_create_single_accessory(
self, entity_states: list[State]
) -> HomeAccessory | None:
"""Create a single HomeKit accessory (accessory mode)."""
if not entity_states:
_LOGGER.error(
@ -796,7 +835,9 @@ class HomeKit:
)
return acc
async def _async_create_bridge_accessory(self, entity_states):
async def _async_create_bridge_accessory(
self, entity_states: Iterable[State]
) -> HomeAccessory:
"""Create a HomeKit bridge with accessories. (bridge mode)."""
self.bridge = HomeBridge(self.hass, self.driver, self._name)
for state in entity_states:
@ -820,12 +861,11 @@ class HomeKit:
valid_device_ids,
)
).items():
self.add_bridge_triggers_accessory(
dev_reg.async_get(device_id), device_triggers
)
if device := dev_reg.async_get(device_id):
self.add_bridge_triggers_accessory(device, device_triggers)
return self.bridge
async def _async_create_accessories(self):
async def _async_create_accessories(self) -> bool:
"""Create the accessories."""
entity_states = await self.async_configure_accessories()
if self._homekit_mode == HOMEKIT_MODE_ACCESSORY:
@ -839,7 +879,7 @@ class HomeKit:
self.driver.accessory = acc
return True
async def async_stop(self, *args):
async def async_stop(self, *args: Any) -> None:
"""Stop the accessory driver."""
if self.status != STATUS_RUNNING:
return
@ -848,7 +888,12 @@ class HomeKit:
await self.driver.async_stop()
@callback
def _async_configure_linked_sensors(self, ent_reg_ent, device_lookup, state):
def _async_configure_linked_sensors(
self,
ent_reg_ent: entity_registry.RegistryEntry,
device_lookup: dict[str, dict[tuple[str, str | None], str]],
state: State,
) -> None:
if (
ent_reg_ent is None
or ent_reg_ent.device_id is None
@ -905,7 +950,12 @@ class HomeKit:
current_humidity_sensor_entity_id,
)
async def _async_set_device_info_attributes(self, ent_reg_ent, dev_reg, entity_id):
async def _async_set_device_info_attributes(
self,
ent_reg_ent: entity_registry.RegistryEntry,
dev_reg: device_registry.DeviceRegistry,
entity_id: str,
) -> None:
"""Set attributes that will be used for homekit device info."""
ent_cfg = self._config.setdefault(entity_id, {})
if ent_reg_ent.device_id:
@ -920,7 +970,9 @@ class HomeKit:
except IntegrationNotFound:
ent_cfg[ATTR_INTEGRATION] = ent_reg_ent.platform
def _fill_config_from_device_registry_entry(self, device_entry, config):
def _fill_config_from_device_registry_entry(
self, device_entry: device_registry.DeviceEntry, config: dict[str, Any]
) -> None:
"""Populate a config dict from the registry."""
if device_entry.manufacturer:
config[ATTR_MANUFACTURER] = device_entry.manufacturer
@ -943,7 +995,7 @@ class HomeKitPairingQRView(HomeAssistantView):
name = "api:homekit:pairingqr"
requires_auth = False
async def get(self, request):
async def get(self, request: web.Request) -> web.Response:
"""Retrieve the pairing QRCode image."""
# pylint: disable=no-self-use
if not request.query_string:

121
homeassistant/components/homekit/accessories.py
View File

@ -2,6 +2,8 @@
from __future__ import annotations
import logging
from typing import Any, cast
from uuid import UUID
from pyhap.accessory import Accessory, Bridge
from pyhap.accessory_driver import AccessoryDriver
@ -34,7 +36,15 @@ from homeassistant.const import (
TEMP_FAHRENHEIT,
__version__,
)
from homeassistant.core import Context, callback as ha_callback, split_entity_id
from homeassistant.core import (
CALLBACK_TYPE,
Context,
Event,
HomeAssistant,
State,
callback as ha_callback,
split_entity_id,
)
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.util.decorator import Registry
@ -95,7 +105,9 @@ SWITCH_TYPES = {
TYPES: Registry[str, type[HomeAccessory]] = Registry()
def get_accessory(hass, driver, state, aid, config): # noqa: C901
def get_accessory( # noqa: C901
hass: HomeAssistant, driver: HomeDriver, state: State, aid: int | None, config: dict
) -> HomeAccessory | None:
"""Take state and return an accessory object if supported."""
if not aid:
_LOGGER.warning(
@ -232,22 +244,22 @@ def get_accessory(hass, driver, state, aid, config): # noqa: C901
return TYPES[a_type](hass, driver, name, state.entity_id, aid, config)
class HomeAccessory(Accessory):
class HomeAccessory(Accessory): # type: ignore[misc]
"""Adapter class for Accessory."""
def __init__(
self,
hass,
driver,
name,
entity_id,
aid,
config,
*args,
category=CATEGORY_OTHER,
device_id=None,
**kwargs,
):
hass: HomeAssistant,
driver: HomeDriver,
name: str,
entity_id: str,
aid: int,
config: dict,
*args: Any,
category: str = CATEGORY_OTHER,
device_id: str | None = None,
**kwargs: Any,
) -> None:
"""Initialize a Accessory object."""
super().__init__(
driver=driver,
@ -258,7 +270,7 @@ class HomeAccessory(Accessory):
)
self.config = config or {}
if device_id:
self.device_id = device_id
self.device_id: str | None = device_id
serial_number = device_id
domain = None
else:
@ -285,6 +297,7 @@ class HomeAccessory(Accessory):
sw_version = format_version(self.config[ATTR_SW_VERSION])
if sw_version is None:
sw_version = format_version(__version__)
assert sw_version is not None
hw_version = None
if self.config.get(ATTR_HW_VERSION) is not None:
hw_version = format_version(self.config[ATTR_HW_VERSION])
@ -308,7 +321,7 @@ class HomeAccessory(Accessory):
self.category = category
self.entity_id = entity_id
self.hass = hass
self._subscriptions = []
self._subscriptions: list[CALLBACK_TYPE] = []
if device_id:
return
@ -325,7 +338,9 @@ class HomeAccessory(Accessory):
)
"""Add battery service if available"""
entity_attributes = self.hass.states.get(self.entity_id).attributes
state = self.hass.states.get(self.entity_id)
assert state is not None
entity_attributes = state.attributes
battery_found = entity_attributes.get(ATTR_BATTERY_LEVEL)
if self.linked_battery_sensor:
@ -367,15 +382,15 @@ class HomeAccessory(Accessory):
)
@property
def available(self):
def available(self) -> bool:
"""Return if accessory is available."""
state = self.hass.states.get(self.entity_id)
return state is not None and state.state != STATE_UNAVAILABLE
async def run(self):
async def run(self) -> None:
"""Handle accessory driver started event."""
state = self.hass.states.get(self.entity_id)
self.async_update_state_callback(state)
if state := self.hass.states.get(self.entity_id):
self.async_update_state_callback(state)
self._subscriptions.append(
async_track_state_change_event(
self.hass, [self.entity_id], self.async_update_event_state_callback
@ -384,10 +399,11 @@ class HomeAccessory(Accessory):
battery_charging_state = None
battery_state = None
if self.linked_battery_sensor:
linked_battery_sensor_state = self.hass.states.get(
if self.linked_battery_sensor and (
linked_battery_sensor_state := self.hass.states.get(
self.linked_battery_sensor
)
):
battery_state = linked_battery_sensor_state.state
battery_charging_state = linked_battery_sensor_state.attributes.get(
ATTR_BATTERY_CHARGING
@ -418,12 +434,12 @@ class HomeAccessory(Accessory):
self.async_update_battery(battery_state, battery_charging_state)
@ha_callback
def async_update_event_state_callback(self, event):
def async_update_event_state_callback(self, event: Event) -> None:
"""Handle state change event listener callback."""
self.async_update_state_callback(event.data.get("new_state"))
@ha_callback
def async_update_state_callback(self, new_state):
def async_update_state_callback(self, new_state: State | None) -> None:
"""Handle state change listener callback."""
_LOGGER.debug("New_state: %s", new_state)
if new_state is None:
@ -445,7 +461,7 @@ class HomeAccessory(Accessory):
self.async_update_state(new_state)
@ha_callback
def async_update_linked_battery_callback(self, event):
def async_update_linked_battery_callback(self, event: Event) -> None:
"""Handle linked battery sensor state change listener callback."""
if (new_state := event.data.get("new_state")) is None:
return
@ -456,19 +472,19 @@ class HomeAccessory(Accessory):
self.async_update_battery(new_state.state, battery_charging_state)
@ha_callback
def async_update_linked_battery_charging_callback(self, event):
def async_update_linked_battery_charging_callback(self, event: Event) -> None:
"""Handle linked battery charging sensor state change listener callback."""
if (new_state := event.data.get("new_state")) is None:
return
self.async_update_battery(None, new_state.state == STATE_ON)
@ha_callback
def async_update_battery(self, battery_level, battery_charging):
def async_update_battery(self, battery_level: Any, battery_charging: Any) -> None:
"""Update battery service if available.
Only call this function if self._support_battery_level is True.
"""
if not self._char_battery:
if not self._char_battery or not self._char_low_battery:
# Battery appeared after homekit was started
return
@ -495,7 +511,7 @@ class HomeAccessory(Accessory):
)
@ha_callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Handle state change to update HomeKit value.
Overridden by accessory types.
@ -503,7 +519,13 @@ class HomeAccessory(Accessory):
raise NotImplementedError()
@ha_callback
def async_call_service(self, domain, service, service_data, value=None):
def async_call_service(
self,
domain: str,
service: str,
service_data: dict[str, Any] | None,
value: Any | None = None,
) -> None:
"""Fire event and call service for changes from HomeKit."""
event_data = {
ATTR_ENTITY_ID: self.entity_id,
@ -521,7 +543,7 @@ class HomeAccessory(Accessory):
)
@ha_callback
def async_reset(self):
def async_reset(self) -> None:
"""Reset and recreate an accessory."""
self.hass.async_create_task(
self.hass.services.async_call(
@ -531,16 +553,16 @@ class HomeAccessory(Accessory):
)
)
async def stop(self):
async def stop(self) -> None:
"""Cancel any subscriptions when the bridge is stopped."""
while self._subscriptions:
self._subscriptions.pop(0)()
class HomeBridge(Bridge):
class HomeBridge(Bridge): # type: ignore[misc]
"""Adapter class for Bridge."""
def __init__(self, hass, driver, name):
def __init__(self, hass: HomeAssistant, driver: HomeDriver, name: str) -> None:
"""Initialize a Bridge object."""
super().__init__(driver, name)
self.set_info_service(
@ -551,10 +573,10 @@ class HomeBridge(Bridge):
)
self.hass = hass
def setup_message(self):
def setup_message(self) -> None:
"""Prevent print of pyhap setup message to terminal."""
async def async_get_snapshot(self, info):
async def async_get_snapshot(self, info: dict) -> bytes:
"""Get snapshot from accessory if supported."""
if (acc := self.accessories.get(info["aid"])) is None:
raise ValueError("Requested snapshot for missing accessory")
@ -563,13 +585,20 @@ class HomeBridge(Bridge):
"Got a request for snapshot, but the Accessory "
'does not define a "async_get_snapshot" method'
)
return await acc.async_get_snapshot(info)
return cast(bytes, await acc.async_get_snapshot(info))
class HomeDriver(AccessoryDriver):
class HomeDriver(AccessoryDriver): # type: ignore[misc]
"""Adapter class for AccessoryDriver."""
def __init__(self, hass, entry_id, bridge_name, entry_title, **kwargs):
def __init__(
self,
hass: HomeAssistant,
entry_id: str,
bridge_name: str,
entry_title: str,
**kwargs: Any,
) -> None:
"""Initialize a AccessoryDriver object."""
super().__init__(**kwargs)
self.hass = hass
@ -577,16 +606,18 @@ class HomeDriver(AccessoryDriver):
self._bridge_name = bridge_name
self._entry_title = entry_title
@pyhap_callback
def pair(self, client_uuid, client_public, client_permissions):
@pyhap_callback # type: ignore[misc]
def pair(
self, client_uuid: UUID, client_public: str, client_permissions: int
) -> bool:
"""Override super function to dismiss setup message if paired."""
success = super().pair(client_uuid, client_public, client_permissions)
if success:
async_dismiss_setup_message(self.hass, self._entry_id)
return success
return cast(bool, success)
@pyhap_callback
def unpair(self, client_uuid):
@pyhap_callback # type: ignore[misc]
def unpair(self, client_uuid: UUID) -> None:
"""Override super function to show setup message if unpaired."""
super().unpair(client_uuid)

42
homeassistant/components/homekit/aidmanager.py
View File

@ -9,13 +9,15 @@ can't change the hash without causing breakages for HA users.
This module generates and stores them in a HA storage.
"""
from __future__ import annotations
from collections.abc import Generator
import random
from fnvhash import fnv1a_32
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_registry import RegistryEntry
from homeassistant.helpers.entity_registry import EntityRegistry, RegistryEntry
from homeassistant.helpers.storage import Store
from .util import get_aid_storage_filename_for_entry_id
@ -32,12 +34,12 @@ AID_MIN = 2
AID_MAX = 18446744073709551615
def get_system_unique_id(entity: RegistryEntry):
def get_system_unique_id(entity: RegistryEntry) -> str:
"""Determine the system wide unique_id for an entity."""
return f"{entity.platform}.{entity.domain}.{entity.unique_id}"
def _generate_aids(unique_id: str, entity_id: str) -> int:
def _generate_aids(unique_id: str | None, entity_id: str) -> Generator[int, None, None]:
"""Generate accessory aid."""
if unique_id:
@ -65,39 +67,41 @@ class AccessoryAidStorage:
persist over reboots.
"""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
def __init__(self, hass: HomeAssistant, entry_id: str) -> None:
"""Create a new entity map store."""
self.hass = hass
self.allocations = {}
self.allocated_aids = set()
self._entry = entry
self.store = None
self._entity_registry = None
self.allocations: dict[str, int] = {}
self.allocated_aids: set[int] = set()
self._entry_id = entry_id
self.store: Store | None = None
self._entity_registry: EntityRegistry | None = None
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Load the latest AID data."""
self._entity_registry = (
await self.hass.helpers.entity_registry.async_get_registry()
)
aidstore = get_aid_storage_filename_for_entry_id(self._entry)
aidstore = get_aid_storage_filename_for_entry_id(self._entry_id)
self.store = Store(self.hass, AID_MANAGER_STORAGE_VERSION, aidstore)
if not (raw_storage := await self.store.async_load()):
# There is no data about aid allocations yet
return
assert isinstance(raw_storage, dict)
self.allocations = raw_storage.get(ALLOCATIONS_KEY, {})
self.allocated_aids = set(self.allocations.values())
def get_or_allocate_aid_for_entity_id(self, entity_id: str):
def get_or_allocate_aid_for_entity_id(self, entity_id: str) -> int:
"""Generate a stable aid for an entity id."""
assert self._entity_registry is not None
if not (entity := self._entity_registry.async_get(entity_id)):
return self.get_or_allocate_aid(None, entity_id)
sys_unique_id = get_system_unique_id(entity)
return self.get_or_allocate_aid(sys_unique_id, entity_id)
def get_or_allocate_aid(self, unique_id: str, entity_id: str):
def get_or_allocate_aid(self, unique_id: str | None, entity_id: str) -> int:
"""Allocate (and return) a new aid for an accessory."""
if unique_id and unique_id in self.allocations:
return self.allocations[unique_id]
@ -119,7 +123,7 @@ class AccessoryAidStorage:
f"Unable to generate unique aid allocation for {entity_id} [{unique_id}]"
)
def delete_aid(self, storage_key: str):
def delete_aid(self, storage_key: str) -> None:
"""Delete an aid allocation."""
if storage_key not in self.allocations:
return
@ -129,15 +133,17 @@ class AccessoryAidStorage:
self.async_schedule_save()
@callback
def async_schedule_save(self):
def async_schedule_save(self) -> None:
"""Schedule saving the entity map cache."""
assert self.store is not None
self.store.async_delay_save(self._data_to_save, AID_MANAGER_SAVE_DELAY)
async def async_save(self):
async def async_save(self) -> None:
"""Save the entity map cache."""
assert self.store is not None
return await self.store.async_save(self._data_to_save())
@callback
def _data_to_save(self):
def _data_to_save(self) -> dict:
"""Return data of entity map to store in a file."""
return {ALLOCATIONS_KEY: self.allocations}

106
homeassistant/components/homekit/config_flow.py
View File

@ -2,11 +2,12 @@
from __future__ import annotations
import asyncio
from collections.abc import Iterable
from copy import deepcopy
import random
import re
import string
from typing import Any, Final
from typing import Any
import voluptuous as vol
@ -27,6 +28,7 @@ from homeassistant.const import (
CONF_PORT,
)
from homeassistant.core import HomeAssistant, callback, split_entity_id
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import device_registry, entity_registry
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entityfilter import (
@ -119,7 +121,7 @@ DEFAULT_DOMAINS = [
"water_heater",
]
_EMPTY_ENTITY_FILTER: Final = {
_EMPTY_ENTITY_FILTER: dict[str, list[str]] = {
CONF_INCLUDE_DOMAINS: [],
CONF_EXCLUDE_DOMAINS: [],
CONF_INCLUDE_ENTITIES: [],
@ -151,9 +153,9 @@ def _async_build_entites_filter(
return entity_filter
def _async_cameras_from_entities(entities: list[str]) -> set[str]:
def _async_cameras_from_entities(entities: list[str]) -> dict[str, str]:
return {
entity_id
entity_id: entity_id
for entity_id in entities
if entity_id.startswith(CAMERA_ENTITY_PREFIX)
}
@ -181,9 +183,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize config flow."""
self.hk_data = {}
self.hk_data: dict[str, Any] = {}
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose specific domains in bridge mode."""
if user_input is not None:
entity_filter = deepcopy(_EMPTY_ENTITY_FILTER)
@ -205,7 +209,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
),
)
async def async_step_pairing(self, user_input=None):
async def async_step_pairing(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Pairing instructions."""
if user_input is not None:
port = async_find_next_available_port(self.hass, DEFAULT_CONFIG_FLOW_PORT)
@ -227,7 +233,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
description_placeholders={CONF_NAME: self.hk_data[CONF_NAME]},
)
async def _async_add_entries_for_accessory_mode_entities(self, last_assigned_port):
async def _async_add_entries_for_accessory_mode_entities(
self, last_assigned_port: int
) -> None:
"""Generate new flows for entities that need their own instances."""
accessory_mode_entity_ids = _async_get_entity_ids_for_accessory_mode(
self.hass, self.hk_data[CONF_FILTER][CONF_INCLUDE_DOMAINS]
@ -249,12 +257,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
)
async def async_step_accessory(self, accessory_input):
async def async_step_accessory(self, accessory_input: dict) -> FlowResult:
"""Handle creation a single accessory in accessory mode."""
entity_id = accessory_input[CONF_ENTITY_ID]
port = accessory_input[CONF_PORT]
state = self.hass.states.get(entity_id)
assert state is not None
name = state.attributes.get(ATTR_FRIENDLY_NAME) or state.entity_id
entity_filter = _EMPTY_ENTITY_FILTER.copy()
entity_filter[CONF_INCLUDE_ENTITIES] = [entity_id]
@ -274,7 +283,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
title=f"{name}:{entry_data[CONF_PORT]}", data=entry_data
)
async def async_step_import(self, user_input=None):
async def async_step_import(self, user_input: dict) -> FlowResult:
"""Handle import from yaml."""
if not self._async_is_unique_name_port(user_input):
return self.async_abort(reason="port_name_in_use")
@ -283,7 +292,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
@callback
def _async_current_names(self):
def _async_current_names(self) -> set[str]:
"""Return a set of bridge names."""
return {
entry.data[CONF_NAME]
@ -292,7 +301,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
}
@callback
def _async_available_name(self, requested_name):
def _async_available_name(self, requested_name: str) -> str:
"""Return an available for the bridge."""
current_names = self._async_current_names()
valid_mdns_name = re.sub("[^A-Za-z0-9 ]+", " ", requested_name)
@ -301,7 +310,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return valid_mdns_name
acceptable_mdns_chars = string.ascii_uppercase + string.digits
suggested_name = None
suggested_name: str | None = None
while not suggested_name or suggested_name in current_names:
trailer = "".join(random.choices(acceptable_mdns_chars, k=2))
suggested_name = f"{valid_mdns_name} {trailer}"
@ -309,7 +318,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return suggested_name
@callback
def _async_is_unique_name_port(self, user_input):
def _async_is_unique_name_port(self, user_input: dict[str, str]) -> bool:
"""Determine is a name or port is already used."""
name = user_input[CONF_NAME]
port = user_input[CONF_PORT]
@ -320,7 +329,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
@ -331,10 +342,12 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
self.hk_options = {}
self.included_cameras = set()
self.hk_options: dict[str, Any] = {}
self.included_cameras: dict[str, str] = {}
async def async_step_yaml(self, user_input=None):
async def async_step_yaml(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""No options for yaml managed entries."""
if user_input is not None:
# Apparently not possible to abort an options flow
@ -343,7 +356,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
return self.async_show_form(step_id="yaml")
async def async_step_advanced(self, user_input=None):
async def async_step_advanced(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose advanced options."""
if (
not self.show_advanced_options
@ -352,17 +367,16 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
):
if user_input:
self.hk_options.update(user_input)
if (
self.show_advanced_options
and self.hk_options[CONF_HOMEKIT_MODE] == HOMEKIT_MODE_BRIDGE
):
self.hk_options[CONF_DEVICES] = user_input[CONF_DEVICES]
for key in (CONF_DOMAINS, CONF_ENTITIES):
if key in self.hk_options:
del self.hk_options[key]
if (
self.show_advanced_options
and self.hk_options[CONF_HOMEKIT_MODE] == HOMEKIT_MODE_BRIDGE
):
self.hk_options[CONF_DEVICES] = user_input[CONF_DEVICES]
if CONF_INCLUDE_EXCLUDE_MODE in self.hk_options:
del self.hk_options[CONF_INCLUDE_EXCLUDE_MODE]
@ -386,7 +400,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
),
)
async def async_step_cameras(self, user_input=None):
async def async_step_cameras(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose camera config."""
if user_input is not None:
entity_config = self.hk_options[CONF_ENTITY_CONFIG]
@ -433,7 +449,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
)
return self.async_show_form(step_id="cameras", data_schema=data_schema)
async def async_step_accessory(self, user_input=None):
async def async_step_accessory(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose entity for the accessory."""
domains = self.hk_options[CONF_DOMAINS]
@ -470,7 +488,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
),
)
async def async_step_include(self, user_input=None):
async def async_step_include(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose entities to include from the domain on the bridge."""
domains = self.hk_options[CONF_DOMAINS]
if user_input is not None:
@ -507,7 +527,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
),
)
async def async_step_exclude(self, user_input=None):
async def async_step_exclude(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Choose entities to exclude from the domain on the bridge."""
domains = self.hk_options[CONF_DOMAINS]
@ -516,13 +538,13 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
entities = cv.ensure_list(user_input[CONF_ENTITIES])
entity_filter[CONF_INCLUDE_DOMAINS] = domains
entity_filter[CONF_EXCLUDE_ENTITIES] = entities
self.included_cameras = set()
self.included_cameras = {}
if CAMERA_DOMAIN in entity_filter[CONF_INCLUDE_DOMAINS]:
camera_entities = _async_get_matching_entities(
self.hass, [CAMERA_DOMAIN]
)
self.included_cameras = {
entity_id
entity_id: entity_id
for entity_id in camera_entities
if entity_id not in entities
}
@ -571,7 +593,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
),
)
async def async_step_init(self, user_input=None):
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle options flow."""
if self.config_entry.source == SOURCE_IMPORT:
return await self.async_step_yaml(user_input)
@ -615,16 +639,16 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
)
async def _async_get_supported_devices(hass):
async def _async_get_supported_devices(hass: HomeAssistant) -> dict[str, str]:
"""Return all supported devices."""
results = await device_automation.async_get_device_automations(
hass, device_automation.DeviceAutomationType.TRIGGER
)
dev_reg = device_registry.async_get(hass)
unsorted = {
device_id: dev_reg.async_get(device_id).name or device_id
for device_id in results
}
unsorted: dict[str, str] = {}
for device_id in results:
entry = dev_reg.async_get(device_id)
unsorted[device_id] = entry.name or device_id if entry else device_id
return dict(sorted(unsorted.items(), key=lambda item: item[1]))
@ -641,13 +665,15 @@ def _async_get_matching_entities(
}
def _domains_set_from_entities(entity_ids):
def _domains_set_from_entities(entity_ids: Iterable[str]) -> set[str]:
"""Build a set of domains for the given entity ids."""
return {split_entity_id(entity_id)[0] for entity_id in entity_ids}
@callback
def _async_get_entity_ids_for_accessory_mode(hass, include_domains):
def _async_get_entity_ids_for_accessory_mode(
hass: HomeAssistant, include_domains: Iterable[str]
) -> list[str]:
"""Build a list of entities that should be paired in accessory mode."""
accessory_mode_domains = {
domain for domain in include_domains if domain in DOMAINS_NEED_ACCESSORY_MODE
@ -664,7 +690,7 @@ def _async_get_entity_ids_for_accessory_mode(hass, include_domains):
@callback
def _async_entity_ids_with_accessory_mode(hass):
def _async_entity_ids_with_accessory_mode(hass: HomeAssistant) -> set[str]:
"""Return a set of entity ids that have config entries in accessory mode."""
entity_ids = set()

4
homeassistant/components/homekit/diagnostics.py
View File

@ -18,7 +18,6 @@ async def async_get_config_entry_diagnostics(
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
homekit: HomeKit = hass.data[DOMAIN][entry.entry_id][HOMEKIT]
driver: AccessoryDriver = homekit.driver
data: dict[str, Any] = {
"status": homekit.status,
"config-entry": {
@ -28,8 +27,9 @@ async def async_get_config_entry_diagnostics(
"options": dict(entry.options),
},
}
if not driver:
if not hasattr(homekit, "driver"):
return data
driver: AccessoryDriver = homekit.driver
data.update(driver.get_accessories())
state: State = driver.state
data.update(

12
homeassistant/components/homekit/logbook.py
View File

@ -1,16 +1,22 @@
"""Describe logbook events."""
from collections.abc import Callable
from typing import Any
from homeassistant.const import ATTR_ENTITY_ID, ATTR_SERVICE
from homeassistant.core import callback
from homeassistant.core import Event, HomeAssistant, callback
from .const import ATTR_DISPLAY_NAME, ATTR_VALUE, DOMAIN, EVENT_HOMEKIT_CHANGED
@callback
def async_describe_events(hass, async_describe_event):
def async_describe_events(
hass: HomeAssistant,
async_describe_event: Callable[[str, str, Callable[[Event], dict[str, Any]]], None],
) -> None:
"""Describe logbook events."""
@callback
def async_describe_logbook_event(event):
def async_describe_logbook_event(event: Event) -> dict[str, Any]:
"""Describe a logbook event."""
data = event.data
entity_id = data.get(ATTR_ENTITY_ID)

30
homeassistant/components/homekit/type_triggers.py
View File

@ -1,8 +1,12 @@
"""Class to hold all sensor accessories."""
from __future__ import annotations
import logging
from typing import Any
from pyhap.const import CATEGORY_SENSOR
from homeassistant.core import CALLBACK_TYPE, Context
from homeassistant.helpers.trigger import async_initialize_triggers
from .accessories import TYPES, HomeAccessory
@ -22,14 +26,21 @@ _LOGGER = logging.getLogger(__name__)
class DeviceTriggerAccessory(HomeAccessory):
"""Generate a Programmable switch."""
def __init__(self, *args, device_triggers=None, device_id=None):
def __init__(
self,
*args: Any,
device_triggers: list[dict[str, Any]] | None = None,
device_id: str | None = None,
) -> None:
"""Initialize a Programmable switch accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR, device_id=device_id)
assert device_triggers is not None
self._device_triggers = device_triggers
self._remove_triggers = None
self._remove_triggers: CALLBACK_TYPE | None = None
self.triggers = []
assert device_triggers is not None
for idx, trigger in enumerate(device_triggers):
type_ = trigger.get("type")
type_ = trigger["type"]
subtype = trigger.get("subtype")
trigger_name = (
f"{type_.title()} {subtype.title()}" if subtype else type_.title()
@ -53,7 +64,12 @@ class DeviceTriggerAccessory(HomeAccessory):
serv_service_label.configure_char(CHAR_SERVICE_LABEL_NAMESPACE, value=1)
serv_stateless_switch.add_linked_service(serv_service_label)
async def async_trigger(self, run_variables, context=None, skip_condition=False):
async def async_trigger(
self,
run_variables: dict,
context: Context | None = None,
skip_condition: bool = False,
) -> None:
"""Trigger button press.
This method is a coroutine.
@ -67,7 +83,7 @@ class DeviceTriggerAccessory(HomeAccessory):
# Attach the trigger using the helper in async run
# and detach it in async stop
async def run(self):
async def run(self) -> None:
"""Handle accessory driver started event."""
self._remove_triggers = await async_initialize_triggers(
self.hass,
@ -78,12 +94,12 @@ class DeviceTriggerAccessory(HomeAccessory):
_LOGGER.log,
)
async def stop(self):
async def stop(self) -> None:
"""Handle accessory driver stop event."""
if self._remove_triggers:
self._remove_triggers()
@property
def available(self):
def available(self) -> bool:
"""Return available."""
return True

53
homeassistant/components/homekit/util.py
View File

@ -8,7 +8,9 @@ import os
import re
import secrets
import socket
from typing import Any, cast
from pyhap.accessory import Accessory
import pyqrcode
import voluptuous as vol
@ -34,7 +36,7 @@ from homeassistant.const import (
CONF_TYPE,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant, callback, split_entity_id
from homeassistant.core import HomeAssistant, State, callback, split_entity_id
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.storage import STORAGE_DIR
import homeassistant.util.temperature as temp_util
@ -242,7 +244,7 @@ HOMEKIT_CHAR_TRANSLATIONS = {
}
def validate_entity_config(values):
def validate_entity_config(values: dict) -> dict[str, dict]:
"""Validate config entry for CONF_ENTITY."""
if not isinstance(values, dict):
raise vol.Invalid("expected a dictionary")
@ -288,7 +290,7 @@ def validate_entity_config(values):
return entities
def get_media_player_features(state):
def get_media_player_features(state: State) -> list[str]:
"""Determine features for media players."""
features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
@ -306,7 +308,7 @@ def get_media_player_features(state):
return supported_modes
def validate_media_player_features(state, feature_list):
def validate_media_player_features(state: State, feature_list: str) -> bool:
"""Validate features for media players."""
if not (supported_modes := get_media_player_features(state)):
_LOGGER.error("%s does not support any media_player features", state.entity_id)
@ -329,7 +331,9 @@ def validate_media_player_features(state, feature_list):
return True
def async_show_setup_message(hass, entry_id, bridge_name, pincode, uri):
def async_show_setup_message(
hass: HomeAssistant, entry_id: str, bridge_name: str, pincode: bytes, uri: str
) -> None:
"""Display persistent notification with setup information."""
pin = pincode.decode()
_LOGGER.info("Pincode: %s", pin)
@ -351,12 +355,12 @@ def async_show_setup_message(hass, entry_id, bridge_name, pincode, uri):
persistent_notification.async_create(hass, message, "HomeKit Pairing", entry_id)
def async_dismiss_setup_message(hass, entry_id):
def async_dismiss_setup_message(hass: HomeAssistant, entry_id: str) -> None:
"""Dismiss persistent notification and remove QR code."""
persistent_notification.async_dismiss(hass, entry_id)
def convert_to_float(state):
def convert_to_float(state: Any) -> float | None:
"""Return float of state, catch errors."""
try:
return float(state)
@ -384,17 +388,17 @@ def cleanup_name_for_homekit(name: str | None) -> str:
return name.translate(HOMEKIT_CHAR_TRANSLATIONS)[:MAX_NAME_LENGTH]
def temperature_to_homekit(temperature, unit):
def temperature_to_homekit(temperature: float | int, unit: str) -> float:
"""Convert temperature to Celsius for HomeKit."""
return round(temp_util.convert(temperature, unit, TEMP_CELSIUS), 1)
def temperature_to_states(temperature, unit):
def temperature_to_states(temperature: float | int, unit: str) -> float:
"""Convert temperature back from Celsius to Home Assistant unit."""
return round(temp_util.convert(temperature, TEMP_CELSIUS, unit) * 2) / 2
def density_to_air_quality(density):
def density_to_air_quality(density: float) -> int:
"""Map PM2.5 density to HomeKit AirQuality level."""
if density <= 35:
return 1
@ -407,7 +411,7 @@ def density_to_air_quality(density):
return 5
def density_to_air_quality_pm10(density):
def density_to_air_quality_pm10(density: float) -> int:
"""Map PM10 density to HomeKit AirQuality level."""
if density <= 40:
return 1
@ -420,7 +424,7 @@ def density_to_air_quality_pm10(density):
return 5
def density_to_air_quality_pm25(density):
def density_to_air_quality_pm25(density: float) -> int:
"""Map PM2.5 density to HomeKit AirQuality level."""
if density <= 25:
return 1
@ -433,22 +437,22 @@ def density_to_air_quality_pm25(density):
return 5
def get_persist_filename_for_entry_id(entry_id: str):
def get_persist_filename_for_entry_id(entry_id: str) -> str:
"""Determine the filename of the homekit state file."""
return f"{DOMAIN}.{entry_id}.state"
def get_aid_storage_filename_for_entry_id(entry_id: str):
def get_aid_storage_filename_for_entry_id(entry_id: str) -> str:
"""Determine the ilename of homekit aid storage file."""
return f"{DOMAIN}.{entry_id}.aids"
def get_persist_fullpath_for_entry_id(hass: HomeAssistant, entry_id: str):
def get_persist_fullpath_for_entry_id(hass: HomeAssistant, entry_id: str) -> str:
"""Determine the path to the homekit state file."""
return hass.config.path(STORAGE_DIR, get_persist_filename_for_entry_id(entry_id))
def get_aid_storage_fullpath_for_entry_id(hass: HomeAssistant, entry_id: str):
def get_aid_storage_fullpath_for_entry_id(hass: HomeAssistant, entry_id: str) -> str:
"""Determine the path to the homekit aid storage file."""
return hass.config.path(
STORAGE_DIR, get_aid_storage_filename_for_entry_id(entry_id)
@ -459,7 +463,7 @@ def _format_version_part(version_part: str) -> str:
return str(max(0, min(MAX_VERSION_PART, coerce_int(version_part))))
def format_version(version):
def format_version(version: str) -> str | None:
"""Extract the version string in a format homekit can consume."""
split_ver = str(version).replace("-", ".").replace(" ", ".")
num_only = NUMBERS_ONLY_RE.sub("", split_ver)
@ -469,12 +473,12 @@ def format_version(version):
return None if _is_zero_but_true(value) else value
def _is_zero_but_true(value):
def _is_zero_but_true(value: Any) -> bool:
"""Zero but true values can crash apple watches."""
return convert_to_float(value) == 0
def remove_state_files_for_entry_id(hass: HomeAssistant, entry_id: str):
def remove_state_files_for_entry_id(hass: HomeAssistant, entry_id: str) -> bool:
"""Remove the state files from disk."""
persist_file_path = get_persist_fullpath_for_entry_id(hass, entry_id)
aid_storage_path = get_aid_storage_fullpath_for_entry_id(hass, entry_id)
@ -484,7 +488,7 @@ def remove_state_files_for_entry_id(hass: HomeAssistant, entry_id: str):
return True
def _get_test_socket():
def _get_test_socket() -> socket.socket:
"""Create a socket to test binding ports."""
test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
test_socket.setblocking(False)
@ -527,9 +531,10 @@ def _async_find_next_available_port(start_port: int, exclude_ports: set) -> int:
if port == MAX_PORT:
raise
continue
raise RuntimeError("unreachable")
def pid_is_alive(pid) -> bool:
def pid_is_alive(pid: int) -> bool:
"""Check to see if a process is alive."""
try:
os.kill(pid, 0)
@ -539,14 +544,14 @@ def pid_is_alive(pid) -> bool:
return False
def accessory_friendly_name(hass_name, accessory):
def accessory_friendly_name(hass_name: str, accessory: Accessory) -> str:
"""Return the combined name for the accessory.
The mDNS name and the Home Assistant config entry
name are usually different which means they need to
see both to identify the accessory.
"""
accessory_mdns_name = accessory.display_name
accessory_mdns_name = cast(str, accessory.display_name)
if hass_name.casefold().startswith(accessory_mdns_name.casefold()):
return hass_name
if accessory_mdns_name.casefold().startswith(hass_name.casefold()):
@ -554,7 +559,7 @@ def accessory_friendly_name(hass_name, accessory):
return f"{hass_name} ({accessory_mdns_name})"
def state_needs_accessory_mode(state):
def state_needs_accessory_mode(state: State) -> bool:
"""Return if the entity represented by the state must be paired in accessory mode."""
if state.domain in (CAMERA_DOMAIN, LOCK_DOMAIN):
return True

97
mypy.ini
View File

@ -891,6 +891,94 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.accessories]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.aidmanager]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.config_flow]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.diagnostics]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.logbook]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.type_triggers]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit.util]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.homekit_controller]
check_untyped_defs = true
disallow_incomplete_defs = true
@ -2566,15 +2654,6 @@ ignore_errors = true
[mypy-homeassistant.components.home_plus_control.api]
ignore_errors = true
[mypy-homeassistant.components.homekit.aidmanager]
ignore_errors = true
[mypy-homeassistant.components.homekit.config_flow]
ignore_errors = true
[mypy-homeassistant.components.homekit.util]
ignore_errors = true
[mypy-homeassistant.components.honeywell.climate]
ignore_errors = true

3
script/hassfest/mypy_config.py
View File

@ -60,9 +60,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.here_travel_time.sensor",
"homeassistant.components.home_plus_control",
"homeassistant.components.home_plus_control.api",
"homeassistant.components.homekit.aidmanager",
"homeassistant.components.homekit.config_flow",
"homeassistant.components.homekit.util",
"homeassistant.components.honeywell.climate",
"homeassistant.components.icloud",
"homeassistant.components.icloud.account",