from __future__ import annotations
-from collections.abc import AsyncGenerator
-from typing import TYPE_CHECKING
-
+import asyncio
+from collections.abc import AsyncGenerator, Awaitable, Callable
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+from tidalapi import Album as TidalAlbum
+from tidalapi import Artist as TidalArtist
+from tidalapi import Config as TidalConfig
+from tidalapi import Playlist as TidalPlaylist
+from tidalapi import Quality as TidalQuality
from tidalapi import Session as TidalSession
+from tidalapi import Track as TidalTrack
+from music_assistant.common.helpers.uri import create_uri
+from music_assistant.common.helpers.util import create_sort_name
from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
-from music_assistant.common.models.enums import ConfigEntryType, MediaType, ProviderFeature
+from music_assistant.common.models.enums import (
+ AlbumType,
+ ConfigEntryType,
+ ImageType,
+ MediaType,
+ ProviderFeature,
+)
from music_assistant.common.models.errors import LoginFailed, MediaNotFoundError
from music_assistant.common.models.media_items import (
Album,
Artist,
ContentType,
+ ItemMapping,
+ MediaItemImage,
Playlist,
+ ProviderMapping,
SearchResults,
StreamDetails,
Track,
from music_assistant.server.models.music_provider import MusicProvider
from .helpers import (
+ DEFAULT_LIMIT,
add_remove_playlist_tracks,
create_playlist,
get_album,
get_playlist,
get_playlist_tracks,
get_similar_tracks,
- get_tidal_session,
get_track,
get_track_url,
library_items_add_remove,
- parse_album,
- parse_artist,
- parse_playlist,
- parse_track,
search,
- tidal_code_login,
)
if TYPE_CHECKING:
return prov
+async def tidal_code_login(auth_helper: AuthenticationHelper) -> TidalSession:
+ """Async wrapper around the tidalapi Session function."""
+
+ def inner() -> TidalSession:
+ config = TidalConfig(quality=TidalQuality.lossless, item_limit=10000, alac=False)
+ session = TidalSession(config=config)
+ login, future = session.login_oauth()
+ auth_helper.send_url(f"https://{login.verification_uri_complete}")
+ future.result()
+ return session
+
+ return await asyncio.to_thread(inner)
+
+
async def get_config_entries(
mass: MusicAssistant,
instance_id: str | None = None, # noqa: ARG001
)
+async def iter_items(func: Awaitable | Callable, *args, **kwargs) -> AsyncGenerator[Any, None]:
+ """Yield all items from a larger listing."""
+ offset = 0
+ while True:
+ if asyncio.iscoroutinefunction(func):
+ chunk = await func(*args, **kwargs, offset=offset)
+ else:
+ chunk = await asyncio.to_thread(func, *args, **kwargs, offset=offset)
+ offset += len(chunk)
+ for item in chunk:
+ yield item
+ if len(chunk) < DEFAULT_LIMIT:
+ break
+
+
class TidalProvider(MusicProvider):
"""Implementation of a Tidal MusicProvider."""
async def handle_setup(self) -> None:
"""Handle async initialization of the provider."""
self._tidal_user_id = self.config.get_value(CONF_USER_ID)
- self._tidal_session = await get_tidal_session(self)
+ self._tidal_session = await self._get_tidal_session()
@property
def supported_features(self) -> tuple[ProviderFeature, ...]:
:param media_types: A list of media_types to include. All types if None.
:param limit: Number of items to return in the search (per type).
"""
+ tidal_session = await self._get_tidal_session()
search_query = search_query.replace("'", "")
- results = await search(self, search_query, media_types, limit)
+ results = await search(tidal_session, search_query, media_types, limit)
parsed_results = SearchResults()
if results["artists"]:
for artist in results["artists"]:
- parsed_results.artists.append(await parse_artist(artist))
+ parsed_results.artists.append(await self._parse_artist(artist_obj=artist))
if results["albums"]:
for album in results["albums"]:
- parsed_results.albums.append(await parse_album(album))
+ parsed_results.albums.append(await self._parse_album(album_obj=album))
if results["playlists"]:
for playlist in results["playlists"]:
- parsed_results.playlists.append(await parse_playlist(playlist))
+ parsed_results.playlists.append(await self._parse_playlist(playlist_obj=playlist))
if results["tracks"]:
for track in results["tracks"]:
- parsed_results.tracks.append(await parse_track(track))
+ parsed_results.tracks.append(await self._parse_track(track_obj=track))
return parsed_results
async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
"""Retrieve all library artists from Tidal."""
- artists_obj = await get_library_artists(self, self._tidal_user_id)
- for artist in artists_obj:
- yield parse_artist(tidal_provider=self, artist_obj=artist)
+ tidal_session = await self._get_tidal_session()
+ artist: TidalArtist # satisfy the type checker
+ async for artist in iter_items(
+ get_library_artists, tidal_session, self._tidal_user_id, limit=DEFAULT_LIMIT
+ ):
+ yield await self._parse_artist(artist_obj=artist)
async def get_library_albums(self) -> AsyncGenerator[Album, None]:
"""Retrieve all library albums from Tidal."""
- albums_obj = await get_library_albums(self, self._tidal_user_id)
- for album in albums_obj:
- yield parse_album(tidal_provider=self, album_obj=album)
+ tidal_session = await self._get_tidal_session()
+ album: TidalAlbum # satisfy the type checker
+ async for album in iter_items(
+ get_library_albums, tidal_session, self._tidal_user_id, limit=DEFAULT_LIMIT
+ ):
+ yield await self._parse_album(album_obj=album)
async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
"""Retrieve library tracks from Tidal."""
- tracks_obj = await get_library_tracks(self, self._tidal_user_id)
- for track in tracks_obj:
- if track.available:
- yield parse_track(tidal_provider=self, track_obj=track)
+ tidal_session = await self._get_tidal_session()
+ track: TidalTrack # satisfy the type checker
+ async for track in iter_items(
+ get_library_tracks, tidal_session, self._tidal_user_id, limit=DEFAULT_LIMIT
+ ):
+ yield await self._parse_track(track_obj=track)
async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
"""Retrieve all library playlists from the provider."""
- playlists_obj = await get_library_playlists(self, self._tidal_user_id)
- for playlist in playlists_obj:
- yield parse_playlist(tidal_provider=self, playlist_obj=playlist)
+ tidal_session = await self._get_tidal_session()
+ playlist: TidalPlaylist # satisfy the type checker
+ async for playlist in iter_items(get_library_playlists, tidal_session, self._tidal_user_id):
+ yield await self._parse_playlist(playlist_obj=playlist)
async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
"""Get album tracks for given album id."""
+ tidal_session = await self._get_tidal_session()
result = []
- tracks = await get_album_tracks(self, prov_album_id)
+ tracks = await get_album_tracks(tidal_session, prov_album_id)
for index, track_obj in enumerate(tracks, 1):
if track_obj.available:
- track = parse_track(tidal_provider=self, track_obj=track_obj)
+ track = await self._parse_track(track_obj=track_obj)
track.position = index
result.append(track)
return result
- async def get_artist_albums(self, prov_artist_id) -> list[Album]:
+ async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
"""Get a list of all albums for the given artist."""
+ tidal_session = await self._get_tidal_session()
result = []
- albums = await get_artist_albums(self, prov_artist_id)
+ albums = await get_artist_albums(tidal_session, prov_artist_id)
for album_obj in albums:
- album = parse_album(tidal_provider=self, album_obj=album_obj)
+ album = await self._parse_album(album_obj=album_obj)
result.append(album)
return result
- async def get_artist_toptracks(self, prov_artist_id) -> list[Track]:
+ async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
"""Get a list of 10 most popular tracks for the given artist."""
+ tidal_session = await self._get_tidal_session()
result = []
- tracks = await get_artist_toptracks(self, prov_artist_id)
+ tracks = await get_artist_toptracks(tidal_session, prov_artist_id)
for index, track_obj in enumerate(tracks, 1):
if track_obj.available:
- track = parse_track(tidal_provider=self, track_obj=track_obj)
+ track = await self._parse_track(track_obj=track_obj)
track.position = index
result.append(track)
return result
- async def get_playlist_tracks(self, prov_playlist_id) -> AsyncGenerator[Track, None]:
+ async def get_playlist_tracks(self, prov_playlist_id: str) -> AsyncGenerator[Track, None]:
"""Get all playlist tracks for given playlist id."""
- tracks = await get_playlist_tracks(self, prov_playlist_id=prov_playlist_id)
- for index, track_obj in enumerate(tracks):
- if track_obj.available:
- track = parse_track(tidal_provider=self, track_obj=track_obj)
- track.position = index + 1
- yield track
-
- async def get_similar_tracks(self, prov_track_id, limit=25) -> list[Track]:
+ tidal_session = await self._get_tidal_session()
+ total_playlist_tracks = 0
+ track: TidalTrack # satisfy the type checker
+ async for track_obj in iter_items(
+ get_playlist_tracks, tidal_session, prov_playlist_id, limit=DEFAULT_LIMIT
+ ):
+ total_playlist_tracks += 1
+ track = await self._parse_track(track_obj=track_obj)
+ track.position = total_playlist_tracks
+ yield track
+
+ async def get_similar_tracks(self, prov_track_id: str, limit=25) -> list[Track]:
"""Get similar tracks for given track id."""
- similar_tracks_obj = await get_similar_tracks(self, prov_track_id, limit)
+ tidal_session = await self._get_tidal_session()
+ similar_tracks_obj = await get_similar_tracks(tidal_session, prov_track_id, limit)
tracks = []
for track_obj in similar_tracks_obj:
if track_obj.available:
- track = parse_track(tidal_provider=self, track_obj=track_obj)
+ track = await self._parse_track(track_obj=track_obj)
tracks.append(track)
return tracks
- async def library_add(self, prov_item_id, media_type: MediaType):
+ async def library_add(self, prov_item_id: str, media_type: MediaType):
"""Add item to library."""
return await library_items_add_remove(
self,
add=True,
)
- async def library_remove(self, prov_item_id, media_type: MediaType):
+ async def library_remove(self, prov_item_id: str, media_type: MediaType):
"""Remove item from library."""
return await library_items_add_remove(
self,
async def create_playlist(self, name: str) -> Playlist:
"""Create a new playlist on provider with given name."""
- playlist_obj = await create_playlist(self, self._tidal_user_id, name)
- playlist = parse_playlist(tidal_provider=self, playlist_obj=playlist_obj)
- return await self.mass.music.playlists.add_db_item(playlist)
+ tidal_session = await self._get_tidal_session()
+ playlist_obj = await create_playlist(tidal_session, self._tidal_user_id, name)
+ playlist = await self._parse_playlist(playlist_obj=playlist_obj)
+ return playlist
async def get_stream_details(self, item_id: str) -> StreamDetails:
"""Return the content details for the given track when it will be streamed."""
# make sure a valid track is requested.
- track = await get_track(self, item_id)
- url = await get_track_url(self, item_id)
+ tidal_session = await self._get_tidal_session()
+ track = await get_track(tidal_session, item_id)
+ url = await get_track_url(tidal_session, item_id)
if not track:
raise MediaNotFoundError(f"track {item_id} not found")
- # make sure that the token is still valid by just requesting it
- await get_tidal_session(self)
return StreamDetails(
item_id=track.id,
provider=self.instance_id,
async def get_artist(self, prov_artist_id: str) -> Artist:
"""Get artist details for given artist id."""
+ tidal_session = await self._get_tidal_session()
try:
- artist = parse_artist(
- tidal_provider=self,
- artist_obj=await get_artist(self, prov_artist_id),
+ artist = await self._parse_artist(
+ artist_obj=await get_artist(tidal_session, prov_artist_id), full_details=True
)
except MediaNotFoundError as err:
raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
async def get_album(self, prov_album_id: str) -> Album:
"""Get album details for given album id."""
+ tidal_session = await self._get_tidal_session()
try:
- album = parse_album(
- tidal_provider=self,
- album_obj=await get_album(self, prov_album_id),
+ album = await self._parse_album(
+ album_obj=await get_album(tidal_session, prov_album_id), full_details=True
)
except MediaNotFoundError as err:
raise MediaNotFoundError(f"Album {prov_album_id} not found") from err
async def get_track(self, prov_track_id: str) -> Track:
"""Get track details for given track id."""
+ tidal_session = await self._get_tidal_session()
try:
- track = parse_track(
- tidal_provider=self,
- track_obj=await get_track(self, prov_track_id),
+ track = await self._parse_track(
+ track_obj=await get_track(tidal_session, prov_track_id), full_details=True
)
except MediaNotFoundError as err:
raise MediaNotFoundError(f"Track {prov_track_id} not found") from err
async def get_playlist(self, prov_playlist_id: str) -> Playlist:
"""Get playlist details for given playlist id."""
+ tidal_session = await self._get_tidal_session()
try:
- playlist = parse_playlist(
- tidal_provider=self,
- playlist_obj=await get_playlist(self, prov_playlist_id),
+ playlist = await self._parse_playlist(
+ playlist_obj=await get_playlist(tidal_session, prov_playlist_id), full_details=True
)
except MediaNotFoundError as err:
raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found") from err
return playlist
+
+ def get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
+ """Create a generic item mapping."""
+ return ItemMapping(
+ media_type,
+ key,
+ self.instance_id,
+ name,
+ create_uri(media_type, self.instance_id, key),
+ create_sort_name(self.name),
+ )
+
+ async def _get_tidal_session(self) -> TidalSession:
+ """Ensure the current token is valid and return a tidal session."""
+ if (
+ self._tidal_session
+ and self._tidal_session.access_token
+ and datetime.fromisoformat(self.config.get_value(CONF_EXPIRY_TIME))
+ > (datetime.now() + timedelta(days=1))
+ ):
+ return self._tidal_session
+ self._tidal_session = await self._load_tidal_session(
+ token_type="Bearer",
+ access_token=self.config.get_value(CONF_AUTH_TOKEN),
+ refresh_token=self.config.get_value(CONF_REFRESH_TOKEN),
+ expiry_time=datetime.fromisoformat(self.config.get_value(CONF_EXPIRY_TIME)),
+ )
+ await self.mass.config.set_provider_config_value(
+ self.config.instance_id,
+ CONF_AUTH_TOKEN,
+ self._tidal_session.access_token,
+ )
+ await self.mass.config.set_provider_config_value(
+ self.config.instance_id,
+ CONF_REFRESH_TOKEN,
+ self._tidal_session.refresh_token,
+ )
+ await self.mass.config.set_provider_config_value(
+ self.config.instance_id,
+ CONF_EXPIRY_TIME,
+ self._tidal_session.expiry_time.isoformat(),
+ )
+ return self._tidal_session
+
+ async def _load_tidal_session(
+ self, token_type, access_token, refresh_token=None, expiry_time=None
+ ) -> TidalSession:
+ """Load the tidalapi Session."""
+
+ def inner() -> TidalSession:
+ config = TidalConfig(quality=TidalQuality.lossless, item_limit=10000, alac=False)
+ session = TidalSession(config=config)
+ session.load_oauth_session(token_type, access_token, refresh_token, expiry_time)
+ return session
+
+ return await asyncio.to_thread(inner)
+
+ # Parsers
+
+ async def _parse_artist(self, artist_obj: TidalArtist, full_details: bool = False) -> Artist:
+ """Parse tidal artist object to generic layout."""
+ artist_id = artist_obj.id
+ artist = Artist(item_id=artist_id, provider=self.instance_id, name=artist_obj.name)
+ artist.add_provider_mapping(
+ ProviderMapping(
+ item_id=str(artist_id),
+ provider_domain=self.domain,
+ provider_instance=self.instance_id,
+ url=f"http://www.tidal.com/artist/{artist_id}",
+ )
+ )
+ # metadata
+ if full_details:
+ image_url = None
+ if artist_obj.name != "Various Artists":
+ try:
+ image_url = await asyncio.to_thread(artist_obj.image(750))
+ except Exception:
+ self.logger.info(f"Artist {artist_obj.id} has no available picture")
+ artist.metadata.images = [
+ MediaItemImage(
+ ImageType.THUMB,
+ image_url,
+ )
+ ]
+ return artist
+
+ async def _parse_album(self, album_obj: TidalAlbum, full_details: bool = False) -> Album:
+ """Parse tidal album object to generic layout."""
+ name = album_obj.name
+ version = album_obj.version if album_obj.version is not None else None
+ album_id = album_obj.id
+ album = Album(item_id=album_id, provider=self.instance_id, name=name, version=version)
+ for artist_obj in album_obj.artists:
+ album.artists.append(await self._parse_artist(artist_obj=artist_obj))
+ if album_obj.type == "ALBUM":
+ album.album_type = AlbumType.ALBUM
+ elif album_obj.type == "COMPILATION":
+ album.album_type = AlbumType.COMPILATION
+ elif album_obj.type == "EP":
+ album.album_type = AlbumType.EP
+ elif album_obj.type == "SINGLE":
+ album.album_type = AlbumType.SINGLE
+
+ album.upc = album_obj.universal_product_number
+ album.year = int(album_obj.year)
+ album.add_provider_mapping(
+ ProviderMapping(
+ item_id=album_id,
+ provider_domain=self.domain,
+ provider_instance=self.instance_id,
+ content_type=ContentType.FLAC,
+ url=f"http://www.tidal.com/album/{album_id}",
+ )
+ )
+ # metadata
+ album.metadata.copyright = album_obj.copyright
+ album.metadata.explicit = album_obj.explicit
+ album.metadata.popularity = album_obj.popularity
+ if full_details:
+ image_url = None
+ try:
+ image_url = await asyncio.to_thread(album_obj.image(1280))
+ except Exception:
+ self.logger.info(f"Album {album_obj.id} has no available picture")
+ album.metadata.images = [
+ MediaItemImage(
+ ImageType.THUMB,
+ image_url,
+ )
+ ]
+ return album
+
+ async def _parse_track(self, track_obj: TidalTrack, full_details: bool = False) -> Track:
+ """Parse tidal track object to generic layout."""
+ version = track_obj.version if track_obj.version is not None else None
+ track_id = str(track_obj.id)
+ track = Track(
+ item_id=track_id,
+ provider=self.instance_id,
+ name=track_obj.name,
+ version=version,
+ duration=track_obj.duration,
+ disc_number=track_obj.volume_num,
+ track_number=track_obj.track_num,
+ )
+ track.isrc.add(track_obj.isrc)
+ track.album = self.get_item_mapping(
+ media_type=MediaType.ALBUM,
+ key=track_obj.album.id,
+ name=track_obj.album.name,
+ )
+ track.artists = []
+ for track_artist in track_obj.artists:
+ artist = await self._parse_artist(artist_obj=track_artist)
+ track.artists.append(artist)
+ available = track_obj.available
+ track.add_provider_mapping(
+ ProviderMapping(
+ item_id=track_id,
+ provider_domain=self.domain,
+ provider_instance=self.instance_id,
+ content_type=ContentType.FLAC,
+ sample_rate=44100,
+ bit_depth=16,
+ url=f"http://www.tidal.com/tracks/{track_id}",
+ available=available,
+ )
+ )
+ # metadata
+ track.metadata.explicit = track_obj.explicit
+ track.metadata.popularity = track_obj.popularity
+ track.metadata.copyright = track_obj.copyright
+ if full_details:
+ try:
+ if lyrics_obj := await asyncio.to_thread(track_obj.lyrics):
+ track.metadata.lyrics = lyrics_obj.text
+ except Exception:
+ self.logger.info(f"Track {track_obj.id} has no available lyrics")
+ return track
+
+ async def _parse_playlist(
+ self, playlist_obj: TidalPlaylist, full_details: bool = False
+ ) -> Playlist:
+ """Parse tidal playlist object to generic layout."""
+ playlist_id = playlist_obj.id
+ creator_id = playlist_obj.creator.id if playlist_obj.creator else None
+ creator_name = playlist_obj.creator.name if playlist_obj.creator else "Tidal"
+ playlist = Playlist(
+ item_id=playlist_id,
+ provider=self.instance_id,
+ name=playlist_obj.name,
+ owner=creator_name,
+ )
+ playlist.add_provider_mapping(
+ ProviderMapping(
+ item_id=playlist_id,
+ provider_domain=self.domain,
+ provider_instance=self.instance_id,
+ url=f"http://www.tidal.com/playlists/{playlist_id}",
+ )
+ )
+ is_editable = bool(creator_id and str(creator_id) == self._tidal_user_id)
+ playlist.is_editable = is_editable
+ # metadata
+ playlist.metadata.checksum = str(playlist_obj.last_updated)
+ playlist.metadata.popularity = playlist_obj.popularity
+ if full_details:
+ image_url = None
+ try:
+ image_url = await asyncio.to_thread(playlist_obj.image(1080))
+ except Exception:
+ self.logger.info(f"Playlist {playlist_obj.id} has no available picture")
+ playlist.metadata.images = [
+ MediaItemImage(
+ ImageType.THUMB,
+ image_url,
+ )
+ ]
+
+ return playlist
"""
import asyncio
-from datetime import datetime, timedelta
-from functools import partial, wraps
from requests import HTTPError
from tidalapi import Album as TidalAlbum
from tidalapi import Artist as TidalArtist
-from tidalapi import Config as TidalConfig
from tidalapi import Favorites as TidalFavorites
from tidalapi import LoggedInUser
from tidalapi import Playlist as TidalPlaylist
-from tidalapi import Quality as TidalQuality
from tidalapi import Session as TidalSession
from tidalapi import Track as TidalTrack
from tidalapi import UserPlaylist as TidalUserPlaylist
-from music_assistant.common.helpers.uri import create_uri
-from music_assistant.common.helpers.util import create_sort_name
-from music_assistant.common.models.enums import AlbumType, ContentType, ImageType, MediaType
+from music_assistant.common.models.enums import MediaType
from music_assistant.common.models.errors import MediaNotFoundError
-from music_assistant.common.models.media_items import (
- Album,
- Artist,
- ItemMapping,
- MediaItemImage,
- MediaItemMetadata,
- Playlist,
- ProviderMapping,
- Track,
-)
-from music_assistant.server.helpers.auth import AuthenticationHelper
-
-CONF_AUTH_TOKEN = "auth_token"
-CONF_REFRESH_TOKEN = "refresh_token"
-CONF_USER_ID = "user_id"
-CONF_EXPIRY_TIME = "expiry_time"
-
-
-# Parsers
-
-
-def parse_artist(tidal_provider, artist_obj: TidalArtist) -> Artist:
- """Parse tidal artist object to generic layout."""
- artist_id = artist_obj.id
- artist = Artist(item_id=artist_id, provider=tidal_provider.instance_id, name=artist_obj.name)
- artist.add_provider_mapping(
- ProviderMapping(
- item_id=str(artist_id),
- provider_domain=tidal_provider.domain,
- provider_instance=tidal_provider.instance_id,
- url=f"http://www.tidal.com/artist/{artist_id}",
- )
- )
- artist.metadata = parse_artist_metadata(tidal_provider, artist_obj)
- return artist
+DEFAULT_LIMIT = 50
-def parse_artist_metadata(tidal_provider, artist_obj: TidalArtist) -> MediaItemMetadata:
- """Parse tidal artist object to MA metadata."""
- metadata = MediaItemMetadata()
- image_url = None
- if artist_obj.name != "Various Artists":
- try:
- image_url = artist_obj.image(750)
- except Exception:
- tidal_provider.logger.info(f"Artist {artist_obj.id} has no available picture")
- metadata.images = [
- MediaItemImage(
- ImageType.THUMB,
- image_url,
- )
- ]
- return metadata
-
-
-def parse_album(tidal_provider, album_obj: TidalAlbum) -> Album:
- """Parse tidal album object to generic layout."""
- name = album_obj.name
- version = album_obj.version if album_obj.version is not None else None
- album_id = album_obj.id
- album = Album(item_id=album_id, provider=tidal_provider.instance_id, name=name, version=version)
- for artist_obj in album_obj.artists:
- album.artists.append(parse_artist(tidal_provider=tidal_provider, artist_obj=artist_obj))
- if album_obj.type == "ALBUM":
- album.album_type = AlbumType.ALBUM
- elif album_obj.type == "COMPILATION":
- album.album_type = AlbumType.COMPILATION
- elif album_obj.type == "EP":
- album.album_type = AlbumType.EP
- elif album_obj.type == "SINGLE":
- album.album_type = AlbumType.SINGLE
-
- album.upc = album_obj.universal_product_number
- album.year = int(album_obj.year)
- album.add_provider_mapping(
- ProviderMapping(
- item_id=album_id,
- provider_domain=tidal_provider.domain,
- provider_instance=tidal_provider.instance_id,
- content_type=ContentType.FLAC,
- url=f"http://www.tidal.com/album/{album_id}",
- )
- )
- album.metadata = parse_album_metadata(tidal_provider, album_obj)
- return album
-
-
-def parse_album_metadata(tidal_provider, album_obj: TidalAlbum) -> MediaItemMetadata:
- """Parse tidal album object to MA metadata."""
- metadata = MediaItemMetadata()
- image_url = None
- try:
- image_url = album_obj.image(1280)
- except Exception:
- tidal_provider.logger.info(f"Album {album_obj.id} has no available picture")
- metadata.images = [
- MediaItemImage(
- ImageType.THUMB,
- image_url,
- )
- ]
- metadata.copyright = album_obj.copyright
- metadata.explicit = album_obj.explicit
- metadata.popularity = album_obj.popularity
- return metadata
-
-
-def parse_track(tidal_provider, track_obj: TidalTrack) -> Track:
- """Parse tidal track object to generic layout."""
- version = track_obj.version if track_obj.version is not None else None
- track_id = str(track_obj.id)
- track = Track(
- item_id=track_id,
- provider=tidal_provider.instance_id,
- name=track_obj.name,
- version=version,
- duration=track_obj.duration,
- disc_number=track_obj.volume_num,
- track_number=track_obj.track_num,
- )
- track.isrc.add(track_obj.isrc)
- track.album = get_item_mapping(
- tidal_provider=tidal_provider,
- media_type=MediaType.ALBUM,
- key=track_obj.album.id,
- name=track_obj.album.name,
- )
- track.artists = []
- for track_artist in track_obj.artists:
- artist = parse_artist(tidal_provider=tidal_provider, artist_obj=track_artist)
- track.artists.append(artist)
- available = track_obj.available
- track.add_provider_mapping(
- ProviderMapping(
- item_id=track_id,
- provider_domain=tidal_provider.domain,
- provider_instance=tidal_provider.instance_id,
- content_type=ContentType.FLAC,
- sample_rate=44100,
- bit_depth=16,
- url=f"http://www.tidal.com/tracks/{track_id}",
- available=available,
- )
- )
- track.metadata = parse_track_metadata(tidal_provider, track_obj)
- return track
-
-
-def parse_track_metadata(tidal_provider, track_obj: TidalTrack) -> MediaItemMetadata:
- """Parse tidal track object to MA metadata."""
- metadata = MediaItemMetadata()
- try:
- metadata.lyrics = track_obj.lyrics().text
- except Exception:
- tidal_provider.logger.info(f"Track {track_obj.id} has no available lyrics")
- metadata.explicit = track_obj.explicit
- metadata.popularity = track_obj.popularity
- metadata.copyright = track_obj.copyright
- return metadata
-
-
-def parse_playlist(tidal_provider, playlist_obj: TidalPlaylist) -> Playlist:
- """Parse tidal playlist object to generic layout."""
- playlist_id = playlist_obj.id
- creator_id = playlist_obj.creator.id if playlist_obj.creator else None
- creator_name = playlist_obj.creator.name if playlist_obj.creator else "Tidal"
- playlist = Playlist(
- item_id=playlist_id,
- provider=tidal_provider.instance_id,
- name=playlist_obj.name,
- owner=creator_name,
- )
- playlist.add_provider_mapping(
- ProviderMapping(
- item_id=playlist_id,
- provider_domain=tidal_provider.domain,
- provider_instance=tidal_provider.instance_id,
- url=f"http://www.tidal.com/playlists/{playlist_id}",
- )
- )
- is_editable = bool(creator_id and creator_id == tidal_provider._tidal_user_id)
- playlist.is_editable = is_editable
- playlist.metadata = parse_playlist_metadata(tidal_provider, playlist_obj)
- return playlist
-
-
-def parse_playlist_metadata(tidal_provider, playlist_obj: TidalPlaylist) -> MediaItemMetadata:
- """Parse tidal playlist object to MA metadata."""
- metadata = MediaItemMetadata()
- image_url = None
- try:
- image_url = playlist_obj.image(1080)
- except Exception:
- tidal_provider.logger.info(f"Playlist {playlist_obj.id} has no available picture")
- metadata.images = [
- MediaItemImage(
- ImageType.THUMB,
- image_url,
- )
- ]
- metadata.checksum = str(playlist_obj.last_updated)
- metadata.popularity = playlist_obj.popularity
- return metadata
-
-
-# Helper functions
-
-
-def async_wrap(func):
- """Async decorator for all tidalapi functions."""
-
- @wraps(func)
- async def run(*args, loop=None, executor=None, **kwargs):
- if loop is None:
- loop = asyncio.get_event_loop()
- pfunc = partial(func, *args, **kwargs)
- return await loop.run_in_executor(executor, pfunc)
-
- return run
-
-
-# Login and session management
-def get_session(func):
- """Async decorator to get a tidal session."""
-
- @wraps(func)
- async def wrapper(tidal_provider, *args, **kwargs):
- return await func(await get_tidal_session(tidal_provider), *args, **kwargs)
-
- return wrapper
-
-
-async def get_tidal_session(tidal_provider) -> TidalSession:
- """Ensure the current token is valid and return a tidal session."""
- if (
- tidal_provider._tidal_session
- and tidal_provider._tidal_session.access_token
- and datetime.fromisoformat(tidal_provider.config.get_value(CONF_EXPIRY_TIME))
- > (datetime.now() + timedelta(days=1))
- ):
- return tidal_provider._tidal_session
- tidal_provider._tidal_session = await load_tidal_session(
- token_type="Bearer",
- access_token=tidal_provider.config.get_value(CONF_AUTH_TOKEN),
- refresh_token=tidal_provider.config.get_value(CONF_REFRESH_TOKEN),
- expiry_time=datetime.fromisoformat(tidal_provider.config.get_value(CONF_EXPIRY_TIME)),
- )
- await tidal_provider.mass.config.set_provider_config_value(
- tidal_provider.config.instance_id,
- CONF_AUTH_TOKEN,
- tidal_provider._tidal_session.access_token,
- )
- await tidal_provider.mass.config.set_provider_config_value(
- tidal_provider.config.instance_id,
- CONF_REFRESH_TOKEN,
- tidal_provider._tidal_session.refresh_token,
- )
- await tidal_provider.mass.config.set_provider_config_value(
- tidal_provider.config.instance_id,
- CONF_EXPIRY_TIME,
- tidal_provider._tidal_session.expiry_time.isoformat(),
- )
- return tidal_provider._tidal_session
-
-
-@async_wrap
-def tidal_code_login(auth_helper: AuthenticationHelper) -> TidalSession:
- """Async wrapper around the tidalapi Session function."""
- config = TidalConfig(quality=TidalQuality.lossless, item_limit=10000, alac=False)
- session = TidalSession(config=config)
- login, future = session.login_oauth()
- auth_helper.send_url(f"https://{login.verification_uri_complete}")
- future.result()
- return session
-
-
-@async_wrap
-def load_tidal_session(
- token_type, access_token, refresh_token=None, expiry_time=None
-) -> TidalSession:
- """Async wrapper around the tidalapi Session function."""
- config = TidalConfig(quality=TidalQuality.lossless, item_limit=10000, alac=False)
- session = TidalSession(config=config)
- session.load_oauth_session(token_type, access_token, refresh_token, expiry_time)
- return session
-
-
-@get_session
-@async_wrap
-def get_library_artists(session: TidalSession, user_id: str) -> dict[str, str]:
+
+async def get_library_artists(
+ session: TidalSession, user_id: str, limit: int = DEFAULT_LIMIT, offset: int = 0
+) -> list[TidalArtist]:
"""Async wrapper around the tidalapi Favorites.artists function."""
- return TidalFavorites(session, user_id).artists(limit=9999)
+ def inner() -> list[TidalArtist]:
+ return TidalFavorites(session, user_id).artists(limit=limit, offset=offset)
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def library_items_add_remove(
+
+async def library_items_add_remove(
session: TidalSession, user_id: str, item_id: str, media_type: MediaType, add: bool = True
) -> None:
"""Async wrapper around the tidalapi Favorites.items add/remove function."""
- match media_type:
- case MediaType.ARTIST:
- TidalFavorites(session, user_id).add_artist(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_artist(item_id)
- case MediaType.ALBUM:
- TidalFavorites(session, user_id).add_album(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_album(item_id)
- case MediaType.TRACK:
- TidalFavorites(session, user_id).add_track(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_track(item_id)
- case MediaType.PLAYLIST:
- TidalFavorites(session, user_id).add_playlist(item_id) if add else TidalFavorites(
- session, user_id
- ).remove_playlist(item_id)
- case MediaType.UNKNOWN:
- return
-
-
-@get_session
-@async_wrap
-def get_artist(session: TidalSession, prov_artist_id: str) -> TidalArtist:
+
+ def inner() -> None:
+ match media_type:
+ case MediaType.ARTIST:
+ TidalFavorites(session, user_id).add_artist(item_id) if add else TidalFavorites(
+ session, user_id
+ ).remove_artist(item_id)
+ case MediaType.ALBUM:
+ TidalFavorites(session, user_id).add_album(item_id) if add else TidalFavorites(
+ session, user_id
+ ).remove_album(item_id)
+ case MediaType.TRACK:
+ TidalFavorites(session, user_id).add_track(item_id) if add else TidalFavorites(
+ session, user_id
+ ).remove_track(item_id)
+ case MediaType.PLAYLIST:
+ TidalFavorites(session, user_id).add_playlist(item_id) if add else TidalFavorites(
+ session, user_id
+ ).remove_playlist(item_id)
+ case MediaType.UNKNOWN:
+ return
+
+ return await asyncio.to_thread(inner)
+
+
+async def get_artist(session: TidalSession, prov_artist_id: str) -> TidalArtist:
"""Async wrapper around the tidalapi Artist function."""
- try:
- artist_obj = TidalArtist(session, prov_artist_id)
- except HTTPError as err:
- raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
- return artist_obj
+ def inner() -> TidalArtist:
+ try:
+ artist_obj = TidalArtist(session, prov_artist_id)
+ except HTTPError as err:
+ raise MediaNotFoundError(f"Artist {prov_artist_id} not found") from err
+ return artist_obj
-@get_session
-@async_wrap
-def get_artist_albums(session: TidalSession, prov_artist_id: str) -> list[TidalAlbum]:
+ return await asyncio.to_thread(inner)
+
+
+async def get_artist_albums(session: TidalSession, prov_artist_id: str) -> list[TidalAlbum]:
"""Async wrapper around 3 tidalapi album functions."""
- all_albums = []
- albums = TidalArtist(session, prov_artist_id).get_albums(limit=9999)
- eps_singles = TidalArtist(session, prov_artist_id).get_albums_ep_singles(limit=9999)
- compilations = TidalArtist(session, prov_artist_id).get_albums_other(limit=9999)
- all_albums.extend(albums)
- all_albums.extend(eps_singles)
- all_albums.extend(compilations)
- return all_albums
-
-
-@get_session
-@async_wrap
-def get_artist_toptracks(session: TidalSession, prov_artist_id: str) -> list[TidalTrack]:
+
+ def inner() -> list[TidalAlbum]:
+ all_albums = []
+ albums = TidalArtist(session, prov_artist_id).get_albums(limit=DEFAULT_LIMIT)
+ eps_singles = TidalArtist(session, prov_artist_id).get_albums_ep_singles(
+ limit=DEFAULT_LIMIT
+ )
+ compilations = TidalArtist(session, prov_artist_id).get_albums_other(limit=DEFAULT_LIMIT)
+ all_albums.extend(albums)
+ all_albums.extend(eps_singles)
+ all_albums.extend(compilations)
+ return all_albums
+
+ return await asyncio.to_thread(inner)
+
+
+async def get_artist_toptracks(
+ session: TidalSession, prov_artist_id: str, limit: int = 10, offset: int = 0
+) -> list[TidalTrack]:
"""Async wrapper around the tidalapi Artist.get_top_tracks function."""
- return TidalArtist(session, prov_artist_id).get_top_tracks(limit=10)
+
+ def inner() -> list[TidalTrack]:
+ return TidalArtist(session, prov_artist_id).get_top_tracks(limit=limit, offset=offset)
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_library_albums(session: TidalSession, user_id: str) -> list[TidalAlbum]:
+async def get_library_albums(
+ session: TidalSession, user_id: str, limit: int = DEFAULT_LIMIT, offset: int = 0
+) -> list[TidalAlbum]:
"""Async wrapper around the tidalapi Favorites.albums function."""
- return TidalFavorites(session, user_id).albums(limit=9999)
+ def inner() -> list[TidalAlbum]:
+ return TidalFavorites(session, user_id).albums(limit=limit, offset=offset)
-@get_session
-@async_wrap
-def get_album(session: TidalSession, prov_album_id: str) -> TidalAlbum:
+ return await asyncio.to_thread(inner)
+
+
+async def get_album(session: TidalSession, prov_album_id: str) -> TidalAlbum:
"""Async wrapper around the tidalapi Album function."""
- try:
- album_obj = TidalAlbum(session, prov_album_id)
- except HTTPError as err:
- raise MediaNotFoundError(f"Album {prov_album_id} not found") from err
- return album_obj
+
+ def inner() -> TidalAlbum:
+ try:
+ album_obj = TidalAlbum(session, prov_album_id)
+ except HTTPError as err:
+ raise MediaNotFoundError(f"Album {prov_album_id} not found") from err
+ return album_obj
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_track(session: TidalSession, prov_track_id: str) -> TidalTrack:
+async def get_track(session: TidalSession, prov_track_id: str) -> TidalTrack:
"""Async wrapper around the tidalapi Track function."""
- try:
- track_obj = TidalTrack(session, prov_track_id)
- except HTTPError as err:
- raise MediaNotFoundError(f"Track {prov_track_id} not found") from err
- return track_obj
+ def inner() -> TidalTrack:
+ try:
+ track_obj = TidalTrack(session, prov_track_id)
+ except HTTPError as err:
+ raise MediaNotFoundError(f"Track {prov_track_id} not found") from err
+ return track_obj
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_track_url(session: TidalSession, prov_track_id: str) -> dict[str, str]:
+
+async def get_track_url(session: TidalSession, prov_track_id: str) -> dict[str, str]:
"""Async wrapper around the tidalapi Track.get_url function."""
- return TidalTrack(session, prov_track_id).get_url()
+
+ def inner() -> dict[str, str]:
+ return TidalTrack(session, prov_track_id).get_url()
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_album_tracks(session: TidalSession, prov_album_id: str) -> list[TidalTrack]:
+async def get_album_tracks(session: TidalSession, prov_album_id: str) -> list[TidalTrack]:
"""Async wrapper around the tidalapi Album.tracks function."""
- return TidalAlbum(session, prov_album_id).tracks()
+ def inner() -> list[TidalTrack]:
+ return TidalAlbum(session, prov_album_id).tracks(limit=DEFAULT_LIMIT)
-@get_session
-@async_wrap
-def get_library_tracks(session: TidalSession, user_id: str) -> list[TidalTrack]:
+ return await asyncio.to_thread(inner)
+
+
+async def get_library_tracks(
+ session: TidalSession, user_id: str, limit: int = DEFAULT_LIMIT, offset: int = 0
+) -> list[TidalTrack]:
"""Async wrapper around the tidalapi Favorites.tracks function."""
- return TidalFavorites(session, user_id).tracks(limit=9999)
+ def inner() -> list[TidalTrack]:
+ return TidalFavorites(session, user_id).tracks(limit=limit, offset=offset)
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_library_playlists(session: TidalSession, user_id: str) -> list[TidalPlaylist]:
+
+async def get_library_playlists(
+ session: TidalSession, user_id: str, offset: int = 0
+) -> list[TidalPlaylist]:
"""Async wrapper around the tidalapi LoggedInUser.playlist_and_favorite_playlists function."""
- return LoggedInUser(session, user_id).playlist_and_favorite_playlists()
+
+ def inner() -> list[TidalPlaylist]:
+ return LoggedInUser(session, user_id).playlist_and_favorite_playlists(offset=offset)
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_playlist(session: TidalSession, prov_playlist_id: str) -> TidalPlaylist:
+async def get_playlist(session: TidalSession, prov_playlist_id: str) -> TidalPlaylist:
"""Async wrapper around the tidal Playlist function."""
- try:
- playlist_obj = TidalPlaylist(session, prov_playlist_id)
- except HTTPError as err:
- raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found") from err
- return playlist_obj
+
+ def inner() -> TidalPlaylist:
+ try:
+ playlist_obj = TidalPlaylist(session, prov_playlist_id)
+ except HTTPError as err:
+ raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found") from err
+ return playlist_obj
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_playlist_tracks(session: TidalSession, prov_playlist_id: str) -> list[TidalTrack]:
+async def get_playlist_tracks(
+ session: TidalSession, prov_playlist_id: str, limit: int = DEFAULT_LIMIT, offset: int = 0
+) -> list[TidalTrack]:
"""Async wrapper around the tidal Playlist.tracks function."""
- return TidalPlaylist(session, prov_playlist_id).tracks(limit=9999)
+ def inner() -> list[TidalTrack]:
+ return TidalPlaylist(session, prov_playlist_id).tracks(limit=limit, offset=offset)
-@get_session
-@async_wrap
-def add_remove_playlist_tracks(
+ return await asyncio.to_thread(inner)
+
+
+async def add_remove_playlist_tracks(
session: TidalSession, prov_playlist_id: str, track_ids: list[str], add: bool = True
) -> None:
"""Async wrapper around the tidal Playlist.add and Playlist.remove function."""
- if add:
- return TidalUserPlaylist(session, prov_playlist_id).add(track_ids)
- for item in track_ids:
- TidalUserPlaylist(session, prov_playlist_id).remove_by_id(int(item))
+ def inner() -> None:
+ if add:
+ return TidalUserPlaylist(session, prov_playlist_id).add(track_ids)
+ for item in track_ids:
+ TidalUserPlaylist(session, prov_playlist_id).remove_by_id(int(item))
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def create_playlist(
+
+async def create_playlist(
session: TidalSession, user_id: str, title: str, description: str = None
) -> TidalPlaylist:
"""Async wrapper around the tidal LoggedInUser.create_playlist function."""
- return LoggedInUser(session, user_id).create_playlist(title, description)
+
+ def inner() -> TidalPlaylist:
+ return LoggedInUser(session, user_id).create_playlist(title, description)
+
+ return await asyncio.to_thread(inner)
-@get_session
-@async_wrap
-def get_similar_tracks(session: TidalSession, prov_track_id, limit: int) -> list[TidalTrack]:
+async def get_similar_tracks(session: TidalSession, prov_track_id, limit: int) -> list[TidalTrack]:
"""Async wrapper around the tidal Track.get_similar_tracks function."""
- return TidalTrack(session, media_id=prov_track_id).get_track_radio(limit)
+ def inner() -> list[TidalTrack]:
+ return TidalTrack(session, media_id=prov_track_id).get_track_radio(limit)
-@get_session
-@async_wrap
-def search(
+ return await asyncio.to_thread(inner)
+
+
+async def search(
session: TidalSession, query: str, media_types=None, limit=50, offset=0
) -> dict[str, str]:
"""Async wrapper around the tidalapi Search function."""
- search_types = []
- if MediaType.ARTIST in media_types:
- search_types.append(TidalArtist)
- if MediaType.ALBUM in media_types:
- search_types.append(TidalAlbum)
- if MediaType.TRACK in media_types:
- search_types.append(TidalTrack)
- if MediaType.PLAYLIST in media_types:
- search_types.append(TidalPlaylist)
-
- models = search_types if search_types else None
- return session.search(query, models, limit, offset)
-
-
-def get_item_mapping(tidal_provider, media_type: MediaType, key: str, name: str) -> ItemMapping:
- """Create a generic item mapping."""
- return ItemMapping(
- media_type,
- key,
- tidal_provider.instance_id,
- name,
- create_uri(media_type, tidal_provider.instance_id, key),
- create_sort_name(tidal_provider.name),
- )
+
+ def inner() -> dict[str, str]:
+ search_types = []
+ if MediaType.ARTIST in media_types:
+ search_types.append(TidalArtist)
+ if MediaType.ALBUM in media_types:
+ search_types.append(TidalAlbum)
+ if MediaType.TRACK in media_types:
+ search_types.append(TidalTrack)
+ if MediaType.PLAYLIST in media_types:
+ search_types.append(TidalPlaylist)
+
+ models = search_types if search_types else None
+ return session.search(query, models, limit, offset)
+
+ return await asyncio.to_thread(inner)
projects / music-assistant-server.git / commitdiff