Smart playlist plugin

music_assistant/controllers/media/artists.py

@@ -6,7 +6,7 @@
 import contextlib
 from typing import TYPE_CHECKING, Any, cast
 
-from music_assistant_models.enums import AlbumType, MediaType, ProviderFeature
+from music_assistant_models.enums import AlbumType, MediaType, ProviderFeature, ProviderType
 from music_assistant_models.errors import (
     MediaNotFoundError,
     MusicAssistantError,
@@ -25,10 +25,12 @@
 from music_assistant.helpers.compare import compare_artist, compare_strings, create_safe_string
 from music_assistant.helpers.database import UNSET
 from music_assistant.helpers.json import serialize_to_json
+from music_assistant.models.music_provider import MusicProvider
 
 if TYPE_CHECKING:
     from music_assistant import MusicAssistant
-    from music_assistant.models.music_provider import MusicProvider
+    from music_assistant.models.metadata_provider import MetadataProvider
+    from music_assistant.models.plugin import PluginProvider
 
 
 class ArtistsController(MediaControllerBase[Artist]):
@@ -46,6 +48,48 @@ def __init__(self, mass: MusicAssistant) -> None:
         api_base = self.api_base
         self.mass.register_api_command(f"music/{api_base}/artist_albums", self.albums)
         self.mass.register_api_command(f"music/{api_base}/artist_tracks", self.tracks)
+        self.mass.register_api_command(f"music/{api_base}/similar_artists", self.similar_artists)
+
+    async def similar_artists(
+        self,
+        item_id: str,
+        provider_instance_id_or_domain: str,
+        limit: int = 25,
+    ) -> list[Artist]:
+        """
+        Get a list of similar artists for the given artist.
+
+        :param item_id: The item ID of the artist.
+        :param provider_instance_id_or_domain: The provider instance ID or domain.
+        :param limit: Maximum number of similar artists to return.
+        """
+        ref_item = await self.get(item_id, provider_instance_id_or_domain)
+        # Try music providers mapped to the reference artist first.
+        for prov_mapping in ref_item.provider_mappings:
+            prov = self.mass.get_provider(prov_mapping.provider_instance)
+            if prov is None or not isinstance(prov, MusicProvider):
+                continue
+            if ProviderFeature.SIMILAR_ARTISTS not in prov.supported_features:
+                continue
+            try:
+                if result := await prov.get_similar_artists(
+                    prov_artist_id=prov_mapping.item_id, limit=limit
+                ):
+                    return result
+            except NotImplementedError:
+                continue
+        # Fallback: metadata / plugin providers.
+        for prov in self.mass.get_providers_supporting_feature(
+            ProviderFeature.SIMILAR_ARTISTS,
+            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
+        ):
+            try:
+                cross_prov = cast("MetadataProvider | PluginProvider", prov)
+                if result := await cross_prov.get_similar_artists(ref_item, limit=limit):
+                    return result
+            except NotImplementedError:
+                continue
+        return []
 
     async def library_count(
         self, favorite_only: bool = False, album_artists_only: bool = False

music_assistant/controllers/media/base.py

@@ -43,6 +43,7 @@
 
     from music_assistant import MusicAssistant
     from music_assistant.models.music_provider import MusicProvider
+    from music_assistant.models.plugin import PluginProvider
 
 
 ItemCls = TypeVar("ItemCls", bound="MediaItemType")
@@ -583,7 +584,7 @@ async def get_provider_item(
         if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
             raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
         if provider := self.mass.get_provider(provider_instance_id_or_domain):
-            provider = cast("MusicProvider", provider)
+            provider = cast("MusicProvider | PluginProvider", provider)
             with suppress(MediaNotFoundError):
                 async with self.mass.cache.handle_refresh(force_refresh):
                     return cast("ItemCls", await provider.get_item(self.media_type, item_id))

music_assistant/controllers/media/tracks.py

@@ -4,9 +4,9 @@
 
 import urllib.parse
 from collections.abc import Iterable
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
-from music_assistant_models.enums import MediaType, ProviderFeature
+from music_assistant_models.enums import MediaType, ProviderFeature, ProviderType
 from music_assistant_models.errors import (
     InvalidDataError,
     MusicAssistantError,
@@ -42,6 +42,8 @@
 
 if TYPE_CHECKING:
     from music_assistant import MusicAssistant
+    from music_assistant.models.metadata_provider import MetadataProvider
+    from music_assistant.models.plugin import PluginProvider
 
 
 class TracksController(MediaControllerBase[Track]):
@@ -367,12 +369,21 @@ def sort_key(mapping: ProviderMapping) -> tuple[int, int]:
                     prov_track_id=prov_mapping.item_id, limit=limit
                 )
 
+        # Fallback: consult metadata/plugin providers that claim SIMILAR_TRACKS
+        for prov in self.mass.get_providers_supporting_feature(
+            ProviderFeature.SIMILAR_TRACKS,
+            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
+        ):
+            try:
+                cross_prov = cast("MetadataProvider | PluginProvider", prov)
+                if result := await cross_prov.get_similar_tracks(ref_item, limit=limit):
+                    return result
+            except NotImplementedError:
+                continue
+
         if not allow_lookup:
             return []
 
-        # check if we have any provider that supports dynamic tracks
-        # TODO: query metadata provider(s) (such as lastfm?)
-        # to get similar tracks (or tracks from similar artists)
         music_prov: MusicProvider | None = None
         for prov in self.mass.music.providers:
             if ProviderFeature.SIMILAR_TRACKS in prov.supported_features:

music_assistant/controllers/music.py

@@ -103,6 +103,8 @@
     from music_assistant_models.media_items import Audiobook, PodcastEpisode
 
     from music_assistant import MusicAssistant
+    from music_assistant.models.metadata_provider import MetadataProvider
+    from music_assistant.models.plugin import PluginProvider
 
 
 CONF_RESET_DB = "reset_db"
@@ -551,11 +553,9 @@ async def search_library(
     async def browse(self, path: str | None = None) -> Sequence[MediaItemType | BrowseFolder]:
         """Browse Music providers."""
         if not path or path == "root":
-            # root level; folder per provider
+            # root level; folder per provider that declares BROWSE
             root_items: list[BrowseFolder] = []
-            for prov in self.providers:
-                if ProviderFeature.BROWSE not in prov.supported_features:
-                    continue
+            for prov in self.mass.get_providers_supporting_feature(ProviderFeature.BROWSE):
                 root_items.append(
                     BrowseFolder(
                         item_id="root",
@@ -801,9 +801,9 @@ async def get_item_by_uri(
     @api_command("music/recommendations")
     async def recommendations(self) -> list[RecommendationFolder]:
         """Get all recommendations."""
-        recommendation_providers = [
-            x for x in self.providers if ProviderFeature.RECOMMENDATIONS in x.supported_features
-        ]
+        recommendation_providers = self.mass.get_providers_supporting_feature(
+            ProviderFeature.RECOMMENDATIONS,
+        )
         results_per_provider: list[list[RecommendationFolder]] = await asyncio.gather(
             self._get_default_recommendations(),
             *[
@@ -1818,7 +1818,7 @@ async def _get_default_recommendations(self) -> list[RecommendationFolder]:
         ]
 
     async def _get_provider_recommendations(
-        self, provider: MusicProvider
+        self, provider: MusicProvider | MetadataProvider | PluginProvider
     ) -> list[RecommendationFolder]:
         """Return recommendations from a provider."""
         try:

music_assistant/mass.py

@@ -58,7 +58,6 @@
 from music_assistant.models.audio_analysis_provider import AudioAnalysisProvider
 from music_assistant.models.music_provider import MusicProvider
 from music_assistant.models.player_provider import PlayerProvider
-from music_assistant.models.plugin import PluginProvider
 
 if TYPE_CHECKING:
     from types import TracebackType
@@ -432,18 +431,39 @@ def get_provider_instances(
             and (return_unavailable or prov.available)
         ]
 
-    def get_plugins_by_feature(self, feature: ProviderFeature) -> list[PluginProvider]:
-        """Return all available PluginProvider instances that support the given feature."""
-        return cast(
-            "list[PluginProvider]",
-            [
-                prov
-                for prov in list(self._providers.values())
-                if prov.available
-                and isinstance(prov, PluginProvider)
-                and feature in prov.supported_features
-            ],
-        )
+    def get_providers_supporting_feature(
+        self,
+        feature: ProviderFeature,
+        priority: tuple[ProviderType, ...] = (
+            ProviderType.MUSIC,
+            ProviderType.METADATA,
+            ProviderType.PLUGIN,
+        ),
+    ) -> list[ProviderInstanceType]:
+        """
+        Return all available providers that support the given feature.
+
+        Results are grouped by provider type in the order given by ``priority``,
+        and sorted within each tier by the provider's ``priority`` attribute
+        (lower value = higher priority).
+
+        :param feature: The ProviderFeature to query for.
+        :param priority: Ordered tuple of ProviderType values indicating tier order.
+            Types omitted from this tuple are excluded from the results.
+        """
+        by_tier: dict[ProviderType, list[ProviderInstanceType]] = {ptype: [] for ptype in priority}
+        for prov in self.get_providers():
+            if not prov.available:
+                continue
+            if prov.type not in by_tier:
+                continue
+            if feature not in prov.supported_features:
+                continue
+            by_tier[prov.type].append(prov)
+        result: list[ProviderInstanceType] = []
+        for ptype in priority:
+            result.extend(sorted(by_tier[ptype], key=lambda p: getattr(p, "priority", 50)))
+        return result
 
     def signal_event(
         self,

music_assistant/models/metadata_provider.py

@@ -9,7 +9,13 @@
 from .provider import Provider
 
 if TYPE_CHECKING:
-    from music_assistant_models.media_items import Album, Artist, MediaItemMetadata, Track
+    from music_assistant_models.media_items import (
+        Album,
+        Artist,
+        MediaItemMetadata,
+        RecommendationFolder,
+        Track,
+    )
 
 
 class MetadataProvider(Provider):
@@ -41,6 +47,42 @@ async def get_track_metadata(self, track: Track) -> MediaItemMetadata | None:
             raise NotImplementedError
         return None
 
+    async def get_similar_tracks(self, track: Track, limit: int = 25) -> list[Track]:
+        """
+        Retrieve a list of similar tracks for the given track.
+
+        Will only be called if ProviderFeature.SIMILAR_TRACKS is declared.
+
+        :param track: The reference track.
+        :param limit: Maximum number of similar tracks to return.
+        """
+        if ProviderFeature.SIMILAR_TRACKS in self.supported_features:
+            raise NotImplementedError
+        return []
+
+    async def get_similar_artists(self, artist: Artist, limit: int = 25) -> list[Artist]:
+        """
+        Retrieve a list of similar artists for the given artist.
+
+        Will only be called if ProviderFeature.SIMILAR_ARTISTS is declared.
+
+        :param artist: The reference artist.
+        :param limit: Maximum number of similar artists to return.
+        """
+        if ProviderFeature.SIMILAR_ARTISTS in self.supported_features:
+            raise NotImplementedError
+        return []
+
+    async def recommendations(self) -> list[RecommendationFolder]:
+        """
+        Retrieve a list of recommendation folders from this metadata provider.
+
+        Will only be called if ProviderFeature.RECOMMENDATIONS is declared.
+        """
+        if ProviderFeature.RECOMMENDATIONS in self.supported_features:
+            raise NotImplementedError
+        return []
+
     async def resolve_image(self, path: str) -> str | bytes:
         """
         Resolve an image from an image path.

music_assistant/models/music_provider.py

@@ -398,6 +398,13 @@ async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[
         """
         raise NotImplementedError
 
+    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
+        """Retrieve a dynamic list of similar artists based on the provided artist.
+
+        Only called if provider supports ProviderFeature.SIMILAR_ARTISTS.
+        """
+        raise NotImplementedError
+
     async def get_resume_position(
         self, item_id: str, media_type: MediaType
     ) -> tuple[bool, int, datetime | None]: