Add timesync and restart functionality to linkplay (#130167)
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>pull/131616/head
parent
db198d4da2
commit
17466684a6
|
@ -0,0 +1,82 @@
|
|||
"""Support for LinkPlay buttons."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from linkplay.bridge import LinkPlayBridge
|
||||
|
||||
from homeassistant.components.button import (
|
||||
ButtonDeviceClass,
|
||||
ButtonEntity,
|
||||
ButtonEntityDescription,
|
||||
)
|
||||
from homeassistant.const import EntityCategory
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import LinkPlayConfigEntry
|
||||
from .entity import LinkPlayBaseEntity, exception_wrap
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class LinkPlayButtonEntityDescription(ButtonEntityDescription):
|
||||
"""Class describing LinkPlay button entities."""
|
||||
|
||||
remote_function: Callable[[LinkPlayBridge], Coroutine[Any, Any, None]]
|
||||
|
||||
|
||||
BUTTON_TYPES: tuple[LinkPlayButtonEntityDescription, ...] = (
|
||||
LinkPlayButtonEntityDescription(
|
||||
key="timesync",
|
||||
translation_key="timesync",
|
||||
remote_function=lambda linkplay_bridge: linkplay_bridge.device.timesync(),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
LinkPlayButtonEntityDescription(
|
||||
key="restart",
|
||||
device_class=ButtonDeviceClass.RESTART,
|
||||
remote_function=lambda linkplay_bridge: linkplay_bridge.device.reboot(),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: LinkPlayConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the LinkPlay buttons from config entry."""
|
||||
|
||||
# add entities
|
||||
async_add_entities(
|
||||
LinkPlayButton(config_entry.runtime_data.bridge, description)
|
||||
for description in BUTTON_TYPES
|
||||
)
|
||||
|
||||
|
||||
class LinkPlayButton(LinkPlayBaseEntity, ButtonEntity):
|
||||
"""Representation of LinkPlay button."""
|
||||
|
||||
entity_description: LinkPlayButtonEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
bridge: LinkPlayBridge,
|
||||
description: LinkPlayButtonEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize LinkPlay button."""
|
||||
super().__init__(bridge)
|
||||
self.entity_description = description
|
||||
self._attr_unique_id = f"{bridge.device.uuid}-{description.key}"
|
||||
|
||||
@exception_wrap
|
||||
async def async_press(self) -> None:
|
||||
"""Press the button."""
|
||||
await self.entity_description.remote_function(self._bridge)
|
|
@ -8,5 +8,5 @@ from homeassistant.util.hass_dict import HassKey
|
|||
DOMAIN = "linkplay"
|
||||
CONTROLLER = "controller"
|
||||
CONTROLLER_KEY: HassKey[LinkPlayController] = HassKey(CONTROLLER)
|
||||
PLATFORMS = [Platform.MEDIA_PLAYER]
|
||||
PLATFORMS = [Platform.BUTTON, Platform.MEDIA_PLAYER]
|
||||
DATA_SESSION = "session"
|
||||
|
|
|
@ -0,0 +1,57 @@
|
|||
"""BaseEntity to support multiple LinkPlay platforms."""
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from typing import Any, Concatenate
|
||||
|
||||
from linkplay.bridge import LinkPlayBridge
|
||||
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from . import DOMAIN, LinkPlayRequestException
|
||||
from .utils import MANUFACTURER_GENERIC, get_info_from_project
|
||||
|
||||
|
||||
def exception_wrap[_LinkPlayEntityT: LinkPlayBaseEntity, **_P, _R](
|
||||
func: Callable[Concatenate[_LinkPlayEntityT, _P], Coroutine[Any, Any, _R]],
|
||||
) -> Callable[Concatenate[_LinkPlayEntityT, _P], Coroutine[Any, Any, _R]]:
|
||||
"""Define a wrapper to catch exceptions and raise HomeAssistant errors."""
|
||||
|
||||
async def _wrap(self: _LinkPlayEntityT, *args: _P.args, **kwargs: _P.kwargs) -> _R:
|
||||
try:
|
||||
return await func(self, *args, **kwargs)
|
||||
except LinkPlayRequestException as err:
|
||||
raise HomeAssistantError(
|
||||
f"Exception occurred when communicating with API {func}: {err}"
|
||||
) from err
|
||||
|
||||
return _wrap
|
||||
|
||||
|
||||
class LinkPlayBaseEntity(Entity):
|
||||
"""Representation of a LinkPlay base entity."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(self, bridge: LinkPlayBridge) -> None:
|
||||
"""Initialize the LinkPlay media player."""
|
||||
|
||||
self._bridge = bridge
|
||||
|
||||
manufacturer, model = get_info_from_project(bridge.device.properties["project"])
|
||||
model_id = None
|
||||
if model != MANUFACTURER_GENERIC:
|
||||
model_id = bridge.device.properties["project"]
|
||||
|
||||
self._attr_device_info = dr.DeviceInfo(
|
||||
configuration_url=bridge.endpoint,
|
||||
connections={(dr.CONNECTION_NETWORK_MAC, bridge.device.properties["MAC"])},
|
||||
hw_version=bridge.device.properties["hardware"],
|
||||
identifiers={(DOMAIN, bridge.device.uuid)},
|
||||
manufacturer=manufacturer,
|
||||
model=model,
|
||||
model_id=model_id,
|
||||
name=bridge.device.name,
|
||||
sw_version=bridge.device.properties["firmware"],
|
||||
)
|
|
@ -1,4 +1,11 @@
|
|||
{
|
||||
"entity": {
|
||||
"button": {
|
||||
"timesync": {
|
||||
"default": "mdi:clock"
|
||||
}
|
||||
}
|
||||
},
|
||||
"services": {
|
||||
"play_preset": {
|
||||
"service": "mdi:play-box-outline"
|
||||
|
|
|
@ -2,9 +2,8 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
import logging
|
||||
from typing import Any, Concatenate
|
||||
from typing import Any
|
||||
|
||||
from linkplay.bridge import LinkPlayBridge
|
||||
from linkplay.consts import EqualizerMode, LoopMode, PlayingMode, PlayingStatus
|
||||
|
@ -28,7 +27,6 @@ from homeassistant.core import HomeAssistant
|
|||
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
|
||||
from homeassistant.helpers import (
|
||||
config_validation as cv,
|
||||
device_registry as dr,
|
||||
entity_platform,
|
||||
entity_registry as er,
|
||||
)
|
||||
|
@ -37,7 +35,7 @@ from homeassistant.util.dt import utcnow
|
|||
|
||||
from . import LinkPlayConfigEntry, LinkPlayData
|
||||
from .const import CONTROLLER_KEY, DOMAIN
|
||||
from .utils import MANUFACTURER_GENERIC, get_info_from_project
|
||||
from .entity import LinkPlayBaseEntity, exception_wrap
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
STATE_MAP: dict[PlayingStatus, MediaPlayerState] = {
|
||||
|
@ -145,58 +143,24 @@ async def async_setup_entry(
|
|||
async_add_entities([LinkPlayMediaPlayerEntity(entry.runtime_data.bridge)])
|
||||
|
||||
|
||||
def exception_wrap[_LinkPlayEntityT: LinkPlayMediaPlayerEntity, **_P, _R](
|
||||
func: Callable[Concatenate[_LinkPlayEntityT, _P], Coroutine[Any, Any, _R]],
|
||||
) -> Callable[Concatenate[_LinkPlayEntityT, _P], Coroutine[Any, Any, _R]]:
|
||||
"""Define a wrapper to catch exceptions and raise HomeAssistant errors."""
|
||||
|
||||
async def _wrap(self: _LinkPlayEntityT, *args: _P.args, **kwargs: _P.kwargs) -> _R:
|
||||
try:
|
||||
return await func(self, *args, **kwargs)
|
||||
except LinkPlayRequestException as err:
|
||||
raise HomeAssistantError(
|
||||
f"Exception occurred when communicating with API {func}: {err}"
|
||||
) from err
|
||||
|
||||
return _wrap
|
||||
|
||||
|
||||
class LinkPlayMediaPlayerEntity(MediaPlayerEntity):
|
||||
class LinkPlayMediaPlayerEntity(LinkPlayBaseEntity, MediaPlayerEntity):
|
||||
"""Representation of a LinkPlay media player."""
|
||||
|
||||
_attr_sound_mode_list = list(EQUALIZER_MAP.values())
|
||||
_attr_device_class = MediaPlayerDeviceClass.RECEIVER
|
||||
_attr_media_content_type = MediaType.MUSIC
|
||||
_attr_has_entity_name = True
|
||||
_attr_name = None
|
||||
|
||||
def __init__(self, bridge: LinkPlayBridge) -> None:
|
||||
"""Initialize the LinkPlay media player."""
|
||||
|
||||
self._bridge = bridge
|
||||
super().__init__(bridge)
|
||||
self._attr_unique_id = bridge.device.uuid
|
||||
|
||||
self._attr_source_list = [
|
||||
SOURCE_MAP[playing_mode] for playing_mode in bridge.device.playmode_support
|
||||
]
|
||||
|
||||
manufacturer, model = get_info_from_project(bridge.device.properties["project"])
|
||||
model_id = None
|
||||
if model != MANUFACTURER_GENERIC:
|
||||
model_id = bridge.device.properties["project"]
|
||||
|
||||
self._attr_device_info = dr.DeviceInfo(
|
||||
configuration_url=bridge.endpoint,
|
||||
connections={(dr.CONNECTION_NETWORK_MAC, bridge.device.properties["MAC"])},
|
||||
hw_version=bridge.device.properties["hardware"],
|
||||
identifiers={(DOMAIN, bridge.device.uuid)},
|
||||
manufacturer=manufacturer,
|
||||
model=model,
|
||||
model_id=model_id,
|
||||
name=bridge.device.name,
|
||||
sw_version=bridge.device.properties["firmware"],
|
||||
)
|
||||
|
||||
@exception_wrap
|
||||
async def async_update(self) -> None:
|
||||
"""Update the state of the media player."""
|
||||
|
|
|
@ -35,6 +35,13 @@
|
|||
}
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"button": {
|
||||
"timesync": {
|
||||
"name": "Sync time"
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"invalid_grouping_entity": {
|
||||
"message": "Entity with id {entity_id} can't be added to the LinkPlay multiroom. Is the entity a LinkPlay mediaplayer?"
|
||||
|
|
Loading…
Reference in New Issue