starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 01:11:51 -04:00
Code Issues Packages Projects Releases Wiki Activity

Make LinkPlay followers mirror and control the group leader (#175589)

Browse Source 
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Mickael Goubin
2026-09-17 11:13:09 +00:00
committed by GitHub
co-authored by Claude Opus 5 Joost Lekkerkerker
parent c8f3999c3c
commit 3ee09ed808
3 changed files with 221 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/linkplay/media_player.py
+37 -20
View File
@@ -5,7 +5,7 @@ from datetime import timedelta
import logging
from typing import TYPE_CHECKING, Any, override
from linkplay.bridge import LinkPlayBridge
from linkplay.bridge import LinkPlayBridge, LinkPlayPlayer
from linkplay.consts import EqualizerMode, LoopMode, PlayingMode, PlayingStatus
from linkplay.controller import LinkPlayController, LinkPlayMultiroom
from linkplay.exceptions import LinkPlayRequestException
@@ -189,37 +189,37 @@ class LinkPlayMediaPlayerEntity(LinkPlayBaseEntity, MediaPlayerEntity):
@override
async def async_media_pause(self) -> None:
"""Send pause command."""
await self._bridge.player.pause()
await self._active_player.pause()
@exception_wrap
@override
async def async_media_play(self) -> None:
"""Send play command."""
await self._bridge.player.resume()
await self._active_player.resume()
@exception_wrap
@override
async def async_media_stop(self) -> None:
"""Send stop command."""
await self._bridge.player.stop()
await self._active_player.stop()
@exception_wrap
@override
async def async_media_next_track(self) -> None:
"""Send next command."""
await self._bridge.player.next()
await self._active_player.next()
@exception_wrap
@override
async def async_media_previous_track(self) -> None:
"""Send previous command."""
await self._bridge.player.previous()
await self._active_player.previous()
@exception_wrap
@override
async def async_set_repeat(self, repeat: RepeatMode) -> None:
"""Set repeat mode."""
await self._bridge.player.set_loop_mode(REPEAT_MAP_INV[repeat])
await self._active_player.set_loop_mode(REPEAT_MAP_INV[repeat])
@override
async def async_browse_media(
@@ -253,13 +253,13 @@ class LinkPlayMediaPlayerEntity(LinkPlayBaseEntity, MediaPlayerEntity):
media_id = play_item.url
url = async_process_play_media_url(self.hass, media_id)
await self._bridge.player.play(url)
await self._active_player.play(url)
@exception_wrap
async def async_play_preset(self, preset_number: int) -> None:
"""Play preset number."""
try:
await self._bridge.player.play_preset(preset_number)
await self._active_player.play_preset(preset_number)
except ValueError as err:
raise HomeAssistantError(err) from err
@@ -267,7 +267,7 @@ class LinkPlayMediaPlayerEntity(LinkPlayBaseEntity, MediaPlayerEntity):
@override
async def async_media_seek(self, position: float) -> None:
"""Seek to a position."""
await self._bridge.player.seek(round(position))
await self._active_player.seek(round(position))
@exception_wrap
@override
@@ -326,12 +326,25 @@ class LinkPlayMediaPlayerEntity(LinkPlayBaseEntity, MediaPlayerEntity):
assert leader_id is not None
return [leader_id, *followers]
@property
def _active_player(self) -> LinkPlayPlayer:
"""Return the player that holds the active media info.
A follower in a multiroom group does not expose the media info of the
stream it is playing; that info is only available on the group leader.
"""
multiroom = self._bridge.multiroom
if multiroom is not None and multiroom.leader is not self._bridge:
return multiroom.leader.player
return self._bridge.player
@property
@override
def media_image_url(self) -> str | None:
"""Image url of playing media."""
if self._bridge.player.status in [PlayingStatus.PLAYING, PlayingStatus.PAUSED]:
return str(self._bridge.player.album_art)
player = self._active_player
if player.status in [PlayingStatus.PLAYING, PlayingStatus.PAUSED]:
return str(player.album_art)
return None
@exception_wrap
@@ -352,24 +365,28 @@ class LinkPlayMediaPlayerEntity(LinkPlayBaseEntity, MediaPlayerEntity):
self._attr_state = STATE_MAP[self._bridge.player.status]
self._attr_volume_level = self._bridge.player.volume / 100
self._attr_is_volume_muted = self._bridge.player.muted
self._attr_repeat = REPEAT_MAP[self._bridge.player.loop_mode]
self._attr_shuffle = self._bridge.player.loop_mode == LoopMode.RANDOM_PLAYBACK
self._attr_repeat = REPEAT_MAP[self._active_player.loop_mode]
self._attr_shuffle = self._active_player.loop_mode == LoopMode.RANDOM_PLAYBACK
self._attr_sound_mode = self._bridge.player.equalizer_mode.value
self._attr_supported_features = DEFAULT_FEATURES
if self._bridge.player.status == PlayingStatus.PLAYING:
if self._bridge.player.total_length != 0:
# A follower mirrors the media info from the group leader and routes
# its transport controls to the leader, so the seekable features
# follow the leader's player as well.
player = self._active_player
if player.total_length != 0:
self._attr_supported_features = (
self._attr_supported_features | SEEKABLE_FEATURES
)
self._attr_source = SOURCE_MAP.get(self._bridge.player.play_mode, "other")
self._attr_media_position = self._bridge.player.current_position_in_seconds
self._attr_media_position = player.current_position_in_seconds
self._attr_media_position_updated_at = utcnow()
self._attr_media_duration = self._bridge.player.total_length_in_seconds
self._attr_media_artist = self._bridge.player.artist
self._attr_media_title = self._bridge.player.title
self._attr_media_album_name = self._bridge.player.album
self._attr_media_duration = player.total_length_in_seconds
self._attr_media_artist = player.artist
self._attr_media_title = player.title
self._attr_media_album_name = player.album
elif self._bridge.player.status == PlayingStatus.STOPPED:
self._attr_media_position = None
self._attr_media_position_updated_at = None
tests/components/linkplay/fixtures/getPlayerEx_follower.json
+19
View File
@@ -0,0 +1,19 @@
{
"type": "0",
"ch": "0",
"mode": "99",
"loop": "0",
"eq": "0",
"status": "play",
"curpos": "0",
"offset_pts": "0",
"totlen": "0",
"Title": "",
"Artist": "",
"Album": "",
"alarmflag": "0",
"plicount": "0",
"plicurr": "0",
"vol": "80",
"mute": "0"
}
tests/components/linkplay/test_media_player.py
+165
View File
@@ -0,0 +1,165 @@
"""Tests for the LinkPlay media player."""
from collections.abc import AsyncGenerator
from typing import Any
from unittest.mock import AsyncMock, patch
from linkplay.bridge import (
LinkPlayBridge,
LinkPlayDevice,
LinkPlayMultiroom,
LinkPlayPlayer,
)
from linkplay.consts import API_ENDPOINT, LoopMode, PlayingStatus
import pytest
from homeassistant.components.linkplay.const import DOMAIN, SHARED_DATA
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_MEDIA_ALBUM_NAME,
ATTR_MEDIA_ARTIST,
ATTR_MEDIA_DURATION,
ATTR_MEDIA_POSITION,
ATTR_MEDIA_REPEAT,
ATTR_MEDIA_SEEK_POSITION,
ATTR_MEDIA_SHUFFLE,
ATTR_MEDIA_TITLE,
ATTR_MEDIA_VOLUME_LEVEL,
DOMAIN as MEDIA_PLAYER_DOMAIN,
RepeatMode,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_MEDIA_SEEK,
SERVICE_MEDIA_STOP,
SERVICE_REPEAT_SET,
STATE_PLAYING,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_component import async_update_entity
from . import setup_integration
from .conftest import HOST, mock_lp_aiohttp_client
from tests.common import MockConfigEntry, async_load_fixture
ENTITY_ID = "media_player.smart_zone_1_54b9"
LEADER_ENTITY_ID = "media_player.leader"
LEADER_UUID = "FF31F09E-5001-FBDE-0546-2DBFFF31F0AA"
@pytest.fixture
async def leader_player(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> AsyncGenerator[AsyncMock]:
"""Set up a device that is a follower in a multiroom group and mock its leader."""
with (
mock_lp_aiohttp_client() as mock_session,
patch.object(LinkPlayMultiroom, "update_status", return_value=None),
):
for endpoint in (f"https://{HOST}", f"http://{HOST}"):
mock_session.get(
API_ENDPOINT.format(endpoint, "getPlayerStatusEx"),
text=await async_load_fixture(
hass, "getPlayerEx_follower.json", DOMAIN
),
)
mock_session.get(
API_ENDPOINT.format(endpoint, "getStatusEx"),
text=await async_load_fixture(hass, "getStatusEx.json", DOMAIN),
)
await setup_integration(hass, mock_config_entry)
player = AsyncMock(spec=LinkPlayPlayer)
player.status = PlayingStatus.PLAYING
player.title = "Zelda's Lullaby"
player.artist = "Spiritual Concepts"
player.album = "Cello Covers"
player.current_position_in_seconds = 17
player.total_length = 62000
player.total_length_in_seconds = 62
player.loop_mode = LoopMode.RANDOM_PLAYBACK
leader = AsyncMock(spec=LinkPlayBridge)
leader.player = player
leader.device = AsyncMock(spec=LinkPlayDevice)
leader.device.uuid = LEADER_UUID
hass.data[DOMAIN][SHARED_DATA].entity_to_bridge[LEADER_ENTITY_ID] = LEADER_UUID
bridge = mock_config_entry.runtime_data.bridge
bridge.multiroom = LinkPlayMultiroom(leader)
bridge.multiroom.followers = [bridge]
await async_update_entity(hass, ENTITY_ID)
yield player
@pytest.mark.usefixtures("leader_player")
async def test_follower_mirrors_leader_media_info(hass: HomeAssistant) -> None:
"""Test that a follower shows the media info of its group leader."""
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_PLAYING
assert state.attributes[ATTR_MEDIA_TITLE] == "Zelda's Lullaby"
assert state.attributes[ATTR_MEDIA_ARTIST] == "Spiritual Concepts"
assert state.attributes[ATTR_MEDIA_ALBUM_NAME] == "Cello Covers"
assert state.attributes[ATTR_MEDIA_POSITION] == 17
assert state.attributes[ATTR_MEDIA_DURATION] == 62
assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.ALL
assert state.attributes[ATTR_MEDIA_SHUFFLE] is True
# volume and source stay on the follower itself
assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.8
assert state.attributes[ATTR_INPUT_SOURCE] == "Follower"
@pytest.mark.parametrize(
("service", "service_data", "method", "method_args"),
[
pytest.param(SERVICE_MEDIA_PAUSE, {}, "pause", (), id="pause"),
pytest.param(SERVICE_MEDIA_PLAY, {}, "resume", (), id="play"),
pytest.param(SERVICE_MEDIA_STOP, {}, "stop", (), id="stop"),
pytest.param(SERVICE_MEDIA_NEXT_TRACK, {}, "next", (), id="next_track"),
pytest.param(
SERVICE_MEDIA_PREVIOUS_TRACK, {}, "previous", (), id="previous_track"
),
pytest.param(
SERVICE_MEDIA_SEEK,
{ATTR_MEDIA_SEEK_POSITION: 42},
"seek",
(42,),
id="seek",
),
pytest.param(
SERVICE_REPEAT_SET,
{ATTR_MEDIA_REPEAT: RepeatMode.ONE},
"set_loop_mode",
(LoopMode.CONTINOUS_PLAY_ONE_SONG,),
id="repeat_set",
),
],
)
async def test_follower_transport_commands_go_to_leader(
hass: HomeAssistant,
leader_player: AsyncMock,
service: str,
service_data: dict[str, Any],
method: str,
method_args: tuple[Any, ...],
) -> None:
"""Test that transport commands on a follower control the group leader."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: ENTITY_ID, **service_data},
blocking=True,
)
getattr(leader_player, method).assert_awaited_once_with(*method_args)