from __future__ import annotations
import asyncio
+import itertools
import statistics
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
raise ProviderUnavailableError(f"Provider {provider_id} is not available")
async def search(
- self, search_query, media_types: List[MediaType], limit: int = 10
+ self,
+ search_query,
+ media_types: List[MediaType] = MediaType.ALL,
+ limit: int = 10,
) -> List[MediaItemType]:
"""
Perform global search for media items on all providers.
# include results from all music providers
provider_ids = [item.id for item in self.providers]
# TODO: sort by name and filter out duplicates ?
- return await asyncio.gather(
- *[
- self.search_provider(
- search_query, media_types, provider_id=prov_id, limit=limit
- )
- for prov_id in provider_ids
- ]
+ return itertools.chain.from_iterable(
+ await asyncio.gather(
+ *[
+ self.search_provider(
+ search_query, media_types, provider_id=prov_id, limit=limit
+ )
+ for prov_id in provider_ids
+ ]
+ )
)
async def search_provider(
self,
search_query: str,
- media_types: List[MediaType],
+ media_types: List[MediaType] = MediaType.ALL,
provider: Optional[ProviderType] = None,
provider_id: Optional[str] = None,
limit: int = 10,
# prefer cache items (if any)
cache_key = f"{prov.type.value}.search.{search_query}.{limit}"
- cache_key += "".join(media_types)
+ cache_key += "".join((x.value for x in media_types))
if cache := await self.mass.cache.get(cache_key):
return [media_from_dict(x) for x in cache]
"""All enums used by the Music Assistant models."""
from enum import Enum, IntEnum
+from typing import List
class MediaType(Enum):
ANNOUNCEMENT = "announcement"
UNKNOWN = "unknown"
+ @classmethod
+ @property
+ def ALL(cls) -> List["MediaType"]: # pylint: disable=invalid-name
+ """Return all (default) MediaTypes as list."""
+ return [
+ MediaType.ARTIST,
+ MediaType.ALBUM,
+ MediaType.TRACK,
+ MediaType.PLAYLIST,
+ MediaType.RADIO,
+ ]
+
class MediaQuality(IntEnum):
"""Enum for Media Quality."""