Improve harmony typing (#107447)

homeassistant/components/harmony/__init__.py

@@ -4,7 +4,7 @@ import logging
 from homeassistant.components.remote import ATTR_ACTIVITY, ATTR_DELAY_SECS
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STOP
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import Event, HomeAssistant, callback
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 
@@ -38,7 +38,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     cancel_listener = entry.add_update_listener(_update_listener)
 
-    async def _async_on_stop(event):
+    async def _async_on_stop(event: Event) -> None:
         await data.shutdown()
 
     cancel_stop = hass.bus.async_listen(EVENT_HOMEASSISTANT_STOP, _async_on_stop)
@@ -56,11 +56,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
 async def _migrate_old_unique_ids(
     hass: HomeAssistant, entry_id: str, data: HarmonyData
-):
+) -> None:
     names_to_ids = {activity["label"]: activity["id"] for activity in data.activities}
 
     @callback
-    def _async_migrator(entity_entry: er.RegistryEntry):
+    def _async_migrator(entity_entry: er.RegistryEntry) -> dict[str, str] | None:
         # Old format for switches was {remote_unique_id}-{activity_name}
         # New format is activity_{activity_id}
         parts = entity_entry.unique_id.split("-", 1)
@@ -82,7 +82,9 @@ async def _migrate_old_unique_ids(
 
 
 @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 = dict(entry.options)
     modified = 0
     for importable_option in (ATTR_ACTIVITY, ATTR_DELAY_SECS):

homeassistant/components/harmony/config_flow.py

@@ -35,7 +35,7 @@ DATA_SCHEMA = vol.Schema(
 )
 
 
-async def validate_input(data):
+async def validate_input(data: dict[str, Any]) -> dict[str, Any]:
     """Validate the user input allows us to connect.
 
     Data has the keys from DATA_SCHEMA with values provided by the user.
@@ -60,9 +60,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         """Initialize the Harmony config flow."""
         self.harmony_config: 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:
         """Handle the initial step."""
-        errors = {}
+        errors: dict[str, str] = {}
         if user_input is not None:
             try:
                 validated = await validate_input(user_input)
@@ -116,9 +118,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         self.harmony_config[UNIQUE_ID] = unique_id
         return await self.async_step_link()
 
-    async def async_step_link(self, user_input=None):
+    async def async_step_link(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Attempt to link with the Harmony."""
-        errors = {}
+        errors: dict[str, str] = {}
 
         if user_input is not None:
             # Everything was validated in async_step_ssdp
@@ -145,7 +149,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         """Get the options flow for this handler."""
         return OptionsFlowHandler(config_entry)
 
-    async def _async_create_entry_from_valid_input(self, validated, user_input):
+    async def _async_create_entry_from_valid_input(
+        self, validated: dict[str, Any], user_input: dict[str, Any]
+    ) -> FlowResult:
         """Single path to create the config entry from validated input."""
 
         data = {
@@ -159,8 +165,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         return self.async_create_entry(title=validated[CONF_NAME], data=data)
 
 
-def _options_from_user_input(user_input):
-    options = {}
+def _options_from_user_input(user_input: dict[str, Any]) -> dict[str, Any]:
+    options: dict[str, Any] = {}
     if ATTR_ACTIVITY in user_input:
         options[ATTR_ACTIVITY] = user_input[ATTR_ACTIVITY]
     if ATTR_DELAY_SECS in user_input:
@@ -175,7 +181,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
         """Initialize options flow."""
         self.config_entry = config_entry
 
-    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 user_input is not None:
             return self.async_create_entry(title="", data=user_input)

homeassistant/components/harmony/data.py

@@ -45,7 +45,7 @@ class HarmonyData(HarmonySubscriberMixin):
         ]
 
     @property
-    def activity_names(self):
+    def activity_names(self) -> list[str]:
         """Names of all the remotes activities."""
         activity_infos = self.activities
         activities = [activity["label"] for activity in activity_infos]
@@ -61,7 +61,7 @@ class HarmonyData(HarmonySubscriberMixin):
         return devices
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Return the Harmony device's name."""
         return self._name
 
@@ -138,7 +138,7 @@ class HarmonyData(HarmonySubscriberMixin):
                 f"{self._name}: Unable to connect to HUB at: {self._address}:8088"
             )
 
