diff --git a/.strict-typing b/.strict-typing index d9c5e77f66d..6d428fdcb2f 100644 --- a/.strict-typing +++ b/.strict-typing @@ -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 diff --git a/homeassistant/components/homekit/__init__.py b/homeassistant/components/homekit/__init__.py index e2847a597eb..af21bbdaf84 100644 --- a/homeassistant/components/homekit/__init__.py +++ b/homeassistant/components/homekit/__init__.py @@ -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: diff --git a/homeassistant/components/homekit/accessories.py b/homeassistant/components/homekit/accessories.py index c77fa96a532..1d06fa04a57 100644 --- a/homeassistant/components/homekit/accessories.py +++ b/homeassistant/components/homekit/accessories.py @@ -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) diff --git a/homeassistant/components/homekit/aidmanager.py b/homeassistant/components/homekit/aidmanager.py index c7ddc29a788..ddba9d02bcd 100644 --- a/homeassistant/components/homekit/aidmanager.py +++ b/homeassistant/components/homekit/aidmanager.py @@ -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} diff --git a/homeassistant/components/homekit/config_flow.py b/homeassistant/components/homekit/config_flow.py index d0c0f73ce07..a147c8dcb5d 100644 --- a/homeassistant/components/homekit/config_flow.py +++ b/homeassistant/components/homekit/config_flow.py @@ -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() diff --git a/homeassistant/components/homekit/diagnostics.py b/homeassistant/components/homekit/diagnostics.py index 2a54c1ef543..f717f02ea02 100644 --- a/homeassistant/components/homekit/diagnostics.py +++ b/homeassistant/components/homekit/diagnostics.py @@ -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( diff --git a/homeassistant/components/homekit/logbook.py b/homeassistant/components/homekit/logbook.py index 0ea5a5d542a..b6805f8cf6c 100644 --- a/homeassistant/components/homekit/logbook.py +++ b/homeassistant/components/homekit/logbook.py @@ -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) diff --git a/homeassistant/components/homekit/type_triggers.py b/homeassistant/components/homekit/type_triggers.py index 6d5f67f9915..4b3a7e73cac 100644 --- a/homeassistant/components/homekit/type_triggers.py +++ b/homeassistant/components/homekit/type_triggers.py @@ -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 diff --git a/homeassistant/components/homekit/util.py b/homeassistant/components/homekit/util.py index b31c55db767..be5c166b71d 100644 --- a/homeassistant/components/homekit/util.py +++ b/homeassistant/components/homekit/util.py @@ -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 diff --git a/mypy.ini b/mypy.ini index f39c982fb0b..cc114633787 100644 --- a/mypy.ini +++ b/mypy.ini @@ -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 diff --git a/script/hassfest/mypy_config.py b/script/hassfest/mypy_config.py index 279e78fb44f..d58406f67f9 100644 --- a/script/hassfest/mypy_config.py +++ b/script/hassfest/mypy_config.py @@ -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",