Add base entity to Russound RIO integration (#123842)

* Add base entity to Russound RIO integration

* Set entity back to primary mac addr

* Switch to type shorthand
pull/122636/head^2
Noah Husby 2024-08-13 17:40:51 -04:00 committed by GitHub
parent 2b6949f3c7
commit 29887c2a17
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 75 additions and 54 deletions

View File

@ -0,0 +1,70 @@
"""Base entity for Russound RIO integration."""
from collections.abc import Awaitable, Callable, Coroutine
from functools import wraps
from typing import Any, Concatenate
from aiorussound import Controller
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import DOMAIN, RUSSOUND_RIO_EXCEPTIONS
def command[_EntityT: RussoundBaseEntity, **_P](
func: Callable[Concatenate[_EntityT, _P], Awaitable[None]],
) -> Callable[Concatenate[_EntityT, _P], Coroutine[Any, Any, None]]:
"""Wrap async calls to raise on request error."""
@wraps(func)
async def decorator(self: _EntityT, *args: _P.args, **kwargs: _P.kwargs) -> None:
"""Wrap all command methods."""
try:
await func(self, *args, **kwargs)
except RUSSOUND_RIO_EXCEPTIONS as exc:
raise HomeAssistantError(
f"Error executing {func.__name__} on entity {self.entity_id},"
) from exc
return decorator
class RussoundBaseEntity(Entity):
"""Russound Base Entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self,
controller: Controller,
) -> None:
"""Initialize the entity."""
self._instance = controller.instance
self._controller = controller
self._primary_mac_address = (
controller.mac_address or controller.parent_controller.mac_address
)
self._device_identifier = (
self._controller.mac_address
or f"{self._primary_mac_address}-{self._controller.controller_id}"
)
self._attr_device_info = DeviceInfo(
# Use MAC address of Russound device as identifier
identifiers={(DOMAIN, self._device_identifier)},
manufacturer="Russound",
name=controller.controller_type,
model=controller.controller_type,
sw_version=controller.firmware_version,
)
if controller.parent_controller:
self._attr_device_info["via_device"] = (
DOMAIN,
controller.parent_controller.mac_address,
)
else:
self._attr_device_info["connections"] = {
(CONNECTION_NETWORK_MAC, controller.mac_address)
}

View File

@ -2,10 +2,7 @@
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from functools import wraps
import logging
from typing import Any, Concatenate
from aiorussound import Source, Zone
@ -20,14 +17,13 @@ from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import RUSSOUND_RIO_EXCEPTIONS, RussoundConfigEntry
from . import RussoundConfigEntry
from .const import DOMAIN, MP_FEATURES_BY_FLAG
from .entity import RussoundBaseEntity, command
_LOGGER = logging.getLogger(__name__)
@ -111,31 +107,11 @@ async def async_setup_entry(
async_add_entities(entities)
def command[_T: RussoundZoneDevice, **_P](
func: Callable[Concatenate[_T, _P], Awaitable[None]],
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]:
"""Wrap async calls to raise on request error."""
@wraps(func)
async def decorator(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None:
"""Wrap all command methods."""
try:
await func(self, *args, **kwargs)
except RUSSOUND_RIO_EXCEPTIONS as exc:
raise HomeAssistantError(
f"Error executing {func.__name__} on entity {self.entity_id},"
) from exc
return decorator
class RussoundZoneDevice(MediaPlayerEntity):
class RussoundZoneDevice(RussoundBaseEntity, MediaPlayerEntity):
"""Representation of a Russound Zone."""
_attr_device_class = MediaPlayerDeviceClass.SPEAKER
_attr_media_content_type = MediaType.MUSIC
_attr_should_poll = False
_attr_has_entity_name = True
_attr_supported_features = (
MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_STEP
@ -146,36 +122,11 @@ class RussoundZoneDevice(MediaPlayerEntity):
def __init__(self, zone: Zone, sources: dict[int, Source]) -> None:
"""Initialize the zone device."""
self._controller = zone.controller
super().__init__(zone.controller)
self._zone = zone
self._sources = sources
self._attr_name = zone.name
primary_mac_address = (
self._controller.mac_address
or self._controller.parent_controller.mac_address
)
self._attr_unique_id = f"{primary_mac_address}-{zone.device_str()}"
device_identifier = (
self._controller.mac_address
or f"{primary_mac_address}-{self._controller.controller_id}"
)
self._attr_device_info = DeviceInfo(
# Use MAC address of Russound device as identifier
identifiers={(DOMAIN, device_identifier)},
manufacturer="Russound",
name=self._controller.controller_type,
model=self._controller.controller_type,
sw_version=self._controller.firmware_version,
)
if self._controller.parent_controller:
self._attr_device_info["via_device"] = (
DOMAIN,
self._controller.parent_controller.mac_address,
)
else:
self._attr_device_info["connections"] = {
(CONNECTION_NETWORK_MAC, self._controller.mac_address)
}
self._attr_unique_id = f"{self._primary_mac_address}-{zone.device_str()}"
for flag, feature in MP_FEATURES_BY_FLAG.items():
if flag in zone.instance.supported_features:
self._attr_supported_features |= feature