diff --git a/homeassistant/components/denonavr/config_flow.py b/homeassistant/components/denonavr/config_flow.py index fe6c05b3aca..2ec58df8126 100644 --- a/homeassistant/components/denonavr/config_flow.py +++ b/homeassistant/components/denonavr/config_flow.py @@ -90,14 +90,14 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize the Denon AVR flow.""" - self.host = None - self.serial_number = None - self.model_name = None + self.host: str | None = None + self.serial_number: str | None = None + self.model_name: str | None = None self.timeout = DEFAULT_TIMEOUT self.show_all_sources = DEFAULT_SHOW_SOURCES self.zone2 = DEFAULT_ZONE2 self.zone3 = DEFAULT_ZONE3 - self.d_receivers = [] + self.d_receivers: list[dict[str, Any]] = [] @staticmethod @callback @@ -138,7 +138,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Handle multiple receivers found.""" - errors = {} + errors: dict[str, str] = {} if user_input is not None: self.host = user_input["select_host"] return await self.async_step_connect() @@ -169,6 +169,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Connect to the receiver.""" + assert self.host connect_denonavr = ConnectDenonAVR( self.host, self.timeout, @@ -185,6 +186,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if not success: return self.async_abort(reason="cannot_connect") receiver = connect_denonavr.receiver + assert receiver if not self.serial_number: self.serial_number = receiver.serial_number @@ -238,6 +240,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): "*", "" ) self.serial_number = discovery_info.upnp[ssdp.ATTR_UPNP_SERIAL] + assert discovery_info.ssdp_location is not None self.host = urlparse(discovery_info.ssdp_location).hostname if self.model_name in IGNORED_MODELS: @@ -260,6 +263,6 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_confirm() @staticmethod - def construct_unique_id(model_name: str, serial_number: str) -> str: + def construct_unique_id(model_name: str | None, serial_number: str | None) -> str: """Construct the unique id from the ssdp discovery or user_step.""" return f"{model_name}-{serial_number}" diff --git a/homeassistant/components/denonavr/media_player.py b/homeassistant/components/denonavr/media_player.py index 85e28c29d7c..10c3cb4f6b0 100644 --- a/homeassistant/components/denonavr/media_player.py +++ b/homeassistant/components/denonavr/media_player.py @@ -1,10 +1,11 @@ """Support for Denon AVR receivers using their HTTP interface.""" from __future__ import annotations -from collections.abc import Coroutine +from collections.abc import Awaitable, Callable, Coroutine from datetime import timedelta from functools import wraps import logging +from typing import Any, TypeVar from denonavr import DenonAVR from denonavr.const import POWER_ON @@ -15,6 +16,7 @@ from denonavr.exceptions import ( AvrTimoutError, DenonAvrError, ) +from typing_extensions import Concatenate, ParamSpec import voluptuous as vol from homeassistant.components.media_player import ( @@ -79,6 +81,10 @@ SERVICE_GET_COMMAND = "get_command" SERVICE_SET_DYNAMIC_EQ = "set_dynamic_eq" SERVICE_UPDATE_AUDYSSEY = "update_audyssey" +_DenonDeviceT = TypeVar("_DenonDeviceT", bound="DenonDevice") +_R = TypeVar("_R") +_P = ParamSpec("_P") + async def async_setup_entry( hass: HomeAssistant, @@ -131,6 +137,79 @@ async def async_setup_entry( async_add_entities(entities, update_before_add=True) +def async_log_errors( + func: Callable[Concatenate[_DenonDeviceT, _P], Awaitable[_R]], +) -> Callable[Concatenate[_DenonDeviceT, _P], Coroutine[Any, Any, _R | None]]: + """ + Log errors occurred when calling a Denon AVR receiver. + + Decorates methods of DenonDevice class. + Declaration of staticmethod for this method is at the end of this class. + """ + + @wraps(func) + async def wrapper( + self: _DenonDeviceT, *args: _P.args, **kwargs: _P.kwargs + ) -> _R | None: + # pylint: disable=protected-access + available = True + try: + return await func(self, *args, **kwargs) + except AvrTimoutError: + available = False + if self._available is True: + _LOGGER.warning( + "Timeout connecting to Denon AVR receiver at host %s. " + "Device is unavailable", + self._receiver.host, + ) + self._available = False + except AvrNetworkError: + available = False + if self._available is True: + _LOGGER.warning( + "Network error connecting to Denon AVR receiver at host %s. " + "Device is unavailable", + self._receiver.host, + ) + self._available = False + except AvrForbiddenError: + available = False + if self._available is True: + _LOGGER.warning( + "Denon AVR receiver at host %s responded with HTTP 403 error. " + "Device is unavailable. Please consider power cycling your " + "receiver", + self._receiver.host, + ) + self._available = False + except AvrCommandError as err: + available = False + _LOGGER.error( + "Command %s failed with error: %s", + func.__name__, + err, + ) + except DenonAvrError as err: + available = False + _LOGGER.error( + "Error %s occurred in method %s for Denon AVR receiver", + err, + func.__name__, + exc_info=True, + ) + finally: + if available is True and self._available is False: + _LOGGER.info( + "Denon AVR receiver at host %s is available again", + self._receiver.host, + ) + self._available = True + return None + + return wrapper + + class DenonDevice(MediaPlayerEntity): """Representation of a Denon Media Player Device.""" @@ -144,6 +223,7 @@ class DenonDevice(MediaPlayerEntity): """Initialize the device.""" self._attr_name = receiver.name self._attr_unique_id = unique_id + assert config_entry.unique_id self._attr_device_info = DeviceInfo( configuration_url=f"http://{config_entry.data[CONF_HOST]}/", identifiers={(DOMAIN, config_entry.unique_id)}, @@ -163,71 +243,6 @@ class DenonDevice(MediaPlayerEntity): ) self._available = True - def async_log_errors( - func: Coroutine, - ) -> Coroutine: - """ - Log errors occurred when calling a Denon AVR receiver. - - Decorates methods of DenonDevice class. - Declaration of staticmethod for this method is at the end of this class. - """ - - @wraps(func) - async def wrapper(self, *args, **kwargs): - # pylint: disable=protected-access - available = True - try: - return await func(self, *args, **kwargs) - except AvrTimoutError: - available = False - if self._available is True: - _LOGGER.warning( - "Timeout connecting to Denon AVR receiver at host %s. Device is unavailable", - self._receiver.host, - ) - self._available = False - except AvrNetworkError: - available = False - if self._available is True: - _LOGGER.warning( - "Network error connecting to Denon AVR receiver at host %s. Device is unavailable", - self._receiver.host, - ) - self._available = False - except AvrForbiddenError: - available = False - if self._available is True: - _LOGGER.warning( - "Denon AVR receiver at host %s responded with HTTP 403 error. Device is unavailable. Please consider power cycling your receiver", - self._receiver.host, - ) - self._available = False - except AvrCommandError as err: - available = False - _LOGGER.error( - "Command %s failed with error: %s", - func.__name__, - err, - ) - except DenonAvrError as err: - available = False - _LOGGER.error( - "Error %s occurred in method %s for Denon AVR receiver", - err, - func.__name__, - exc_info=True, - ) - finally: - if available is True and self._available is False: - _LOGGER.info( - "Denon AVR receiver at host %s is available again", - self._receiver.host, - ) - self._available = True - - return wrapper - @async_log_errors async def async_update(self) -> None: """Get the latest status information from device.""" @@ -466,8 +481,3 @@ class DenonDevice(MediaPlayerEntity): if self._update_audyssey: await self._receiver.async_update_audyssey() - - # Decorator defined before is a staticmethod - async_log_errors = staticmethod( # pylint: disable=no-staticmethod-decorator - async_log_errors - ) diff --git a/homeassistant/components/denonavr/receiver.py b/homeassistant/components/denonavr/receiver.py index 5c15468e6d4..28969d25792 100644 --- a/homeassistant/components/denonavr/receiver.py +++ b/homeassistant/components/denonavr/receiver.py @@ -23,12 +23,12 @@ class ConnectDenonAVR: ) -> None: """Initialize the class.""" self._async_client_getter = async_client_getter - self._receiver = None + self._receiver: DenonAVR | None = None self._host = host self._show_all_inputs = show_all_inputs self._timeout = timeout - self._zones = {} + self._zones: dict[str, str | None] = {} if zone2: self._zones["Zone2"] = None if zone3: @@ -42,6 +42,7 @@ class ConnectDenonAVR: async def async_connect_receiver(self) -> bool: """Connect to the DenonAVR receiver.""" await self.async_init_receiver_class() + assert self._receiver if ( self._receiver.manufacturer is None @@ -70,7 +71,7 @@ class ConnectDenonAVR: return True - async def async_init_receiver_class(self) -> bool: + async def async_init_receiver_class(self) -> None: """Initialize the DenonAVR class asynchronously.""" receiver = DenonAVR( host=self._host, diff --git a/mypy.ini b/mypy.ini index 5c5675bcb37..c4a135322fe 100644 --- a/mypy.ini +++ b/mypy.ini @@ -2644,15 +2644,6 @@ ignore_errors = true [mypy-homeassistant.components.conversation.default_agent] ignore_errors = true -[mypy-homeassistant.components.denonavr.config_flow] -ignore_errors = true - -[mypy-homeassistant.components.denonavr.media_player] -ignore_errors = true - -[mypy-homeassistant.components.denonavr.receiver] -ignore_errors = true - [mypy-homeassistant.components.evohome] ignore_errors = true diff --git a/script/hassfest/mypy_config.py b/script/hassfest/mypy_config.py index a1955c51993..e7c5e0ed129 100644 --- a/script/hassfest/mypy_config.py +++ b/script/hassfest/mypy_config.py @@ -23,9 +23,6 @@ IGNORED_MODULES: Final[list[str]] = [ "homeassistant.components.cloud.http_api", "homeassistant.components.conversation", "homeassistant.components.conversation.default_agent", - "homeassistant.components.denonavr.config_flow", - "homeassistant.components.denonavr.media_player", - "homeassistant.components.denonavr.receiver", "homeassistant.components.evohome", "homeassistant.components.evohome.climate", "homeassistant.components.evohome.water_heater",