From 0155df6778f7903093d9dfdf47016ad309c2941f Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 13 Oct 2023 09:28:23 -1000 Subject: [PATCH] Add some more typing to HomeKit --- .../components/homekit/type_media_players.py | 58 ++++++++++--------- .../components/homekit/type_remotes.py | 14 +++-- .../homekit/type_security_systems.py | 10 ++-- .../components/homekit/type_sensors.py | 58 ++++++++++--------- 4 files changed, 77 insertions(+), 63 deletions(-) diff --git a/homeassistant/components/homekit/type_media_players.py b/homeassistant/components/homekit/type_media_players.py index da7fdceede31e5..23fbd5b454dbb7 100644 --- a/homeassistant/components/homekit/type_media_players.py +++ b/homeassistant/components/homekit/type_media_players.py @@ -2,6 +2,7 @@ import logging from typing import Any +from pyhap.characteristic import Characteristic from pyhap.const import CATEGORY_SWITCH from homeassistant.components.media_player import ( @@ -32,7 +33,7 @@ STATE_STANDBY, STATE_UNKNOWN, ) -from homeassistant.core import callback +from homeassistant.core import State, callback from .accessories import TYPES, HomeAccessory from .const import ( @@ -82,11 +83,12 @@ class MediaPlayer(HomeAccessory): """Generate a Media Player accessory.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a Switch accessory object.""" super().__init__(*args, category=CATEGORY_SWITCH) state = self.hass.states.get(self.entity_id) - self.chars = { + assert state + self.chars: dict[str, Characteristic | None] = { FEATURE_ON_OFF: None, FEATURE_PLAY_PAUSE: None, FEATURE_PLAY_STOP: None, @@ -137,20 +139,20 @@ def __init__(self, *args): ) self.async_update_state(state) - def generate_service_name(self, mode): + def generate_service_name(self, mode: str) -> str: """Generate name for individual service.""" return cleanup_name_for_homekit( f"{self.display_name} {MODE_FRIENDLY_NAME[mode]}" ) - def set_on_off(self, value): + def set_on_off(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug('%s: Set switch state for "on_off" to %s', self.entity_id, value) service = SERVICE_TURN_ON if value else SERVICE_TURN_OFF params = {ATTR_ENTITY_ID: self.entity_id} self.async_call_service(DOMAIN, service, params) - def set_play_pause(self, value): + def set_play_pause(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug( '%s: Set switch state for "play_pause" to %s', self.entity_id, value @@ -159,7 +161,7 @@ def set_play_pause(self, value): params = {ATTR_ENTITY_ID: self.entity_id} self.async_call_service(DOMAIN, service, params) - def set_play_stop(self, value): + def set_play_stop(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug( '%s: Set switch state for "play_stop" to %s', self.entity_id, value @@ -168,7 +170,7 @@ def set_play_stop(self, value): params = {ATTR_ENTITY_ID: self.entity_id} self.async_call_service(DOMAIN, service, params) - def set_toggle_mute(self, value): + def set_toggle_mute(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug( '%s: Set switch state for "toggle_mute" to %s', self.entity_id, value @@ -177,43 +179,43 @@ def set_toggle_mute(self, value): self.async_call_service(DOMAIN, SERVICE_VOLUME_MUTE, 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 - if self.chars[FEATURE_ON_OFF]: + if on_off_char := self.chars[FEATURE_ON_OFF]: hk_state = current_state not in MEDIA_PLAYER_OFF_STATES _LOGGER.debug( '%s: Set current state for "on_off" to %s', self.entity_id, hk_state ) - self.chars[FEATURE_ON_OFF].set_value(hk_state) + on_off_char.set_value(hk_state) - if self.chars[FEATURE_PLAY_PAUSE]: + if play_pause_char := self.chars[FEATURE_PLAY_PAUSE]: hk_state = current_state == STATE_PLAYING _LOGGER.debug( '%s: Set current state for "play_pause" to %s', self.entity_id, hk_state, ) - self.chars[FEATURE_PLAY_PAUSE].set_value(hk_state) + play_pause_char.set_value(hk_state) - if self.chars[FEATURE_PLAY_STOP]: + if play_stop_char := self.chars[FEATURE_PLAY_STOP]: hk_state = current_state == STATE_PLAYING _LOGGER.debug( '%s: Set current state for "play_stop" to %s', self.entity_id, hk_state, ) - self.chars[FEATURE_PLAY_STOP].set_value(hk_state) + play_stop_char.set_value(hk_state) - if self.chars[FEATURE_TOGGLE_MUTE]: - current_state = bool(new_state.attributes.get(ATTR_MEDIA_VOLUME_MUTED)) + if toggle_mute_char := self.chars[FEATURE_TOGGLE_MUTE]: + mute_state = bool(new_state.attributes.get(ATTR_MEDIA_VOLUME_MUTED)) _LOGGER.debug( '%s: Set current state for "toggle_mute" to %s', self.entity_id, - current_state, + mute_state, ) - self.chars[FEATURE_TOGGLE_MUTE].set_value(current_state) + toggle_mute_char.set_value(mute_state) @TYPES.register("TelevisionMediaPlayer") @@ -278,14 +280,14 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.async_update_state(state) - def set_on_off(self, value): + def set_on_off(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug('%s: Set switch state for "on_off" to %s', self.entity_id, value) service = SERVICE_TURN_ON if value else SERVICE_TURN_OFF params = {ATTR_ENTITY_ID: self.entity_id} self.async_call_service(DOMAIN, service, params) - def set_mute(self, value): + def set_mute(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug( '%s: Set switch state for "toggle_mute" to %s', self.entity_id, value @@ -293,27 +295,27 @@ def set_mute(self, value): params = {ATTR_ENTITY_ID: self.entity_id, ATTR_MEDIA_VOLUME_MUTED: value} self.async_call_service(DOMAIN, SERVICE_VOLUME_MUTE, params) - def set_volume(self, value): + def set_volume(self, value: bool) -> None: """Send volume step value if call came from HomeKit.""" _LOGGER.debug("%s: Set volume to %s", self.entity_id, value) params = {ATTR_ENTITY_ID: self.entity_id, ATTR_MEDIA_VOLUME_LEVEL: value} self.async_call_service(DOMAIN, SERVICE_VOLUME_SET, params) - def set_volume_step(self, value): + def set_volume_step(self, value: bool) -> None: """Send volume step value if call came from HomeKit.""" _LOGGER.debug("%s: Step volume by %s", self.entity_id, value) service = SERVICE_VOLUME_DOWN if value else SERVICE_VOLUME_UP params = {ATTR_ENTITY_ID: self.entity_id} self.async_call_service(DOMAIN, service, params) - def set_input_source(self, value): + def set_input_source(self, value: int) -> None: """Send input set value if call came from HomeKit.""" _LOGGER.debug("%s: Set current input to %s", self.entity_id, value) source_name = self._mapped_sources[self.sources[value]] params = {ATTR_ENTITY_ID: self.entity_id, ATTR_INPUT_SOURCE: source_name} self.async_call_service(DOMAIN, SERVICE_SELECT_SOURCE, params) - def set_remote_key(self, value): + def set_remote_key(self, value: int) -> None: """Send remote key value if call came from HomeKit.""" _LOGGER.debug("%s: Set remote key to %s", self.entity_id, value) if (key_name := REMOTE_KEYS.get(value)) is None: @@ -322,7 +324,9 @@ def set_remote_key(self, value): if key_name == KEY_PLAY_PAUSE and self._supports_play_pause: # Handle Play Pause by directly updating the media player entity. - state = self.hass.states.get(self.entity_id).state + state_obj = self.hass.states.get(self.entity_id) + assert state_obj + state = state_obj.state if state in (STATE_PLAYING, STATE_PAUSED): service = ( SERVICE_MEDIA_PLAY if state == STATE_PAUSED else SERVICE_MEDIA_PAUSE @@ -340,7 +344,7 @@ def set_remote_key(self, value): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update Television state after state changed.""" current_state = new_state.state diff --git a/homeassistant/components/homekit/type_remotes.py b/homeassistant/components/homekit/type_remotes.py index 5dfc9777964a3a..0f6e22abe1dc0a 100644 --- a/homeassistant/components/homekit/type_remotes.py +++ b/homeassistant/components/homekit/type_remotes.py @@ -219,7 +219,7 @@ def _async_update_input_state(self, hk_state, new_state): class ActivityRemote(RemoteInputSelectAccessory): """Generate a Activity Remote accessory.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a Activity Remote accessory object.""" super().__init__( RemoteEntityFeature.ACTIVITY, @@ -227,23 +227,25 @@ def __init__(self, *args): ATTR_ACTIVITY_LIST, *args, ) - self.async_update_state(self.hass.states.get(self.entity_id)) + state = self.hass.states.get(self.entity_id) + assert state + self.async_update_state(state) - def set_on_off(self, value): + def set_on_off(self, value: bool) -> None: """Move switch state to value if call came from HomeKit.""" _LOGGER.debug('%s: Set switch state for "on_off" to %s', self.entity_id, value) service = SERVICE_TURN_ON if value else SERVICE_TURN_OFF params = {ATTR_ENTITY_ID: self.entity_id} self.async_call_service(REMOTE_DOMAIN, service, params) - def set_input_source(self, value): + def set_input_source(self, value: int) -> None: """Send input set value if call came from HomeKit.""" _LOGGER.debug("%s: Set current input to %s", self.entity_id, value) source = self._mapped_sources[self.sources[value]] params = {ATTR_ENTITY_ID: self.entity_id, ATTR_ACTIVITY: source} self.async_call_service(REMOTE_DOMAIN, SERVICE_TURN_ON, params) - def set_remote_key(self, value): + def set_remote_key(self, value: int) -> None: """Send remote key value if call came from HomeKit.""" _LOGGER.debug("%s: Set remote key to %s", self.entity_id, value) if (key_name := REMOTE_KEYS.get(value)) is None: @@ -255,7 +257,7 @@ def set_remote_key(self, value): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update Television remote state after state changed.""" current_state = new_state.state # Power state remote diff --git a/homeassistant/components/homekit/type_security_systems.py b/homeassistant/components/homekit/type_security_systems.py index f9c881339cea92..de2c463bfb2127 100644 --- a/homeassistant/components/homekit/type_security_systems.py +++ b/homeassistant/components/homekit/type_security_systems.py @@ -1,5 +1,6 @@ """Class to hold all alarm control panel accessories.""" import logging +from typing import Any from pyhap.const import CATEGORY_ALARM_SYSTEM @@ -23,7 +24,7 @@ STATE_ALARM_DISARMED, STATE_ALARM_TRIGGERED, ) -from homeassistant.core import callback +from homeassistant.core import State, callback from .accessories import TYPES, HomeAccessory from .const import ( @@ -78,10 +79,11 @@ class SecuritySystem(HomeAccessory): """Generate an SecuritySystem accessory for an alarm control panel.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a SecuritySystem accessory object.""" super().__init__(*args, category=CATEGORY_ALARM_SYSTEM) state = self.hass.states.get(self.entity_id) + assert state self._alarm_code = self.config.get(ATTR_CODE) supported_states = state.attributes.get( @@ -143,7 +145,7 @@ def __init__(self, *args): # GET to avoid an event storm after homekit startup self.async_update_state(state) - def set_security_state(self, value): + def set_security_state(self, value: int) -> None: """Move security state to value if call came from HomeKit.""" _LOGGER.debug("%s: Set security state to %d", self.entity_id, value) service = HK_TO_SERVICE[value] @@ -153,7 +155,7 @@ def set_security_state(self, value): 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 security state after state changed.""" hass_state = new_state.state if (current_state := HASS_TO_HOMEKIT_CURRENT.get(hass_state)) is not None: diff --git a/homeassistant/components/homekit/type_sensors.py b/homeassistant/components/homekit/type_sensors.py index 240cdd888d29f5..dbf2808a55a814 100644 --- a/homeassistant/components/homekit/type_sensors.py +++ b/homeassistant/components/homekit/type_sensors.py @@ -3,7 +3,7 @@ from collections.abc import Callable import logging -from typing import NamedTuple +from typing import Any, NamedTuple from pyhap.const import CATEGORY_SENSOR from pyhap.service import Service @@ -16,7 +16,7 @@ STATE_ON, UnitOfTemperature, ) -from homeassistant.core import callback +from homeassistant.core import State, callback from .accessories import TYPES, HomeAccessory from .const import ( @@ -112,10 +112,11 @@ class TemperatureSensor(HomeAccessory): Sensor entity must return temperature in °C, °F. """ - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a TemperatureSensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) + assert state serv_temp = self.add_preload_service(SERV_TEMPERATURE_SENSOR) self.char_temp = serv_temp.configure_char( CHAR_CURRENT_TEMPERATURE, value=0, properties=PROP_CELSIUS @@ -125,7 +126,7 @@ def __init__(self, *args): self.async_update_state(state) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update temperature after state changed.""" unit = new_state.attributes.get( ATTR_UNIT_OF_MEASUREMENT, UnitOfTemperature.CELSIUS @@ -142,10 +143,11 @@ def async_update_state(self, new_state): class HumiditySensor(HomeAccessory): """Generate a HumiditySensor accessory as humidity sensor.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a HumiditySensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) + assert state serv_humidity = self.add_preload_service(SERV_HUMIDITY_SENSOR) self.char_humidity = serv_humidity.configure_char( CHAR_CURRENT_HUMIDITY, value=0 @@ -155,7 +157,7 @@ def __init__(self, *args): self.async_update_state(state) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" if (humidity := convert_to_float(new_state.state)) is not None: self.char_humidity.set_value(humidity) @@ -166,18 +168,18 @@ def async_update_state(self, new_state): class AirQualitySensor(HomeAccessory): """Generate a AirQualitySensor accessory as air quality sensor.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a AirQualitySensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) - + assert state self.create_services() # Set the state so it is in sync on initial # GET to avoid an event storm after homekit startup self.async_update_state(state) - def create_services(self): + def create_services(self) -> None: """Initialize a AirQualitySensor accessory object.""" serv_air_quality = self.add_preload_service( SERV_AIR_QUALITY_SENSOR, [CHAR_AIR_PARTICULATE_DENSITY] @@ -188,7 +190,7 @@ def create_services(self): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" if (density := convert_to_float(new_state.state)) is not None: if self.char_density.value != density: @@ -203,7 +205,7 @@ def async_update_state(self, new_state): class PM10Sensor(AirQualitySensor): """Generate a PM10Sensor accessory as PM 10 sensor.""" - def create_services(self): + def create_services(self) -> None: """Override the init function for PM 10 Sensor.""" serv_air_quality = self.add_preload_service( SERV_AIR_QUALITY_SENSOR, [CHAR_PM10_DENSITY] @@ -212,7 +214,7 @@ def create_services(self): self.char_density = serv_air_quality.configure_char(CHAR_PM10_DENSITY, value=0) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" density = convert_to_float(new_state.state) if density is None: @@ -230,7 +232,7 @@ def async_update_state(self, new_state): class PM25Sensor(AirQualitySensor): """Generate a PM25Sensor accessory as PM 2.5 sensor.""" - def create_services(self): + def create_services(self) -> None: """Override the init function for PM 2.5 Sensor.""" serv_air_quality = self.add_preload_service( SERV_AIR_QUALITY_SENSOR, [CHAR_PM25_DENSITY] @@ -239,7 +241,7 @@ def create_services(self): self.char_density = serv_air_quality.configure_char(CHAR_PM25_DENSITY, value=0) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" density = convert_to_float(new_state.state) if density is None: @@ -257,7 +259,7 @@ def async_update_state(self, new_state): class NitrogenDioxideSensor(AirQualitySensor): """Generate a NitrogenDioxideSensor accessory as NO2 sensor.""" - def create_services(self): + def create_services(self) -> None: """Override the init function for PM 2.5 Sensor.""" serv_air_quality = self.add_preload_service( SERV_AIR_QUALITY_SENSOR, [CHAR_NITROGEN_DIOXIDE_DENSITY] @@ -268,7 +270,7 @@ def create_services(self): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" density = convert_to_float(new_state.state) if density is None: @@ -289,7 +291,7 @@ class VolatileOrganicCompoundsSensor(AirQualitySensor): Sensor entity must return VOC in µg/m3. """ - def create_services(self): + def create_services(self) -> None: """Override the init function for VOC Sensor.""" serv_air_quality: Service = self.add_preload_service( SERV_AIR_QUALITY_SENSOR, [CHAR_VOC_DENSITY] @@ -305,7 +307,7 @@ def create_services(self): ) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" density = convert_to_float(new_state.state) if density is None: @@ -323,10 +325,11 @@ def async_update_state(self, new_state): class CarbonMonoxideSensor(HomeAccessory): """Generate a CarbonMonoxidSensor accessory as CO sensor.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a CarbonMonoxideSensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) + assert state serv_co = self.add_preload_service( SERV_CARBON_MONOXIDE_SENSOR, [CHAR_CARBON_MONOXIDE_LEVEL, CHAR_CARBON_MONOXIDE_PEAK_LEVEL], @@ -343,7 +346,7 @@ def __init__(self, *args): self.async_update_state(state) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" if (value := convert_to_float(new_state.state)) is not None: self.char_level.set_value(value) @@ -358,10 +361,11 @@ def async_update_state(self, new_state): class CarbonDioxideSensor(HomeAccessory): """Generate a CarbonDioxideSensor accessory as CO2 sensor.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a CarbonDioxideSensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) + assert state serv_co2 = self.add_preload_service( SERV_CARBON_DIOXIDE_SENSOR, [CHAR_CARBON_DIOXIDE_LEVEL, CHAR_CARBON_DIOXIDE_PEAK_LEVEL], @@ -378,7 +382,7 @@ def __init__(self, *args): self.async_update_state(state) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" if (value := convert_to_float(new_state.state)) is not None: self.char_level.set_value(value) @@ -393,10 +397,11 @@ def async_update_state(self, new_state): class LightSensor(HomeAccessory): """Generate a LightSensor accessory as light sensor.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a LightSensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) + assert state serv_light = self.add_preload_service(SERV_LIGHT_SENSOR) self.char_light = serv_light.configure_char( CHAR_CURRENT_AMBIENT_LIGHT_LEVEL, value=0 @@ -406,7 +411,7 @@ def __init__(self, *args): self.async_update_state(state) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" if (luminance := convert_to_float(new_state.state)) is not None: self.char_light.set_value(luminance) @@ -417,10 +422,11 @@ def async_update_state(self, new_state): class BinarySensor(HomeAccessory): """Generate a BinarySensor accessory as binary sensor.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a BinarySensor accessory object.""" super().__init__(*args, category=CATEGORY_SENSOR) state = self.hass.states.get(self.entity_id) + assert state device_class = state.attributes.get(ATTR_DEVICE_CLASS) service_char = ( BINARY_SENSOR_SERVICE_MAP[device_class] @@ -439,7 +445,7 @@ def __init__(self, *args): self.async_update_state(state) @callback - def async_update_state(self, new_state): + def async_update_state(self, new_state: State) -> None: """Update accessory after state change.""" state = new_state.state detected = self.format(state in (STATE_ON, STATE_HOME))