from mashumaro import DataClassDictMixin
from .enums import MediaType
-from .media_items import ItemMapping, Radio, StreamDetails, Track
+from .media_items import ItemMapping, MediaItemImage, Radio, StreamDetails, Track
if TYPE_CHECKING:
- from music_assistant.server import MusicAssistant
+ pass
@dataclass
sort_index: int = 0
streamdetails: StreamDetails | None = None
media_item: Track | Radio | None = None
- image_url: str | None = None
+ image: MediaItemImage | None = None
def __post_init__(self):
"""Set default values."""
return MediaType.UNKNOWN
@classmethod
- def from_media_item(cls, queue_id: str, media_item: Track | Radio):
+ def from_media_item(cls, queue_id: str, media_item: Track | Radio) -> QueueItem:
"""Construct QueueItem from track/radio item."""
if media_item.media_type == MediaType.TRACK:
artists = "/".join(x.name for x in media_item.artists)
name=name,
duration=media_item.duration,
media_item=media_item,
- image_url=media_item.image.path if media_item.image else None,
+ image=get_image(media_item),
)
- async def resolve_image_url(self, mass: MusicAssistant) -> None:
- """Resolve Image URL for the MediaItem."""
- if self.image_url and self.image_url.startswith("http"):
- return
- if not self.media_item:
- return
- self.image_url = await mass.metadata.get_image_url_for_item(self.media_item, resolve=True)
+
+def get_image(media_item: Track | Radio | None) -> MediaItemImage | None:
+ """Find the Image for the MediaItem."""
+ if not media_item:
+ return None
+ if media_item.image:
+ return media_item.image
+ if isinstance(media_item, Track) and media_item.album and getattr(media_item.album, "image"):
+ return media_item.album.image
+ return None
if img.provider != "url" and not resolve:
continue
if img.provider != "url" and resolve:
- # return imageproxy url for images that need to be resolved
- # the original path is double encoded
- encoded_url = urllib.parse.quote(urllib.parse.quote(img.path))
- return f"{self.mass.webserver.base_url}/imageproxy?path={encoded_url}"
+ return self.get_image_url(img)
return img.path
# retry with track's album
return None
+ def get_image_url(self, image: MediaItemImage) -> str:
+ """Get (proxied) URL for MediaItemImage."""
+ if image.provider != "url":
+ # return imageproxy url for images that need to be resolved
+ # the original path is double encoded
+ encoded_url = urllib.parse.quote(urllib.parse.quote(image.path))
+ return f"{self.mass.webserver.base_url}/imageproxy?path={encoded_url}"
+ return image.path
+
async def get_thumbnail(
self, path: str, size: int | None = None, provider: str = "url", base64: bool = False
) -> bytes | str:
player_prov = self.mass.players.get_player_provider(queue_id)
flow_mode = self.mass.config.get_player_config_value(queue.queue_id, CONF_FLOW_MODE)
queue.flow_mode = flow_mode.value
- # make sure that the queue item image is resolved
- await queue_item.resolve_image_url(self.mass)
await player_prov.cmd_play_media(
queue_id,
queue_item=queue_item,
next_item = self.get_item(queue.queue_id, next_index)
if not next_item:
raise QueueEmpty("No more tracks left in the queue.")
- # make sure that the queue item image is resolved
- await next_item.resolve_image_url(self.mass)
queue.index_in_buffer = next_index
# work out crossfade
crossfade = queue.crossfade_enabled
if TYPE_CHECKING:
from music_assistant.common.models.queue_item import QueueItem
+ from music_assistant.server import MusicAssistant
# ruff: noqa: E501
-def create_didl_metadata(url: str, queue_item: QueueItem, flow_mode: bool = False) -> str:
+def create_didl_metadata(
+ mass: MusicAssistant, url: str, queue_item: QueueItem, flow_mode: bool = False
+) -> str:
"""Create DIDL metadata string from url and QueueItem."""
ext = url.split(".")[-1]
is_radio = queue_item.media_type != MediaType.TRACK or not queue_item.duration
+ image_url = mass.metadata.get_image_url(queue_item.image) if queue_item.image else ""
if flow_mode:
return (
if is_radio:
# radio or other non-track item
- image = queue_item.image_url if queue_item.image_url else ""
return (
'<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/" xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/" xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:dlna="urn:schemas-dlna-org:metadata-1-0/">'
f'<item id="{queue_item.queue_item_id}" parentID="0" restricted="1">'
f"<dc:title>{_escape_str(queue_item.name)}</dc:title>"
- f"<upnp:albumArtURI>{_escape_str(queue_item.image_url)}</upnp:albumArtURI>"
- f"<dc:queueItemId>{image}</dc:queueItemId>"
+ f"<upnp:albumArtURI>{_escape_str(image_url)}</upnp:albumArtURI>"
+ f"<dc:queueItemId>{queue_item.queue_item_id}</dc:queueItemId>"
"<upnp:class>object.item.audioItem.audioBroadcast</upnp:class>"
f"<upnp:mimeType>audio/{ext}</upnp:mimeType>"
f'<res protocolInfo="http-get:*:audio/{ext}:DLNA.ORG_OP=00;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=0d500000000000000000000000000000">{url}</res>'
f"<upnp:duration>{queue_item.duration}</upnp:duration>"
"<upnp:playlistTitle>Music Assistant</upnp:playlistTitle>"
f"<dc:queueItemId>{queue_item.queue_item_id}</dc:queueItemId>"
- f"<upnp:albumArtURI>{queue_item.image_url}</upnp:albumArtURI>"
+ f"<upnp:albumArtURI>{image_url}</upnp:albumArtURI>"
f"<upnp:class>{item_class}</upnp:class>"
f"<upnp:mimeType>audio/{ext}</upnp:mimeType>"
f'<res duration="{duration_str}" protocolInfo="http-get:*:audio/{ext}:DLNA.ORG_OP=00;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=0d500000000000000000000000000000">{url}</res>'
castplayer.status_listener = None
self.castplayers.pop(castplayer.player_id, None)
- @staticmethod
- def _create_queue_item(queue_item: QueueItem, stream_url: str):
+ def _create_queue_item(self, queue_item: QueueItem, stream_url: str):
"""Create CC queue item from MA QueueItem."""
duration = int(queue_item.duration) if queue_item.duration else None
+ image_url = self.mass.metadata.get_image_url(queue_item.image) if queue_item.image else ""
if queue_item.media_type == MediaType.TRACK and queue_item.media_item:
stream_type = STREAM_TYPE_BUFFERED
metadata = {
if queue_item.media_item.artists
else "",
"title": queue_item.name,
- "images": [{"url": queue_item.image_url}] if queue_item.image_url else None,
+ "images": [{"url": image_url}] if image_url else None,
}
else:
stream_type = STREAM_TYPE_LIVE
metadata = {
"metadataType": 0,
"title": queue_item.name,
- "images": [{"url": queue_item.image_url}] if queue_item.image_url else None,
+ "images": [{"url": image_url}] if image_url else None,
}
return {
"autoplay": True,
flow_mode=flow_mode,
)
- didl_metadata = create_didl_metadata(url, queue_item, flow_mode)
+ didl_metadata = create_didl_metadata(self.mass, url, queue_item, flow_mode)
await dlna_player.device.async_set_transport_uri(url, queue_item.name, didl_metadata)
# Play it
await dlna_player.device.async_wait_for_can_play(10)
# DLNA pre-caches pretty aggressively so do not yet start the runner
auto_start_runner=False,
)
- didl_metadata = create_didl_metadata(url, next_item)
+ didl_metadata = create_didl_metadata(self.mass, url, next_item)
try:
await dlna_player.device.async_set_next_transport_uri(
url, next_item.name, didl_metadata
start_index : start_index + limit
]
# we ignore the tags, just always send all info
- return player_status_from_mass(player=player, queue=queue, queue_items=queue_items)
+ return player_status_from_mass(
+ self.mass, player=player, queue=queue, queue_items=queue_items
+ )
def _handle_mixer(
self,
from music_assistant.common.models.player import Player
from music_assistant.common.models.player_queue import PlayerQueue
from music_assistant.common.models.queue_item import QueueItem
+ from music_assistant.server import MusicAssistant
# ruff: noqa: UP013
)
-def playlist_item_from_mass(queue_item: QueueItem, index: int = 0) -> PlaylistItem:
+def playlist_item_from_mass(
+ mass: MusicAssistant, queue_item: QueueItem, index: int = 0
+) -> PlaylistItem:
"""Parse PlaylistItem for the Json RPC interface from MA QueueItem."""
if queue_item.media_item and queue_item.media_type == MediaType.TRACK:
artist = queue_item.media_item.artists[0].name if queue_item.media_item.artists else ""
artist = ""
album = ""
title = queue_item.name
+ image_url = mass.metadata.get_image_url(queue_item.image) if queue_item.image else ""
return {
"playlist index": index,
"id": queue_item.queue_item_id,
"genre": "",
"remote": 0,
"remote_title": queue_item.streamdetails.stream_title if queue_item.streamdetails else "",
- "artwork_url": queue_item.image_url or "",
+ "artwork_url": image_url,
"bitrate": "",
"duration": queue_item.duration or 0,
"coverid": "-94099753136392",
def player_status_from_mass(
- player: Player, queue: PlayerQueue, queue_items: list[QueueItem]
+ mass: MusicAssistant, player: Player, queue: PlayerQueue, queue_items: list[QueueItem]
) -> PlayerStatusResponse:
"""Parse PlayerStatusResponse for the Json RPC interface from MA info."""
return {
"rate": 1,
"playlist_tracks": queue.items,
"playlist_loop": [
- playlist_item_from_mass(item, queue.current_index + index)
+ playlist_item_from_mass(mass, item, queue.current_index + index)
for index, item in enumerate(queue_items)
],
}
if radio_mode:
sonos_player.radio_mode_started = time.time()
url = url.replace("http", "x-rincon-mp3radio")
- metadata = create_didl_metadata(url, queue_item, flow_mode)
+ metadata = create_didl_metadata(self.mass, url, queue_item, flow_mode)
# sonos does multiple get requests if no duration is known
# our stream engine does not like that, hence the workaround
self.mass.streams.workaround_players.add(sonos_player.player_id)
flow_mode: bool = False,
) -> None:
"""Enqueue a queue item to the Sonos player Queue."""
- metadata = create_didl_metadata(url, queue_item, flow_mode)
+ metadata = create_didl_metadata(self.mass, url, queue_item, flow_mode)
await asyncio.to_thread(
sonos_player.soco.avTransport.AddURIToQueue,
[
projects / music-assistant-server.git / commitdiff