import coloredlogs
from aiorun import run
+from music_assistant.common.helpers.json import json_loads
from music_assistant.server import MusicAssistant
data_dir = args.config
if not os.path.isdir(data_dir):
os.makedirs(data_dir)
+
+ # TEMP: override options though hass config file
+ hass_options_file = os.path.join(data_dir, "options.json")
+ if os.path.isfile(hass_options_file):
+ with open(hass_options_file, "rb") as _file:
+ hass_options = json_loads(_file.read())
+ else:
+ hass_options = {}
+
+ log_level = hass_options.get("log_level", args.log_level).upper()
+
# setup logger
- log_level = args.log_level.upper()
logger = setup_logger(data_dir, log_level)
mass = MusicAssistant(data_dir)
:param media_types: A list of media_types to include.
:param limit: number of items to return in the search (per type).
"""
- # include results from all music providers
- provider_instances = (item.instance_id for item in self.providers)
+ # include results from all (unique) music providers
+ provider_domains = {item.domain for item in self.providers}
results_per_provider: list[SearchResults] = await asyncio.gather(
*[
self.search_provider(
search_query,
media_types,
- provider_instance=provider_instance,
+ provider_domain=provider_domain,
limit=limit,
)
- for provider_instance in provider_instances
+ for provider_domain in provider_domains
]
)
# return result from all providers while keeping index
provider_instance: str | None = None,
) -> None:
"""Add an item to the library."""
- ctrl = self.get_controller(media_type)
- await ctrl.add_to_library(
+ # make sure we have a full db item
+ full_item = await self.get_item(
+ media_type,
item_id,
provider_domain=provider_domain,
provider_instance=provider_instance,
+ lazy=False,
+ add_to_db=True,
+ )
+ ctrl = self.get_controller(media_type)
+ await ctrl.add_to_library(
+ full_item.item_id,
+ provider_domain=full_item.provider,
)
@api_command("music/library/add_items")
return self.playlists
return None
+ def get_unique_providers(self) -> set[str]:
+ """
+ Return all unique MusicProvider instance ids.
+
+ This will return all filebased instances but only one instance
+ for streaming providers.
+ """
+ instances = set()
+ domains = set()
+ for provider in self.providers:
+ if provider.domain not in domains or provider.is_file():
+ instances.add(provider.instance_id)
+ domains.add(provider.domain)
+ return instances
+
def _start_provider_sync(self, provider_instance: str, media_types: tuple[MediaType, ...]):
"""Start sync task on provider and track progress."""
# check if we're not already running a sync task for this provider/mediatype