Add some more typing to HomeKit (#101959)

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,42 +280,42 @@ 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
         )
         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
 

homeassistant/components/homekit/type_remotes.py

@@ -219,31 +219,33 @@ 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,
             ATTR_CURRENT_ACTIVITY,
             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

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: