from tidalapi import Track as TidalTrack
from tidalapi.media import Lyrics as TidalLyrics
-from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
+from music_assistant.common.models.config_entries import (
+ ConfigEntry,
+ ConfigValueOption,
+ ConfigValueType,
+)
from music_assistant.common.models.enums import (
AlbumType,
ConfigEntryType,
Track,
)
from music_assistant.server.helpers.auth import AuthenticationHelper
+from music_assistant.server.helpers.tags import AudioTags, parse_tags
from music_assistant.server.models.music_provider import MusicProvider
from .helpers import (
CONF_REFRESH_TOKEN = "refresh_token"
CONF_USER_ID = "user_id"
CONF_EXPIRY_TIME = "expiry_time"
+CONF_QUALITY = "quality"
async def setup(
return prov
-async def tidal_code_login(auth_helper: AuthenticationHelper) -> TidalSession:
+async def tidal_code_login(auth_helper: AuthenticationHelper, quality: str) -> TidalSession:
"""Async wrapper around the tidalapi Session function."""
def inner() -> TidalSession:
- config = TidalConfig(quality=TidalQuality.lossless, item_limit=10000, alac=False)
+ config = TidalConfig(quality=TidalQuality[quality], item_limit=10000, alac=False)
session = TidalSession(config=config)
login, future = session.login_oauth()
auth_helper.send_url(f"https://{login.verification_uri_complete}")
# config flow auth action/step (authenticate button clicked)
if action == CONF_ACTION_AUTH:
async with AuthenticationHelper(mass, values["session_id"]) as auth_helper:
- tidal_session = await tidal_code_login(auth_helper)
+ tidal_session = await tidal_code_login(auth_helper, values.get(CONF_QUALITY))
if not tidal_session.check_login():
raise LoginFailed("Authentication to Tidal failed")
# set the retrieved token on the values object to pass along
values[CONF_EXPIRY_TIME] = tidal_session.expiry_time.isoformat()
values[CONF_USER_ID] = str(tidal_session.user.id)
+ # config flow auth action/step to pick the library to use
+ # because this call is very slow, we only show/calculate the dropdown if we do
+ # not yet have this info or we/user invalidated it.
+
# return the collected config entries
return (
+ ConfigEntry(
+ key=CONF_QUALITY,
+ type=ConfigEntryType.STRING,
+ label="Quality",
+ required=True,
+ description="The Tidal Quality you wish to use",
+ options=[
+ ConfigValueOption(
+ title=TidalQuality.low_96k.value, value=TidalQuality.low_96k.name
+ ),
+ ConfigValueOption(
+ title=TidalQuality.low_320k.value, value=TidalQuality.low_320k.name
+ ),
+ ConfigValueOption(
+ title=TidalQuality.high_lossless.value, value=TidalQuality.high_lossless.name
+ ),
+ ConfigValueOption(title=TidalQuality.hi_res.value, value=TidalQuality.hi_res.name),
+ ],
+ default_value=TidalQuality.high_lossless.name,
+ value=values.get(CONF_QUALITY) if values else None,
+ ),
ConfigEntry(
key=CONF_AUTH_TOKEN,
type=ConfigEntryType.SECURE_STRING,
label="Authentication token for Tidal",
description="You need to link Music Assistant to your Tidal account.",
action=CONF_ACTION_AUTH,
+ depends_on=CONF_QUALITY,
action_label="Authenticate on Tidal.com",
value=values.get(CONF_AUTH_TOKEN) if values else None,
),
)
yield track
- async def get_similar_tracks(self, prov_track_id: str, limit=25) -> list[Track]: # noqa: ARG002
+ async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
"""Get similar tracks for given track id."""
tidal_session = await self._get_tidal_session()
async with self._throttler:
return [
await self._parse_track(track_obj=track)
- # Re-add limit here after tidalapi supports it, and remove noqa above
- for track in await get_similar_tracks(tidal_session, prov_track_id)
+ for track in await get_similar_tracks(tidal_session, prov_track_id, limit)
]
async def library_add(self, prov_item_id: str, media_type: MediaType):
tidal_session = await self._get_tidal_session()
track = await get_track(tidal_session, item_id)
url = await get_track_url(tidal_session, item_id)
+ media_info = await self._get_media_info(item_id=item_id, url=url)
if not track:
raise MediaNotFoundError(f"track {item_id} not found")
return StreamDetails(
item_id=track.id,
provider=self.instance_id,
audio_format=AudioFormat(
- content_type=ContentType.FLAC,
- sample_rate=44100,
- bit_depth=16,
+ content_type=ContentType.try_parse(media_info.format),
+ sample_rate=media_info.sample_rate,
+ bit_depth=media_info.bits_per_sample,
+ channels=media_info.channels,
),
duration=track.duration,
direct=url,
return self._tidal_session
self._tidal_session = await self._load_tidal_session(
token_type="Bearer",
+ quality=self.config.get_value(CONF_QUALITY),
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)),
return self._tidal_session
async def _load_tidal_session(
- self, token_type, access_token, refresh_token=None, expiry_time=None
+ self, token_type, quality: TidalQuality, 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)
+ config = TidalConfig(quality=TidalQuality[quality], item_limit=10000, alac=False)
session = TidalSession(config=config)
session.load_oauth_session(token_type, access_token, refresh_token, expiry_time)
return session
provider_instance=self.instance_id,
audio_format=AudioFormat(
content_type=ContentType.FLAC,
- sample_rate=44100,
- bit_depth=16,
+ bit_depth=24 if self._is_hi_res(track_obj=track_obj) else 16,
),
isrc=track_obj.isrc,
url=f"http://www.tidal.com/tracks/{track_id}",
yield item
if len(chunk) < DEFAULT_LIMIT:
break
+
+ async def _get_media_info(
+ self, item_id: str, url: str, force_refresh: bool = False
+ ) -> AudioTags:
+ """Retrieve (cached) mediainfo for track."""
+ cache_key = f"{self.instance_id}.media_info.{item_id}"
+ # do we have some cached info for this url ?
+ cached_info = await self.mass.cache.get(cache_key)
+ if cached_info and not force_refresh:
+ media_info = AudioTags.parse(cached_info)
+ else:
+ # parse info with ffprobe (and store in cache)
+ media_info = await parse_tags(url)
+ await self.mass.cache.set(cache_key, media_info.raw)
+ return media_info
+
+ def _is_hi_res(self, track_obj: TidalTrack) -> bool:
+ """Check if track is hi-res."""
+ return track_obj.audio_quality.value == "HI_RES"
projects / music-assistant-server.git / commitdiff