Add support for getting playlist tracks.

custom_components/mass_queue/actions.py

@@ -62,6 +62,7 @@
     MOVE_QUEUE_ITEM_NEXT_SERVICE_SCHEMA,
     MOVE_QUEUE_ITEM_UP_SERVICE_SCHEMA,
     PLAY_QUEUE_ITEM_SERVICE_SCHEMA,
+    PLAYLIST_ITEM_SCHEMA,
     QUEUE_ITEM_SCHEMA,
     QUEUE_ITEMS_SERVICE_SCHEMA,
     REMOVE_QUEUE_ITEM_SERVICE_SCHEMA,
@@ -70,6 +71,7 @@
 )
 from .utils import (
     find_image,
+    parse_uri,
 )
 
 if TYPE_CHECKING:
@@ -344,6 +346,48 @@ async def unfavorite_item(self, call: ServiceCall) -> ServiceResponse:
             library_item_id=item_id,
         )
 
+    async def get_playlist_items(self, playlist_uri: str):
+        """Retrieves all playlist items."""
+        provider, item_id = parse_uri(playlist_uri)
+        LOGGER.debug(
+            f"Getting playlist items for provider {provider}, item_id {item_id}",
+        )
+        resp = await self._client.music.get_playlist_tracks(item_id, provider)
+        LOGGER.debug(f"Got response with {len(resp) if resp else 0} items")
+        result = [self.format_playlist_item(item.to_dict()) for item in resp]
+        msg = f"Got response {result[0]}"
+        if len(msg) > 200:
+            msg = f"{msg[180]}..." + "}"
+        LOGGER.debug(msg)
+        return result
+
+    def format_playlist_item(self, playlist_item: dict) -> dict:
+        """Processes the individual items in a playlist."""
+        media_title = playlist_item.get("name") or "N/A"
+        media_album = playlist_item.get("album") or "N/A"
+        media_album_name = "" if media_album is None else media_album.get("name", "")
+        media_content_id = playlist_item["uri"]
+        media_image = find_image(playlist_item) or ""
+        local_image_encoded = playlist_item.get(ATTR_LOCAL_IMAGE_ENCODED)
+        favorite = playlist_item["favorite"]
+
+        artists = playlist_item["artists"]
+        artist_names = [artist["name"] for artist in artists]
+        media_artist = ", ".join(artist_names)
+        response: ServiceResponse = PLAYLIST_ITEM_SCHEMA(
+            {
+                ATTR_MEDIA_TITLE: media_title,
+                ATTR_MEDIA_ALBUM_NAME: media_album_name,
+                ATTR_MEDIA_ARTIST: media_artist,
+                ATTR_MEDIA_CONTENT_ID: media_content_id,
+                ATTR_MEDIA_IMAGE: media_image,
+                ATTR_FAVORITE: favorite,
+            },
+        )
+        if local_image_encoded:
+            response[ATTR_LOCAL_IMAGE_ENCODED] = local_image_encoded
+        return response
+
 
 @callback
 def get_music_assistant_client(

custom_components/mass_queue/const.py

@@ -11,6 +11,7 @@
 DEFAULT_NAME = "Music Assistant Queue Items"
 SERVICE_CLEAR_QUEUE_FROM_HERE = "clear_queue_from_here"
 SERVICE_GET_GROUP_VOLUME = "get_group_volume"
+SERVICE_GET_PLAYLIST_TRACKS = "get_playlist_tracks"
 SERVICE_GET_QUEUE_ITEMS = "get_queue_items"
 SERVICE_GET_RECOMMENDATIONS = "get_recommendations"
 SERVICE_PLAY_QUEUE_ITEM = "play_queue_item"
@@ -37,6 +38,7 @@
 ATTR_MEDIA_TITLE = "media_title"
 ATTR_OFFSET = "offset"
 ATTR_PLAYER_ENTITY = "entity"
+ATTR_URI = "uri"
 ATTR_PROVIDERS = "providers"
 ATTR_QUEUE_ID = "active_queue"
 ATTR_QUEUE_ITEM_ID = "queue_item_id"

custom_components/mass_queue/icons.json

@@ -1,5 +1,6 @@
 {
   "services": {
+    "get_playlist_tracks": { "service": "mdi:playlist-music"},
     "get_queue_items": { "service": "mdi:playlist-music" },
     "move_queue_item_down": { "service": "mdi:arrow-down" },
     "move_queue_item_next": { "service": "mdi:arrow-collapse-up" },

custom_components/mass_queue/schemas.py

@@ -24,6 +24,7 @@
     ATTR_PROVIDERS,
     ATTR_QUEUE_ITEM_ID,
     ATTR_QUEUE_ITEMS,
+    ATTR_URI,
     ATTR_VOLUME_LEVEL,
 )
 
@@ -52,6 +53,18 @@
     },
 )
 
