import asyncio
import base64
import pickle
+from collections.abc import Callable
from contextlib import suppress
from datetime import datetime, timedelta
from enum import StrEnum
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, ParamSpec, TypeVar, cast
from tidalapi import Album as TidalAlbum
from tidalapi import Artist as TidalArtist
ProviderMapping,
SearchResults,
Track,
+ UniqueList,
)
from music_assistant.common.models.streamdetails import StreamDetails
from music_assistant.server.helpers.auth import AuthenticationHelper
)
if TYPE_CHECKING:
- from collections.abc import AsyncGenerator, Awaitable, Callable
+ from collections.abc import AsyncGenerator, Awaitable
from tidalapi.media import Lyrics as TidalLyrics
from tidalapi.media import Stream as TidalStream
BROWSE_URL = "https://tidal.com/browse"
RESOURCES_URL = "https://resources.tidal.com/images"
+_R = TypeVar("_R")
+_P = ParamSpec("_P")
+
class TidalQualityEnum(StrEnum):
"""Enum for Tidal Quality."""
action: [optional] action key called from config entries UI.
values: the (intermediate) raw values for config entries sent with the action.
"""
+ assert values is not None
+
if action == CONF_ACTION_START_PKCE_LOGIN:
async with AuthenticationHelper(mass, cast(str, values["session_id"])) as auth_helper:
- quality: str = values.get(CONF_QUALITY) if values else None
- base64_session = await tidal_auth_url(auth_helper, cast(str, quality))
+ quality = str(values.get(CONF_QUALITY))
+ base64_session = await tidal_auth_url(auth_helper, quality)
values[CONF_TEMP_SESSION] = base64_session
# Tidal is (ab)using the AuthenticationHelper just to send the user to an URL
# there is no actual oauth callback happening, instead the user is redirected
await asyncio.sleep(15)
if action == CONF_ACTION_COMPLETE_PKCE_LOGIN:
- quality: str = values.get(CONF_QUALITY) if values else None
- pkce_url: str = values.get(CONF_OOPS_URL) if values else None
- base64_session = values.get(CONF_TEMP_SESSION) if values else None
+ quality = str(values.get(CONF_QUALITY))
+ pkce_url = str(values.get(CONF_OOPS_URL))
+ base64_session = str(values.get(CONF_TEMP_SESSION))
tidal_session = await tidal_pkce_login(base64_session, pkce_url)
if not tidal_session.check_login():
msg = "Authentication to Tidal failed"
values[CONF_AUTH_TOKEN] = None
if values.get(CONF_AUTH_TOKEN):
- auth_entries = (
+ auth_entries: tuple[ConfigEntry, ...] = (
ConfigEntry(
key="label_ok",
type=ConfigEntryType.LABEL,
"""Implementation of a Tidal MusicProvider."""
_tidal_session: TidalSession | None = None
- _tidal_user_id: str | None = None
+ _tidal_user_id: str
# rate limiter needs to be specified on provider-level,
# so make it an instance attribute
throttler = ThrottlerManager(rate_limit=1, period=2)
async def handle_async_init(self) -> None:
"""Handle async initialization of the provider."""
- self._tidal_user_id: str = self.config.get_value(CONF_USER_ID)
+ self._tidal_user_id = str(self.config.get_value(CONF_USER_ID))
try:
self._tidal_session = await self._get_tidal_session()
except Exception as err:
results = await search(tidal_session, search_query, media_types, limit)
if results["artists"]:
- for artist in results["artists"]:
- parsed_results.artists.append(self._parse_artist(artist))
+ parsed_results.artists = [self._parse_artist(artist) for artist in results["artists"]]
if results["albums"]:
- for album in results["albums"]:
- parsed_results.albums.append(self._parse_album(album))
+ parsed_results.albums = [self._parse_album(album) for album in results["albums"]]
if results["playlists"]:
- for playlist in results["playlists"]:
- parsed_results.playlists.append(self._parse_playlist(playlist))
+ parsed_results.playlists = [
+ self._parse_playlist(playlist) for playlist in results["playlists"]
+ ]
if results["tracks"]:
- for track in results["tracks"]:
- parsed_results.tracks.append(self._parse_track(track))
+ parsed_results.tracks = [self._parse_track(track) for track in results["tracks"]]
return parsed_results
async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
track = self._parse_track(track_obj)
# get some extra details for the full track info
with suppress(tidal_exceptions.MetadataNotAvailable, AttributeError):
- lyrics: TidalLyrics = await asyncio.to_thread(track.lyrics)
+ lyrics: TidalLyrics = await asyncio.to_thread(track_obj.lyrics)
track.metadata.lyrics = lyrics.text
return track
except tidal_exceptions.ObjectNotFound as err:
return self._tidal_session
self._tidal_session = await self._load_tidal_session(
token_type="Bearer",
- quality=self.config.get_value(CONF_QUALITY),
+ quality=str(self.config.get_value(CONF_QUALITY)),
access_token=str(self.config.get_value(CONF_AUTH_TOKEN)),
refresh_token=str(self.config.get_value(CONF_REFRESH_TOKEN)),
expiry_time=datetime.fromisoformat(str(self.config.get_value(CONF_EXPIRY_TIME))),
if artist_obj.picture:
picture_id = artist_obj.picture.replace("-", "/")
image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
- artist.metadata.images = [
- MediaItemImage(
- type=ImageType.THUMB,
- path=image_url,
- provider=self.lookup_key,
- remotely_accessible=True,
- )
- ]
+ artist.metadata.images = UniqueList(
+ [
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=image_url,
+ provider=self.lookup_key,
+ remotely_accessible=True,
+ )
+ ]
+ )
return artist
if album_obj.cover:
picture_id = album_obj.cover.replace("-", "/")
image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
- album.metadata.images = [
- MediaItemImage(
- type=ImageType.THUMB,
- path=image_url,
- provider=self.lookup_key,
- remotely_accessible=True,
- )
- ]
+ album.metadata.images = UniqueList(
+ [
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=image_url,
+ provider=self.lookup_key,
+ remotely_accessible=True,
+ )
+ ]
+ )
return album
)
if track_obj.isrc:
track.external_ids.add((ExternalID.ISRC, track_obj.isrc))
- track.artists = []
+ track.artists = UniqueList()
for track_artist in track_obj.artists:
artist = self._parse_artist(track_artist)
track.artists.append(artist)
if track_obj.album.cover:
picture_id = track_obj.album.cover.replace("-", "/")
image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
- track.metadata.images = [
- MediaItemImage(
- type=ImageType.THUMB,
- path=image_url,
- provider=self.lookup_key,
- remotely_accessible=True,
- )
- ]
+ track.metadata.images = UniqueList(
+ [
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=image_url,
+ provider=self.lookup_key,
+ remotely_accessible=True,
+ )
+ ]
+ )
return track
def _parse_playlist(self, playlist_obj: TidalPlaylist) -> Playlist:
if picture := (playlist_obj.square_picture or playlist_obj.picture):
picture_id = picture.replace("-", "/")
image_url = f"{RESOURCES_URL}/{picture_id}/750x750.jpg"
- playlist.metadata.images = [
- MediaItemImage(
- type=ImageType.THUMB,
- path=image_url,
- provider=self.lookup_key,
- remotely_accessible=True,
- )
- ]
+ playlist.metadata.images = UniqueList(
+ [
+ MediaItemImage(
+ type=ImageType.THUMB,
+ path=image_url,
+ provider=self.lookup_key,
+ remotely_accessible=True,
+ )
+ ]
+ )
return playlist
async def _iter_items(
- self, func: Awaitable | Callable, *args, **kwargs
- ) -> AsyncGenerator[Any, None]:
+ self,
+ func: Callable[_P, list[_R]] | Callable[_P, Awaitable[list[_R]]],
+ *args: _P.args,
+ **kwargs: _P.kwargs,
+ ) -> AsyncGenerator[_R, None]:
"""Yield all items from a larger listing."""
offset = 0
while True:
if asyncio.iscoroutinefunction(func):
- chunk = await func(*args, **kwargs, offset=offset)
+ chunk = await func(*args, **kwargs, offset=offset) # type: ignore[arg-type]
else:
- chunk = await asyncio.to_thread(func, *args, **kwargs, offset=offset)
+ chunk = await asyncio.to_thread(func, *args, **kwargs, offset=offset) # type: ignore[arg-type]
offset += len(chunk)
for item in chunk:
yield item
projects / music-assistant-server.git / commitdiff