Source code for tidalpy.quality

"""Quality profiles, the playback downgrade ladder, and codec helpers.

Tidal's playback API accepts a small set of *API qualities*
(``LOW``/``HIGH``/``LOSSLESS``/``HI_RES_LOSSLESS``/``HI_RES``). Hosts
think in *profiles* (what the user asked for). This module maps between
the two and decides, in order, which API qualities to attempt when the
requested one cannot be delivered.

The ladder encodes real Tidal behaviour (subscribed tier, per-track
availability, spatial-audio quirks) and is the single place it lives.
"""

from __future__ import annotations

import re
from enum import Enum

SPATIAL_TAGS = frozenset({"DOLBY_ATMOS", "SONY_360RA"})
# Raw manifest spellings plus the canonical name normalize_codec()
# produces for them (StreamInfo.codec carries the canonical form).
SPATIAL_CODECS = ("ac4", "ac-4", "mha1", "eac3", "ec-3", "mqa", "dolby")
LOSSLESS_API_QUALITIES = frozenset({"LOSSLESS", "HI_RES", "HI_RES_LOSSLESS"})


[docs] class Quality(str, Enum): """Host-facing quality profiles, ordered from lossy to highest lossless.""" LOW = "LOW" HIGH = "HIGH" LOSSLESS = "LOSSLESS" HI_RES_LOSSLESS = "HI_RES_LOSSLESS"
LOSSLESS_PROFILES = frozenset({Quality.LOSSLESS, Quality.HI_RES_LOSSLESS}) _ALIASES: dict[str, Quality] = { "low": Quality.LOW, "normal": Quality.LOW, "standard": Quality.LOW, "128": Quality.LOW, "aac128": Quality.LOW, "high": Quality.HIGH, "320": Quality.HIGH, "aac320": Quality.HIGH, "aac": Quality.HIGH, "lossless": Quality.LOSSLESS, "hifi": Quality.LOSSLESS, "hi-fi": Quality.LOSSLESS, "flac": Quality.LOSSLESS, "flac16": Quality.LOSSLESS, "16/44": Quality.LOSSLESS, "16/44100": Quality.LOSSLESS, "hi_res": Quality.HI_RES_LOSSLESS, "hi-res": Quality.HI_RES_LOSSLESS, "hi_res_lossless": Quality.HI_RES_LOSSLESS, "hi-res-lossless": Quality.HI_RES_LOSSLESS, "master": Quality.HI_RES_LOSSLESS, "max": Quality.HI_RES_LOSSLESS, "24/192": Quality.HI_RES_LOSSLESS, "24/96": Quality.HI_RES_LOSSLESS, }
[docs] def parse_quality( value: str | Quality | None, default: Quality = Quality.LOSSLESS ) -> Quality: """Parse a human quality label into a :class:`Quality` profile. Accepts enum members, API quality names, and common aliases (``"Master"``, ``"hifi"``, ``"24/192"``, ...). Falls back to ``default`` for empty/unknown input. """ if isinstance(value, Quality): return value if value is None: return default text = str(value).strip().lower().replace(" ", "_") if not text: return default if text in _ALIASES: return _ALIASES[text] if text.upper() in Quality.__members__: return Quality[text.upper()] return default
#: Short human labels for API qualities — display and the #: ``{quality}`` naming placeholder. Mirrors the reference client's #: label set (its ``"Master (legacy)"`` becomes ``"HiRes"``: the old #: label is confusing in a file name). QUALITY_LABELS: dict[str, str] = { "LOW": "Low", "HIGH": "High", "LOSSLESS": "HiFi", "HI_RES": "HiRes", "HI_RES_LOSSLESS": "Max", }
[docs] def quality_label(quality: str | None) -> str: """A short human label for an API quality (``"HI_RES_LOSSLESS"`` -> ``"Max"``). Unknown non-empty values pass through unchanged; empty/None give ``""`` — never ``"Unknown"`` in a file name.""" if not quality: return "" text = str(quality) return QUALITY_LABELS.get(text.upper(), text)
[docs] def normalize_codec(codec: str | None) -> str: """Canonical short codec name (``AAC``/``FLAC``/``ALAC``/...).""" if not codec: return "" lowered = codec.lower() if "mp4a" in lowered or lowered.strip() == "aac": return "AAC" if "flac" in lowered: return "FLAC" if "alac" in lowered: return "ALAC" if "mqa" in lowered: return "MQA" if ( "ac4" in lowered or "ac-4" in lowered or "mha1" in lowered or "eac3" in lowered or "ec-3" in lowered ): return "DOLBY" return codec.strip().upper()
[docs] def is_lossless_codec(codec: str | None) -> bool: lowered = (codec or "").lower() return "flac" in lowered or "alac" in lowered
[docs] def is_spatial_codec(codec: str | None) -> bool: lowered = (codec or "").lower() return any(item in lowered for item in SPATIAL_CODECS)
[docs] def stereo_media_tags(tags: list[str] | tuple[str, ...] | None) -> list[str]: """Media tags with spatial-audio tags removed (stereo delivery only).""" if not tags: return [] return [t for t in tags if t not in SPATIAL_TAGS]
[docs] def resolve_api_quality( profile: Quality, media_tags: list[str] | tuple[str, ...] | None = None, catalog_quality: str | None = None, ) -> str: """The primary API quality to request for a profile and a track. ``catalog_quality`` is the track/album's advertised ``audioQuality`` (e.g. ``HI_RES_LOSSLESS``); ``media_tags`` are its ``mediaMetadata.tags``. When the catalog already says the track is not hi-res, requesting ``HI_RES_LOSSLESS`` just burns a 404, so the primary tier is lowered before any network call. When the catalog is unknown (playlist tracks are not album-bound), the primary tier stays ``HI_RES_LOSSLESS`` — deliberately optimistic where the reference client falls back to ``LOSSLESS``: hi-res playlist tracks would be silently downgraded otherwise, and the v2-first FLAC manifest keeps the cost of a miss to one extra request. """ tags = stereo_media_tags(media_tags) if profile is Quality.LOW: return "LOW" if profile is Quality.HIGH: return "HIGH" if profile is Quality.LOSSLESS: return "LOSSLESS" # HI_RES_LOSSLESS: honour the catalog when it says "not hi-res". if "HIRES_LOSSLESS" in tags or "HI_RES_LOSSLESS" in tags: return "HI_RES_LOSSLESS" if catalog_quality in ("LOSSLESS",): return "LOSSLESS" if catalog_quality == "HI_RES": return "HI_RES" return "HI_RES_LOSSLESS"
[docs] def playback_attempts( profile: Quality, media_tags: list[str] | tuple[str, ...] | None = None, catalog_quality: str | None = None, ) -> list[str]: """Ordered API qualities to attempt for a playback request.""" primary = resolve_api_quality(profile, media_tags, catalog_quality) attempts = [primary] if primary == "HI_RES_LOSSLESS": attempts.extend(["HI_RES", "LOSSLESS"]) elif primary == "HI_RES": attempts.append("LOSSLESS") ordered: list[str] = [] for item in attempts: if item not in ordered: ordered.append(item) return ordered
[docs] def needs_stereo_fallback(requested_api_quality: str, codec: str | None) -> bool: """A lossless request that came back spatial/proprietary/lossy. Tidal sometimes answers a lossless request with a spatial or proprietary (MQA/E-AC3) rendition, which is useless for the DJ/CDJ use case; the caller should retry at ``LOSSLESS``. """ if requested_api_quality not in LOSSLESS_API_QUALITIES: return False if is_spatial_codec(codec): return True lowered = (codec or "").lower() return ( "mp4a" in lowered or lowered == "aac" or "mqa" in lowered or "eac3" in lowered )
[docs] def quality_meets_profile( profile: Quality, delivered_quality: str | None, codec: str | None ) -> bool: """Whether a delivered stream satisfies the host's profile. Lossy profiles are always satisfied (Tidal may deliver ``HIGH`` for a ``LOW`` request, which is fine). Lossless profiles require an actually lossless delivery (API quality or codec). """ if profile not in LOSSLESS_PROFILES: return True if (delivered_quality or "").upper() in LOSSLESS_API_QUALITIES: return True return is_lossless_codec(codec)
[docs] def representation_key(rep: dict) -> tuple: """Fidelity ranking for DASH representations. Higher is better: lossless first, then bit depth, sample rate, bandwidth, channels. Selection uses this key instead of trusting the XML order Tidal emits. """ codec = str(rep.get("codec") or "").lower() return ( int(is_lossless_codec(codec)), int(rep.get("bit_depth") or 0), int(rep.get("sample_rate") or 0), int(rep.get("bandwidth") or 0), int(rep.get("channels") or 0), )
[docs] def select_representation( representations: list[dict], requested_api_quality: str ) -> dict: """Pick the best representation for a requested API quality. - non-audio / spatial representations are filtered out - lossless requests get the best lossless representation (for ``HI_RES_LOSSLESS`` the highest-rate FLAC specifically) - everything else falls back to the best AAC, then the best overall """ if not representations: raise ValueError("No audio representations in manifest") candidates = [ rep for rep in representations if not is_spatial_codec(rep.get("codec", "")) ] if not candidates: candidates = list(representations) if requested_api_quality in LOSSLESS_API_QUALITIES: lossless = [ rep for rep in candidates if is_lossless_codec(rep.get("codec", "")) ] if lossless: if requested_api_quality == "HI_RES_LOSSLESS": flac = [ rep for rep in lossless if "flac" in str(rep.get("codec") or "").lower() ] if flac: return max(flac, key=representation_key) return max(lossless, key=representation_key) for rep in candidates: codec = str(rep.get("codec") or "").lower() if codec == "aac" or "mp4a" in codec: return rep return max(candidates, key=representation_key)
[docs] def parse_iso8601_duration(value: str | int | float | None) -> int: """``PT4M48S`` -> whole seconds (0 for missing/invalid). Accepts a plain seconds value too (int/float, or a bare number string): the API paths disagree on the shape (v2 sends ISO-8601, v1 and search rows send plain seconds), and a bare number must not parse as 0 the way the unit regex would. """ if value is None or value is False: return 0 if isinstance(value, (int, float)): return int(value) text = str(value).strip() if not text: return 0 if re.fullmatch(r"\d+(?:\.\d+)?", text): return int(float(text)) total = 0.0 for amount, unit in re.findall(r"(\d+(?:\.\d+)?)([HMS])", text): total += float(amount) * {"H": 3600, "M": 60, "S": 1}[unit] return int(total)