+PLAYLIST_ITEM_SCHEMA = vol.Schema(
+    {
+        vol.Required(ATTR_MEDIA_TITLE): str,
+        vol.Required(ATTR_MEDIA_ALBUM_NAME): str,
+        vol.Required(ATTR_MEDIA_ARTIST): str,
+        vol.Required(ATTR_MEDIA_CONTENT_ID): str,
+        vol.Required(ATTR_MEDIA_IMAGE): str,
+        vol.Required(ATTR_FAVORITE): bool,
+        vol.Optional(ATTR_LOCAL_IMAGE_ENCODED): str,
+    },
+)
+
 QUEUE_DETAILS_SCHEMA = vol.Schema(
     {
         vol.Required(ATTR_QUEUE_ITEMS): vol.All(
@@ -109,6 +122,13 @@
     },
 )
 
+GET_PLAYLIST_TRACKS_SERVICE_SCHEMA = vol.Schema(
+    {
+        vol.Required(ATTR_CONFIG_ENTRY_ID): str,
+        vol.Required(ATTR_URI): str,
+    },
+)
+
 GET_RECOMMENDATIONS_SERVICE_SCHEMA = vol.Schema(
     {
         vol.Required(ATTR_PLAYER_ENTITY): str,

custom_components/mass_queue/services.py

@@ -12,10 +12,12 @@
     ATTR_CONFIG_ENTRY_ID,
     ATTR_PLAYER_ENTITY,
     ATTR_QUEUE_ITEM_ID,
+    ATTR_URI,
     DOMAIN,
     LOGGER,
     SERVICE_CLEAR_QUEUE_FROM_HERE,
     SERVICE_GET_GROUP_VOLUME,
+    SERVICE_GET_PLAYLIST_TRACKS,
     SERVICE_GET_QUEUE_ITEMS,
     SERVICE_GET_RECOMMENDATIONS,
     SERVICE_MOVE_QUEUE_ITEM_DOWN,
@@ -30,6 +32,7 @@
 from .schemas import (
     CLEAR_QUEUE_FROM_HERE_SERVICE_SCHEMA,
     GET_GROUP_VOLUME_SERVICE_SCHEMA,
+    GET_PLAYLIST_TRACKS_SERVICE_SCHEMA,
     GET_RECOMMENDATIONS_SERVICE_SCHEMA,
     MOVE_QUEUE_ITEM_DOWN_SERVICE_SCHEMA,
     MOVE_QUEUE_ITEM_NEXT_SERVICE_SCHEMA,
@@ -131,6 +134,13 @@ def register_actions(hass) -> None:
         schema=CLEAR_QUEUE_FROM_HERE_SERVICE_SCHEMA,
         supports_response=SupportsResponse.NONE,
     )
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_GET_PLAYLIST_TRACKS,
+        get_playlist_tracks,
+        schema=GET_PLAYLIST_TRACKS_SERVICE_SCHEMA,
+        supports_response=SupportsResponse.ONLY,
+    )
 
 
 async def get_queue_items(call: ServiceCall):
@@ -252,3 +262,15 @@ async def clear_queue_from_here(call: ServiceCall):
     for item in items:
         queue_item_id = item[ATTR_QUEUE_ITEM_ID]
         await client.player_queues.queue_command_delete(queue_id, queue_item_id)
+
+
+async def get_playlist_tracks(call: ServiceCall):
+    """Gets all tracks in a playlist."""
+    config_entry = call.data[ATTR_CONFIG_ENTRY_ID]
+    uri = call.data[ATTR_URI]
+    hass = call.hass
+    entry = hass.config_entries.async_get_entry(config_entry)
+    actions = entry.runtime_data.actions
+    return {
+        "tracks": await actions.get_playlist_items(uri),
+    }

custom_components/mass_queue/services.yaml

@@ -217,3 +217,18 @@ clear_queue_from_here:
         entity:
           domain: media_player
           integration: music_assistant
+get_playlist_tracks:
+  fields:
+    config_entry_id:
+      name: Config Entry ID
+      required: true
+      selector:
+        config_entry:
+          integration: mass_queue
+    uri:
+      name: Playlist URI
+      description: URI for the playlist
+      selector:
+        text:
+      example: "library://playlist/12"
+      required: true

custom_components/mass_queue/strings.json

@@ -242,6 +242,20 @@
           "description": "Music Assistant Media Player Entity."
         }
       }