-    async def shutdown(self):
+    async def shutdown(self) -> None:
         """Close connection on shutdown."""
         _LOGGER.debug("%s: Closing Harmony Hub", self._name)
         try:
@@ -146,7 +146,7 @@ class HarmonyData(HarmonySubscriberMixin):
         except aioexc.TimeOut:
             _LOGGER.warning("%s: Disconnect timed-out", self._name)
 
-    async def async_start_activity(self, activity: str):
+    async def async_start_activity(self, activity: str) -> None:
         """Start an activity from the Harmony device."""
 
         if not activity:
@@ -189,7 +189,7 @@ class HarmonyData(HarmonySubscriberMixin):
             _LOGGER.error("%s: Starting activity %s timed-out", self.name, activity)
             self.async_unlock_start_activity()
 
-    async def async_power_off(self):
+    async def async_power_off(self) -> None:
         """Start the PowerOff activity."""
         _LOGGER.debug("%s: Turn Off", self.name)
         try:
@@ -204,7 +204,7 @@ class HarmonyData(HarmonySubscriberMixin):
         num_repeats: int,
         delay_secs: float,
         hold_secs: float,
-    ):
+    ) -> None:
         """Send a list of commands to one device."""
         device_id = None
         if device.isdigit():
@@ -259,7 +259,7 @@ class HarmonyData(HarmonySubscriberMixin):
                 result.msg,
             )
 
-    async def change_channel(self, channel: int):
+    async def change_channel(self, channel: int) -> None:
         """Change the channel using Harmony remote."""
         _LOGGER.debug("%s: Changing channel to %s", self.name, channel)
         try:

homeassistant/components/harmony/entity.py

@@ -1,4 +1,8 @@
 """Base class Harmony entities."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import datetime
 import logging
 
 from homeassistant.helpers.entity import Entity
@@ -17,7 +21,7 @@ class HarmonyEntity(Entity):
     def __init__(self, data: HarmonyData) -> None:
         """Initialize the Harmony base entity."""
         super().__init__()
-        self._unsub_mark_disconnected = None
+        self._unsub_mark_disconnected: Callable[[], None] | None = None
         self._name = data.name
         self._data = data
         self._attr_should_poll = False
@@ -27,14 +31,14 @@ class HarmonyEntity(Entity):
         """Return True if we're connected to the Hub, otherwise False."""
         return self._data.available
 
-    async def async_got_connected(self, _=None):
+    async def async_got_connected(self, _: str | None = None) -> None:
         """Notification that we're connected to the HUB."""
         _LOGGER.debug("%s: connected to the HUB", self._name)
         self.async_write_ha_state()
 
         self._clear_disconnection_delay()
 
-    async def async_got_disconnected(self, _=None):
+    async def async_got_disconnected(self, _: str | None = None) -> None:
         """Notification that we're disconnected from the HUB."""
         _LOGGER.debug("%s: disconnected from the HUB", self._name)
         # We're going to wait for 10 seconds before announcing we're
@@ -43,12 +47,12 @@ class HarmonyEntity(Entity):
             self.hass, TIME_MARK_DISCONNECTED, self._mark_disconnected_if_unavailable
         )
 
-    def _clear_disconnection_delay(self):
+    def _clear_disconnection_delay(self) -> None:
         if self._unsub_mark_disconnected:
             self._unsub_mark_disconnected()
             self._unsub_mark_disconnected = None
 
-    def _mark_disconnected_if_unavailable(self, _):
+    def _mark_disconnected_if_unavailable(self, _: datetime) -> None:
         self._unsub_mark_disconnected = None
         if not self.available:
             # Still disconnected. Let the state engine know.

homeassistant/components/harmony/remote.py

@@ -1,4 +1,6 @@
 """Support for Harmony Hub devices."""
