diff --git a/homeassistant/components/homekit/iidmanager.py b/homeassistant/components/homekit/iidmanager.py index 2bd50821138..f44d76d3ee7 100644 --- a/homeassistant/components/homekit/iidmanager.py +++ b/homeassistant/components/homekit/iidmanager.py @@ -36,7 +36,7 @@ class IIDStorage(Store): old_major_version: int, old_minor_version: int, old_data: dict, - ): + ) -> dict: """Migrate to the new version.""" if old_major_version == 1: # Convert v1 to v2 format which uses a unique iid set per accessory diff --git a/homeassistant/components/homekit/type_cameras.py b/homeassistant/components/homekit/type_cameras.py index 63b2bc023da..6bc8e785c7f 100644 --- a/homeassistant/components/homekit/type_cameras.py +++ b/homeassistant/components/homekit/type_cameras.py @@ -2,6 +2,7 @@ import asyncio from datetime import timedelta import logging +from typing import Any from haffmpeg.core import FFMPEG_STDERR, HAFFmpeg from pyhap.camera import ( @@ -14,7 +15,7 @@ from pyhap.const import CATEGORY_CAMERA from homeassistant.components import camera from homeassistant.components.ffmpeg import get_ffmpeg_manager from homeassistant.const import STATE_ON -from homeassistant.core import State, callback +from homeassistant.core import HomeAssistant, State, callback from homeassistant.helpers.event import ( EventStateChangedData, async_track_state_change_event, @@ -22,7 +23,7 @@ from homeassistant.helpers.event import ( ) from homeassistant.helpers.typing import EventType -from .accessories import TYPES, HomeAccessory +from .accessories import TYPES, HomeAccessory, HomeDriver from .const import ( CHAR_MOTION_DETECTED, CHAR_MUTE, @@ -141,7 +142,15 @@ CONFIG_DEFAULTS = { class Camera(HomeAccessory, PyhapCamera): """Generate a Camera accessory.""" - def __init__(self, hass, driver, name, entity_id, aid, config): + def __init__( + self, + hass: HomeAssistant, + driver: HomeDriver, + name: str, + entity_id: str, + aid: int, + config: dict[str, Any], + ) -> None: """Initialize a Camera accessory object.""" self._ffmpeg = get_ffmpeg_manager(hass) for config_key, conf in CONFIG_DEFAULTS.items(): @@ -242,12 +251,13 @@ class Camera(HomeAccessory, PyhapCamera): self._async_update_doorbell_state(state) - async def run(self): + async def run(self) -> None: """Handle accessory driver started event. Run inside the Home Assistant event loop. """ if self._char_motion_detected: + assert self.linked_motion_sensor self._subscriptions.append( async_track_state_change_event( self.hass, @@ -257,6 +267,7 @@ class Camera(HomeAccessory, PyhapCamera): ) if self._char_doorbell_detected: + assert self.linked_doorbell_sensor self._subscriptions.append( async_track_state_change_event( self.hass, @@ -282,6 +293,7 @@ class Camera(HomeAccessory, PyhapCamera): return detected = new_state.state == STATE_ON + assert self._char_motion_detected if self._char_motion_detected.value == detected: return @@ -307,6 +319,8 @@ class Camera(HomeAccessory, PyhapCamera): if not new_state: return + assert self._char_doorbell_detected + assert self._char_doorbell_detected_switch if new_state.state == STATE_ON: self._char_doorbell_detected.set_value(DOORBELL_SINGLE_PRESS) self._char_doorbell_detected_switch.set_value(DOORBELL_SINGLE_PRESS) @@ -318,11 +332,10 @@ class Camera(HomeAccessory, PyhapCamera): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State | None) -> None: """Handle state change to update HomeKit value.""" - pass # pylint: disable=unnecessary-pass - async def _async_get_stream_source(self): + async def _async_get_stream_source(self) -> str | None: """Find the camera stream source url.""" if stream_source := self.config.get(CONF_STREAM_SOURCE): return stream_source @@ -337,7 +350,9 @@ class Camera(HomeAccessory, PyhapCamera): ) return stream_source - async def start_stream(self, session_info, stream_config): + async def start_stream( + self, session_info: dict[str, Any], stream_config: dict[str, Any] + ) -> bool: """Start a new stream with the given configuration.""" _LOGGER.debug( "[%s] Starting stream with the following parameters: %s", @@ -418,7 +433,9 @@ class Camera(HomeAccessory, PyhapCamera): return await self._async_ffmpeg_watch(session_info["id"]) - async def _async_log_stderr_stream(self, stderr_reader): + async def _async_log_stderr_stream( + self, stderr_reader: asyncio.StreamReader + ) -> None: """Log output from ffmpeg.""" _LOGGER.debug("%s: ffmpeg: started", self.display_name) while True: @@ -428,7 +445,7 @@ class Camera(HomeAccessory, PyhapCamera): _LOGGER.debug("%s: ffmpeg: %s", self.display_name, line.rstrip()) - async def _async_ffmpeg_watch(self, session_id): + async def _async_ffmpeg_watch(self, session_id: str) -> bool: """Check to make sure ffmpeg is still running and cleanup if not.""" ffmpeg_pid = self.sessions[session_id][FFMPEG_PID] if pid_is_alive(ffmpeg_pid): @@ -440,7 +457,7 @@ class Camera(HomeAccessory, PyhapCamera): return False @callback - def _async_stop_ffmpeg_watch(self, session_id): + def _async_stop_ffmpeg_watch(self, session_id: str) -> None: """Cleanup a streaming session after stopping.""" if FFMPEG_WATCHER not in self.sessions[session_id]: return @@ -448,7 +465,7 @@ class Camera(HomeAccessory, PyhapCamera): self.sessions[session_id].pop(FFMPEG_LOGGER).cancel() @callback - def async_stop(self): + def async_stop(self) -> None: """Stop any streams when the accessory is stopped.""" for session_info in self.sessions.values(): self.hass.async_create_background_task( @@ -456,7 +473,7 @@ class Camera(HomeAccessory, PyhapCamera): ) super().async_stop() - async def stop_stream(self, session_info): + async def stop_stream(self, session_info: dict[str, Any]) -> None: """Stop the stream for the given ``session_id``.""" session_id = session_info["id"] if not (stream := session_info.get("stream")): @@ -467,7 +484,7 @@ class Camera(HomeAccessory, PyhapCamera): if not pid_is_alive(stream.process.pid): _LOGGER.info("[%s] Stream already stopped", session_id) - return True + return for shutdown_method in ("close", "kill"): _LOGGER.info("[%s] %s stream", session_id, shutdown_method) @@ -479,11 +496,13 @@ class Camera(HomeAccessory, PyhapCamera): "[%s] Failed to %s stream", session_id, shutdown_method ) - async def reconfigure_stream(self, session_info, stream_config): + async def reconfigure_stream( + self, session_info: dict[str, Any], stream_config: dict[str, Any] + ) -> bool: """Reconfigure the stream so that it uses the given ``stream_config``.""" return True - async def async_get_snapshot(self, image_size): + async def async_get_snapshot(self, image_size: dict[str, int]) -> bytes: """Return a jpeg of a snapshot from the camera.""" image = await camera.async_get_image( self.hass, diff --git a/homeassistant/components/homekit/type_humidifiers.py b/homeassistant/components/homekit/type_humidifiers.py index de25717877c..d296b293820 100644 --- a/homeassistant/components/homekit/type_humidifiers.py +++ b/homeassistant/components/homekit/type_humidifiers.py @@ -1,5 +1,6 @@ """Class to hold all thermostat accessories.""" import logging +from typing import Any from pyhap.const import CATEGORY_HUMIDIFIER @@ -73,7 +74,7 @@ HC_STATE_DEHUMIDIFYING = 3 class HumidifierDehumidifier(HomeAccessory): """Generate a HumidifierDehumidifier accessory for a humidifier.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a HumidifierDehumidifier accessory object.""" super().__init__(*args, category=CATEGORY_HUMIDIFIER) self._reload_on_change_attrs.extend( @@ -83,8 +84,9 @@ class HumidifierDehumidifier(HomeAccessory): ) ) - self.chars = [] + self.chars: list[str] = [] state = self.hass.states.get(self.entity_id) + assert state device_class = state.attributes.get( ATTR_DEVICE_CLASS, HumidifierDeviceClass.HUMIDIFIER ) @@ -151,7 +153,7 @@ class HumidifierDehumidifier(HomeAccessory): if humidity_state: self._async_update_current_humidity(humidity_state) - async def run(self): + async def run(self) -> None: """Handle accessory driver started event. Run inside the Home Assistant event loop. @@ -205,7 +207,8 @@ class HumidifierDehumidifier(HomeAccessory): ex, ) - def _set_chars(self, char_values): + def _set_chars(self, char_values: dict[str, Any]) -> None: + """Set characteristics based on the data coming from HomeKit.""" _LOGGER.debug("HumidifierDehumidifier _set_chars: %s", char_values) if CHAR_TARGET_HUMIDIFIER_DEHUMIDIFIER in char_values: @@ -225,6 +228,7 @@ class HumidifierDehumidifier(HomeAccessory): if self._target_humidity_char_name in char_values: state = self.hass.states.get(self.entity_id) + assert state max_humidity = state.attributes.get(ATTR_MAX_HUMIDITY, DEFAULT_MAX_HUMIDITY) max_humidity = round(max_humidity) max_humidity = min(max_humidity, 100) @@ -232,6 +236,11 @@ class HumidifierDehumidifier(HomeAccessory): min_humidity = state.attributes.get(ATTR_MIN_HUMIDITY, DEFAULT_MIN_HUMIDITY) min_humidity = round(min_humidity) min_humidity = max(min_humidity, 0) + # The min/max humidity values here should be clamped to the HomeKit + # min/max that was set when the accessory was added to HomeKit so + # that the user cannot set a value outside of the range that was + # originally set as it could cause HomeKit to report the accessory + # as not responding. humidity = round(char_values[self._target_humidity_char_name]) @@ -252,7 +261,7 @@ class HumidifierDehumidifier(HomeAccessory): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update state without rechecking the device features.""" is_active = new_state.state == STATE_ON diff --git a/homeassistant/components/homekit/type_lights.py b/homeassistant/components/homekit/type_lights.py index e8272358633..13301c3f507 100644 --- a/homeassistant/components/homekit/type_lights.py +++ b/homeassistant/components/homekit/type_lights.py @@ -1,7 +1,9 @@ """Class to hold all light accessories.""" from __future__ import annotations +from datetime import datetime import logging +from typing import Any from pyhap.const import CATEGORY_LIGHTBULB @@ -29,7 +31,7 @@ from homeassistant.const import ( SERVICE_TURN_ON, STATE_ON, ) -from homeassistant.core import callback +from homeassistant.core import CALLBACK_TYPE, State, callback from homeassistant.helpers.event import async_call_later from homeassistant.util.color import ( color_temperature_kelvin_to_mired, @@ -68,7 +70,7 @@ class Light(HomeAccessory): Currently supports: state, brightness, color temperature, rgb_color. """ - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a new Light accessory object.""" super().__init__(*args, category=CATEGORY_LIGHTBULB) self._reload_on_change_attrs.extend( @@ -79,10 +81,11 @@ class Light(HomeAccessory): ) ) self.chars = [] - self._event_timer = None - self._pending_events = {} + self._event_timer: CALLBACK_TYPE | None = None + self._pending_events: dict[str, Any] = {} state = self.hass.states.get(self.entity_id) + assert state attributes = state.attributes self.color_modes = color_modes = ( attributes.get(ATTR_SUPPORTED_COLOR_MODES) or [] @@ -140,7 +143,7 @@ class Light(HomeAccessory): self.async_update_state(state) serv_light.setter_callback = self._set_chars - def _set_chars(self, char_values): + def _set_chars(self, char_values: dict[str, Any]) -> None: _LOGGER.debug("Light _set_chars: %s", char_values) # Newest change always wins if CHAR_COLOR_TEMPERATURE in self._pending_events and ( @@ -159,14 +162,14 @@ class Light(HomeAccessory): ) @callback - def _async_send_events(self, *_): + def _async_send_events(self, _now: datetime) -> None: """Process all changes at once.""" _LOGGER.debug("Coalesced _set_chars: %s", self._pending_events) char_values = self._pending_events self._pending_events = {} events = [] service = SERVICE_TURN_ON - params = {ATTR_ENTITY_ID: self.entity_id} + params: dict[str, Any] = {ATTR_ENTITY_ID: self.entity_id} if CHAR_ON in char_values: if not char_values[CHAR_ON]: @@ -231,7 +234,7 @@ class Light(HomeAccessory): self.async_call_service(DOMAIN, service, params, ", ".join(events)) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update light after state change.""" # Handle State state = new_state.state diff --git a/homeassistant/components/homekit/type_switches.py b/homeassistant/components/homekit/type_switches.py index f79185c64b1..5c0c2c74f0a 100644 --- a/homeassistant/components/homekit/type_switches.py +++ b/homeassistant/components/homekit/type_switches.py @@ -2,8 +2,9 @@ from __future__ import annotations import logging -from typing import NamedTuple +from typing import Any, NamedTuple +from pyhap.characteristic import Characteristic from pyhap.const import ( CATEGORY_FAUCET, CATEGORY_OUTLET, @@ -30,7 +31,7 @@ from homeassistant.const import ( SERVICE_TURN_ON, STATE_ON, ) -from homeassistant.core import callback, split_entity_id +from homeassistant.core import State, callback, split_entity_id from homeassistant.helpers.event import async_call_later from .accessories import TYPES, HomeAccessory @@ -78,10 +79,11 @@ ACTIVATE_ONLY_RESET_SECONDS = 10 class Outlet(HomeAccessory): """Generate an Outlet accessory.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize an Outlet accessory object.""" super().__init__(*args, category=CATEGORY_OUTLET) state = self.hass.states.get(self.entity_id) + assert state serv_outlet = self.add_preload_service(SERV_OUTLET) self.char_on = serv_outlet.configure_char( @@ -94,7 +96,7 @@ class Outlet(HomeAccessory): # GET to avoid an event storm after homekit startup self.async_update_state(state) - def set_state(self, value): + def set_state(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug("%s: Set switch state to %s", self.entity_id, value) params = {ATTR_ENTITY_ID: self.entity_id} @@ -102,7 +104,7 @@ class Outlet(HomeAccessory): self.async_call_service(DOMAIN, service, params) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update switch state after state changed.""" current_state = new_state.state == STATE_ON _LOGGER.debug("%s: Set current state to %s", self.entity_id, current_state) @@ -113,13 +115,14 @@ class Outlet(HomeAccessory): class Switch(HomeAccessory): """Generate a Switch accessory.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a Switch accessory object.""" super().__init__(*args, category=CATEGORY_SWITCH) self._domain, self._object_id = split_entity_id(self.entity_id) state = self.hass.states.get(self.entity_id) + assert state - self.activate_only = self.is_activate(self.hass.states.get(self.entity_id)) + self.activate_only = self.is_activate(state) serv_switch = self.add_preload_service(SERV_SWITCH) self.char_on = serv_switch.configure_char( @@ -129,16 +132,16 @@ class Switch(HomeAccessory): # GET to avoid an event storm after homekit startup self.async_update_state(state) - def is_activate(self, state): + def is_activate(self, state: State) -> bool: """Check if entity is activate only.""" return self._domain in ACTIVATE_ONLY_SWITCH_DOMAINS - def reset_switch(self, *args): + def reset_switch(self, *args: Any) -> None: """Reset switch to emulate activate click.""" _LOGGER.debug("%s: Reset switch to off", self.entity_id) self.char_on.set_value(False) - def set_state(self, value): + def set_state(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug("%s: Set switch state to %s", self.entity_id, value) if self.activate_only and not value: @@ -162,7 +165,7 @@ class Switch(HomeAccessory): async_call_later(self.hass, ACTIVATE_ONLY_RESET_SECONDS, self.reset_switch) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update switch state after state changed.""" self.activate_only = self.is_activate(new_state) if self.activate_only: @@ -180,10 +183,12 @@ class Switch(HomeAccessory): class Vacuum(Switch): """Generate a Switch accessory.""" - def set_state(self, value): + def set_state(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug("%s: Set switch state to %s", self.entity_id, value) state = self.hass.states.get(self.entity_id) + assert state + features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0) if value: @@ -198,7 +203,7 @@ class Vacuum(Switch): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update switch state after state changed.""" current_state = new_state.state in (STATE_CLEANING, STATE_ON) _LOGGER.debug("%s: Set current state to %s", self.entity_id, current_state) @@ -209,10 +214,12 @@ class Vacuum(Switch): class Valve(HomeAccessory): """Generate a Valve accessory.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a Valve accessory object.""" super().__init__(*args) state = self.hass.states.get(self.entity_id) + assert state + valve_type = self.config[CONF_TYPE] self.category = VALVE_TYPE[valve_type].category @@ -228,7 +235,7 @@ class Valve(HomeAccessory): # GET to avoid an event storm after homekit startup self.async_update_state(state) - def set_state(self, value): + def set_state(self, value: bool) -> None: """Move value state to value if call came from HomeKit.""" _LOGGER.debug("%s: Set switch state to %s", self.entity_id, value) self.char_in_use.set_value(value) @@ -237,7 +244,7 @@ class Valve(HomeAccessory): self.async_call_service(DOMAIN, service, params) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update switch state after state changed.""" current_state = 1 if new_state.state == STATE_ON else 0 _LOGGER.debug("%s: Set active state to %s", self.entity_id, current_state) @@ -250,12 +257,14 @@ class Valve(HomeAccessory): class SelectSwitch(HomeAccessory): """Generate a Switch accessory that contains multiple switches.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a Switch accessory object.""" super().__init__(*args, category=CATEGORY_SWITCH) self.domain = split_entity_id(self.entity_id)[0] state = self.hass.states.get(self.entity_id) - self.select_chars = {} + assert state + + self.select_chars: dict[str, Characteristic] = {} options = state.attributes[ATTR_OPTIONS] for option in options: serv_option = self.add_preload_service( @@ -275,14 +284,14 @@ class SelectSwitch(HomeAccessory): # GET to avoid an event storm after homekit startup self.async_update_state(state) - def select_option(self, option): + def select_option(self, option: str) -> None: """Set option from HomeKit.""" _LOGGER.debug("%s: Set option to %s", self.entity_id, option) params = {ATTR_ENTITY_ID: self.entity_id, "option": option} self.async_call_service(self.domain, SERVICE_SELECT_OPTION, params) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update switch state after state changed.""" current_option = cleanup_name_for_homekit(new_state.state) for option, char in self.select_chars.items():