github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-08-02 23:40:32 +02:00
Code Releases Activity

Plex media browser improvements (#56312)

Browse Source
This commit is contained in:
jjlawren 2021-09-29 13:17:55 -05:00 committed by GitHub
parent ef13e473cf
commit 8f4ba564d4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 244 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
homeassistant/components/plex/helpers.py Normal file
View File

@ -0,0 +1,24 @@
"""Helper methods for common Plex integration operations."""
def pretty_title(media, short_name=False):
"""Return a formatted title for the given media item."""
year = None
if media.type == "album":
title = f"{media.parentTitle} - {media.title}"
elif media.type == "episode":
title = f"{media.seasonEpisode.upper()} - {media.title}"
if not short_name:
title = f"{media.grandparentTitle} - {title}"
elif media.type == "track":
title = f"{media.index}. {media.title}"
else:
title = media.title
if media.type in ["album", "movie", "season"]:
year = media.year
if year:
title += f" ({year!s})"
return title

33
homeassistant/components/plex/media_browser.py
View File

@ -1,4 +1,5 @@
"""Support to interface with the Plex API."""
from itertools import islice
import logging
from homeassistant.components.media_player import BrowseMedia
@ -17,6 +18,7 @@ from homeassistant.components.media_player.const import (
from homeassistant.components.media_player.errors import BrowseError
from .const import DOMAIN
from .helpers import pretty_title
class UnknownMediaType(BrowseError):
@ -32,9 +34,10 @@ PLAYLISTS_BROWSE_PAYLOAD = {
"can_play": False,
"can_expand": True,
}
SPECIAL_METHODS = {
"On Deck": "onDeck",
"Recently Added": "recentlyAdded",
LIBRARY_PREFERRED_LIBTYPE = {
"show": "episode",
"artist": "album",
}
ITEM_TYPE_MEDIA_CLASS = {
@ -57,7 +60,7 @@ def browse_media( # noqa: C901
):
"""Implement the websocket media browsing helper."""
def item_payload(item):
def item_payload(item, short_name=False):
"""Create response payload for a single media item."""
try:
media_class = ITEM_TYPE_MEDIA_CLASS[item.type]
@ -65,7 +68,7 @@ def browse_media( # noqa: C901
_LOGGER.debug("Unknown type received: %s", item.type)
raise UnknownMediaType from err
payload = {
"title": item.title,
"title": pretty_title(item, short_name),
"media_class": media_class,
"media_content_id": str(item.ratingKey),
"media_content_type": item.type,
@ -129,7 +132,7 @@ def browse_media( # noqa: C901
media_info.children = []
for item in media:
try:
media_info.children.append(item_payload(item))
media_info.children.append(item_payload(item, short_name=True))
except UnknownMediaType:
continue
return media_info
@ -180,8 +183,22 @@ def browse_media( # noqa: C901
"children_media_class": children_media_class,
}
method = SPECIAL_METHODS[special_folder]
items = getattr(library_or_section, method)()
if special_folder == "On Deck":
items = library_or_section.onDeck()
elif special_folder == "Recently Added":
if library_or_section.TYPE:
libtype = LIBRARY_PREFERRED_LIBTYPE.get(
library_or_section.TYPE, library_or_section.TYPE
)
items = library_or_section.recentlyAdded(libtype=libtype)
else:
recent_iter = (
x
for x in library_or_section.search(sort="addedAt:desc", limit=100)
if x.type in ["album", "episode", "movie"]
)
items = list(islice(recent_iter, 30))
for item in items:
try:
payload["children"].append(item_payload(item))

215
tests/components/plex/test_browse_media.py
View File

@ -1,15 +1,74 @@
"""Tests for Plex media browser."""
from unittest.mock import patch
from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
)
from homeassistant.components.plex.const import CONF_SERVER_IDENTIFIER
from homeassistant.components.plex.media_browser import SPECIAL_METHODS
from homeassistant.components.websocket_api.const import ERR_UNKNOWN_ERROR, TYPE_RESULT
from .const import DEFAULT_DATA
class MockPlexShow:
"""Mock a plexapi Season instance."""
ratingKey = 30
title = "TV Show"
type = "show"
def __iter__(self):
"""Iterate over episodes."""
yield MockPlexSeason()
class MockPlexSeason:
"""Mock a plexapi Season instance."""
ratingKey = 20
title = "Season 1"
type = "season"
year = 2021
def __iter__(self):
"""Iterate over episodes."""
yield MockPlexEpisode()
class MockPlexEpisode:
"""Mock a plexapi Episode instance."""
ratingKey = 10
title = "Episode 1"
grandparentTitle = "TV Show"
seasonEpisode = "s01e01"
type = "episode"
class MockPlexAlbum:
"""Mock a plexapi Album instance."""
ratingKey = 200
parentTitle = "Artist"
title = "Album"
type = "album"
year = 2001
def __iter__(self):
"""Iterate over tracks."""
yield MockPlexTrack()
class MockPlexTrack:
"""Mock a plexapi Track instance."""
index = 1
ratingKey = 100
title = "Track 1"
type = "track"
async def test_browse_media(
hass,
hass_ws_client,
@ -58,15 +117,13 @@ async def test_browse_media(
result = msg["result"]
assert result[ATTR_MEDIA_CONTENT_TYPE] == "server"
assert result[ATTR_MEDIA_CONTENT_ID] == DEFAULT_DATA[CONF_SERVER_IDENTIFIER]
# Library Sections + Special Sections + Playlists
assert (
len(result["children"])
== len(mock_plex_server.library.sections()) + len(SPECIAL_METHODS) + 1
)
# Library Sections + On Deck + Recently Added + Playlists
assert len(result["children"]) == len(mock_plex_server.library.sections()) + 3
music = next(iter(x for x in result["children"] if x["title"] == "Music"))
tvshows = next(iter(x for x in result["children"] if x["title"] == "TV Shows"))
playlists = next(iter(x for x in result["children"] if x["title"] == "Playlists"))
special_keys = list(SPECIAL_METHODS.keys())
special_keys = ["On Deck", "Recently Added"]
# Browse into a special folder (server)
msg_id += 1
@ -144,23 +201,34 @@ async def test_browse_media(
result = msg["result"]
assert result[ATTR_MEDIA_CONTENT_TYPE] == "library"
result_id = int(result[ATTR_MEDIA_CONTENT_ID])
assert len(result["children"]) == len(
mock_plex_server.library.sectionByID(result_id).all()
) + len(SPECIAL_METHODS)
# All items in section + On Deck + Recently Added
assert (
len(result["children"])
== len(mock_plex_server.library.sectionByID(result_id).all()) + 2
)
# Browse into a Plex TV show
msg_id += 1
await websocket_client.send_json(
{
"id": msg_id,
"type": "media_player/browse_media",
"entity_id": media_players[0],
ATTR_MEDIA_CONTENT_TYPE: result["children"][-1][ATTR_MEDIA_CONTENT_TYPE],
ATTR_MEDIA_CONTENT_ID: str(result["children"][-1][ATTR_MEDIA_CONTENT_ID]),
}
)
mock_show = MockPlexShow()
mock_season = next(iter(mock_show))
with patch.object(
mock_plex_server, "fetch_item", return_value=mock_show
) as mock_fetch:
await websocket_client.send_json(
{
"id": msg_id,
"type": "media_player/browse_media",
"entity_id": media_players[0],
ATTR_MEDIA_CONTENT_TYPE: result["children"][-1][
ATTR_MEDIA_CONTENT_TYPE
],
ATTR_MEDIA_CONTENT_ID: str(
result["children"][-1][ATTR_MEDIA_CONTENT_ID]
),
}
)
msg = await websocket_client.receive_json()
msg = await websocket_client.receive_json()
assert msg["id"] == msg_id
assert msg["type"] == TYPE_RESULT
assert msg["success"]
@ -168,6 +236,90 @@ async def test_browse_media(
assert result[ATTR_MEDIA_CONTENT_TYPE] == "show"
result_id = int(result[ATTR_MEDIA_CONTENT_ID])
assert result["title"] == mock_plex_server.fetch_item(result_id).title
assert result["children"][0]["title"] == f"{mock_season.title} ({mock_season.year})"
# Browse into a Plex TV show season
msg_id += 1
mock_episode = next(iter(mock_season))
with patch.object(
mock_plex_server, "fetch_item", return_value=mock_season
) as mock_fetch:
await websocket_client.send_json(
{
"id": msg_id,
"type": "media_player/browse_media",
"entity_id": media_players[0],
ATTR_MEDIA_CONTENT_TYPE: result["children"][0][ATTR_MEDIA_CONTENT_TYPE],
ATTR_MEDIA_CONTENT_ID: str(
result["children"][0][ATTR_MEDIA_CONTENT_ID]
),
}
)
msg = await websocket_client.receive_json()
assert mock_fetch.called
assert msg["id"] == msg_id
assert msg["type"] == TYPE_RESULT
assert msg["success"]
result = msg["result"]
assert result[ATTR_MEDIA_CONTENT_TYPE] == "season"
result_id = int(result[ATTR_MEDIA_CONTENT_ID])
assert result["title"] == f"{mock_season.title} ({mock_season.year})"
assert (
result["children"][0]["title"]
== f"{mock_episode.seasonEpisode.upper()} - {mock_episode.title}"
)
# Browse into a Plex music library
msg_id += 1
await websocket_client.send_json(
{
"id": msg_id,
"type": "media_player/browse_media",
"entity_id": media_players[0],
ATTR_MEDIA_CONTENT_TYPE: music[ATTR_MEDIA_CONTENT_TYPE],
ATTR_MEDIA_CONTENT_ID: str(music[ATTR_MEDIA_CONTENT_ID]),
}
)
msg = await websocket_client.receive_json()
assert msg["success"]
result = msg["result"]
result_id = int(result[ATTR_MEDIA_CONTENT_ID])
assert result[ATTR_MEDIA_CONTENT_TYPE] == "library"
assert result["title"] == "Music"
# Browse into a Plex album
msg_id += 1
mock_album = MockPlexAlbum()
with patch.object(
mock_plex_server, "fetch_item", return_value=mock_album
) as mock_fetch:
await websocket_client.send_json(
{
"id": msg_id,
"type": "media_player/browse_media",
"entity_id": media_players[0],
ATTR_MEDIA_CONTENT_TYPE: result["children"][-1][
ATTR_MEDIA_CONTENT_TYPE
],
ATTR_MEDIA_CONTENT_ID: str(
result["children"][-1][ATTR_MEDIA_CONTENT_ID]
),
}
)
msg = await websocket_client.receive_json()
assert mock_fetch.called
assert msg["success"]
result = msg["result"]
result_id = int(result[ATTR_MEDIA_CONTENT_ID])
assert result[ATTR_MEDIA_CONTENT_TYPE] == "album"
assert (
result["title"]
== f"{mock_album.parentTitle} - {mock_album.title} ({mock_album.year})"
)
# Browse into a non-existent TV season
unknown_key = 99999999999999
@ -211,3 +363,26 @@ async def test_browse_media(
result = msg["result"]
assert result[ATTR_MEDIA_CONTENT_TYPE] == "playlists"
result_id = result[ATTR_MEDIA_CONTENT_ID]
# Browse recently added items
msg_id += 1
mock_items = [MockPlexAlbum(), MockPlexEpisode(), MockPlexSeason(), MockPlexTrack()]
with patch("plexapi.library.Library.search", return_value=mock_items) as mock_fetch:
await websocket_client.send_json(
{
"id": msg_id,
"type": "media_player/browse_media",
"entity_id": media_players[0],
ATTR_MEDIA_CONTENT_TYPE: "server",
ATTR_MEDIA_CONTENT_ID: f"{DEFAULT_DATA[CONF_SERVER_IDENTIFIER]}:{special_keys[1]}",
}
)
msg = await websocket_client.receive_json()
assert msg["success"]
result = msg["result"]
assert result[ATTR_MEDIA_CONTENT_TYPE] == "server"
result_id = result[ATTR_MEDIA_CONTENT_ID]
for child in result["children"]:
assert child["media_content_type"] in ["album", "episode"]
assert child["media_content_type"] not in ["season", "track"]