+from __future__ import annotations
+
 from collections.abc import Iterable
 import json
 import logging
@@ -36,6 +38,7 @@ from .const import (
     SERVICE_CHANGE_CHANNEL,
     SERVICE_SYNC,
 )
+from .data import HarmonyData
 from .entity import HarmonyEntity
 from .subscriber import HarmonyCallback
 
@@ -56,12 +59,12 @@ async def async_setup_entry(
 ) -> None:
     """Set up the Harmony config entry."""
 
-    data = hass.data[DOMAIN][entry.entry_id][HARMONY_DATA]
+    data: HarmonyData = hass.data[DOMAIN][entry.entry_id][HARMONY_DATA]
 
     _LOGGER.debug("HarmonyData : %s", data)
 
-    default_activity = entry.options.get(ATTR_ACTIVITY)
-    delay_secs = entry.options.get(ATTR_DELAY_SECS, DEFAULT_DELAY_SECS)
+    default_activity: str | None = entry.options.get(ATTR_ACTIVITY)
+    delay_secs: float = entry.options.get(ATTR_DELAY_SECS, DEFAULT_DELAY_SECS)
 
     harmony_conf_file = hass.config.path(f"harmony_{entry.unique_id}.conf")
     device = HarmonyRemote(data, default_activity, delay_secs, harmony_conf_file)
@@ -84,10 +87,12 @@ class HarmonyRemote(HarmonyEntity, RemoteEntity, RestoreEntity):
 
     _attr_supported_features = RemoteEntityFeature.ACTIVITY
 
-    def __init__(self, data, activity, delay_secs, out_path):
+    def __init__(
+        self, data: HarmonyData, activity: str | None, delay_secs: float, out_path: str
+    ) -> None:
         """Initialize HarmonyRemote class."""
         super().__init__(data=data)
-        self._state = None
+        self._state: bool | None = None
         self._current_activity = ACTIVITY_POWER_OFF
         self.default_activity = activity
         self._activity_starting = None
@@ -99,7 +104,7 @@ class HarmonyRemote(HarmonyEntity, RemoteEntity, RestoreEntity):
         self._attr_device_info = self._data.device_info(DOMAIN)
         self._attr_name = data.name
 
-    async def _async_update_options(self, data):
+    async def _async_update_options(self, data: dict[str, Any]) -> None:
         """Change options when the options flow does."""
         if ATTR_DELAY_SECS in data:
             self.delay_secs = data[ATTR_DELAY_SECS]
@@ -170,7 +175,7 @@ class HarmonyRemote(HarmonyEntity, RemoteEntity, RestoreEntity):
         return self._data.activity_names
 
     @property
-    def extra_state_attributes(self):
+    def extra_state_attributes(self) -> dict[str, Any]:
         """Add platform specific attributes."""
         return {
             ATTR_ACTIVITY_STARTING: self._activity_starting,
@@ -179,7 +184,7 @@ class HarmonyRemote(HarmonyEntity, RemoteEntity, RestoreEntity):
         }
 
     @property
-    def is_on(self):
+    def is_on(self) -> bool:
         """Return False if PowerOff is the current activity, otherwise True."""
         return self._current_activity not in [None, "PowerOff"]
 
@@ -201,7 +206,7 @@ class HarmonyRemote(HarmonyEntity, RemoteEntity, RestoreEntity):
         self._state = bool(activity_id != -1)
         self.async_write_ha_state()
 
-    async def async_new_config(self, _=None):
+    async def async_new_config(self, _: dict | None = None) -> None:
         """Call for updating the current activity."""
         _LOGGER.debug("%s: configuration has been updated", self.name)
         self.async_new_activity(self._data.current_activity)
@@ -242,16 +247,16 @@ class HarmonyRemote(HarmonyEntity, RemoteEntity, RestoreEntity):
             command, device, num_repeats, delay_secs, hold_secs
         )
 
-    async def change_channel(self, channel):
+    async def change_channel(self, channel: int) -> None:
         """Change the channel using Harmony remote."""
         await self._data.change_channel(channel)
 
