Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove denonavr from mypy ignore list ()

pull/74604/head
epenet 2022-07-07 12:14:46 +02:00 committed by GitHub
parent 4604694255
commit f19c542d6d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 94 additions and 92 deletions
homeassistant/components/denonavr
script/hassfest

15
homeassistant/components/denonavr/config_flow.py
View File

@ -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}"

152
homeassistant/components/denonavr/media_player.py
View File

@ -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
)

7
homeassistant/components/denonavr/receiver.py
View File

@ -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,

9
mypy.ini
View File

@ -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

3
script/hassfest/mypy_config.py
View File

@ -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",