+    },
+    "get_playlist_tracks": {
+      "name": "Get Playlist Tracks",
+      "description": "Returns all tracks for the playlist given.",
+      "fields": {
+        "config_entry_id": {
+          "name": "Config Entry ID",
+          "description": "Config Entry ID for the Music Assistant Queue Items instance."
+        },
+        "uri": {
+          "name": "Playlist URI",
+          "description": "URI for the playlist."
+        }
+      }
     }
   }
 }

custom_components/mass_queue/translations/en.json

@@ -218,6 +218,20 @@
           "description": "Music Assistant Media Player Entity."
         }
       }
+    },
+    "get_playlist_tracks": {
+      "name": "Get Playlist Tracks",
+      "description": "Returns all tracks for the playlist given.",
+      "fields": {
+        "config_entry_id": {
+          "name": "Config Entry ID",
+          "description": "Config Entry ID for the Music Assistant Queue Items instance."
+        },
+        "uri": {
+          "name": "Playlist URI",
+          "description": "URI for the playlist."
+        }
+      }
     }
   }
 }

custom_components/mass_queue/utils.py

@@ -147,8 +147,7 @@ def find_image_from_image(data: dict, remotely_accessible: bool):
 
 def find_image_from_metadata(data: dict, remotely_accessible: bool):
     """Attempts to find the image via the metadata key."""
-    media_item = data.get("media_item", {})
-    metadata = media_item.get("metadata", {})
+    metadata = data.get("metadata", {})
     img_data = metadata.get("images")
     if img_data is None:
         return None
@@ -168,16 +167,19 @@ def find_image_from_album(data: dict, remotely_accessible: bool):
 def find_image_from_artists(data: dict, remotely_accessible: bool):
     """Attempts to find the image via the artists key."""
     artist = data.get("artist", {})
-    img_data = artist.get("image")
-    if img_data is list:
+    img_data = artist.get("image") or []
+    img_data += artist.get("metadata") or []
+    if len(img_data):
         return search_image_list(img_data, remotely_accessible)
     return return_image_or_none(img_data, remotely_accessible)
 
 
 def find_image(data: dict, remotely_accessible: bool = True):
     """Returns None if image is not present or not remotely accessible."""
+    media_item = data.get("media_item", data)
+
     from_image = find_image_from_image(data, remotely_accessible)
-    from_metadata = find_image_from_metadata(data, remotely_accessible)
+    from_metadata = find_image_from_metadata(media_item, remotely_accessible)
     from_album = find_image_from_album(data, remotely_accessible)
     from_artists = find_image_from_artists(data, remotely_accessible)
     return from_image or from_metadata or from_album or from_artists
@@ -354,3 +356,10 @@ def get_entity_info(hass: HomeAssistant, entity_id: str):
         "synced_to": synced_to,
         "type": player_type,
     }
+
+
+def parse_uri(uri):
+    """Parse a URI and split to provider and item ID."""
+    provider = uri.split("://")[0]
+    item_id = uri.split("/")[-1]
+    return [provider, item_id]

custom_components/mass_queue/websocket_commands.py

@@ -90,3 +90,18 @@ async def api_download_images(
         image["encoded"] = img
         result.append(image)
     connection.send_result(msg["id"], result)
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "mass_queue/get_playlist_items",
+        vol.Required("playlist_uri"): str,
+    },
+)
+@websocket_api.async_response
+async def get_playlist_items(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict,
+) -> None:
+    """Retrieves all playlist items."""