Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-05-27 18:56:26 +01:00
Code Activity

Add browse and play media support to Yoto

Browse Source
This commit is contained in:
Paul Bottein
2026-05-27 00:20:42 +02:00
parent f6e8394771
commit 00ebccf168
5 changed files with 619 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/yoto/media_player.py
+240 -2
View File
@@ -4,22 +4,28 @@ from collections.abc import Awaitable, Callable
from datetime import datetime
from typing import Any
from yoto_api import Card, PlaybackStatus, YotoError, YotoPlayer
from yoto_api import Card, Chapter, PlaybackStatus, Track, YotoError, YotoPlayer
from homeassistant.components.media_player import (
BrowseError,
BrowseMedia,
MediaClass,
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import YotoConfigEntry, YotoDataUpdateCoordinator
from .entity import YotoEntity
URI_SCHEME = "yoto"
PARALLEL_UPDATES = 0
# Yoto players expose 16 hardware volume steps.
@@ -56,6 +62,8 @@ class YotoMediaPlayer(YotoEntity, MediaPlayerEntity):
MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.PLAY_MEDIA
| MediaPlayerEntityFeature.BROWSE_MEDIA
| MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.PREVIOUS_TRACK
@@ -169,6 +177,204 @@ class YotoMediaPlayer(YotoEntity, MediaPlayerEntity):
"""Skip to the previous track on the active card."""
await self._async_run(self.coordinator.client.previous_track, self._player_id)
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play a Yoto card, chapter, or track from the browse tree."""
try:
card_id, chapter_key, track_key = _parse_uri(media_id)
except ValueError as err:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_media_id",
translation_placeholders={"media_id": media_id},
) from err
client = self.coordinator.client
card = client.library.get(card_id)
if card is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unknown_card",
translation_placeholders={"card_id": card_id},
)
if chapter_key is not None:
chapter = (card.chapters or {}).get(chapter_key)
if chapter is None:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unknown_chapter",
translation_placeholders={
"chapter_key": chapter_key,
"card_id": card_id,
},
)
# Playing a chapter means playing it from its first track.
if track_key is None and chapter.tracks:
track_key = next(iter(chapter.tracks))
# Targeted plays (chapter or track) always start from the
# beginning; only a bare card play honours the card's own
# resume setting.
seconds_in = 0 if track_key is not None else None
try:
await client.play_card(
self._player_id,
card_id,
chapter_key=chapter_key,
track_key=track_key,
seconds_in=seconds_in,
)
except YotoError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="play_failed",
translation_placeholders={"error": str(err)},
) from err
async def async_browse_media(
self,
media_content_type: MediaType | str | None = None,
media_content_id: str | None = None,
) -> BrowseMedia:
"""Browse the Yoto card library."""
if not media_content_id:
return self._browse_root()
try:
card_id, chapter_key, _ = _parse_uri(media_content_id)
except ValueError as err:
raise BrowseError(
translation_domain=DOMAIN,
translation_key="invalid_media_id",
translation_placeholders={"media_id": media_content_id},
) from err
card = self.coordinator.client.library.get(card_id)
if card is None:
raise BrowseError(
translation_domain=DOMAIN,
translation_key="unknown_card",
translation_placeholders={"card_id": card_id},
)
if not card.chapters:
try:
await self.coordinator.client.update_card_detail(card_id)
except YotoError as err:
raise BrowseError(
translation_domain=DOMAIN,
translation_key="card_detail_failed",
translation_placeholders={"error": str(err)},
) from err
if chapter_key is not None:
chapter = (card.chapters or {}).get(chapter_key)
if chapter is None:
raise BrowseError(
translation_domain=DOMAIN,
translation_key="unknown_chapter",
translation_placeholders={
"chapter_key": chapter_key,
"card_id": card_id,
},
)
return self._browse_chapter(card_id, chapter_key, chapter)
return self._browse_card(card)
def _browse_root(self) -> BrowseMedia:
"""List every card in the user's library."""
return BrowseMedia(
media_class=MediaClass.DIRECTORY,
media_content_id="",
media_content_type=MediaType.MUSIC,
title="Yoto library",
can_play=False,
can_expand=True,
children=[
self._card_node(card)
for card in self.coordinator.client.library.values()
],
children_media_class=MediaClass.ALBUM,
)
def _browse_card(self, card: Card) -> BrowseMedia:
"""List a card's chapters, collapsing single-chapter cards to tracks."""
chapters = card.chapters or {}
# Single-chapter cards skip the chapter level: the card expands
# straight into tracks, avoiding a one-item submenu for the common
# single-story case.
if len(chapters) == 1:
chapter_key, chapter = next(iter(chapters.items()))
children = [
self._track_node(card.id, chapter_key, track_key, track)
for track_key, track in (chapter.tracks or {}).items()
]
else:
children = [
self._chapter_node(card.id, key, chapter)
for key, chapter in chapters.items()
]
node = self._card_node(card)
node.children = children
node.children_media_class = MediaClass.MUSIC
return node
def _browse_chapter(
self, card_id: str, chapter_key: str, chapter: Chapter
) -> BrowseMedia:
"""List the tracks of a chapter."""
node = self._chapter_node(card_id, chapter_key, chapter)
node.can_expand = True
node.children = [
self._track_node(card_id, chapter_key, track_key, track)
for track_key, track in (chapter.tracks or {}).items()
]
node.children_media_class = MediaClass.MUSIC
return node
def _card_node(self, card: Card) -> BrowseMedia:
"""Build a browse node for a card."""
return BrowseMedia(
media_class=MediaClass.ALBUM,
media_content_id=_build_uri(card.id),
media_content_type=MediaType.ALBUM,
title=card.title or card.id,
can_play=True,
can_expand=True,
thumbnail=card.cover_image_large,
)
def _chapter_node(
self, card_id: str, chapter_key: str, chapter: Chapter
) -> BrowseMedia:
"""Build a browse node for a chapter."""
return BrowseMedia(
media_class=MediaClass.MUSIC,
media_content_id=_build_uri(card_id, chapter_key),
media_content_type=MediaType.MUSIC,
title=chapter.title or chapter_key,
can_play=True,
can_expand=False,
thumbnail=chapter.icon,
)
def _track_node(
self, card_id: str, chapter_key: str, track_key: str, track: Track
) -> BrowseMedia:
"""Build a browse node for a track."""
return BrowseMedia(
media_class=MediaClass.MUSIC,
media_content_id=_build_uri(card_id, chapter_key, track_key),
media_content_type=MediaType.MUSIC,
title=track.title or track_key,
can_play=True,
can_expand=False,
thumbnail=track.icon,
)
async def _async_run(
self, func: Callable[..., Awaitable[Any]], /, *args: Any
) -> None:
@@ -181,3 +387,35 @@ class YotoMediaPlayer(YotoEntity, MediaPlayerEntity):
translation_key="command_failed",
translation_placeholders={"error": str(err)},
) from err
def _build_uri(
card_id: str,
chapter_key: str | None = None,
track_key: str | None = None,
) -> str:
"""Build a yoto:// URI from card/chapter/track parts."""
segments = [card_id]
if chapter_key is not None:
segments.append(chapter_key)
if track_key is not None:
segments.append(track_key)
return f"{URI_SCHEME}://{'/'.join(segments)}"
def _parse_uri(media_id: str) -> tuple[str, str | None, str | None]:
"""Parse a yoto:// URI into card/chapter/track parts.
Parsed manually so card IDs keep their original casing (URL parsers
lower-case the authority component per RFC 3986).
"""
prefix = f"{URI_SCHEME}://"
if not media_id.startswith(prefix):
raise ValueError(f"Not a Yoto media identifier: {media_id}")
parts = [segment for segment in media_id[len(prefix) :].split("/") if segment]
if not parts:
raise ValueError(f"Not a Yoto media identifier: {media_id}")
card_id = parts[0]
chapter_key = parts[1] if len(parts) > 1 else None
track_key = parts[2] if len(parts) > 2 else None
return card_id, chapter_key, track_key
homeassistant/components/yoto/strings.json
+15
View File
@@ -31,12 +31,27 @@
}
},
"exceptions": {
"card_detail_failed": {
"message": "Could not load Yoto card details: {error}"
},
"command_failed": {
"message": "Yoto command failed: {error}"
},
"invalid_media_id": {
"message": "Not a Yoto media identifier: {media_id}"
},
"oauth2_implementation_unavailable": {
"message": "[%key:common::exceptions::oauth2_implementation_unavailable::message%]"
},
"play_failed": {
"message": "Failed to play Yoto card: {error}"
},
"unknown_card": {
"message": "Unknown Yoto card: {card_id}"
},
"unknown_chapter": {
"message": "Unknown chapter {chapter_key} on card {card_id}"
},
"update_error": {
"message": "Error communicating with Yoto: {error}"
}
tests/components/yoto/conftest.py
+23 -1
View File
@@ -9,11 +9,13 @@ import jwt
import pytest
from yoto_api import (
Card,
Chapter,
Device,
PlaybackEvent,
PlaybackStatus,
PlayerInfo,
PlayerStatus,
Track,
YotoPlayer,
)
@@ -36,12 +38,32 @@ ACCESS_TOKEN = jwt.encode({"sub": USER_ID}, "test-secret-long-enough-for-hmac-sh
def _build_card() -> Card:
"""Build a representative Yoto library card."""
"""Build a representative Yoto library card with chapters and tracks."""
return Card(
id=CARD_ID,
title="Outer Space",
author="Ladybird Audio Adventures",
cover_image_large="https://example.test/cover.jpg",
chapters={
"01": Chapter(
key="01",
title="Introduction",
icon="https://example.test/ch01.png",
tracks={
"01-INT": Track(key="01-INT", title="Welcome", duration=120),
"01-MAIN": Track(
key="01-MAIN", title="The Story Begins", duration=240
),
},
),
"02": Chapter(
key="02",
title="Planets",
tracks={
"02-MER": Track(key="02-MER", title="Mercury", duration=180),
},
),
},
)
tests/components/yoto/snapshots/test_media_player.ambr
+2 -2
View File
@@ -31,7 +31,7 @@
'platform': 'yoto',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 21559>,
'supported_features': <MediaPlayerEntityFeature: 153143>,
'translation_key': None,
'unique_id': 'player-test',
'unit_of_measurement': None,
@@ -50,7 +50,7 @@
'media_position': 120,
'media_position_updated_at': datetime.datetime(2026, 5, 8, 12, 0, tzinfo=datetime.timezone.utc),
'media_title': 'Introduction',
'supported_features': <MediaPlayerEntityFeature: 21559>,
'supported_features': <MediaPlayerEntityFeature: 153143>,
'volume_level': 0.5,
}),
'context': <ANY>,
tests/components/yoto/test_media_player.py
+339 -6
View File
@@ -1,11 +1,12 @@
"""Tests for the Yoto media player platform."""
from typing import Any
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from yoto_api import YotoError
from yoto_api import Chapter, YotoError
from homeassistant.components.media_player import (
ATTR_MEDIA_SEEK_POSITION,
@@ -17,16 +18,19 @@ from homeassistant.components.media_player import (
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_MEDIA_SEEK,
SERVICE_MEDIA_STOP,
SERVICE_PLAY_MEDIA,
SERVICE_VOLUME_SET,
MediaPlayerState,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
from tests.typing import WebSocketGenerator
ENTITY_ID = "media_player.nursery_yoto"
@@ -160,22 +164,351 @@ async def test_state_idle_before_first_event(
state = hass.states.get(ENTITY_ID)
assert state is not None
assert state.state == "idle"
assert state.state == MediaPlayerState.IDLE
@pytest.mark.parametrize(
("media_content_id", "expected_call"),
[
(
"yoto://card-test",
{"chapter_key": None, "track_key": None, "seconds_in": None},
),
(
"yoto://card-test/01",
{"chapter_key": "01", "track_key": "01-INT", "seconds_in": 0},
),
(
"yoto://card-test/01/01-INT",
{"chapter_key": "01", "track_key": "01-INT", "seconds_in": 0},
),
],
)
async def test_play_media(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
media_content_id: str,
expected_call: dict[str, Any],
) -> None:
"""play_media routes a yoto:// URI to the right play_card call."""
await setup_integration(hass, mock_config_entry)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: ENTITY_ID,
"media_content_type": "music",
"media_content_id": media_content_id,
},
blocking=True,
)
mock_yoto_client.play_card.assert_called_once_with(
"player-test", "card-test", **expected_call
)
@pytest.mark.parametrize(
"media_content_id",
["spotify:track:abc", "yoto://"],
)
async def test_play_media_invalid_uri_raises(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
media_content_id: str,
) -> None:
"""A media_id that isn't a complete yoto:// URI is rejected."""
await setup_integration(hass, mock_config_entry)
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: ENTITY_ID,
"media_content_type": "music",
"media_content_id": media_content_id,
},
blocking=True,
)
@pytest.mark.parametrize(
"media_content_id",
[
pytest.param("yoto://does-not-exist", id="unknown_card"),
pytest.param("yoto://card-test/does-not-exist", id="unknown_chapter"),
],
)
async def test_play_media_unknown_target_raises(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
media_content_id: str,
) -> None:
"""A yoto:// URI pointing at unknown content is rejected."""
await setup_integration(hass, mock_config_entry)
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: ENTITY_ID,
"media_content_type": "music",
"media_content_id": media_content_id,
},
blocking=True,
)
mock_yoto_client.play_card.assert_not_called()
async def test_browse_media_root_lists_cards(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Browsing without a content id lists every library card."""
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{"id": 1, "type": "media_player/browse_media", "entity_id": ENTITY_ID}
)
response = await client.receive_json()
assert response["success"]
children = response["result"]["children"]
assert len(children) == 1
assert children[0]["title"] == "Outer Space"
assert children[0]["media_content_id"] == "yoto://card-test"
assert children[0]["can_play"] is True
assert children[0]["can_expand"] is True
async def test_browse_media_card_shows_chapters(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Browsing a multi-chapter card shows its chapters."""
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "album",
"media_content_id": "yoto://card-test",
}
)
response = await client.receive_json()
assert response["success"]
children = response["result"]["children"]
assert [c["title"] for c in children] == ["Introduction", "Planets"]
assert children[0]["media_content_id"] == "yoto://card-test/01"
async def test_browse_media_single_chapter_card_collapses_to_tracks(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""A card with a single chapter shows its tracks directly."""
card = mock_yoto_client.library["card-test"]
card.chapters = {"01": card.chapters["01"]}
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "album",
"media_content_id": "yoto://card-test",
}
)
response = await client.receive_json()
assert response["success"]
children = response["result"]["children"]
assert [c["title"] for c in children] == ["Welcome", "The Story Begins"]
assert children[0]["media_content_id"] == "yoto://card-test/01/01-INT"
async def test_browse_media_chapter_shows_tracks(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Browsing a chapter lists its tracks."""
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "playlist",
"media_content_id": "yoto://card-test/01",
}
)
response = await client.receive_json()
assert response["success"]
children = response["result"]["children"]
assert [c["title"] for c in children] == ["Welcome", "The Story Begins"]
assert children[0]["media_content_id"] == "yoto://card-test/01/01-INT"
async def test_browse_media_fetches_card_detail_lazily(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Browsing a card without loaded chapters triggers update_card_detail."""
card = mock_yoto_client.library["card-test"]
card.chapters = None
async def _populate(card_id: str) -> None:
card.chapters = {"01": Chapter(key="01", title="Intro", tracks={})}
mock_yoto_client.update_card_detail.side_effect = _populate
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "album",
"media_content_id": "yoto://card-test",
}
)
response = await client.receive_json()
assert response["success"]
mock_yoto_client.update_card_detail.assert_called_once_with("card-test")
async def test_browse_media_unknown_card_raises(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Browsing a card that's not in the library returns a browse error."""
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "album",
"media_content_id": "yoto://does-not-exist",
}
)
response = await client.receive_json()
assert response["success"] is False
async def test_browse_media_unknown_chapter_raises(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Browsing a chapter that's not in the card returns a browse error."""
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "playlist",
"media_content_id": "yoto://card-test/does-not-exist",
}
)
response = await client.receive_json()
assert response["success"] is False
async def test_browse_media_card_detail_failure_raises(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""A failure fetching card chapters bubbles up as a browse error."""
card = mock_yoto_client.library["card-test"]
card.chapters = None
mock_yoto_client.update_card_detail.side_effect = YotoError("offline")
await setup_integration(hass, mock_config_entry)
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "media_player/browse_media",
"entity_id": ENTITY_ID,
"media_content_type": "album",
"media_content_id": "yoto://card-test",
}
)
response = await client.receive_json()
assert response["success"] is False
@pytest.mark.parametrize(
("client_method", "service", "service_data"),
[
pytest.param("pause", SERVICE_MEDIA_PAUSE, {}, id="playback"),
pytest.param(
"play_card",
SERVICE_PLAY_MEDIA,
{"media_content_type": "music", "media_content_id": "yoto://card-test"},
id="play_media",
),
],
)
async def test_command_error_raises(
hass: HomeAssistant,
mock_yoto_client: MagicMock,
mock_config_entry: MockConfigEntry,
client_method: str,
service: str,
service_data: dict[str, Any],
) -> None:
"""Yoto command failures surface as HomeAssistantError."""
await setup_integration(hass, mock_config_entry)
mock_yoto_client.pause.side_effect = YotoError("nope")
getattr(mock_yoto_client, client_method).side_effect = YotoError("nope")
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PAUSE,
{ATTR_ENTITY_ID: ENTITY_ID},
service,
{ATTR_ENTITY_ID: ENTITY_ID, **service_data},
blocking=True,
)