jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 00:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Going async with denonavr (#47920)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Oliver 2021-04-02 19:47:16 +02:00 committed by GitHub
parent 212d9aa748
commit eed3bfc762
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 305 additions and 401 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -174,6 +174,7 @@ omit =
homeassistant/components/deluge/sensor.py
homeassistant/components/deluge/switch.py
homeassistant/components/denon/media_player.py
homeassistant/components/denonavr/__init__.py
homeassistant/components/denonavr/media_player.py
homeassistant/components/denonavr/receiver.py
homeassistant/components/deutsche_bahn/sensor.py

47
homeassistant/components/denonavr/__init__.py
View File

@ -1,13 +1,13 @@
"""The denonavr component."""
import logging
import voluptuous as vol
from denonavr.exceptions import AvrNetworkError, AvrTimoutError
from homeassistant import config_entries, core
from homeassistant.const import ATTR_COMMAND, ATTR_ENTITY_ID, CONF_HOST
from homeassistant.const import CONF_HOST
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.httpx_client import get_async_client
from .config_flow import (
CONF_SHOW_ALL_SOURCES,
@ -23,34 +23,9 @@ from .receiver import ConnectDenonAVR
CONF_RECEIVER = "receiver"
UNDO_UPDATE_LISTENER = "undo_update_listener"
SERVICE_GET_COMMAND = "get_command"
_LOGGER = logging.getLogger(__name__)
CALL_SCHEMA = vol.Schema({vol.Required(ATTR_ENTITY_ID): cv.comp_entity_ids})
GET_COMMAND_SCHEMA = CALL_SCHEMA.extend({vol.Required(ATTR_COMMAND): cv.string})
SERVICE_TO_METHOD = {
SERVICE_GET_COMMAND: {"method": "get_command", "schema": GET_COMMAND_SCHEMA}
}
def setup(hass: core.HomeAssistant, config: dict):
"""Set up the denonavr platform."""
def service_handler(service):
method = SERVICE_TO_METHOD.get(service.service)
data = service.data.copy()
data["method"] = method["method"]
dispatcher_send(hass, DOMAIN, data)
for service in SERVICE_TO_METHOD:
schema = SERVICE_TO_METHOD[service]["schema"]
hass.services.register(DOMAIN, service, service_handler, schema=schema)
return True
async def async_setup_entry(
hass: core.HomeAssistant, entry: config_entries.ConfigEntry
@ -60,15 +35,18 @@ async def async_setup_entry(
# Connect to receiver
connect_denonavr = ConnectDenonAVR(
hass,
entry.data[CONF_HOST],
DEFAULT_TIMEOUT,
entry.options.get(CONF_SHOW_ALL_SOURCES, DEFAULT_SHOW_SOURCES),
entry.options.get(CONF_ZONE2, DEFAULT_ZONE2),
entry.options.get(CONF_ZONE3, DEFAULT_ZONE3),
lambda: get_async_client(hass),
entry.state,
)
if not await connect_denonavr.async_connect_receiver():
raise ConfigEntryNotReady
try:
await connect_denonavr.async_connect_receiver()
except (AvrNetworkError, AvrTimoutError) as ex:
raise ConfigEntryNotReady from ex
receiver = connect_denonavr.receiver
undo_listener = entry.add_update_listener(update_listener)
@ -98,8 +76,9 @@ async def async_unload_entry(
# Remove zone2 and zone3 entities if needed
entity_registry = await er.async_get_registry(hass)
entries = er.async_entries_for_config_entry(entity_registry, config_entry.entry_id)
zone2_id = f"{config_entry.unique_id}-Zone2"
zone3_id = f"{config_entry.unique_id}-Zone3"
unique_id = config_entry.unique_id or config_entry.entry_id
zone2_id = f"{unique_id}-Zone2"
zone3_id = f"{unique_id}-Zone3"
for entry in entries:
if entry.unique_id == zone2_id and not config_entry.options.get(CONF_ZONE2):
entity_registry.async_remove(entry.entity_id)

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

@ -1,17 +1,17 @@
"""Config flow to configure Denon AVR receivers using their HTTP interface."""
from functools import partial
import logging
from typing import Any, Dict, Optional
from urllib.parse import urlparse
import denonavr
from getmac import get_mac_address
from denonavr.exceptions import AvrNetworkError, AvrTimoutError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_TYPE
from homeassistant.const import CONF_HOST, CONF_TYPE
from homeassistant.core import callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.httpx_client import get_async_client
from .receiver import ConnectDenonAVR
@ -44,7 +44,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
"""Init object."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: Optional[Dict[str, Any]] = None):
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
@ -90,11 +90,13 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry) -> OptionsFlowHandler:
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow."""
return OptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
async def async_step_user(self, user_input: Optional[Dict[str, Any]] = None):
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
@ -105,7 +107,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_connect()
# discovery using denonavr library
self.d_receivers = await self.hass.async_add_executor_job(denonavr.discover)
self.d_receivers = await denonavr.async_discover()
# More than one receiver could be discovered by that method
if len(self.d_receivers) == 1:
self.host = self.d_receivers[0]["host"]
@ -120,7 +122,9 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=CONFIG_SCHEMA, errors=errors
)
async def async_step_select(self, user_input=None):
async def async_step_select(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Handle multiple receivers found."""
errors = {}
if user_input is not None:
@ -139,29 +143,37 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
step_id="select", data_schema=select_scheme, errors=errors
)
async def async_step_confirm(self, user_input=None):
async def async_step_confirm(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Allow the user to confirm adding the device."""
if user_input is not None:
return await self.async_step_connect()
self._set_confirm_only()
return self.async_show_form(step_id="confirm")
async def async_step_connect(self, user_input=None):
async def async_step_connect(
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Connect to the receiver."""
connect_denonavr = ConnectDenonAVR(
self.hass,
self.host,
self.timeout,
self.show_all_sources,
self.zone2,
self.zone3,
lambda: get_async_client(self.hass),
)
if not await connect_denonavr.async_connect_receiver():
try:
success = await connect_denonavr.async_connect_receiver()
except (AvrNetworkError, AvrTimoutError):
success = False
if not success:
return self.async_abort(reason="cannot_connect")
receiver = connect_denonavr.receiver
mac_address = await self.async_get_mac(self.host)
if not self.serial_number:
self.serial_number = receiver.serial_number
if not self.model_name:
@ -185,7 +197,6 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
title=receiver.name,
data={
CONF_HOST: self.host,
CONF_MAC: mac_address,
CONF_TYPE: receiver.receiver_type,
CONF_MODEL: self.model_name,
CONF_MANUFACTURER: receiver.manufacturer,
@ -193,7 +204,7 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
},
)
async def async_step_ssdp(self, discovery_info):
async def async_step_ssdp(self, discovery_info: Dict[str, Any]) -> Dict[str, Any]:
"""Handle a discovered Denon AVR.
This flow is triggered by the SSDP component. It will check if the
@ -235,24 +246,6 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm()
@staticmethod
def construct_unique_id(model_name, serial_number):
def construct_unique_id(model_name: str, serial_number: str) -> str:
"""Construct the unique id from the ssdp discovery or user_step."""
return f"{model_name}-{serial_number}"
async def async_get_mac(self, host):
"""Get the mac address of the DenonAVR receiver."""
try:
mac_address = await self.hass.async_add_executor_job(
partial(get_mac_address, **{"ip": host})
)
if not mac_address:
mac_address = await self.hass.async_add_executor_job(
partial(get_mac_address, **{"hostname": host})
)
except Exception as err: # pylint: disable=broad-except
_LOGGER.error("Unable to get mac address: %s", err)
mac_address = None
if mac_address is not None:
mac_address = format_mac(mac_address)
return mac_address

2
homeassistant/components/denonavr/manifest.json
View File

@ -3,7 +3,7 @@
"name": "Denon AVR Network Receivers",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/denonavr",
"requirements": ["denonavr==0.9.10", "getmac==0.8.2"],
"requirements": ["denonavr==0.10.5"],
"codeowners": ["@scarface-4711", "@starkillerOG"],
"ssdp": [
{

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

@ -1,8 +1,22 @@
"""Support for Denon AVR receivers using their HTTP interface."""
from contextlib import suppress
from datetime import timedelta
from functools import wraps
import logging
from typing import Coroutine
from denonavr import DenonAVR
from denonavr.const import POWER_ON
from denonavr.exceptions import (
AvrCommandError,
AvrForbiddenError,
AvrNetworkError,
AvrTimoutError,
DenonAvrError,
)
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.media_player import MediaPlayerEntity
from homeassistant.components.media_player.const import (
MEDIA_TYPE_CHANNEL,
@ -20,18 +34,9 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_SET,
SUPPORT_VOLUME_STEP,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_MAC,
ENTITY_MATCH_ALL,
ENTITY_MATCH_NONE,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
STATE_PLAYING,
)
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.const import ATTR_COMMAND, STATE_PAUSED, STATE_PLAYING
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from . import CONF_RECEIVER
from .config_flow import (
@ -64,8 +69,18 @@ SUPPORT_MEDIA_MODES = (
| SUPPORT_PLAY
)
SCAN_INTERVAL = timedelta(seconds=10)
PARALLEL_UPDATES = 1
async def async_setup_entry(hass, config_entry, async_add_entities):
# Services
SERVICE_GET_COMMAND = "get_command"
async def async_setup_entry(
hass: HomeAssistant,
config_entry: config_entries.ConfigEntry,
async_add_entities: entity_platform.EntityPlatform.async_add_entities,
):
"""Set up the DenonAVR receiver from a config entry."""
entities = []
receiver = hass.data[DOMAIN][config_entry.entry_id][CONF_RECEIVER]
@ -73,93 +88,116 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
if config_entry.data[CONF_SERIAL_NUMBER] is not None:
unique_id = f"{config_entry.unique_id}-{receiver_zone.zone}"
else:
unique_id = None
unique_id = f"{config_entry.entry_id}-{receiver_zone.zone}"
await receiver_zone.async_setup()
entities.append(DenonDevice(receiver_zone, unique_id, config_entry))
_LOGGER.debug(
"%s receiver at host %s initialized", receiver.manufacturer, receiver.host
)
async_add_entities(entities)
# Register additional services
platform = entity_platform.current_platform.get()
platform.async_register_entity_service(
SERVICE_GET_COMMAND,
{vol.Required(ATTR_COMMAND): cv.string},
f"async_{SERVICE_GET_COMMAND}",
)
async_add_entities(entities, update_before_add=True)
class DenonDevice(MediaPlayerEntity):
"""Representation of a Denon Media Player Device."""
def __init__(self, receiver, unique_id, config_entry):
def __init__(
self,
receiver: DenonAVR,
unique_id: str,
config_entry: config_entries.ConfigEntry,
):
"""Initialize the device."""
self._receiver = receiver
self._name = self._receiver.name
self._unique_id = unique_id
self._config_entry = config_entry
self._muted = self._receiver.muted
self._volume = self._receiver.volume
self._current_source = self._receiver.input_func
self._source_list = self._receiver.input_func_list
self._state = self._receiver.state
self._power = self._receiver.power
self._media_image_url = self._receiver.image_url
self._title = self._receiver.title
self._artist = self._receiver.artist
self._album = self._receiver.album
self._band = self._receiver.band
self._frequency = self._receiver.frequency
self._station = self._receiver.station
self._sound_mode_support = self._receiver.support_sound_mode
if self._sound_mode_support:
self._sound_mode = self._receiver.sound_mode
self._sound_mode_raw = self._receiver.sound_mode_raw
self._sound_mode_list = self._receiver.sound_mode_list
else:
self._sound_mode = None
self._sound_mode_raw = None
self._sound_mode_list = None
self._supported_features_base = SUPPORT_DENON
self._supported_features_base |= (
self._sound_mode_support and SUPPORT_SELECT_SOUND_MODE
self._receiver.support_sound_mode and SUPPORT_SELECT_SOUND_MODE
)
self._available = True
async def async_added_to_hass(self):
"""Register signal handler."""
self.async_on_remove(
async_dispatcher_connect(self.hass, DOMAIN, self.signal_handler)
)
def async_log_errors( # pylint: disable=no-self-argument
func: Coroutine,
) -> Coroutine:
"""
Log errors occurred when calling a Denon AVR receiver.
def signal_handler(self, data):
"""Handle domain-specific signal by calling appropriate method."""
entity_ids = data[ATTR_ENTITY_ID]
Decorates methods of DenonDevice class.
Declaration of staticmethod for this method is at the end of this class.
"""
if entity_ids == ENTITY_MATCH_NONE:
return
@wraps(func)
async def wrapper(self, *args, **kwargs):
# pylint: disable=protected-access
available = True
try:
return await func(self, *args, **kwargs) # pylint: disable=not-callable
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:
_LOGGER.error(
"Command %s failed with error: %s",
func.__name__,
err,
)
except DenonAvrError as err:
_LOGGER.error(
"Error %s occurred in method %s for Denon AVR receiver",
err,
func.__name__, # pylint: disable=no-member
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
if entity_ids == ENTITY_MATCH_ALL or self.entity_id in entity_ids:
params = {
key: value
for key, value in data.items()
if key not in ["entity_id", "method"]
}
getattr(self, data["method"])(**params)
return wrapper
def update(self):
@async_log_errors
async def async_update(self) -> None:
"""Get the latest status information from device."""
self._receiver.update()
self._name = self._receiver.name
self._muted = self._receiver.muted
self._volume = self._receiver.volume
self._current_source = self._receiver.input_func
self._source_list = self._receiver.input_func_list
self._state = self._receiver.state
self._power = self._receiver.power
self._media_image_url = self._receiver.image_url
self._title = self._receiver.title
self._artist = self._receiver.artist
self._album = self._receiver.album
self._band = self._receiver.band
self._frequency = self._receiver.frequency
self._station = self._receiver.station
if self._sound_mode_support:
self._sound_mode = self._receiver.sound_mode
self._sound_mode_raw = self._receiver.sound_mode_raw
await self._receiver.async_update()
@property
def available(self):
"""Return True if entity is available."""
return self._available
@property
def unique_id(self):
@ -177,60 +215,59 @@ class DenonDevice(MediaPlayerEntity):
"manufacturer": self._config_entry.data[CONF_MANUFACTURER],
"name": self._config_entry.title,
"model": f"{self._config_entry.data[CONF_MODEL]}-{self._config_entry.data[CONF_TYPE]}",
"serial_number": self._config_entry.data[CONF_SERIAL_NUMBER],
}
if self._config_entry.data[CONF_MAC] is not None:
device_info["connections"] = {
(dr.CONNECTION_NETWORK_MAC, self._config_entry.data[CONF_MAC])
}
return device_info
@property
def name(self):
"""Return the name of the device."""
return self._name
return self._receiver.name
@property
def state(self):
"""Return the state of the device."""
return self._state
return self._receiver.state
@property
def is_volume_muted(self):
"""Return boolean if volume is currently muted."""
return self._muted
return self._receiver.muted
@property
def volume_level(self):
"""Volume level of the media player (0..1)."""
# Volume is sent in a format like -50.0. Minimum is -80.0,
# maximum is 18.0
return (float(self._volume) + 80) / 100
if self._receiver.volume is None:
return None
return (float(self._receiver.volume) + 80) / 100
@property
def source(self):
"""Return the current input source."""
return self._current_source
return self._receiver.input_func
@property
def source_list(self):
"""Return a list of available input sources."""
return self._source_list
return self._receiver.input_func_list
@property
def sound_mode(self):
"""Return the current matched sound mode."""
return self._sound_mode
return self._receiver.sound_mode
@property
def sound_mode_list(self):
"""Return a list of available sound modes."""
return self._sound_mode_list
return self._receiver.sound_mode_list
@property
def supported_features(self):
"""Flag media player features that are supported."""
if self._current_source in self._receiver.netaudio_func_list:
if self._receiver.input_func in self._receiver.netaudio_func_list:
return self._supported_features_base | SUPPORT_MEDIA_MODES
return self._supported_features_base
@ -242,7 +279,10 @@ class DenonDevice(MediaPlayerEntity):
@property
def media_content_type(self):
"""Content type of current playing media."""
if self._state == STATE_PLAYING or self._state == STATE_PAUSED:
if (
self._receiver.state == STATE_PLAYING
or self._receiver.state == STATE_PAUSED
):
return MEDIA_TYPE_MUSIC
return MEDIA_TYPE_CHANNEL
@ -254,32 +294,32 @@ class DenonDevice(MediaPlayerEntity):
@property
def media_image_url(self):
"""Image url of current playing media."""
if self._current_source in self._receiver.playing_func_list:
return self._media_image_url
if self._receiver.input_func in self._receiver.playing_func_list:
return self._receiver.image_url
return None
@property
def media_title(self):
"""Title of current playing media."""
if self._current_source not in self._receiver.playing_func_list:
return self._current_source
if self._title is not None:
return self._title
return self._frequency
if self._receiver.input_func not in self._receiver.playing_func_list:
return self._receiver.input_func
if self._receiver.title is not None:
return self._receiver.title
return self._receiver.frequency
@property
def media_artist(self):
"""Artist of current playing media, music track only."""
if self._artist is not None:
return self._artist
return self._band
if self._receiver.artist is not None:
return self._receiver.artist
return self._receiver.band
@property
def media_album_name(self):
"""Album name of current playing media, music track only."""
if self._album is not None:
return self._album
return self._station
if self._receiver.album is not None:
return self._receiver.album
return self._receiver.station
@property
def media_album_artist(self):
@ -310,77 +350,92 @@ class DenonDevice(MediaPlayerEntity):
def extra_state_attributes(self):
"""Return device specific state attributes."""
if (
self._sound_mode_raw is not None
and self._sound_mode_support
and self._power == "ON"
self._receiver.sound_mode_raw is not None
and self._receiver.support_sound_mode
and self._receiver.power == POWER_ON
):
return {ATTR_SOUND_MODE_RAW: self._sound_mode_raw}
return {ATTR_SOUND_MODE_RAW: self._receiver.sound_mode_raw}
return {}
def media_play_pause(self):
@async_log_errors
async def async_media_play_pause(self):
"""Play or pause the media player."""
return self._receiver.toggle_play_pause()
await self._receiver.async_toggle_play_pause()
def media_play(self):
@async_log_errors
async def async_media_play(self):
"""Send play command."""
return self._receiver.play()
await self._receiver.async_play()
def media_pause(self):
@async_log_errors
async def async_media_pause(self):
"""Send pause command."""
return self._receiver.pause()
await self._receiver.async_pause()
def media_previous_track(self):
@async_log_errors
async def async_media_previous_track(self):
"""Send previous track command."""
return self._receiver.previous_track()
await self._receiver.async_previous_track()
def media_next_track(self):
@async_log_errors
async def async_media_next_track(self):
"""Send next track command."""
return self._receiver.next_track()
await self._receiver.async_next_track()
def select_source(self, source):
@async_log_errors
async def async_select_source(self, source: str):
"""Select input source."""
# Ensure that the AVR is turned on, which is necessary for input
# switch to work.
self.turn_on()
return self._receiver.set_input_func(source)
await self.async_turn_on()
await self._receiver.async_set_input_func(source)
def select_sound_mode(self, sound_mode):
@async_log_errors
async def async_select_sound_mode(self, sound_mode: str):
"""Select sound mode."""
return self._receiver.set_sound_mode(sound_mode)
await self._receiver.async_set_sound_mode(sound_mode)
def turn_on(self):
@async_log_errors
async def async_turn_on(self):
"""Turn on media player."""
if self._receiver.power_on():
self._state = STATE_ON
await self._receiver.async_power_on()
def turn_off(self):
@async_log_errors
async def async_turn_off(self):
"""Turn off media player."""
if self._receiver.power_off():
self._state = STATE_OFF
await self._receiver.async_power_off()
def volume_up(self):
@async_log_errors
async def async_volume_up(self):
"""Volume up the media player."""
return self._receiver.volume_up()
await self._receiver.async_volume_up()
def volume_down(self):
@async_log_errors
async def async_volume_down(self):
"""Volume down media player."""
return self._receiver.volume_down()
await self._receiver.async_volume_down()
def set_volume_level(self, volume):
@async_log_errors
async def async_set_volume_level(self, volume: int):
"""Set volume level, range 0..1."""
# Volume has to be sent in a format like -50.0. Minimum is -80.0,
# maximum is 18.0
volume_denon = float((volume * 100) - 80)
if volume_denon > 18:
volume_denon = float(18)
with suppress(ValueError):
if self._receiver.set_volume(volume_denon):
self._volume = volume_denon
await self._receiver.async_set_volume(volume_denon)
def mute_volume(self, mute):
@async_log_errors
async def async_mute_volume(self, mute: bool):
"""Send mute command."""
return self._receiver.mute(mute)
await self._receiver.async_mute(mute)
def get_command(self, command, **kwargs):
@async_log_errors
async def async_get_command(self, command: str, **kwargs):
"""Send generic command."""
self._receiver.send_get_command(command)
return await self._receiver.async_get_command(command)
# Decorator defined before is a staticmethod
async_log_errors = staticmethod( # pylint: disable=no-staticmethod-decorator
async_log_errors
)

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

@ -1,7 +1,8 @@
"""Code to handle a DenonAVR receiver."""
import logging
from typing import Callable, Optional
import denonavr
from denonavr import DenonAVR
_LOGGER = logging.getLogger(__name__)
@ -9,13 +10,23 @@ _LOGGER = logging.getLogger(__name__)
class ConnectDenonAVR:
"""Class to async connect to a DenonAVR receiver."""
def __init__(self, hass, host, timeout, show_all_inputs, zone2, zone3):
def __init__(
self,
host: str,
timeout: float,
show_all_inputs: bool,
zone2: bool,
zone3: bool,
async_client_getter: Callable,
entry_state: Optional[str] = None,
):
"""Initialize the class."""
self._hass = hass
self._async_client_getter = async_client_getter
self._receiver = None
self._host = host
self._show_all_inputs = show_all_inputs
self._timeout = timeout
self._entry_state = entry_state
self._zones = {}
if zone2:
@ -24,14 +35,13 @@ class ConnectDenonAVR:
self._zones["Zone3"] = None
@property
def receiver(self):
def receiver(self) -> Optional[DenonAVR]:
"""Return the class containing all connections to the receiver."""
return self._receiver
async def async_connect_receiver(self):
async def async_connect_receiver(self) -> bool:
"""Connect to the DenonAVR receiver."""
if not await self._hass.async_add_executor_job(self.init_receiver_class):
return False
await self.async_init_receiver_class()
if (
self._receiver.manufacturer is None
@ -60,19 +70,16 @@ class ConnectDenonAVR:
return True
def init_receiver_class(self):
"""Initialize the DenonAVR class in a way that can called by async_add_executor_job."""
try:
self._receiver = denonavr.DenonAVR(
host=self._host,
show_all_inputs=self._show_all_inputs,
timeout=self._timeout,
add_zones=self._zones,
)
except ConnectionError:
_LOGGER.error(
"ConnectionError during setup of denonavr with host %s", self._host
)
return False
async def async_init_receiver_class(self) -> bool:
"""Initialize the DenonAVR class asynchronously."""
receiver = DenonAVR(
host=self._host,
show_all_inputs=self._show_all_inputs,
timeout=self._timeout,
add_zones=self._zones,
)
# Use httpx.AsyncClient getter provided by Home Assistant
receiver.set_async_client_getter(self._async_client_getter)
await receiver.async_setup()
return True
self._receiver = receiver

2
homeassistant/components/denonavr/services.yaml
View File

@ -1,4 +1,4 @@
# Describes the format for available webostv services
# Describes the format for available denonavr services
get_command:
description: "Send a generic HTTP get command."

3
requirements_all.txt
View File

@ -476,7 +476,7 @@ defusedxml==0.6.0
deluge-client==1.7.1
# homeassistant.components.denonavr
denonavr==0.9.10
denonavr==0.10.5
# homeassistant.components.devolo_home_control
devolo-home-control-api==0.17.1
@ -647,7 +647,6 @@ georss_ign_sismologia_client==0.2
# homeassistant.components.qld_bushfire
georss_qld_bushfire_alert_client==0.3
# homeassistant.components.denonavr
# homeassistant.components.huawei_lte
# homeassistant.components.kef
# homeassistant.components.minecraft_server

3
requirements_test_all.txt
View File

@ -258,7 +258,7 @@ debugpy==1.2.1
defusedxml==0.6.0
# homeassistant.components.denonavr
denonavr==0.9.10
denonavr==0.10.5
# homeassistant.components.devolo_home_control
devolo-home-control-api==0.17.1
@ -344,7 +344,6 @@ georss_ign_sismologia_client==0.2
# homeassistant.components.qld_bushfire
georss_qld_bushfire_alert_client==0.3
# homeassistant.components.denonavr
# homeassistant.components.huawei_lte
# homeassistant.components.kef
# homeassistant.components.minecraft_server

178
tests/components/denonavr/test_config_flow.py
View File

@ -14,13 +14,13 @@ from homeassistant.components.denonavr.config_flow import (
CONF_ZONE2,
CONF_ZONE3,
DOMAIN,
AvrTimoutError,
)
from homeassistant.const import CONF_HOST, CONF_MAC
from homeassistant.const import CONF_HOST
from tests.common import MockConfigEntry
TEST_HOST = "1.2.3.4"
TEST_MAC = "ab:cd:ef:gh"
TEST_HOST2 = "5.6.7.8"
TEST_NAME = "Test_Receiver"
TEST_MODEL = "model5"
@ -38,41 +38,29 @@ TEST_DISCOVER_2_RECEIVER = [{CONF_HOST: TEST_HOST}, {CONF_HOST: TEST_HOST2}]
def denonavr_connect_fixture():
"""Mock denonavr connection and entry setup."""
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR._update_input_func_list",
"homeassistant.components.denonavr.receiver.DenonAVR.async_setup",
return_value=None,
), patch(
"homeassistant.components.denonavr.receiver.DenonAVR.async_update",
return_value=None,
), patch(
"homeassistant.components.denonavr.receiver.DenonAVR.support_sound_mode",
return_value=True,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR._get_receiver_name",
return_value=TEST_NAME,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR._get_support_sound_mode",
return_value=True,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR._update_avr_2016",
return_value=True,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR._update_avr",
return_value=True,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.get_device_info",
return_value=True,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.name",
"homeassistant.components.denonavr.receiver.DenonAVR.name",
TEST_NAME,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.model_name",
"homeassistant.components.denonavr.receiver.DenonAVR.model_name",
TEST_MODEL,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.serial_number",
"homeassistant.components.denonavr.receiver.DenonAVR.serial_number",
TEST_SERIALNUMBER,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.manufacturer",
"homeassistant.components.denonavr.receiver.DenonAVR.manufacturer",
TEST_MANUFACTURER,
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.receiver_type",
"homeassistant.components.denonavr.receiver.DenonAVR.receiver_type",
TEST_RECEIVER_TYPE,
), patch(
"homeassistant.components.denonavr.config_flow.get_mac_address",
return_value=TEST_MAC,
), patch(
"homeassistant.components.denonavr.async_setup_entry", return_value=True
):
@ -102,7 +90,6 @@ async def test_config_flow_manual_host_success(hass):
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -125,7 +112,7 @@ async def test_config_flow_manual_discover_1_success(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.config_flow.denonavr.ssdp.identify_denonavr_receivers",
"homeassistant.components.denonavr.config_flow.denonavr.async_discover",
return_value=TEST_DISCOVER_1_RECEIVER,
):
result = await hass.config_entries.flow.async_configure(
@ -137,7 +124,6 @@ async def test_config_flow_manual_discover_1_success(hass):
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -160,7 +146,7 @@ async def test_config_flow_manual_discover_2_success(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.config_flow.denonavr.ssdp.identify_denonavr_receivers",
"homeassistant.components.denonavr.config_flow.denonavr.async_discover",
return_value=TEST_DISCOVER_2_RECEIVER,
):
result = await hass.config_entries.flow.async_configure(
@ -181,7 +167,6 @@ async def test_config_flow_manual_discover_2_success(hass):
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST2,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -204,7 +189,7 @@ async def test_config_flow_manual_discover_error(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.config_flow.denonavr.ssdp.identify_denonavr_receivers",
"homeassistant.components.denonavr.config_flow.denonavr.async_discover",
return_value=[],
):
result = await hass.config_entries.flow.async_configure(
@ -232,7 +217,7 @@ async def test_config_flow_manual_host_no_serial(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.serial_number",
"homeassistant.components.denonavr.receiver.DenonAVR.serial_number",
None,
):
result = await hass.config_entries.flow.async_configure(
@ -244,118 +229,6 @@ async def test_config_flow_manual_host_no_serial(hass):
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
CONF_SERIAL_NUMBER: None,
}
async def test_config_flow_manual_host_no_mac(hass):
"""
Successful flow manually initialized by the user.
Host specified and an error getting the mac address.
"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.config_flow.get_mac_address",
return_value=None,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: TEST_HOST},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: None,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
CONF_SERIAL_NUMBER: TEST_SERIALNUMBER,
}
async def test_config_flow_manual_host_no_serial_no_mac(hass):
"""
Successful flow manually initialized by the user.
Host specified and an error getting the serial number and mac address.
"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.serial_number",
None,
), patch(
"homeassistant.components.denonavr.config_flow.get_mac_address",
return_value=None,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: TEST_HOST},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: None,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
CONF_SERIAL_NUMBER: None,
}
async def test_config_flow_manual_host_no_serial_no_mac_exception(hass):
"""
Successful flow manually initialized by the user.
Host specified and an error getting the serial number and exception getting mac address.
"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.serial_number",
None,
), patch(
"homeassistant.components.denonavr.config_flow.get_mac_address",
side_effect=OSError,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_HOST: TEST_HOST},
)
assert result["type"] == "create_entry"
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: None,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -378,10 +251,10 @@ async def test_config_flow_manual_host_connection_error(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.get_device_info",
side_effect=ConnectionError,
"homeassistant.components.denonavr.receiver.DenonAVR.async_setup",
side_effect=AvrTimoutError("Timeout", "async_setup"),
), patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.receiver_type",
"homeassistant.components.denonavr.receiver.DenonAVR.receiver_type",
None,
):
result = await hass.config_entries.flow.async_configure(
@ -408,7 +281,7 @@ async def test_config_flow_manual_host_no_device_info(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.receiver_type",
"homeassistant.components.denonavr.receiver.DenonAVR.receiver_type",
None,
):
result = await hass.config_entries.flow.async_configure(
@ -445,7 +318,6 @@ async def test_config_flow_ssdp(hass):
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -521,7 +393,6 @@ async def test_options_flow(hass):
unique_id=TEST_UNIQUE_ID,
data={
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -567,7 +438,7 @@ async def test_config_flow_manual_host_no_serial_double_config(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.serial_number",
"homeassistant.components.denonavr.receiver.DenonAVR.serial_number",
None,
):
result = await hass.config_entries.flow.async_configure(
@ -579,7 +450,6 @@ async def test_config_flow_manual_host_no_serial_double_config(hass):
assert result["title"] == TEST_NAME
assert result["data"] == {
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -595,7 +465,7 @@ async def test_config_flow_manual_host_no_serial_double_config(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR.serial_number",
"homeassistant.components.denonavr.receiver.DenonAVR.serial_number",
None,
):
result = await hass.config_entries.flow.async_configure(

15
tests/components/denonavr/test_media_player.py
View File

@ -4,7 +4,6 @@ from unittest.mock import patch
import pytest
from homeassistant.components import media_player
from homeassistant.components.denonavr import ATTR_COMMAND, SERVICE_GET_COMMAND
from homeassistant.components.denonavr.config_flow import (
CONF_MANUFACTURER,
CONF_MODEL,
@ -12,12 +11,15 @@ from homeassistant.components.denonavr.config_flow import (
CONF_TYPE,
DOMAIN,
)
from homeassistant.const import ATTR_ENTITY_ID, CONF_HOST, CONF_MAC
from homeassistant.components.denonavr.media_player import (
ATTR_COMMAND,
SERVICE_GET_COMMAND,
)
from homeassistant.const import ATTR_ENTITY_ID, CONF_HOST
from tests.common import MockConfigEntry
TEST_HOST = "1.2.3.4"
TEST_MAC = "ab:cd:ef:gh"
TEST_NAME = "Test_Receiver"
TEST_MODEL = "model5"
TEST_SERIALNUMBER = "123456789"
@ -36,10 +38,10 @@ ENTITY_ID = f"{media_player.DOMAIN}.{TEST_NAME}"
def client_fixture():
"""Patch of client library for tests."""
with patch(
"homeassistant.components.denonavr.receiver.denonavr.DenonAVR",
"homeassistant.components.denonavr.receiver.DenonAVR",
autospec=True,
) as mock_client_class, patch(
"homeassistant.components.denonavr.receiver.denonavr.discover"
"homeassistant.components.denonavr.config_flow.denonavr.async_discover"
):
mock_client_class.return_value.name = TEST_NAME
mock_client_class.return_value.model_name = TEST_MODEL
@ -57,7 +59,6 @@ async def setup_denonavr(hass):
"""Initialize media_player for tests."""
entry_data = {
CONF_HOST: TEST_HOST,
CONF_MAC: TEST_MAC,
CONF_MODEL: TEST_MODEL,
CONF_TYPE: TEST_RECEIVER_TYPE,
CONF_MANUFACTURER: TEST_MANUFACTURER,
@ -92,4 +93,4 @@ async def test_get_command(hass, client):
await hass.services.async_call(DOMAIN, SERVICE_GET_COMMAND, data)
await hass.async_block_till_done()
client.send_get_command.assert_called_with("test_command")
client.async_get_command.assert_awaited_with("test_command")