-    async def sync(self):
+    async def sync(self) -> None:
         """Sync the Harmony device with the web service."""
         if await self._data.sync():
             await self.hass.async_add_executor_job(self.write_config_file)
 
-    def write_config_file(self):
+    def write_config_file(self) -> None:
         """Write Harmony configuration file.
 
         This is a handy way for users to figure out the available commands for automations.

homeassistant/components/harmony/select.py

@@ -23,7 +23,7 @@ async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
 ) -> None:
     """Set up harmony activities select."""
-    data = hass.data[DOMAIN][entry.entry_id][HARMONY_DATA]
+    data: HarmonyData = hass.data[DOMAIN][entry.entry_id][HARMONY_DATA]
     _LOGGER.debug("creating select for %s hub activities", entry.data[CONF_NAME])
     async_add_entities(
         [HarmonyActivitySelect(f"{entry.data[CONF_NAME]} Activities", data)]
@@ -85,5 +85,5 @@ class HarmonyActivitySelect(HarmonyEntity, SelectEntity):
         )
 
     @callback
-    def _async_activity_update(self, activity_info: tuple):
+    def _async_activity_update(self, activity_info: tuple) -> None:
         self.async_write_ha_state()

homeassistant/components/harmony/subscriber.py

@@ -34,7 +34,7 @@ class HarmonySubscriberMixin:
         self._subscriptions: list[HarmonyCallback] = []
         self._activity_lock = asyncio.Lock()
 
-    async def async_lock_start_activity(self):
+    async def async_lock_start_activity(self) -> None:
         """Acquire the lock."""
         await self._activity_lock.acquire()
 
@@ -59,17 +59,17 @@ class HarmonySubscriberMixin:
         """Remove a callback subscriber."""
         self._subscriptions.remove(update_callback)
 
-    def _config_updated(self, _=None) -> None:
+    def _config_updated(self, _: dict | None = None) -> None:
         _LOGGER.debug("config_updated")
         self._call_callbacks("config_updated")
 
-    def _connected(self, _=None) -> None:
+    def _connected(self, _: str | None = None) -> None:
         _LOGGER.debug("connected")
         self.async_unlock_start_activity()
         self._available = True
         self._call_callbacks("connected")
 
-    def _disconnected(self, _=None) -> None:
+    def _disconnected(self, _: str | None = None) -> None:
         _LOGGER.debug("disconnected")
         self.async_unlock_start_activity()
         self._available = False

homeassistant/components/harmony/switch.py

@@ -23,7 +23,7 @@ async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
 ) -> None:
     """Set up harmony activity switches."""
-    data = hass.data[DOMAIN][entry.entry_id][HARMONY_DATA]
+    data: HarmonyData = hass.data[DOMAIN][entry.entry_id][HARMONY_DATA]
     activities = data.activities
 
     switches = []
@@ -49,7 +49,7 @@ class HarmonyActivitySwitch(HarmonyEntity, SwitchEntity):
         self._attr_device_info = self._data.device_info(DOMAIN)
 
     @property
-    def is_on(self):
+    def is_on(self) -> bool:
         """Return if the current activity is the one for this switch."""
         _, activity_name = self._data.current_activity
         return activity_name == self._activity_name
@@ -111,5 +111,5 @@ class HarmonyActivitySwitch(HarmonyEntity, SwitchEntity):
             )
 
     @callback
-    def _async_activity_update(self, activity_info: tuple):
+    def _async_activity_update(self, activity_info: tuple) -> None:
         self.async_write_ha_state()

homeassistant/components/harmony/util.py

@@ -25,7 +25,7 @@ def find_best_name_for_remote(data: dict, harmony: HarmonyAPI):
     return data[CONF_NAME]
 
 
-async def get_harmony_client_if_available(ip_address: str):
+async def get_harmony_client_if_available(ip_address: str) -> HarmonyAPI | None:
     """Connect to a harmony hub and fetch info."""
     harmony = HarmonyAPI(ip_address=ip_address)