/
/
/
1"""
2Playlist Metadata Provider for Music Assistant.
3
4Generates metadata for library playlists, including custom artwork using artist images
5and album covers, with support for multiple configurable layout templates.
6
7Future enhancements may include automatic genre detection and AI-generated descriptions.
8
9Supported artwork templates:
10- artist_mosaic: Dominant artist large in centre, others as smaller tiles around it
11- artist_grid: Equal-sized grid of unique artist images (up to 4)
12- album_grid: Classic album-cover grid (same as the built-in collage, kept as fallback)
13- artist_radio: Artist-focused layout with main artist centered on a solid background with overlapping circles of secondary artists
14- artist_banner: Full-bleed artist image with playlist name text overlay (Tidal / Apple Music Essentials style)
15- album_fan: Up to three album covers as framed photo-print cards in a rotated stack (Apple Music collage style)
16- album_grid_tilted: Album-cover grid with white gaps between tiles, the whole composition rotated ~15° (Apple Music tilted collage style)
17
18The metadata controller calls this provider during its regular refresh cycle.
19"""
20
21from __future__ import annotations
22
23import asyncio
24import contextlib
25import itertools
26import logging
27import math
28import os
29import random
30import tempfile
31from collections import Counter
32from io import BytesIO
33from pathlib import Path
34from time import time
35from typing import TYPE_CHECKING, Any, Final
36
37from music_assistant_models.background_task import TaskSchedule
38from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
39from music_assistant_models.enums import ConfigEntryType, ImageType, MediaType, ProviderFeature
40from music_assistant_models.errors import ProviderUnavailableError
41from music_assistant_models.media_items import (
42 Artist,
43 ItemMapping,
44 MediaItemImage,
45 MediaItemMetadata,
46 Playlist,
47 Track,
48)
49from music_assistant_models.unique_list import UniqueList
50from PIL import Image, ImageDraw, ImageFont, ImageOps
51
52from music_assistant.helpers.images import get_image_data
53from music_assistant.helpers.uri import parse_uri
54from music_assistant.mass import MusicAssistant
55from music_assistant.models import ProviderInstanceType
56from music_assistant.models.metadata_provider import MetadataProvider
57
58if TYPE_CHECKING:
59 from music_assistant_models.config_entries import ProviderConfig
60 from music_assistant_models.provider import ProviderManifest
61
62LOGGER = logging.getLogger(__name__)
63
64SUPPORTED_FEATURES: Final[set[ProviderFeature]] = {ProviderFeature.PLAYLIST_METADATA}
65
66CONF_TEMPLATE: Final[str] = "template"
67CONF_SKIP_PROVIDER_PLAYLISTS: Final[str] = "skip_provider_playlists"
68CONF_ENABLE_GENRE_DETECTION: Final[str] = "enable_genre_detection"
69CONF_GENRE_MIN_THRESHOLD: Final[str] = "genre_min_threshold"
70CONF_GENRE_MAX_COUNT: Final[str] = "genre_max_count"
71
72TEMPLATE_ARTIST_MOSAIC: Final[str] = "artist_mosaic"
73TEMPLATE_ARTIST_GRID: Final[str] = "artist_grid"
74TEMPLATE_ALBUM_GRID: Final[str] = "album_grid"
75TEMPLATE_ARTIST_RADIO: Final[str] = "artist_radio"
76TEMPLATE_ARTIST_BANNER: Final[str] = "artist_banner"
77TEMPLATE_ALBUM_FAN: Final[str] = "album_fan"
78TEMPLATE_ALBUM_GRID_TILTED: Final[str] = "album_grid_tilted"
79
80# Directory name inside the plugin's cache space
81_IMAGES_DIR: Final[str] = "playlist_metadata_images"
82
83
84async def setup(
85 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
86) -> ProviderInstanceType:
87 """Initialize the provider instance."""
88 return PlaylistMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
89
90
91class PlaylistMetadataProvider(MetadataProvider):
92 """Metadata provider that generates artwork and metadata for library playlists."""
93
94 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
95 """Return config entries for the provider."""
96 return (
97 ConfigEntry(
98 key=CONF_TEMPLATE,
99 type=ConfigEntryType.STRING,
100 required=True,
101 default_value=TEMPLATE_ALBUM_GRID,
102 options=[
103 ConfigValueOption(value=TEMPLATE_ALBUM_GRID),
104 ConfigValueOption(value=TEMPLATE_ALBUM_FAN),
105 ConfigValueOption(value=TEMPLATE_ALBUM_GRID_TILTED),
106 ConfigValueOption(value=TEMPLATE_ARTIST_MOSAIC),
107 ConfigValueOption(value=TEMPLATE_ARTIST_GRID),
108 ConfigValueOption(value=TEMPLATE_ARTIST_RADIO),
109 ConfigValueOption(value=TEMPLATE_ARTIST_BANNER),
110 ],
111 ),
112 ConfigEntry(
113 key=CONF_SKIP_PROVIDER_PLAYLISTS,
114 type=ConfigEntryType.BOOLEAN,
115 required=False,
116 default_value=True,
117 ),
118 ConfigEntry(
119 key=CONF_ENABLE_GENRE_DETECTION,
120 type=ConfigEntryType.BOOLEAN,
121 required=False,
122 default_value=False,
123 ),
124 ConfigEntry(
125 key=CONF_GENRE_MIN_THRESHOLD,
126 type=ConfigEntryType.INTEGER,
127 required=False,
128 default_value=10,
129 range=(5, 50),
130 advanced=True,
131 ),
132 ConfigEntry(
133 key=CONF_GENRE_MAX_COUNT,
134 type=ConfigEntryType.INTEGER,
135 required=False,
136 default_value=3,
137 range=(1, 10),
138 advanced=True,
139 ),
140 )
141
142 @property
143 def priority(self) -> int:
144 """Priority for this provider (lower = more preferred)."""
145 return 90 # Run after theaudiodb/fanart.tv but before builtin collage
146
147 _images_dir: str
148
149 async def loaded_in_mass(self) -> None:
150 """Set up the provider after it has been loaded into Music Assistant."""
151 self._images_dir = os.path.join(self.mass.storage_path, _IMAGES_DIR)
152 await asyncio.to_thread(os.makedirs, self._images_dir, exist_ok=True)
153 self.mass.tasks.register_scheduled_task(
154 task_id=f"{self.instance_id}_cleanup",
155 name="Playlist metadata cleanup",
156 handler=self._cleanup_stale_images,
157 schedule=TaskSchedule.hourly(every=2),
158 initial_delay=120,
159 )
160
161 async def unload(self, is_removed: bool = False) -> None:
162 """Unload the provider."""
163 self.mass.tasks.unregister_scheduled_task(
164 f"{self.instance_id}_cleanup", clear_persisted_state=is_removed
165 )
166
167 async def resolve_image(self, path: str) -> str | bytes:
168 """Resolve a playlist art image path to raw image bytes."""
169 file_path = os.path.join(self._images_dir, path)
170 # Validate the path stays inside our images directory (path traversal guard)
171 real_images_dir = os.path.realpath(self._images_dir)
172 real_file_path = os.path.realpath(file_path)
173 if not real_file_path.startswith(real_images_dir + os.sep):
174 msg = f"Invalid image path: {path}"
175 raise FileNotFoundError(msg)
176 if not Path(real_file_path).is_file():
177 msg = f"Playlist art image not found: {path}"
178 raise FileNotFoundError(msg)
179 # Return bytes directly so MA never passes a bare filename to ffmpeg
180 return await asyncio.to_thread(_read_bytes, real_file_path)
181
182 async def get_playlist_metadata(self, playlist: Playlist) -> MediaItemMetadata | None:
183 """
184 Generate playlist metadata (primarily artwork).
185
186 :param playlist: The playlist to generate metadata for.
187 """
188 skip_provider = self.config.get_value(CONF_SKIP_PROVIDER_PLAYLISTS)
189 if skip_provider:
190 has_builtin = any(pm.provider_domain == "builtin" for pm in playlist.provider_mappings)
191 has_smart_playlist = any(
192 pm.provider_domain == "smart_playlist" for pm in playlist.provider_mappings
193 )
194 if not has_builtin and not has_smart_playlist:
195 # Only skip if playlist has a provider-supplied image (not our generated one)
196 if playlist.metadata.images:
197 for img in playlist.metadata.images:
198 if img.type == ImageType.THUMB and not self._is_our_image(img):
199 return None
200
201 generated_images: list[MediaItemImage] = []
202 detected_genres: set[str] | None = None
203
204 if thumb_image := await self._generate_and_write(playlist, fanart=False):
205 generated_images.append(thumb_image)
206
207 if fanart_image := await self._generate_and_write(playlist, fanart=True):
208 generated_images.append(fanart_image)
209
210 # Aggregate genres from playlist tracks if enabled
211 if self.config.get_value(CONF_ENABLE_GENRE_DETECTION):
212 detected_genres = await self._analyze_playlist_genres(playlist)
213
214 if generated_images or detected_genres:
215 # Only clean up old images when we have new ones to replace them
216 if generated_images:
217 await self._cleanup_old_playlist_images(playlist)
218 metadata = MediaItemMetadata()
219 if generated_images:
220 metadata.images = UniqueList(generated_images)
221 if detected_genres:
222 metadata.genres = detected_genres
223 return metadata
224 return None
225
226 async def _analyze_playlist_genres(self, playlist: Playlist) -> set[str] | None:
227 """
228 Analyze playlist tracks and aggregate most common genres.
229
230 :param playlist: The playlist to analyze.
231 :return: Set of detected genres or None if not enough data.
232 """
233 genre_counter: Counter[str] = Counter()
234 track_count = 0
235 max_tracks = 500 # Analyze up to 500 tracks to avoid performance issues
236
237 try:
238 async for track in self.mass.music.playlists.tracks(
239 playlist.item_id,
240 playlist.provider,
241 ):
242 if not isinstance(track, Track):
243 continue
244
245 track_count += 1
246 if track.metadata and track.metadata.genres:
247 genre_counter.update(track.metadata.genres)
248
249 if track_count >= max_tracks:
250 break
251
252 if track_count == 0:
253 return None
254
255 # Calculate threshold and get top genres
256 min_threshold_val = self.config.get_value(CONF_GENRE_MIN_THRESHOLD) or 10
257 max_genre_val = self.config.get_value(CONF_GENRE_MAX_COUNT) or 3
258 min_threshold_pct = (
259 int(min_threshold_val) if isinstance(min_threshold_val, int | float) else 10
260 )
261 max_genre_count = int(max_genre_val) if isinstance(max_genre_val, int | float) else 3
262 required_count = math.ceil(track_count * min_threshold_pct / 100)
263
264 # Get most common genres that meet the threshold
265 detected = {
266 genre
267 for genre, count in genre_counter.most_common(max_genre_count)
268 if count >= required_count
269 }
270
271 return detected if detected else None
272
273 except (KeyError, AttributeError, TypeError, ValueError) as err:
274 LOGGER.debug("Failed to analyze genres for playlist %s: %s", playlist.name, err)
275 return None
276
277 # ------------------------------------------------------------------
278 # Internal helpers
279 # ------------------------------------------------------------------
280
281 def _is_our_image(self, img: MediaItemImage) -> bool:
282 """
283 Return True if this image was generated by this plugin.
284
285 Checks both the provider field and whether the path lives inside our images directory.
286 Bare filenames without directory separators are treated as builtin assets, not
287 plugin-generated images.
288 """
289 # Remote URLs and data URIs are never ours, regardless of provider
290 if img.path.startswith(("http://", "https://", "data:")):
291 return False
292 if img.provider == self.instance_id:
293 return True
294 # A bare filename without any directory separator (e.g. "logo.png", "fanart.jpg")
295 # is a builtin provider asset, not something we generated.
296 if "/" not in img.path and "\\" not in img.path:
297 return False
298 images_dir = Path(self._images_dir).resolve()
299 img_path = Path(img.path)
300 resolved = (
301 img_path.resolve() if img_path.is_absolute() else (images_dir / img_path).resolve()
302 )
303 with contextlib.suppress(ValueError):
304 resolved.relative_to(images_dir)
305 return True
306 return False
307
308 async def _cleanup_old_playlist_images(self, playlist: Playlist) -> None:
309 """
310 Remove old images generated by this provider for the given playlist.
311
312 This ensures that when the template changes or metadata is regenerated,
313 old images are immediately removed rather than waiting for the periodic cleanup.
314
315 :param playlist: The playlist whose old images should be removed.
316 """
317 if not playlist.metadata.images:
318 return
319
320 images_to_remove = [img for img in playlist.metadata.images if self._is_our_image(img)]
321 if not images_to_remove:
322 return
323
324 images_dir = Path(self._images_dir).resolve()
325 for img in images_to_remove:
326 img_path = Path(img.path)
327 file_path = img_path if img_path.is_absolute() else images_dir / img_path.name
328 with contextlib.suppress(OSError):
329 await asyncio.to_thread(file_path.unlink, missing_ok=True)
330
331 playlist.metadata.images = UniqueList(
332 [img for img in playlist.metadata.images if img not in images_to_remove]
333 )
334
335 async def _cleanup_stale_images(self) -> None:
336 """Remove stale DB references and orphaned image files from storage."""
337 referenced_paths: set[str] = set()
338 images_dir_resolved = Path(self._images_dir).resolve()
339
340 async for playlist in self.mass.music.playlists.iter_library_items():
341 if not playlist.metadata.images:
342 continue
343 stale = []
344 for img in playlist.metadata.images:
345 if not self._is_our_image(img):
346 continue
347 exists = await asyncio.to_thread(
348 os.path.isfile, images_dir_resolved / Path(img.path).name
349 )
350 if not exists:
351 stale.append(img)
352 if stale:
353 stale_paths = {img.path for img in stale}
354 fresh = await self.mass.music.playlists.get_library_item(playlist.item_id)
355 if fresh.metadata.images:
356 fresh.metadata.images = UniqueList(
357 [img for img in fresh.metadata.images if img.path not in stale_paths]
358 )
359 await self.mass.music.playlists.update_item_in_library(
360 fresh.item_id, fresh, overwrite=True
361 )
362
363 # Collect paths still in use after cleanup (use fresh object if we updated it)
364 source = fresh.metadata.images if stale else playlist.metadata.images
365 for img in source or []:
366 if self._is_our_image(img):
367 # Normalise to basename: path may be absolute or relative depending on
368 # which version of the plugin wrote the record.
369 referenced_paths.add(Path(img.path).name)
370
371 # Delete any files in the images directory not referenced by any playlist.
372 # Keep files that were written in the last 5 minutes to cover the race between
373 # file write and DB update (e.g. artwork just generated but metadata not yet refreshed).
374 for fname in await asyncio.to_thread(os.listdir, self._images_dir):
375 if fname in referenced_paths:
376 continue
377 fpath = Path(self._images_dir) / fname
378 try:
379 mtime = await asyncio.to_thread(fpath.stat)
380 except OSError:
381 continue
382 if time() - mtime.st_mtime < 300:
383 continue
384 with contextlib.suppress(OSError):
385 fpath.unlink()
386
387 async def _get_smart_playlist_rules(self, playlist: Playlist) -> dict[str, Any] | None:
388 """
389 Get the Smart Playlist rules for a playlist if it's a smart playlist.
390
391 :param playlist: The playlist to check.
392 :return: The rules dictionary or None if not a smart playlist.
393 """
394 try:
395 smart_playlist_provider = None
396 for provider in self.mass.providers:
397 if provider.domain == "smart_playlist":
398 smart_playlist_provider = provider
399 break
400
401 if not smart_playlist_provider:
402 return None
403
404 smart_mapping = None
405 for pm in playlist.provider_mappings:
406 if pm.provider_domain == "smart_playlist":
407 smart_mapping = pm
408 break
409
410 if not smart_mapping:
411 return None
412
413 if not hasattr(smart_playlist_provider, "get_smart_playlist_rules"):
414 LOGGER.debug("Smart playlist provider not fully initialized yet")
415 return None
416
417 return await smart_playlist_provider.get_smart_playlist_rules( # type: ignore[union-attr, no-any-return]
418 smart_mapping.item_id
419 )
420 except Exception as err:
421 LOGGER.debug("Could not get smart playlist rules for %s: %s", playlist.name, err)
422 return None
423
424 async def _collect_images_from_smart_playlist_rules(
425 self, rules: dict[str, Any]
426 ) -> tuple[list[MediaItemImage], list[MediaItemImage]]:
427 """
428 Collect images based on smart playlist rules.
429
430 Returns (primary_images, secondary_images) where primary are artist/album images
431 from the rules, and secondary are additional images that could be used.
432
433 :param rules: The smart playlist rules dictionary.
434 :return: Tuple of (primary_images, secondary_images).
435 """
436 artist_images: dict[str, MediaItemImage] = {}
437 album_images: dict[str, MediaItemImage] = {}
438
439 for artist_id in rules.get("artist_ids", []):
440 try:
441 artist = await self.mass.music.artists.get_library_item(artist_id)
442 if artist and artist.image:
443 artist_images[artist.name] = artist.image
444 except Exception as err:
445 LOGGER.debug("Could not get artist %s: %s", artist_id, err)
446
447 for uri in rules.get("seed_artist_uris", []):
448 try:
449 media_type, provider_instance_id, item_id = await parse_uri(uri)
450 if media_type != MediaType.ARTIST:
451 continue
452 if not self.mass.get_provider(provider_instance_id):
453 LOGGER.debug("Provider %s not available for URI %s", provider_instance_id, uri)
454 continue
455 artist = await self.mass.music.artists.get(item_id, provider_instance_id)
456 if artist and artist.image:
457 artist_images[artist.name] = artist.image
458 except ProviderUnavailableError:
459 LOGGER.debug("Provider not available for URI %s", uri)
460 except Exception as err:
461 LOGGER.debug("Could not get artist from URI %s: %s", uri, err)
462
463 for album_id in rules.get("album_ids", []):
464 try:
465 album = await self.mass.music.albums.get_library_item(album_id)
466 if album and album.image:
467 album_images[album.name] = album.image
468 except Exception as err:
469 LOGGER.debug("Could not get album %s: %s", album_id, err)
470
471 for uri in rules.get("seed_album_uris", []):
472 try:
473 media_type, provider_instance_id, item_id = await parse_uri(uri)
474 if media_type != MediaType.ALBUM:
475 continue
476 if not self.mass.get_provider(provider_instance_id):
477 LOGGER.debug("Provider %s not available for URI %s", provider_instance_id, uri)
478 continue
479 album = await self.mass.music.albums.get(item_id, provider_instance_id)
480 if album and album.image:
481 album_images[album.name] = album.image
482 except ProviderUnavailableError:
483 LOGGER.debug("Provider not available for URI %s", uri)
484 except Exception as err:
485 LOGGER.debug("Could not get album from URI %s: %s", uri, err)
486
487 # TODO: Could also collect images from genre_ids by getting artists from those genres
488 # For now, we'll use what we have and fall back to track analysis if needed
489
490 return list(artist_images.values()), list(album_images.values())
491
492 async def _generate_and_write( # noqa: PLR0915
493 self,
494 playlist: Playlist,
495 template: str | None = None,
496 fanart: bool = False,
497 ) -> MediaItemImage | None:
498 """
499 Render playlist artwork and write to disk.
500
501 :param playlist: The playlist to generate artwork for.
502 :param template: Template override (None uses configured default).
503 :param fanart: If True, generate FANART image; if False, generate THUMB image.
504 """
505 effective_template = template or str(self.config.get_value(CONF_TEMPLATE))
506
507 artist_images: dict[str, MediaItemImage] = {} # artist_name â image (deduped)
508 artist_count: dict[str, int] = {} # artist_name â track count
509 album_image_map: dict[str, MediaItemImage] = {} # path â image (deduped)
510 album_count: dict[str, int] = {} # path â track count
511
512 smart_rules = await self._get_smart_playlist_rules(playlist)
513 if smart_rules:
514 LOGGER.debug(
515 "Playlist %s is a smart playlist, collecting images from rules", playlist.name
516 )
517 (
518 rules_artist_images,
519 rules_album_images,
520 ) = await self._collect_images_from_smart_playlist_rules(smart_rules)
521
522 if rules_artist_images or rules_album_images:
523 for img in rules_artist_images:
524 key = img.path
525 if key not in artist_images:
526 artist_images[key] = img
527 artist_count[key] = 1
528
529 for img in rules_album_images:
530 _path = img.path
531 if _path not in album_image_map:
532 album_image_map[_path] = img
533 album_count[_path] = 1
534
535 LOGGER.debug(
536 "Smart playlist %s: collected %d artist images, %d album images from rules",
537 playlist.name,
538 len(artist_images),
539 len(album_image_map),
540 )
541 if len(artist_images) < 4 and len(album_image_map) < 4:
542 LOGGER.debug("Not enough images from rules, falling back to track analysis")
543 smart_rules = None
544 artist_images.clear()
545 artist_count.clear()
546 album_image_map.clear()
547 album_count.clear()
548 else:
549 LOGGER.debug(
550 "No images found in smart playlist rules, falling back to track analysis"
551 )
552 smart_rules = None
553
554 if not smart_rules or (len(artist_images) < 4 and len(album_image_map) < 4):
555 async for track in self.mass.music.playlists.tracks(
556 playlist.item_id, playlist.provider
557 ):
558 if not isinstance(track, Track):
559 continue
560
561 # Collect album image, counting how often each album appears
562 if track.image:
563 _path = track.image.path
564 album_image_map[_path] = track.image
565 album_count[_path] = album_count.get(_path, 0) + 1
566
567 # Collect artist images + count occurrences
568 for artist_or_mapping in track.artists:
569 artist_name = artist_or_mapping.name
570 artist_count[artist_name] = artist_count.get(artist_name, 0) + 1
571 if artist_name in artist_images:
572 continue
573 artist_img = await self._get_artist_image(artist_or_mapping)
574 if artist_img:
575 artist_images[artist_name] = artist_img
576
577 await asyncio.sleep(0) # yield to event loop
578
579 # Sort album images by frequency so the most-seen album comes first.
580 # This lets album-grid templates place the dominant album at a prominent position.
581 sorted_album_paths = sorted(
582 album_image_map, key=lambda p: album_count.get(p, 0), reverse=True
583 )
584 album_images = [album_image_map[p] for p in sorted_album_paths]
585
586 if effective_template in (
587 TEMPLATE_ALBUM_GRID,
588 TEMPLATE_ALBUM_FAN,
589 TEMPLATE_ALBUM_GRID_TILTED,
590 ):
591 images = album_images
592 secondary_images: list[MediaItemImage] = []
593 else:
594 # Sort artists by frequency so the most-used artist image comes first
595 sorted_artists = sorted(
596 artist_images, key=lambda n: artist_count.get(n, 0), reverse=True
597 )
598 images = [artist_images[n] for n in sorted_artists]
599 secondary_images = album_images
600 if not images:
601 # Fall back to album images if no artist images were found
602 images = album_images
603 secondary_images = []
604
605 LOGGER.debug(
606 "Playlist %s: %d artist image(s), %d album image(s), template=%s",
607 playlist.name,
608 len(artist_images),
609 len(album_images),
610 effective_template,
611 )
612
613 if not images:
614 return None
615
616 try:
617 img_data = await self._render(
618 effective_template, images, secondary_images, playlist_name=playlist.name
619 )
620 except Exception as err:
621 LOGGER.warning("Failed to generate playlist artwork for %s: %s", playlist.name, err)
622 return None
623
624 # Use timestamp in filename to bust browser cache when tracks change.
625 # Frontend uses absolute path with provider="builtin" so builtin's resolve_image
626 # is bypassed and the file is served directly via os.path.isfile().
627 image_type = ImageType.FANART if fanart else ImageType.THUMB
628 suffix = "fanart" if fanart else "thumb"
629 filename = f"{playlist.item_id}_{int(time())}_{suffix}.jpg"
630 file_path = os.path.join(self._images_dir, filename)
631
632 # mkstemp prevents collisions if multiple calls regenerate the same playlist concurrently
633 tmp_fd, tmp_path = await asyncio.to_thread(
634 tempfile.mkstemp, dir=self._images_dir, suffix=".tmp"
635 )
636 await asyncio.to_thread(os.close, tmp_fd)
637 await asyncio.to_thread(_write_bytes, tmp_path, img_data)
638 await asyncio.to_thread(os.replace, tmp_path, file_path)
639
640 return MediaItemImage(
641 type=image_type,
642 path=file_path,
643 provider=self.instance_id,
644 remotely_accessible=False,
645 )
646
647 async def _get_artist_image(self, artist: Artist | ItemMapping) -> MediaItemImage | None:
648 """Return a THUMB image for the given artist, checking the library first."""
649 # Both Artist and ItemMapping have an .image attribute; check it first
650 if artist.image:
651 return artist.image
652
653 # Exact library lookup by provider mapping (faster and more accurate than name search)
654 try:
655 library_artist = await self.mass.music.artists.get_library_item_by_prov_id(
656 artist.item_id, artist.provider
657 )
658 if library_artist and library_artist.image:
659 return library_artist.image
660 except Exception as err:
661 LOGGER.debug("Artist provider lookup failed for %s: %s", artist.name, err)
662
663 # Fall back to name search in case the provider mapping differs
664 try:
665 results = await self.mass.music.artists.library_items(
666 search=artist.name,
667 limit=1,
668 summary=False,
669 )
670 if results and results[0].image:
671 return results[0].image
672 except Exception as err:
673 LOGGER.debug("Artist library lookup failed for %s: %s", artist.name, err)
674
675 return None
676
677 async def _render(
678 self,
679 template: str,
680 images: list[MediaItemImage],
681 secondary_images: list[MediaItemImage] | None = None,
682 playlist_name: str = "",
683 ) -> bytes:
684 """Dispatch to the correct renderer."""
685 if template == TEMPLATE_ARTIST_MOSAIC:
686 return await _render_artist_mosaic(self.mass, images, secondary_images or [])
687 if template == TEMPLATE_ARTIST_GRID:
688 return await _render_artist_grid(self.mass, images, secondary_images or [])
689 if template == TEMPLATE_ARTIST_RADIO:
690 return await _render_artist_radio(self.mass, images)
691 if template == TEMPLATE_ARTIST_BANNER:
692 return await _render_artist_banner(
693 self.mass, images, secondary_images or [], playlist_name
694 )
695 if template == TEMPLATE_ALBUM_FAN:
696 return await _render_album_fan(self.mass, images)
697 if template == TEMPLATE_ALBUM_GRID_TILTED:
698 return await _render_album_grid_tilted(self.mass, images)
699 # album_grid / fallback
700 return await _render_album_grid(self.mass, images)
701
702
703# ---------------------------------------------------------------------------
704# Image renderers (run in thread pool via asyncio.to_thread)
705# ---------------------------------------------------------------------------
706
707_CANVAS_SIZE: Final[int] = 1500
708_TILE_SIZE: Final[int] = 375 # 4x4 grid
709_MOSAIC_MAIN_SIZE: Final[int] = 1000
710_MOSAIC_SMALL_SIZE: Final[int] = 500
711
712
713async def _render_artist_mosaic(
714 mass: MusicAssistant,
715 images: list[MediaItemImage],
716 secondary_images: list[MediaItemImage],
717) -> bytes:
718 """
719 Render the 'Artist Mosaic' template.
720
721 Dominant artist fills most of the canvas; up to five secondary artists
722 are arranged as smaller squares on the right/bottom edge.
723 If fewer than two artist images are available, album covers fill the secondary slots.
724 """
725 if not images:
726 msg = "No images provided"
727 raise ValueError(msg)
728
729 main_data = await _fetch(mass, images[0])
730
731 # Use remaining artist images for secondary slots; pad with album covers if needed
732 secondary_candidates = images[1:] + secondary_images
733 secondary = []
734 seen: set[str] = {images[0].path}
735 for img in secondary_candidates:
736 if img.path in seen:
737 continue
738 seen.add(img.path)
739 data = await _fetch(mass, img)
740 if data:
741 secondary.append(data)
742 if len(secondary) >= 5:
743 break
744
745 def _compose() -> bytes:
746 canvas = Image.new("RGB", (_CANVAS_SIZE, _CANVAS_SIZE), (30, 30, 30))
747
748 if main_data:
749 main_img = Image.open(BytesIO(main_data)).convert("RGB")
750 # Fill most of the canvas
751 fill_w = _CANVAS_SIZE if not secondary else _MOSAIC_MAIN_SIZE
752 fill_h = _CANVAS_SIZE if not secondary else _MOSAIC_MAIN_SIZE
753 main_img = main_img.resize((fill_w, fill_h))
754 canvas.paste(main_img, (0, 0))
755
756 # Up to 5 secondary images in 500x500 tiles in the remaining corners
757 positions = [
758 (_MOSAIC_MAIN_SIZE, 0),
759 (_MOSAIC_MAIN_SIZE, _MOSAIC_SMALL_SIZE),
760 (0, _MOSAIC_MAIN_SIZE),
761 (_MOSAIC_SMALL_SIZE, _MOSAIC_MAIN_SIZE),
762 (_MOSAIC_MAIN_SIZE, _MOSAIC_MAIN_SIZE),
763 ]
764 for i, sec_data in enumerate(secondary[:5]):
765 sec_img = Image.open(BytesIO(sec_data)).convert("RGB")
766 sec_img = sec_img.resize((_MOSAIC_SMALL_SIZE, _MOSAIC_SMALL_SIZE))
767 canvas.paste(sec_img, positions[i])
768
769 buf = BytesIO()
770 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
771 return buf.getvalue()
772
773 return await asyncio.to_thread(_compose)
774
775
776async def _render_artist_grid(
777 mass: MusicAssistant,
778 images: list[MediaItemImage],
779 secondary_images: list[MediaItemImage],
780) -> bytes:
781 """
782 Render the 'Artist Grid' template.
783
784 Up to four unique artist images in equal-sized tiles (2x2).
785 If fewer than four artist images are available, album covers fill the remaining slots.
786 """
787 random.shuffle(images)
788 # Deduplicate and merge: artist images first, then album covers to fill up to 4 slots
789 seen: set[str] = set()
790 candidates: list[MediaItemImage] = []
791 for img in images + secondary_images:
792 if img.path not in seen:
793 seen.add(img.path)
794 candidates.append(img)
795 if len(candidates) >= 4:
796 break
797 slots = candidates
798 fetched: list[bytes | None] = [await _fetch(mass, img) for img in slots]
799
800 def _compose() -> bytes:
801 canvas = Image.new("RGB", (_CANVAS_SIZE, _CANVAS_SIZE), (30, 30, 30))
802 coords = [
803 (0, 0),
804 (_CANVAS_SIZE // 2, 0),
805 (0, _CANVAS_SIZE // 2),
806 (_CANVAS_SIZE // 2, _CANVAS_SIZE // 2),
807 ]
808 half = _CANVAS_SIZE // 2
809 for data, (x, y) in zip(fetched, coords, strict=False):
810 if not data:
811 continue
812 img = Image.open(BytesIO(data)).convert("RGB").resize((half, half))
813 canvas.paste(img, (x, y))
814
815 buf = BytesIO()
816 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
817 return buf.getvalue()
818
819 return await asyncio.to_thread(_compose)
820
821
822async def _render_album_grid(mass: MusicAssistant, images: list[MediaItemImage]) -> bytes:
823 """
824 Render the classic album-grid template (identical to the built-in collage).
825
826 250x250 tiles tiled across the canvas, with repetition if needed.
827 """
828 tile_size = 250
829 canvas_size = _CANVAS_SIZE
830
831 random.shuffle(images)
832
833 # Pre-fetch exactly as many unique images as tiles needed (with repetition via cycle)
834 tiles_x = math.ceil(canvas_size / tile_size)
835 tiles_y = math.ceil(canvas_size / tile_size)
836 tile_count = tiles_x * tiles_y
837
838 seen: set[str] = set()
839 selected: list[MediaItemImage] = []
840 for img in images:
841 if img.path not in seen:
842 seen.add(img.path)
843 selected.append(img)
844 if len(selected) >= tile_count:
845 break
846
847 fetched: dict[str, bytes | None] = {}
848 for img in selected:
849 fetched[img.path] = await _fetch(mass, img)
850
851 available = [img for img in selected if fetched.get(img.path)]
852 iter_images = itertools.cycle(available) if available else None
853
854 def _compose() -> bytes:
855 canvas = Image.new("RGB", (canvas_size, canvas_size), (30, 30, 30))
856 if iter_images is None:
857 buf = BytesIO()
858 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
859 return buf.getvalue()
860 for x in range(0, canvas_size, tile_size):
861 for y in range(0, canvas_size, tile_size):
862 img_obj = next(iter_images)
863 data = fetched[img_obj.path]
864 tile = Image.open(BytesIO(data)).convert("RGB").resize((tile_size, tile_size)) # type: ignore[arg-type]
865 canvas.paste(tile, (x, y))
866
867 buf = BytesIO()
868 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
869 return buf.getvalue()
870
871 return await asyncio.to_thread(_compose)
872
873
874async def _render_artist_radio(
875 mass: MusicAssistant,
876 images: list[MediaItemImage],
877) -> bytes:
878 """
879 Render the 'Artist Radio' template.
880
881 Replicates the Apple Music artist radio station style: the primary artist fills
882 a large circle centered on a solid background filled with the most prominent
883 color of that artist image. Additional artists appear as smaller circles
884 arranged around the edge.
885 """
886 if not images:
887 msg = "No images provided"
888 raise ValueError(msg)
889
890 fetched: list[bytes | None] = [await _fetch(mass, img) for img in images[:5]]
891 main_data = fetched[0]
892
893 def _dominant_color(data: bytes) -> tuple[int, int, int]:
894 """Return the background color of an image by sampling its corners and edges."""
895 img = Image.open(BytesIO(data)).convert("RGB").resize((100, 100))
896 # Sample the four corners + midpoints of each edge (16 samples total)
897 sample_coords = [
898 (0, 0),
899 (1, 0),
900 (0, 1),
901 (1, 1), # top-left corner
902 (98, 0),
903 (99, 0),
904 (98, 1),
905 (99, 1), # top-right corner
906 (0, 98),
907 (1, 98),
908 (0, 99),
909 (1, 99), # bottom-left corner
910 (98, 98),
911 (99, 98),
912 (98, 99),
913 (99, 99), # bottom-right corner
914 ]
915 r_sum = g_sum = b_sum = 0
916 for x, y in sample_coords:
917 pixel = img.getpixel((x, y))
918 r_sum += pixel[0] # type: ignore[index]
919 g_sum += pixel[1] # type: ignore[index]
920 b_sum += pixel[2] # type: ignore[index]
921 n = len(sample_coords)
922 # Darken slightly so the circles pop
923 factor = 0.75
924 return (int(r_sum / n * factor), int(g_sum / n * factor), int(b_sum / n * factor))
925
926 def _compose() -> bytes:
927 bg_color = _dominant_color(main_data) if main_data else (15, 15, 15)
928 canvas = Image.new("RGB", (_CANVAS_SIZE, _CANVAS_SIZE), bg_color)
929
930 def _paste_circle(data: bytes, cx: int, cy: int, size: int) -> None:
931 """Paste a circular-cropped image centred at (cx, cy)."""
932 img = Image.open(BytesIO(data)).convert("RGBA").resize((size, size))
933 mask = Image.new("L", (size, size), 0)
934 ImageDraw.Draw(mask).ellipse((0, 0, size - 1, size - 1), fill=255)
935 layer = Image.new("RGBA", (_CANVAS_SIZE, _CANVAS_SIZE), (0, 0, 0, 0))
936 layer.paste(img, (cx - size // 2, cy - size // 2), mask)
937 canvas.paste(layer.convert("RGB"), (0, 0), layer.split()[3])
938
939 centre = _CANVAS_SIZE // 2
940
941 # Main artist â large circle
942 if main_data:
943 main_circle = int(_CANVAS_SIZE * 0.65)
944 _paste_circle(main_data, centre, centre, main_circle)
945
946 # Up to 4 secondary artists as small circles around the edge
947 secondary = [d for d in fetched[1:] if d]
948 if secondary:
949 small_size = int(_CANVAS_SIZE * 0.22)
950 radius = int(_CANVAS_SIZE * 0.42)
951 for i, sec_data in enumerate(secondary[:4]):
952 angle = math.pi / 4 + i * (math.pi / 2) # 45°, 135°, 225°, 315°
953 sx = centre + int(radius * math.cos(angle))
954 sy = centre + int(radius * math.sin(angle))
955 _paste_circle(sec_data, sx, sy, small_size)
956
957 buf = BytesIO()
958 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
959 return buf.getvalue()
960
961 return await asyncio.to_thread(_compose)
962
963
964def _get_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
965 """Return a TrueType font at the given size, falling back to Pillow's built-in."""
966 font_search_paths = [
967 "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
968 "/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
969 "/System/Library/Fonts/Helvetica.ttc",
970 "/Library/Fonts/Arial Bold.ttf",
971 "/Windows/Fonts/arialbd.ttf",
972 ]
973 for path in font_search_paths:
974 if Path(path).exists():
975 try:
976 return ImageFont.truetype(path, size)
977 except OSError:
978 pass
979 try:
980 return ImageFont.load_default(size=size)
981 except TypeError:
982 return ImageFont.load_default()
983
984
985async def _render_artist_banner(
986 mass: MusicAssistant,
987 images: list[MediaItemImage],
988 secondary_images: list[MediaItemImage],
989 playlist_name: str,
990) -> bytes:
991 """
992 Render the 'Artist Banner' template.
993
994 The dominant artist image fills the canvas. A dark gradient is applied at the
995 top so that the playlist name drawn in white remains legible â similar to the
996 style used by Tidal artist mixes and Apple Music Essentials playlists.
997 Falls back to album images when no artist image is available.
998 """
999 candidates = images or secondary_images
1000 if not candidates:
1001 msg = "No images provided"
1002 raise ValueError(msg)
1003
1004 main_data = await _fetch(mass, candidates[0])
1005
1006 def _compose() -> bytes:
1007 canvas = Image.new("RGB", (_CANVAS_SIZE, _CANVAS_SIZE), (30, 30, 30))
1008
1009 if main_data:
1010 bg = Image.open(BytesIO(main_data)).convert("RGB")
1011 bg = ImageOps.fit(bg, (_CANVAS_SIZE, _CANVAS_SIZE))
1012 canvas.paste(bg, (0, 0))
1013
1014 # Dark-to-transparent gradient across the top half for text legibility
1015 overlay = Image.new("RGBA", (_CANVAS_SIZE, _CANVAS_SIZE), (0, 0, 0, 0))
1016 gradient_height = _CANVAS_SIZE // 2
1017 draw_ov = ImageDraw.Draw(overlay)
1018 for y in range(gradient_height):
1019 alpha = int(200 * (1 - y / gradient_height) ** 0.7)
1020 draw_ov.line([(0, y), (_CANVAS_SIZE, y)], fill=(0, 0, 0, alpha))
1021 canvas = Image.alpha_composite(canvas.convert("RGBA"), overlay).convert("RGB")
1022
1023 # Draw playlist name at the top left
1024 draw = ImageDraw.Draw(canvas)
1025 margin = int(_CANVAS_SIZE * 0.06)
1026 font = _get_font(int(_CANVAS_SIZE * 0.075))
1027 draw.text((margin, margin), playlist_name, fill=(255, 255, 255), font=font)
1028
1029 buf = BytesIO()
1030 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
1031 return buf.getvalue()
1032
1033 return await asyncio.to_thread(_compose)
1034
1035
1036async def _render_album_fan(
1037 mass: MusicAssistant,
1038 images: list[MediaItemImage],
1039) -> bytes:
1040 """
1041 Render the 'Album Fan' template.
1042
1043 Up to three album covers are framed as photo-print cards (white border) and
1044 composed in a slightly rotated, overlapping stack on a dark background â
1045 similar to the Apple Music playlist collage style.
1046 """
1047 if not images:
1048 msg = "No images provided"
1049 raise ValueError(msg)
1050
1051 fetched: list[bytes | None] = [await _fetch(mass, img) for img in images[:3]]
1052 available = [d for d in fetched if d]
1053
1054 def _compose() -> bytes:
1055 canvas = Image.new("RGB", (_CANVAS_SIZE, _CANVAS_SIZE), (18, 18, 18))
1056
1057 card_content = int(_CANVAS_SIZE * 0.52)
1058 border = int(card_content * 0.06)
1059 card_total = card_content + 2 * border
1060 centre = _CANVAS_SIZE // 2
1061
1062 # Rotation (degrees clockwise) and centre offsets (pixels) for up to 3 cards, back to front
1063 if len(available) == 1:
1064 card_configs: list[tuple[int, int, int]] = [(4, 0, 0)]
1065 elif len(available) == 2:
1066 card_configs = [
1067 (-10, -int(_CANVAS_SIZE * 0.12), int(_CANVAS_SIZE * 0.03)),
1068 (4, int(_CANVAS_SIZE * 0.08), -int(_CANVAS_SIZE * 0.03)),
1069 ]
1070 else:
1071 card_configs = [
1072 (-14, -int(_CANVAS_SIZE * 0.14), int(_CANVAS_SIZE * 0.04)),
1073 (-3, -int(_CANVAS_SIZE * 0.03), int(_CANVAS_SIZE * 0.01)),
1074 (8, int(_CANVAS_SIZE * 0.10), -int(_CANVAS_SIZE * 0.03)),
1075 ]
1076
1077 for data, (angle, x_off, y_off) in zip(available, card_configs, strict=False):
1078 # Build photo-print card: white frame + album image inset
1079 card = Image.new("RGB", (card_total, card_total), (245, 245, 245))
1080 album = Image.open(BytesIO(data)).convert("RGB").resize((card_content, card_content))
1081 card.paste(album, (border, border))
1082
1083 # Rotate with transparent padding so canvas background shows in the corners
1084 rotated = card.convert("RGBA").rotate(
1085 -angle, expand=True, resample=Image.Resampling.BICUBIC
1086 )
1087 rx, ry = rotated.size
1088 px = centre + x_off - rx // 2
1089 py = centre + y_off - ry // 2
1090 canvas.paste(rotated.convert("RGB"), (px, py), rotated.split()[3])
1091
1092 buf = BytesIO()
1093 canvas.convert("RGB").save(buf, "JPEG", optimize=True, quality=85)
1094 return buf.getvalue()
1095
1096 return await asyncio.to_thread(_compose)
1097
1098
1099async def _render_album_grid_tilted( # noqa: PLR0915
1100 mass: MusicAssistant,
1101 images: list[MediaItemImage],
1102) -> bytes:
1103 """
1104 Render the 'Album Grid Tilted' template.
1105
1106 Builds a regular grid of album covers separated by white gutters on a dark background,
1107 then rotates the entire composition ~15° so the grid bleeds off all four edges â
1108 the Apple Music tilted-collage look.
1109 """
1110 if not images:
1111 msg = "No images provided"
1112 raise ValueError(msg)
1113
1114 # 5x5 grid: tile=360, gutter=18, grid=1872 px, which is large enough
1115 # (1500*(cos15+sin15) ~= 1837 px) so no black corners appear after the
1116 # -15 rotation and centre-crop to 1500 px.
1117 # images[0] is the most-frequently-occurring album and goes in the centre cell.
1118 tile = 360
1119 gutter = 18
1120 cols = 5
1121 rows = 5
1122 grid_w = cols * tile + (cols - 1) * gutter
1123 grid_h = rows * tile + (rows - 1) * gutter
1124 center_row = rows // 2
1125 center_col = cols // 2
1126 tile_count = cols * rows
1127
1128 # images is already sorted by frequency (most common first)
1129 seen: set[str] = set()
1130 deduped: list[MediaItemImage] = []
1131 for img in images:
1132 if img.path not in seen:
1133 seen.add(img.path)
1134 deduped.append(img)
1135
1136 fetched: dict[str, bytes | None] = {}
1137 for img in deduped[:tile_count]:
1138 fetched[img.path] = await _fetch(mass, img)
1139
1140 available = [img for img in deduped if fetched.get(img.path)]
1141 center_img = available[0] if available else None
1142 other_imgs = available[1:] if len(available) > 1 else available
1143 random.shuffle(other_imgs)
1144 iter_others = itertools.cycle(other_imgs) if other_imgs else None
1145
1146 def _compose() -> bytes:
1147 # Build the grid on a white surface so gutters are white
1148 grid = Image.new("RGB", (grid_w, grid_h), (255, 255, 255))
1149 for row in range(rows):
1150 for col in range(cols):
1151 if row == center_row and col == center_col:
1152 img_obj = center_img
1153 elif iter_others is not None:
1154 img_obj = next(iter_others)
1155 else:
1156 img_obj = None
1157 if img_obj is None or not fetched.get(img_obj.path):
1158 continue
1159 data = fetched[img_obj.path]
1160 if not data:
1161 continue
1162 cell = Image.open(BytesIO(data)).convert("RGB").resize((tile, tile))
1163 x = col * (tile + gutter)
1164 y = row * (tile + gutter)
1165 grid.paste(cell, (x, y))
1166
1167 # Rotate the grid ~15° with expand so no corner is clipped, then centre-crop
1168 rotated = grid.convert("RGBA").rotate(-15, expand=True, resample=Image.Resampling.BICUBIC)
1169 # Centre-crop to final canvas
1170 rx, ry = rotated.size
1171 cx = (rx - _CANVAS_SIZE) // 2
1172 cy = (ry - _CANVAS_SIZE) // 2
1173 cropped = rotated.crop((cx, cy, cx + _CANVAS_SIZE, cy + _CANVAS_SIZE))
1174
1175 # Composite over dark background so any transparent corners look intentional
1176 canvas = Image.new("RGB", (_CANVAS_SIZE, _CANVAS_SIZE), (20, 20, 20))
1177 canvas.paste(cropped.convert("RGB"), (0, 0), cropped.split()[3])
1178
1179 buf = BytesIO()
1180 canvas.save(buf, "JPEG", optimize=True, quality=85)
1181 return buf.getvalue()
1182
1183 return await asyncio.to_thread(_compose)
1184
1185
1186async def _fetch(mass: MusicAssistant, image: MediaItemImage) -> bytes | None:
1187 """Fetch raw image bytes, returning None on failure."""
1188 try:
1189 return await get_image_data(mass, image.path, image.provider)
1190 except Exception as err:
1191 LOGGER.debug(
1192 "Failed to fetch image path=%s provider=%s: %s", image.path, image.provider, err
1193 )
1194 return None
1195
1196
1197def _write_bytes(path: str, data: bytes) -> None:
1198 """Write bytes to a file (used in thread pool)."""
1199 with open(path, "wb") as fh:
1200 fh.write(data)
1201
1202
1203def _read_bytes(path: str) -> bytes:
1204 """Read bytes from a file (used in thread pool)."""
1205 with open(path, "rb") as fh:
1206 return fh.read()
1207