From fd52172c336b833ce156dd9b3b880bef148d433c Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Sun, 7 Jan 2024 20:35:55 +0100 Subject: [PATCH] Improve harmony typing (#107447) --- homeassistant/components/harmony/__init__.py | 12 ++++---- .../components/harmony/config_flow.py | 26 +++++++++++------ homeassistant/components/harmony/data.py | 14 ++++----- homeassistant/components/harmony/entity.py | 14 +++++---- homeassistant/components/harmony/remote.py | 29 +++++++++++-------- homeassistant/components/harmony/select.py | 4 +-- .../components/harmony/subscriber.py | 8 ++--- homeassistant/components/harmony/switch.py | 6 ++-- homeassistant/components/harmony/util.py | 2 +- 9 files changed, 67 insertions(+), 48 deletions(-) diff --git a/homeassistant/components/harmony/__init__.py b/homeassistant/components/harmony/__init__.py index d861068629ff..327dbad343b0 100644 --- a/homeassistant/components/harmony/__init__.py +++ b/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): diff --git a/homeassistant/components/harmony/config_flow.py b/homeassistant/components/harmony/config_flow.py index f74a19425aba..ad041e75f1a1 100644 --- a/homeassistant/components/harmony/config_flow.py +++ b/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) diff --git a/homeassistant/components/harmony/data.py b/homeassistant/components/harmony/data.py index a1b11189a044..44c0fde19c19 100644 --- a/homeassistant/components/harmony/data.py +++ b/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: diff --git a/homeassistant/components/harmony/entity.py b/homeassistant/components/harmony/entity.py index 24c72a771e74..b1b1599a16cc 100644 --- a/homeassistant/components/harmony/entity.py +++ b/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. diff --git a/homeassistant/components/harmony/remote.py b/homeassistant/components/harmony/remote.py index c1e85c867874..863c3fe5c56a 100644 --- a/homeassistant/components/harmony/remote.py +++ b/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. diff --git a/homeassistant/components/harmony/select.py b/homeassistant/components/harmony/select.py index 0ed3f0ca2750..e98a15c788fd 100644 --- a/homeassistant/components/harmony/select.py +++ b/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() diff --git a/homeassistant/components/harmony/subscriber.py b/homeassistant/components/harmony/subscriber.py index 4804253151f2..8a47e437e177 100644 --- a/homeassistant/components/harmony/subscriber.py +++ b/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 diff --git a/homeassistant/components/harmony/switch.py b/homeassistant/components/harmony/switch.py index 2d072f11f2c0..c5bba39eb95b 100644 --- a/homeassistant/components/harmony/switch.py +++ b/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() diff --git a/homeassistant/components/harmony/util.py b/homeassistant/components/harmony/util.py index 3f126f22f3cb..0bfee32b4146 100644 --- a/homeassistant/components/harmony/util.py +++ b/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)