bit_depth=32,
channels=2,
)
+
+
+# CACHE categories
+
+CACHE_CATEGORY_DEFAULT: Final[int] = 0
+CACHE_CATEGORY_MUSIC_SEARCH: Final[int] = 1
+CACHE_CATEGORY_MUSIC_ALBUM_TRACKS: Final[int] = 2
+CACHE_CATEGORY_MUSIC_ARTIST_TRACKS: Final[int] = 3
+CACHE_CATEGORY_MUSIC_ARTIST_ALBUMS: Final[int] = 4
+CACHE_CATEGORY_MUSIC_PLAYLIST_TRACKS: Final[int] = 5
+CACHE_CATEGORY_MUSIC_PROVIDER_ITEM: Final[int] = 6
+CACHE_CATEGORY_PLAYER_QUEUE_STATE: Final[int] = 7
+CACHE_CATEGORY_MEDIA_INFO: Final[int] = 8
+CACHE_CATEGORY_LIBRARY_ITEMS: Final[int] = 9
+CACHE_CATEGORY_PLAYERS: Final[int] = 10
+
+# CACHE base keys
+CACHE_KEY_PLAYER_POWER: Final[str] = "player_power"
from collections.abc import Iterable
from typing import TYPE_CHECKING, Any
-from music_assistant_models.enums import AlbumType, CacheCategory, MediaType, ProviderFeature
+from music_assistant_models.enums import AlbumType, MediaType, ProviderFeature
from music_assistant_models.errors import InvalidDataError, MediaNotFoundError, MusicAssistantError
from music_assistant_models.media_items import Album, Artist, ItemMapping, Track, UniqueList
-from music_assistant.constants import DB_TABLE_ALBUM_ARTISTS, DB_TABLE_ALBUM_TRACKS, DB_TABLE_ALBUMS
+from music_assistant.constants import (
+ CACHE_CATEGORY_MUSIC_ALBUM_TRACKS,
+ CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
+ DB_TABLE_ALBUM_ARTISTS,
+ DB_TABLE_ALBUM_TRACKS,
+ DB_TABLE_ALBUMS,
+)
from music_assistant.controllers.media.base import MediaControllerBase
from music_assistant.helpers.compare import (
compare_album,
if prov is None:
return []
# prefer cache items (if any) - for streaming providers only
- cache_category = CacheCategory.MUSIC_ALBUM_TRACKS
+ cache_category = CACHE_CATEGORY_MUSIC_ALBUM_TRACKS
cache_base_key = prov.lookup_key
cache_key = item_id
if (
await self.mass.cache.set(
f"track.{item_id}",
item.to_dict(),
- category=CacheCategory.MUSIC_PROVIDER_ITEM,
+ category=CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
base_key=prov.lookup_key,
)
return items
import contextlib
from typing import TYPE_CHECKING, Any
-from music_assistant_models.enums import AlbumType, CacheCategory, MediaType, ProviderFeature
+from music_assistant_models.enums import AlbumType, MediaType, ProviderFeature
from music_assistant_models.errors import (
MediaNotFoundError,
MusicAssistantError,
from music_assistant_models.media_items import Album, Artist, ItemMapping, Track, UniqueList
from music_assistant.constants import (
+ CACHE_CATEGORY_MUSIC_ARTIST_ALBUMS,
+ CACHE_CATEGORY_MUSIC_ARTIST_TRACKS,
+ CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
DB_TABLE_ALBUM_ARTISTS,
DB_TABLE_ARTISTS,
DB_TABLE_TRACK_ARTISTS,
if prov is None:
return []
# prefer cache items (if any) - for streaming providers
- cache_category = CacheCategory.MUSIC_ARTIST_TRACKS
+ cache_category = CACHE_CATEGORY_MUSIC_ARTIST_TRACKS
cache_base_key = prov.lookup_key
cache_key = item_id
if (
await self.mass.cache.set(
f"track.{item_id}",
item.to_dict(),
- category=CacheCategory.MUSIC_PROVIDER_ITEM,
+ category=CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
base_key=prov.lookup_key,
)
else:
if prov is None:
return []
# prefer cache items (if any)
- cache_category = CacheCategory.MUSIC_ARTIST_ALBUMS
+ cache_category = CACHE_CATEGORY_MUSIC_ARTIST_ALBUMS
cache_base_key = prov.lookup_key
cache_key = item_id
if (
from contextlib import suppress
from typing import TYPE_CHECKING, Any, Generic, TypeVar
-from music_assistant_models.enums import (
- CacheCategory,
- EventType,
- ExternalID,
- MediaType,
- ProviderFeature,
-)
+from music_assistant_models.enums import EventType, ExternalID, MediaType, ProviderFeature
from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
from music_assistant_models.media_items import (
Album,
Track,
)
-from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PROVIDER_MAPPINGS, MASS_LOGGER_NAME
+from music_assistant.constants import (
+ CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
+ CACHE_CATEGORY_MUSIC_SEARCH,
+ DB_TABLE_PLAYLOG,
+ DB_TABLE_PROVIDER_MAPPINGS,
+ MASS_LOGGER_NAME,
+)
from music_assistant.helpers.compare import compare_media_item, create_safe_string
from music_assistant.helpers.json import json_loads, serialize_to_json
return []
# prefer cache items (if any)
- cache_category = CacheCategory.MUSIC_SEARCH
+ cache_category = CACHE_CATEGORY_MUSIC_SEARCH
cache_base_key = prov.lookup_key
cache_key = f"{search_query}.{limit}.{self.media_type.value}"
if (
if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
- cache_category = CacheCategory.MUSIC_PROVIDER_ITEM
+ cache_category = CACHE_CATEGORY_MUSIC_PROVIDER_ITEM
cache_base_key = provider.lookup_key
cache_key = f"{self.media_type.value}.{item_id}"
if not force_refresh and (
from collections.abc import AsyncGenerator
from typing import Any
-from music_assistant_models.enums import CacheCategory, MediaType, ProviderFeature
+from music_assistant_models.enums import MediaType, ProviderFeature
from music_assistant_models.errors import (
InvalidDataError,
MediaNotFoundError,
)
from music_assistant_models.media_items import Playlist, Track
-from music_assistant.constants import DB_TABLE_PLAYLISTS
+from music_assistant.constants import (
+ CACHE_CATEGORY_MUSIC_PLAYLIST_TRACKS,
+ CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
+ DB_TABLE_PLAYLISTS,
+)
from music_assistant.helpers.compare import create_safe_string
from music_assistant.helpers.json import serialize_to_json
from music_assistant.helpers.uri import create_uri, parse_uri
if not provider:
return []
# prefer cache items (if any)
- cache_category = CacheCategory.MUSIC_PLAYLIST_TRACKS
+ cache_category = CACHE_CATEGORY_MUSIC_PLAYLIST_TRACKS
cache_base_key = provider.lookup_key
cache_key = f"{item_id}.{page}"
if (
await self.mass.cache.set(
f"track.{item_id}",
item.to_dict(),
- category=CacheCategory.MUSIC_PROVIDER_ITEM,
+ category=CACHE_CATEGORY_MUSIC_PROVIDER_ITEM,
base_key=provider.lookup_key,
)
return items
from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
from music_assistant_models.enums import (
- CacheCategory,
ConfigEntryType,
EventType,
MediaType,
from music_assistant_models.unique_list import UniqueList
from music_assistant.constants import (
+ CACHE_CATEGORY_MUSIC_SEARCH,
DB_TABLE_ALBUM_ARTISTS,
DB_TABLE_ALBUM_TRACKS,
DB_TABLE_ALBUMS,
# prefer cache items (if any)
media_types_str = ",".join(media_types)
- cache_category = CacheCategory.MUSIC_SEARCH
+ cache_category = CACHE_CATEGORY_MUSIC_SEARCH
cache_base_key = prov.lookup_key
cache_key = f"{search_query}.{limit}.{media_types_str}"
from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
from music_assistant_models.enums import (
- CacheCategory,
ConfigEntryType,
ContentType,
EventType,
from music_assistant_models.queue_item import QueueItem
from music_assistant.constants import (
+ CACHE_CATEGORY_PLAYER_QUEUE_STATE,
CONF_CROSSFADE,
CONF_FLOW_MODE,
MASS_LOGO_ONLINE,
queue = None
# try to restore previous state
if prev_state := await self.mass.cache.get(
- "state", category=CacheCategory.PLAYER_QUEUE_STATE, base_key=queue_id
+ "state", category=CACHE_CATEGORY_PLAYER_QUEUE_STATE, base_key=queue_id
):
try:
queue = PlayerQueue.from_cache(prev_state)
prev_items = await self.mass.cache.get(
"items",
default=[],
- category=CacheCategory.PLAYER_QUEUE_STATE,
+ category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
base_key=queue_id,
)
queue_items = [QueueItem.from_cache(x) for x in prev_items]
self.mass.cache.set(
"items",
[x.to_cache() for x in self._queue_items[queue_id]],
- category=CacheCategory.PLAYER_QUEUE_STATE,
+ category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
base_key=queue_id,
)
)
self.mass.cache.set(
"state",
queue.to_cache(),
- category=CacheCategory.PLAYER_QUEUE_STATE,
+ category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
base_key=queue_id,
)
)
from collections.abc import Sequence
from typing import TYPE_CHECKING, cast
-from music_assistant_models.enums import CacheCategory, MediaType, ProviderFeature
+from music_assistant_models.enums import MediaType, ProviderFeature
from music_assistant_models.errors import (
MediaNotFoundError,
MusicAssistantError,
Track,
)
+from music_assistant.constants import CACHE_CATEGORY_LIBRARY_ITEMS
+
from .provider import Provider
if TYPE_CHECKING:
if subpath == "artists":
library_item_ids = await self.mass.cache.get(
"artist",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
if subpath == "albums":
library_item_ids = await self.mass.cache.get(
"album",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
if subpath == "tracks":
library_item_ids = await self.mass.cache.get(
"track",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
if subpath == "radios":
library_item_ids = await self.mass.cache.get(
"radio",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
if subpath == "playlists":
library_item_ids = await self.mass.cache.get(
"playlist",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
if subpath == "audiobooks":
library_item_ids = await self.mass.cache.get(
"audiobook",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
if subpath == "podcasts":
library_item_ids = await self.mass.cache.get(
"podcast",
- category=CacheCategory.LIBRARY_ITEMS,
+ category=CACHE_CATEGORY_LIBRARY_ITEMS,
base_key=self.instance_id,
)
if not library_item_ids:
)
# process deletions (= no longer in library)
- cache_category = CacheCategory.LIBRARY_ITEMS
+ cache_category = CACHE_CATEGORY_LIBRARY_ITEMS
cache_base_key = self.instance_id
prev_library_items: list[int] | None
import shortuuid
from music_assistant_models.config_entries import ConfigEntry
from music_assistant_models.enums import (
- CacheCategory,
ConfigEntryType,
ContentType,
ImageType,
)
from music_assistant_models.streamdetails import StreamDetails
-from music_assistant.constants import MASS_LOGO, VARIOUS_ARTISTS_FANART
+from music_assistant.constants import CACHE_CATEGORY_MEDIA_INFO, MASS_LOGO, VARIOUS_ARTISTS_FANART
from music_assistant.helpers.tags import AudioTags, async_parse_tags
from music_assistant.helpers.uri import parse_uri
from music_assistant.models.music_provider import MusicProvider
async def _get_media_info(self, url: str, force_refresh: bool = False) -> AudioTags:
"""Retrieve mediainfo for url."""
- cache_category = CacheCategory.MEDIA_INFO
+ cache_category = CACHE_CATEGORY_MEDIA_INFO
cache_base_key = self.lookup_key
# do we have some cached info for this url ?
cached_info = await self.mass.cache.get(
from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
from music_assistant_models.enums import (
AlbumType,
- CacheCategory,
ConfigEntryType,
ContentType,
ExternalID,
from tidalapi import Track as TidalTrack
from tidalapi import exceptions as tidal_exceptions
+from music_assistant.constants import CACHE_CATEGORY_DEFAULT, CACHE_CATEGORY_MEDIA_INFO
from music_assistant.helpers.auth import AuthenticationHelper
from music_assistant.helpers.tags import AudioTags, async_parse_tags
from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
# Try to get from cache first
cache_key = f"isrc_map_{item_id}"
cached_track_id = await self.mass.cache.get(
- cache_key, category=CacheCategory.DEFAULT, base_key=self.lookup_key
+ cache_key, category=CACHE_CATEGORY_DEFAULT, base_key=self.lookup_key
)
if cached_track_id:
except MediaNotFoundError:
# Track no longer exists, invalidate cache
await self.mass.cache.delete(
- cache_key, category=CacheCategory.DEFAULT, base_key=self.lookup_key
+ cache_key, category=CACHE_CATEGORY_DEFAULT, base_key=self.lookup_key
)
# Lookup by ISRC if no cache or cached track not found
# Cache the mapping for future use
await self.mass.cache.set(
- cache_key, tracks[0].id, category=CacheCategory.DEFAULT, base_key=self.lookup_key
+ cache_key, tracks[0].id, category=CACHE_CATEGORY_DEFAULT, base_key=self.lookup_key
)
return tracks[0]
self, item_id: str, url: str, force_refresh: bool = False
) -> AudioTags:
"""Retrieve (cached) mediainfo for track."""
- cache_category = CacheCategory.MEDIA_INFO
+ cache_category = CACHE_CATEGORY_MEDIA_INFO
cache_base_key = self.lookup_key
# do we have some cached info for this url ?
cached_info = await self.mass.cache.get(
projects / music-assistant-server.git / commitdiff