evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add multi-zone support to Anthem AV receiver and distribution solution (#74779)

Browse source 
* Add multi-zone support to Anthem AV receiver and distribution amplifier

* Fix typo in comment

* Convert properties to attribute and add test

* Migrate entity name

* Fix after rebase add strict typing and bump version

* fix typing

* Simplify test

* Small improvement

* remove dispatcher send and use callback
This commit is contained in:
Alex Henry 2022-07-31 00:04:24 +12:00 • committed by GitHub
parent 8181da7090
commit ace359b1bd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 200 additions and 104 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -56,6 +56,7 @@ homeassistant.components.ambee.*
homeassistant.components.ambient_station.*
homeassistant.components.amcrest.*
homeassistant.components.ampio.*
homeassistant.components.anthemav.*
homeassistant.components.aseko_pool_live.*
homeassistant.components.asuswrt.*
homeassistant.components.automation.*

9
homeassistant/components/anthemav/__init__.py
View file

@ -4,6 +4,7 @@ from __future__ import annotations
import logging
import anthemav
from anthemav.device_error import DeviceError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, EVENT_HOMEASSISTANT_STOP, Platform
@ -11,7 +12,7 @@ from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import ANTHEMAV_UDATE_SIGNAL, DOMAIN
from .const import ANTHEMAV_UDATE_SIGNAL, DEVICE_TIMEOUT_SECONDS, DOMAIN
PLATFORMS = [Platform.MEDIA_PLAYER]
@ -22,7 +23,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Anthem A/V Receivers from a config entry."""
@callback
def async_anthemav_update_callback(message):
def async_anthemav_update_callback(message: str) -> None:
"""Receive notification from transport that new data exists."""
_LOGGER.debug("Received update callback from AVR: %s", message)
async_dispatcher_send(hass, f"{ANTHEMAV_UDATE_SIGNAL}_{entry.entry_id}")
@ -34,7 +35,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_callback=async_anthemav_update_callback,
)
except OSError as err:
# Wait for the zones to be initialised based on the model
await avr.protocol.wait_for_device_initialised(DEVICE_TIMEOUT_SECONDS)
except (OSError, DeviceError) as err:
raise ConfigEntryNotReady from err
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = avr

10
homeassistant/components/anthemav/config_flow.py
View file

@ -15,9 +15,13 @@ from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import format_mac
from .const import CONF_MODEL, DEFAULT_NAME, DEFAULT_PORT, DOMAIN
DEVICE_TIMEOUT_SECONDS = 4.0
from .const import (
CONF_MODEL,
DEFAULT_NAME,
DEFAULT_PORT,
DEVICE_TIMEOUT_SECONDS,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)

1
homeassistant/components/anthemav/const.py
View file

@ -5,3 +5,4 @@ DEFAULT_NAME = "Anthem AV"
DEFAULT_PORT = 14999
DOMAIN = "anthemav"
MANUFACTURER = "Anthem"
DEVICE_TIMEOUT_SECONDS = 4.0

2
homeassistant/components/anthemav/manifest.json
View file

@ -2,7 +2,7 @@
"domain": "anthemav",
"name": "Anthem A/V Receivers",
"documentation": "https://www.home-assistant.io/integrations/anthemav",
"requirements": ["anthemav==1.3.2"],
"requirements": ["anthemav==1.4.1"],
"dependencies": ["repairs"],
"codeowners": ["@hyralex"],
"config_flow": true,

121
homeassistant/components/anthemav/media_player.py
View file

@ -2,13 +2,14 @@
from __future__ import annotations
import logging
from typing import Any
from anthemav.connection import Connection
from anthemav.protocol import AVR
import voluptuous as vol
from homeassistant.components.media_player import (
PLATFORM_SCHEMA,
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
)
@ -22,7 +23,7 @@ from homeassistant.const import (
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
@ -88,20 +89,28 @@ async def async_setup_entry(
mac_address = config_entry.data[CONF_MAC]
model = config_entry.data[CONF_MODEL]
avr = hass.data[DOMAIN][config_entry.entry_id]
avr: Connection = hass.data[DOMAIN][config_entry.entry_id]
entity = AnthemAVR(avr, name, mac_address, model, config_entry.entry_id)
entities = []
for zone_number in avr.protocol.zones:
_LOGGER.debug("Initializing Zone %s", zone_number)
entity = AnthemAVR(
avr.protocol, name, mac_address, model, zone_number, config_entry.entry_id
)
entities.append(entity)
_LOGGER.debug("Device data dump: %s", entity.dump_avrdata)
_LOGGER.debug("Connection data dump: %s", avr.dump_conndata)
async_add_entities([entity])
async_add_entities(entities)
class AnthemAVR(MediaPlayerEntity):
"""Entity reading values from Anthem AVR protocol."""
_attr_has_entity_name = True
_attr_should_poll = False
_attr_device_class = MediaPlayerDeviceClass.RECEIVER
_attr_icon = "mdi:audio-video"
_attr_supported_features = (
MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_MUTE
@ -111,23 +120,33 @@ class AnthemAVR(MediaPlayerEntity):
)
def __init__(
self, avr: Connection, name: str, mac_address: str, model: str, entry_id: str
self,
avr: AVR,
name: str,
mac_address: str,
model: str,
zone_number: int,
entry_id: str,
) -> None:
"""Initialize entity with transport."""
super().__init__()
self.avr = avr
self._entry_id = entry_id
self._attr_name = name
self._zone_number = zone_number
self._zone = avr.zones[zone_number]
if zone_number > 1:
self._attr_name = f"zone {zone_number}"
self._attr_unique_id = f"{mac_address}_{zone_number}"
else:
self._attr_unique_id = mac_address
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, mac_address)},
name=name,
manufacturer=MANUFACTURER,
model=model,
)
def _lookup(self, propname: str, dval: Any | None = None) -> Any | None:
return getattr(self.avr.protocol, propname, dval)
self.set_states()
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
@ -135,82 +154,42 @@ class AnthemAVR(MediaPlayerEntity):
async_dispatcher_connect(
self.hass,
f"{ANTHEMAV_UDATE_SIGNAL}_{self._entry_id}",
self.async_write_ha_state,
self.update_states,
)
)
@property
def state(self) -> str | None:
"""Return state of power on/off."""
pwrstate = self._lookup("power")
@callback
def update_states(self) -> None:
"""Update states for the current zone."""
self.set_states()
self.async_write_ha_state()
if pwrstate is True:
return STATE_ON
if pwrstate is False:
return STATE_OFF
return None
@property
def is_volume_muted(self) -> bool | None:
"""Return boolean reflecting mute state on device."""
return self._lookup("mute", False)
@property
def volume_level(self) -> float | None:
"""Return volume level from 0 to 1."""
return self._lookup("volume_as_percentage", 0.0)
@property
def media_title(self) -> str | None:
"""Return current input name (closest we have to media title)."""
return self._lookup("input_name", "No Source")
@property
def app_name(self) -> str | None:
"""Return details about current video and audio stream."""
return (
f"{self._lookup('video_input_resolution_text', '')} "
f"{self._lookup('audio_input_name', '')}"
)
@property
def source(self) -> str | None:
"""Return currently selected input."""
return self._lookup("input_name", "Unknown")
@property
def source_list(self) -> list[str] | None:
"""Return all active, configured inputs."""
return self._lookup("input_list", ["Unknown"])
def set_states(self) -> None:
"""Set all the states from the device to the entity."""
self._attr_state = STATE_ON if self._zone.power is True else STATE_OFF
self._attr_is_volume_muted = self._zone.mute
self._attr_volume_level = self._zone.volume_as_percentage
self._attr_media_title = self._zone.input_name
self._attr_app_name = self._zone.input_format
self._attr_source = self._zone.input_name
self._attr_source_list = self.avr.input_list
async def async_select_source(self, source: str) -> None:
"""Change AVR to the designated source (by name)."""
self._update_avr("input_name", source)
self._zone.input_name = source
async def async_turn_off(self) -> None:
"""Turn AVR power off."""
self._update_avr("power", False)
self._zone.power = False
async def async_turn_on(self) -> None:
"""Turn AVR power on."""
self._update_avr("power", True)
self._zone.power = True
async def async_set_volume_level(self, volume: float) -> None:
"""Set AVR volume (0 to 1)."""
self._update_avr("volume_as_percentage", volume)
self._zone.volume_as_percentage = volume
async def async_mute_volume(self, mute: bool) -> None:
"""Engage AVR mute."""
self._update_avr("mute", mute)
def _update_avr(self, propname: str, value: Any | None) -> None:
"""Update a property in the AVR."""
_LOGGER.debug("Sending command to AVR: set %s to %s", propname, str(value))
setattr(self.avr.protocol, propname, value)
@property
def dump_avrdata(self):
"""Return state of avr object for debugging forensics."""
attrs = vars(self)
items_string = ", ".join(f"{item}: {item}" for item in attrs.items())
return f"dump_avrdata: {items_string}"
self._zone.mute = mute

11
mypy.ini
View file

@ -339,6 +339,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.anthemav.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.aseko_pool_live.*]
check_untyped_defs = true
disallow_incomplete_defs = true

2
requirements_all.txt
View file

@ -313,7 +313,7 @@ androidtv[async]==0.0.67
anel_pwrctrl-homeassistant==0.0.1.dev2
# homeassistant.components.anthemav
anthemav==1.3.2
anthemav==1.4.1
# homeassistant.components.apcupsd
apcaccess==0.0.13

2
requirements_test_all.txt
View file

@ -279,7 +279,7 @@ ambiclimate==0.2.1
androidtv[async]==0.0.67
# homeassistant.components.anthemav
anthemav==1.3.2
anthemav==1.4.1
# homeassistant.components.apprise
apprise==0.9.9

35
tests/components/anthemav/conftest.py
View file

@ -1,10 +1,12 @@
"""Fixtures for anthemav integration tests."""
from typing import Callable
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.anthemav.const import CONF_MODEL, DOMAIN
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_PORT
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -16,17 +18,25 @@ def mock_anthemav() -> AsyncMock:
avr.protocol.macaddress = "000000000001"
avr.protocol.model = "MRX 520"
avr.reconnect = AsyncMock()
avr.protocol.wait_for_device_initialised = AsyncMock()
avr.close = MagicMock()
avr.protocol.input_list = []
avr.protocol.audio_listening_mode_list = []
avr.protocol.power = False
avr.protocol.zones = {1: get_zone(), 2: get_zone()}
return avr
def get_zone() -> MagicMock:
"""Return a mocked zone."""
zone = MagicMock()
zone.power = False
return zone
@pytest.fixture
def mock_connection_create(mock_anthemav: AsyncMock) -> AsyncMock:
"""Return the default mocked connection.create."""
with patch(
"anthemav.Connection.create",
return_value=mock_anthemav,
@ -34,6 +44,12 @@ def mock_connection_create(mock_anthemav: AsyncMock) -> AsyncMock:
yield mock
@pytest.fixture
def update_callback(mock_connection_create: AsyncMock) -> Callable[[str], None]:
"""Return the update_callback used when creating the connection."""
return mock_connection_create.call_args[1]["update_callback"]
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
@ -48,3 +64,18 @@ def mock_config_entry() -> MockConfigEntry:
},
unique_id="00:00:00:00:00:01",
)
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_connection_create: AsyncMock,
) -> MockConfigEntry:
"""Set up the AnthemAv integration for testing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

37
tests/components/anthemav/test_init.py
View file

@ -1,6 +1,10 @@
"""Test the Anthem A/V Receivers config flow."""
from typing import Callable
from unittest.mock import ANY, AsyncMock, patch
from anthemav.device_error import DeviceError
import pytest
from homeassistant import config_entries
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
@ -12,35 +16,31 @@ async def test_load_unload_config_entry(
hass: HomeAssistant,
mock_connection_create: AsyncMock,
mock_anthemav: AsyncMock,
mock_config_entry: MockConfigEntry,
init_integration: MockConfigEntry,
) -> None:
"""Test load and unload AnthemAv component."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# assert avr is created
mock_connection_create.assert_called_with(
host="1.1.1.1", port=14999, update_callback=ANY
)
assert mock_config_entry.state == config_entries.ConfigEntryState.LOADED
assert init_integration.state == config_entries.ConfigEntryState.LOADED
# unload
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.config_entries.async_unload(init_integration.entry_id)
await hass.async_block_till_done()
# assert unload and avr is closed
assert mock_config_entry.state == config_entries.ConfigEntryState.NOT_LOADED
assert init_integration.state == config_entries.ConfigEntryState.NOT_LOADED
mock_anthemav.close.assert_called_once()
async def test_config_entry_not_ready(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
@pytest.mark.parametrize("error", [OSError, DeviceError])
async def test_config_entry_not_ready_when_oserror(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, error: Exception
) -> None:
"""Test AnthemAV configuration entry not ready."""
with patch(
"anthemav.Connection.create",
side_effect=OSError,
side_effect=error,
):
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
@ -52,23 +52,18 @@ async def test_anthemav_dispatcher_signal(
hass: HomeAssistant,
mock_connection_create: AsyncMock,
mock_anthemav: AsyncMock,
mock_config_entry: MockConfigEntry,
init_integration: MockConfigEntry,
update_callback: Callable[[str], None],
) -> None:
"""Test send update signal to dispatcher."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
states = hass.states.get("media_player.anthem_av")
assert states
assert states.state == STATE_OFF
# change state of the AVR
mock_anthemav.protocol.power = True
mock_anthemav.protocol.zones[1].power = True
# get the callback function that trigger the signal to update the state
avr_update_callback = mock_connection_create.call_args[1]["update_callback"]
avr_update_callback("power")
update_callback("power")
await hass.async_block_till_done()

71
tests/components/anthemav/test_media_player.py Normal file
View file

@ -0,0 +1,71 @@
"""Test the Anthem A/V Receivers config flow."""
from typing import Callable
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.media_player.const import (
ATTR_APP_NAME,
ATTR_INPUT_SOURCE,
ATTR_INPUT_SOURCE_LIST,
ATTR_MEDIA_TITLE,
ATTR_MEDIA_VOLUME_MUTED,
)
from homeassistant.components.siren.const import ATTR_VOLUME_LEVEL
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
"entity_id,entity_name",
[
("media_player.anthem_av", "Anthem AV"),
("media_player.anthem_av_zone_2", "Anthem AV zone 2"),
],
)
async def test_zones_loaded(
hass: HomeAssistant,
init_integration: MockConfigEntry,
entity_id: str,
entity_name: str,
) -> None:
"""Test zones are loaded."""
states = hass.states.get(entity_id)
assert states
assert states.state == STATE_OFF
assert states.name == entity_name
async def test_update_states_zone1(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_anthemav: AsyncMock,
update_callback: Callable[[str], None],
) -> None:
"""Test zone states are updated."""
mock_zone = mock_anthemav.protocol.zones[1]
mock_zone.power = True
mock_zone.mute = True
mock_zone.volume_as_percentage = 42
mock_zone.input_name = "TEST INPUT"
mock_zone.input_format = "2.0 PCM"
mock_anthemav.protocol.input_list = ["TEST INPUT", "INPUT 2"]
update_callback("command")
await hass.async_block_till_done()
states = hass.states.get("media_player.anthem_av")
assert states
assert states.state == STATE_ON
assert states.attributes[ATTR_VOLUME_LEVEL] == 42
assert states.attributes[ATTR_MEDIA_VOLUME_MUTED] is True
assert states.attributes[ATTR_INPUT_SOURCE] == "TEST INPUT"
assert states.attributes[ATTR_MEDIA_TITLE] == "TEST INPUT"
assert states.attributes[ATTR_APP_NAME] == "2.0 PCM"
assert states.attributes[ATTR_INPUT_SOURCE_LIST] == ["TEST INPUT", "INPUT 2"]