# ruff: noqa: ARG001, ARG002
+DEFAULT_SUPPORTED_FEATURES = (
+ ProviderFeature.ARTIST_METADATA,
+ ProviderFeature.ALBUM_METADATA,
+ ProviderFeature.TRACK_METADATA,
+ ProviderFeature.GET_ARTIST_MBID,
+)
+
class MetadataProvider(Provider):
"""Base representation of a Metadata Provider (controller).
Metadata Provider implementations should inherit from this base model.
"""
- _attr_supported_features: tuple[ProviderFeature, ...] = (
- ProviderFeature.ARTIST_METADATA,
- ProviderFeature.ALBUM_METADATA,
- ProviderFeature.TRACK_METADATA,
- ProviderFeature.GET_ARTIST_MBID,
- )
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return DEFAULT_SUPPORTED_FEATURES
async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
"""Retrieve metadata for an artist on this Metadata provider."""
class Provider:
"""Base representation of a Provider implementation within Music Assistant."""
- _attr_supported_features: tuple[ProviderFeature, ...] = tuple()
-
def __init__(
self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
) -> None:
@property
def supported_features(self) -> tuple[ProviderFeature, ...]:
- """Return the features supported by this MusicProvider."""
- return self._attr_supported_features
+ """Return the features supported by this Provider."""
+ return tuple()
async def setup(self) -> None:
"""Handle async initialization of the provider.
if TYPE_CHECKING:
from music_assistant.common.models.media_items import Album, Artist
+SUPPORTED_FEATURES = (
+ ProviderFeature.ARTIST_METADATA,
+ ProviderFeature.ALBUM_METADATA,
+)
+
# TODO: add support for personal api keys ?
"""Handle async initialization of the provider."""
self.cache = self.mass.cache
self.throttler = Throttler(rate_limit=2, period=1)
- self._attr_supported_features = (
- ProviderFeature.ARTIST_METADATA,
- ProviderFeature.ALBUM_METADATA,
- )
+
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
"""Retrieve metadata for artist on fanart.tv."""
SUPPORTED_EXTENSIONS = TRACK_EXTENSIONS + PLAYLIST_EXTENSIONS
IMAGE_EXTENSIONS = ("jpg", "jpeg", "JPG", "JPEG", "png", "PNG", "gif", "GIF")
+SUPPORTED_FEATURES = (
+ ProviderFeature.LIBRARY_ARTISTS,
+ ProviderFeature.LIBRARY_ALBUMS,
+ ProviderFeature.LIBRARY_TRACKS,
+ ProviderFeature.LIBRARY_PLAYLISTS,
+ ProviderFeature.PLAYLIST_TRACKS_EDIT,
+ ProviderFeature.PLAYLIST_CREATE,
+ ProviderFeature.BROWSE,
+ ProviderFeature.SEARCH,
+)
+
@dataclass
class FileSystemItem:
Supports having URI's from streaming providers within m3u playlist.
"""
- _attr_supported_features = (
- ProviderFeature.LIBRARY_ARTISTS,
- ProviderFeature.LIBRARY_ALBUMS,
- ProviderFeature.LIBRARY_TRACKS,
- ProviderFeature.LIBRARY_PLAYLISTS,
- ProviderFeature.PLAYLIST_TRACKS_EDIT,
- ProviderFeature.PLAYLIST_CREATE,
- ProviderFeature.BROWSE,
- ProviderFeature.SEARCH,
- )
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
@abstractmethod
async def setup(self) -> None:
LUCENE_SPECIAL = r'([+\-&|!(){}\[\]\^"~*?:\\\/])'
+SUPPORTED_FEATURES = (ProviderFeature.GET_ARTIST_MBID,)
+
class MusicbrainzProvider(MetadataProvider):
"""The Musicbrainz Metadata provider."""
"""Handle async initialization of the provider."""
self.cache = self.mass.cache
self.throttler = Throttler(rate_limit=1, period=1)
- self._attr_supported_features = (ProviderFeature.GET_ARTIST_MBID,)
+
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
async def get_musicbrainz_artist_id(
self, artist: Artist, ref_albums: Iterable[Album], ref_tracks: Iterable[Track]
from music_assistant.server.helpers.app_vars import app_var # pylint: disable=no-name-in-module
from music_assistant.server.models.music_provider import MusicProvider
+SUPPORTED_FEATURES = (
+ ProviderFeature.LIBRARY_ARTISTS,
+ ProviderFeature.LIBRARY_ALBUMS,
+ ProviderFeature.LIBRARY_TRACKS,
+ ProviderFeature.LIBRARY_PLAYLISTS,
+ ProviderFeature.LIBRARY_ARTISTS_EDIT,
+ ProviderFeature.LIBRARY_ALBUMS_EDIT,
+ ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
+ ProviderFeature.LIBRARY_TRACKS_EDIT,
+ ProviderFeature.PLAYLIST_TRACKS_EDIT,
+ ProviderFeature.BROWSE,
+ ProviderFeature.SEARCH,
+ ProviderFeature.ARTIST_ALBUMS,
+ ProviderFeature.ARTIST_TOPTRACKS,
+)
+
class QobuzProvider(MusicProvider):
"""Provider for the Qobux music service."""
async def setup(self) -> None:
"""Handle async initialization of the provider."""
self._throttler = Throttler(rate_limit=4, period=1)
- self._attr_supported_features = (
- ProviderFeature.LIBRARY_ARTISTS,
- ProviderFeature.LIBRARY_ALBUMS,
- ProviderFeature.LIBRARY_TRACKS,
- ProviderFeature.LIBRARY_PLAYLISTS,
- ProviderFeature.LIBRARY_ARTISTS_EDIT,
- ProviderFeature.LIBRARY_ALBUMS_EDIT,
- ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
- ProviderFeature.LIBRARY_TRACKS_EDIT,
- ProviderFeature.PLAYLIST_TRACKS_EDIT,
- ProviderFeature.BROWSE,
- ProviderFeature.SEARCH,
- ProviderFeature.ARTIST_ALBUMS,
- ProviderFeature.ARTIST_TOPTRACKS,
- )
+
if not self.config.get_value(CONF_USERNAME) or not self.config.get_value(CONF_PASSWORD):
raise LoginFailed("Invalid login credentials")
# try to get a token, raise if that fails
if not token:
raise LoginFailed(f"Login failed for user {self.config.get_value(CONF_USERNAME)}")
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
+
async def search(
self, search_query: str, media_types=list[MediaType] | None, limit: int = 5
) -> list[MediaItemType]:
from music_assistant.server.models.music_provider import MusicProvider
CACHE_DIR = gettempdir()
+SUPPORTED_FEATURES = (
+ ProviderFeature.LIBRARY_ARTISTS,
+ ProviderFeature.LIBRARY_ALBUMS,
+ ProviderFeature.LIBRARY_TRACKS,
+ ProviderFeature.LIBRARY_PLAYLISTS,
+ ProviderFeature.LIBRARY_ARTISTS_EDIT,
+ ProviderFeature.LIBRARY_ALBUMS_EDIT,
+ ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
+ ProviderFeature.LIBRARY_TRACKS_EDIT,
+ ProviderFeature.PLAYLIST_TRACKS_EDIT,
+ ProviderFeature.BROWSE,
+ ProviderFeature.SEARCH,
+ ProviderFeature.ARTIST_ALBUMS,
+ ProviderFeature.ARTIST_TOPTRACKS,
+ ProviderFeature.SIMILAR_TRACKS,
+)
class SpotifyProvider(MusicProvider):
self._throttler = Throttler(rate_limit=1, period=0.1)
self._cache_dir = CACHE_DIR
self._ap_workaround = False
- self._attr_supported_features = (
- ProviderFeature.LIBRARY_ARTISTS,
- ProviderFeature.LIBRARY_ALBUMS,
- ProviderFeature.LIBRARY_TRACKS,
- ProviderFeature.LIBRARY_PLAYLISTS,
- ProviderFeature.LIBRARY_ARTISTS_EDIT,
- ProviderFeature.LIBRARY_ALBUMS_EDIT,
- ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
- ProviderFeature.LIBRARY_TRACKS_EDIT,
- ProviderFeature.PLAYLIST_TRACKS_EDIT,
- ProviderFeature.BROWSE,
- ProviderFeature.SEARCH,
- ProviderFeature.ARTIST_ALBUMS,
- ProviderFeature.ARTIST_TOPTRACKS,
- ProviderFeature.SIMILAR_TRACKS,
- )
+
# try to get a token, raise if that fails
self._cache_dir = os.path.join(CACHE_DIR, self.instance_id)
# try login which will raise if it fails
await self.login()
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
+
async def search(
self, search_query: str, media_types=list[MediaType] | None, limit: int = 5
) -> list[MediaItemType]:
if TYPE_CHECKING:
from collections.abc import Iterable
+SUPPORTED_FEATURES = (
+ ProviderFeature.ARTIST_METADATA,
+ ProviderFeature.ALBUM_METADATA,
+ ProviderFeature.TRACK_METADATA,
+ ProviderFeature.GET_ARTIST_MBID,
+)
+
IMG_MAPPING = {
"strArtistThumb": ImageType.THUMB,
"strArtistLogo": ImageType.LOGO,
async def setup(self) -> None:
"""Handle async initialization of the provider."""
self.cache = self.mass.cache
- self._attr_supported_features = (
- ProviderFeature.ARTIST_METADATA,
- ProviderFeature.ALBUM_METADATA,
- ProviderFeature.TRACK_METADATA,
- ProviderFeature.GET_ARTIST_MBID,
- )
self.throttler = Throttler(rate_limit=2, period=1)
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
+
async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
"""Retrieve metadata for artist on theaudiodb."""
if data := await self._get_data("artist-mb.php", i=artist.musicbrainz_id): # noqa: SIM102
from music_assistant.server.helpers.tags import parse_tags
from music_assistant.server.models.music_provider import MusicProvider
+SUPPORTED_FEATURES = (
+ ProviderFeature.LIBRARY_RADIOS,
+ ProviderFeature.BROWSE,
+)
+
class TuneInProvider(MusicProvider):
"""Provider implementation for Tune In."""
_throttler: Throttler
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
+
async def setup(self) -> None:
"""Handle async initialization of the provider."""
self._throttler = Throttler(rate_limit=1, period=1)
- self._attr_supported_features = (
- ProviderFeature.LIBRARY_RADIOS,
- ProviderFeature.BROWSE,
- )
+
if not self.config.get_value(CONF_USERNAME):
raise LoginFailed("Username is invalid")
if "@" in self.config.get_value(CONF_USERNAME):
Called when provider is registered.
"""
- self._attr_available = True
self._full_url = {}
async def get_track(self, prov_track_id: str) -> Track:
search,
)
-# if TYPE_CHECKING:
-# from collections.abc import AsyncGenerator
-
CONF_COOKIE = "cookie"
YT_DOMAIN = "https://www.youtube.com"
YTM_DOMAIN = "https://music.youtube.com"
YTM_BASE_URL = f"{YTM_DOMAIN}/youtubei/v1/"
+SUPPORTED_FEATURES = (
+ ProviderFeature.LIBRARY_ARTISTS,
+ ProviderFeature.LIBRARY_ALBUMS,
+ ProviderFeature.LIBRARY_TRACKS,
+ ProviderFeature.LIBRARY_PLAYLISTS,
+ ProviderFeature.BROWSE,
+ ProviderFeature.SEARCH,
+ ProviderFeature.ARTIST_ALBUMS,
+ ProviderFeature.ARTIST_TOPTRACKS,
+ ProviderFeature.SIMILAR_TRACKS,
+)
+
# TODO: fix disabled tests
# ruff: noqa: PLW2901, RET504
async def setup(self) -> None:
"""Set up the YTMusic provider."""
- self._attr_supported_features = (
- ProviderFeature.LIBRARY_ARTISTS,
- ProviderFeature.LIBRARY_ALBUMS,
- ProviderFeature.LIBRARY_TRACKS,
- ProviderFeature.LIBRARY_PLAYLISTS,
- ProviderFeature.BROWSE,
- ProviderFeature.SEARCH,
- ProviderFeature.ARTIST_ALBUMS,
- ProviderFeature.ARTIST_TOPTRACKS,
- ProviderFeature.SIMILAR_TRACKS,
- )
if not self.config.get_value(CONF_USERNAME) or not self.config.get_value(CONF_COOKIE):
raise LoginFailed("Invalid login credentials")
await self._initialize_headers(cookie=self.config.get_value(CONF_COOKIE))
self._cookies = {"CONSENT": "YES+1"}
self._signature_timestamp = await self._get_signature_timestamp()
+ @property
+ def supported_features(self) -> tuple[ProviderFeature, ...]:
+ """Return the features supported by this Provider."""
+ return SUPPORTED_FEATURES
+
async def search(
self, search_query: str, media_types=list[MediaType] | None, limit: int = 5
) -> list[MediaItemType]:
projects / music-assistant-server.git / commitdiff