from collections.abc import Iterable
from dataclasses import dataclass
from types import NoneType
-from typing import Any
+from typing import Any, Self
from mashumaro import DataClassDictMixin
ConfigEntryType.INTEGER: int,
ConfigEntryType.FLOAT: float,
ConfigEntryType.LABEL: str,
+ ConfigEntryType.DIVIDER: str,
+ ConfigEntryType.ACTION: str,
}
+UI_ONLY = (ConfigEntryType.LABEL, ConfigEntryType.DIVIDER, ConfigEntryType.ACTION)
+
@dataclass
class ConfigValueOption(DataClassDictMixin):
class ConfigEntry(DataClassDictMixin):
"""Model for a Config Entry.
- The definition of something that can be configured for an object (e.g. provider or player)
- within Music Assistant (without the value).
+ The definition of something that can be configured
+ for an object (e.g. provider or player)
+ within Music Assistant.
"""
# key: used as identifier for the entry, also for localization
hidden: bool = False
# advanced: this is an advanced setting (frontend hides it in some corner)
advanced: bool = False
- # encrypt: store string value encrypted and do not send its value in the api
- encrypt: bool = False
-
-
-@dataclass
-class ConfigEntryValue(ConfigEntry):
- """Config Entry with its value parsed."""
-
+ # action: (configentry)action that is needed to get the value for this entry
+ action: str | None = None
+ # action_label: default label for the action when no translation for the action is present
+ action_label: str | None = None
+ # value: set by the config manager/flow (or in rare cases by the provider itself)
value: ConfigValueType = None
- @classmethod
- def parse(
- cls,
- entry: ConfigEntry,
+ def parse_value(
+ self,
value: ConfigValueType,
allow_none: bool = True,
- ) -> ConfigEntryValue:
- """Parse ConfigEntryValue from the config entry and plain value."""
- result = ConfigEntryValue.from_dict(entry.to_dict())
- result.value = value
- expected_type = ConfigEntryTypeMap.get(result.type, NoneType)
- if result.value is None:
- result.value = entry.default_value
- if result.value is None and not entry.required:
+ ) -> ConfigValueType:
+ """Parse value from the config entry details and plain value."""
+ expected_type = ConfigEntryTypeMap.get(self.type, NoneType)
+ if value is None:
+ value = self.default_value
+ if value is None and (not self.required or allow_none):
expected_type = NoneType
- if entry.type == ConfigEntryType.LABEL:
- result.value = result.label
- if not isinstance(result.value, expected_type):
+ if self.type == ConfigEntryType.LABEL:
+ value = self.label
+ if not isinstance(value, expected_type):
# handle common conversions/mistakes
- if expected_type == float and isinstance(result.value, int):
- result.value = float(result.value)
- return result
- if expected_type == int and isinstance(result.value, float):
- result.value = int(result.value)
- return result
- if expected_type == int and isinstance(result.value, str) and result.value.isnumeric():
- result.value = int(result.value)
- return result
- if (
- expected_type == float
- and isinstance(result.value, str)
- and result.value.isnumeric()
- ):
- result.value = float(result.value)
- return result
+ if expected_type == float and isinstance(value, int):
+ self.value = float(value)
+ return self.value
+ if expected_type == int and isinstance(value, float):
+ self.value = int(value)
+ return self.value
+ if expected_type == int and isinstance(value, str) and value.isnumeric():
+ self.value = int(value)
+ return self.value
+ if expected_type == float and isinstance(value, str) and value.isnumeric():
+ self.value = float(value)
+ return self.value
+ if self.type in UI_ONLY:
+ self.value = self.default_value
+ return self.value
# fallback to default
- if result.value is None and allow_none:
- # In some cases we allow this (e.g. create default config)
- result.value = None
- return result
- if entry.default_value:
+ if self.default_value is not None:
LOGGER.warning(
"%s has unexpected type: %s, fallback to default",
- result.key,
- type(result.value),
+ self.key,
+ type(self.value),
)
- result.value = entry.default_value
- return result
- raise ValueError(f"{result.key} has unexpected type: {type(result.value)}")
- return result
+ self.value = self.default_value
+ return self.value
+ raise ValueError(f"{self.key} has unexpected type: {type(value)}")
+ self.value = value
+ return self.value
@dataclass
class Config(DataClassDictMixin):
"""Base Configuration object."""
- values: dict[str, ConfigEntryValue]
+ values: dict[str, ConfigEntry]
def get_value(self, key: str) -> ConfigValueType:
"""Return config value for given key."""
@classmethod
def parse(
- cls,
+ cls: Self,
config_entries: Iterable[ConfigEntry],
raw: dict[str, Any],
- ) -> Config:
+ ) -> Self:
"""Parse Config from the raw values (as stored in persistent storage)."""
- values = {
- x.key: ConfigEntryValue.parse(x, raw.get("values", {}).get(x.key)).to_dict()
- for x in config_entries
- }
- conf = cls.from_dict({**raw, "values": values})
+ conf = cls.from_dict({**raw, "values": {}})
+ for entry in config_entries:
+ # create a copy of the entry
+ conf.values[entry.key] = ConfigEntry.from_dict(entry.to_dict())
+ conf.values[entry.key].parse_value(raw["values"].get(entry.key), allow_none=True)
return conf
def to_raw(self) -> dict[str, Any]:
"""Return minimized/raw dict to store in persistent storage."""
- def _handle_value(value: ConfigEntryValue):
+ def _handle_value(value: ConfigEntry):
if value.type == ConfigEntryType.SECURE_STRING:
assert ENCRYPT_CALLBACK is not None
return ENCRYPT_CALLBACK(value.value)
return {
**self.to_dict(),
"values": {
- x.key: _handle_value(x) for x in self.values.values() if x.value != x.default_value
+ x.key: _handle_value(x)
+ for x in self.values.values()
+ if (x.value != x.default_value and x.type not in UI_ONLY)
},
}
d["values"][key]["value"] = SECURE_STRING_SUBSTITUTE
return d
- def update(self, update: ConfigUpdate) -> set[str]:
+ def update(self, update: dict[str, ConfigValueType]) -> set[str]:
"""Update Config with updated values."""
changed_keys: set[str] = set()
# root values (enabled, name)
- for key in ("enabled", "name"):
- cur_val = getattr(self, key, None)
- new_val = getattr(update, key, None)
- if new_val is None:
+ root_values = ("enabled", "name")
+ for key in root_values:
+ cur_val = getattr(self, key)
+ if key not in update:
continue
+ new_val = update[key]
if new_val == cur_val:
continue
setattr(self, key, new_val)
changed_keys.add(key)
- # update values
- if update.values is not None:
- for key, new_val in update.values.items():
- cur_val = self.values[key].value
- # parse entry to do type validation
- parsed_val = ConfigEntryValue.parse(self.values[key], new_val)
- if cur_val == parsed_val.value:
- continue
- if parsed_val.value is None:
- self.values[key].value = parsed_val.default_value
- else:
- self.values[key].value = parsed_val.value
+ # config entry values
+ for key, new_val in update.items():
+ if key in root_values:
+ continue
+ cur_val = self.values[key].value
+ # parse entry to do type validation
+ parsed_val = self.values[key].parse_value(new_val)
+ if cur_val != parsed_val:
changed_keys.add(f"values/{key}")
return changed_keys
# For now we just use the parse method to check for not allowed None values
# this can be extended later
for value in self.values.values():
- value.parse(value, value.value, allow_none=False)
+ value.parse_value(value.value, allow_none=False)
@dataclass
name: str | None = None
# available: boolean to indicate if the player is available
available: bool = True
- # default_name: default name to use when there is name available
+ # default_name: default name to use when there is no name available
default_name: str | None = None
-@dataclass
-class ConfigUpdate(DataClassDictMixin):
- """Config object to send when updating some/all values through the API."""
-
- enabled: bool | None = None
- name: str | None = None
- values: dict[str, ConfigValueType] | None = None
-
-
DEFAULT_PROVIDER_CONFIG_ENTRIES = (
ConfigEntry(
key=CONF_LOG_LEVEL,
label="Output codec",
options=[
ConfigValueOption("FLAC (lossless, compact file size)", "flac"),
- ConfigValueOption("M4A AAC (lossy, superior quality)", "aac"),
+ ConfigValueOption("AAC (lossy, superior quality)", "aac"),
ConfigValueOption("MP3 (lossy, average quality)", "mp3"),
ConfigValueOption("WAV (lossless, huge file size)", "wav"),
],
PROVIDERS_UPDATED = "providers_updated"
PLAYER_CONFIG_UPDATED = "player_config_updated"
SYNC_TASKS_UPDATED = "sync_tasks_updated"
+ AUTH_SESSION = "auth_session"
class ProviderFeature(StrEnum):
#
# PLAYERPROVIDER FEATURES
#
- CREATE_PLAYER_CONFIG = "create_player_config"
+ # we currently have none ;-)
#
# METADATAPROVIDER FEATURES
INTEGER = "integer"
FLOAT = "float"
LABEL = "label"
+ DIVIDER = "divider"
+ ACTION = "action"
import base64
import logging
import os
+from contextlib import suppress
from typing import TYPE_CHECKING, Any
from uuid import uuid4
from music_assistant.common.models.config_entries import (
DEFAULT_PLAYER_CONFIG_ENTRIES,
DEFAULT_PROVIDER_CONFIG_ENTRIES,
- ConfigEntryValue,
- ConfigUpdate,
+ ConfigEntry,
+ ConfigValueType,
PlayerConfig,
ProviderConfig,
)
if not self._timer_handle:
# no point in forcing a save when there are no changes pending
return
- await self.async_save()
+ await self._async_save()
LOGGER.debug("Stopped.")
def get(self, key: str, default: Any = None) -> Any:
async def get_provider_config(self, instance_id: str) -> ProviderConfig:
"""Return configuration for a single provider."""
if raw_conf := self.get(f"{CONF_PROVIDERS}/{instance_id}", {}):
- for prov in self.mass.get_available_providers():
- if prov.domain != raw_conf["domain"]:
- continue
- prov_mod = await get_provider_module(prov.domain)
- prov_config_entries = await prov_mod.get_config_entries(self.mass, prov)
- config_entries = DEFAULT_PROVIDER_CONFIG_ENTRIES + prov_config_entries
- return ProviderConfig.parse(config_entries, raw_conf)
+ config_entries = await self.get_provider_config_entries(
+ raw_conf["domain"], instance_id=instance_id, values=raw_conf.get("values")
+ )
+ return ProviderConfig.parse(config_entries, raw_conf)
raise KeyError(f"No config found for provider id {instance_id}")
- @api_command("config/providers/update")
- async def update_provider_config(self, instance_id: str, update: ConfigUpdate) -> None:
- """Update ProviderConfig."""
- config = await self.get_provider_config(instance_id)
- changed_keys = config.update(update)
- available = prov.available if (prov := self.mass.get_provider(instance_id)) else False
- if not changed_keys and (config.enabled == available):
- # no changes
- return
- # try to load the provider first to catch errors before we save it.
- if config.enabled:
- await self.mass.load_provider(config)
+ @api_command("config/providers/get_entries")
+ async def get_provider_config_entries(
+ self,
+ provider_domain: str,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
+ ) -> tuple[ConfigEntry, ...]:
+ """
+ Return Config entries to setup/configure a provider.
+
+ provider_domain: (mandatory) domain of the provider.
+ instance_id: id of an existing provider instance (None for new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # lookup provider manifest and module
+ for prov in self.mass.get_available_providers():
+ if prov.domain == provider_domain:
+ prov_mod = await get_provider_module(provider_domain)
+ break
else:
- await self.mass.unload_provider(config.instance_id)
- # load succeeded, save new config
- config.last_error = None
- conf_key = f"{CONF_PROVIDERS}/{instance_id}"
- self.set(conf_key, config.to_raw())
+ raise KeyError(f"Unknown provider domain: {provider_domain}")
+ if values is None:
+ values = self.get(f"{CONF_PROVIDERS}/{instance_id}/values", {}) if instance_id else {}
+ return (
+ await prov_mod.get_config_entries(
+ self.mass, instance_id=instance_id, action=action, values=values
+ )
+ + DEFAULT_PROVIDER_CONFIG_ENTRIES
+ )
- @api_command("config/providers/add")
- async def add_provider_config(
- self, provider_domain: str, config: ProviderConfig | None = None
+ @api_command("config/providers/save")
+ async def save_provider_config(
+ self,
+ provider_domain: str,
+ values: dict[str, ConfigValueType],
+ instance_id: str | None = None,
) -> ProviderConfig:
- """Add new Provider (instance) Config Flow."""
- if not config:
- return await self._get_default_provider_config(provider_domain)
- for key, conf_entry_value in config.values.items():
- # parse entry to do type validation
- parsed_val = ConfigEntryValue.parse(conf_entry_value, conf_entry_value.value)
- conf_entry_value.value = parsed_val.value
- # if provider config is provided, the frontend wants to submit a new provider instance
- # based on the earlier created template config.
- # try to load the provider first to catch errors before we save it.
- await self.mass.load_provider(config)
- # config provided and load success, storeconfig
- conf_key = f"{CONF_PROVIDERS}/{config.instance_id}"
- self.set(conf_key, config.to_raw())
- return config
+ """
+ Save Provider(instance) Config.
+
+ provider_domain: (mandatory) domain of the provider.
+ values: the raw values for config entries that need to be stored/updated.
+ instance_id: id of an existing provider instance (None for new instance setup).
+ action: [optional] action key called from config entries UI.
+ """
+ if instance_id is not None:
+ config = await self._update_provider_config(instance_id, values)
+ else:
+ config = await self._add_provider_config(provider_domain, values)
+ # return full config, just in case
+ return await self.get_provider_config(config.instance_id)
@api_command("config/providers/remove")
async def remove_provider_config(self, instance_id: str) -> None:
# cleanup entries in library
await self.mass.music.cleanup_provider(instance_id)
+ async def remove_provider_config_value(self, instance_id: str, key: str) -> None:
+ """Remove/reset single Provider config value."""
+ conf_key = f"{CONF_PROVIDERS}/{instance_id}/values/{key}"
+ existing = self.get(conf_key)
+ if not existing:
+ return
+ self.remove(conf_key)
+
@api_command("config/providers/reload")
async def reload_provider(self, instance_id: str) -> None:
"""Reload provider."""
raise KeyError(f"No config found for player id {player_id}")
@api_command("config/players/get_value")
- def get_player_config_value(self, player_id: str, key: str) -> ConfigEntryValue:
+ def get_player_config_value(self, player_id: str, key: str) -> ConfigValueType:
"""Return single configentry value for a player."""
conf = self.get(f"{CONF_PLAYERS}/{player_id}")
if not conf:
entries = DEFAULT_PLAYER_CONFIG_ENTRIES + prov.get_player_config_entries(player_id)
for entry in entries:
if entry.key == key:
- return ConfigEntryValue.parse(entry, conf["values"].get(key))
+ # always create a copy to prevent we're altering the base object
+ return ConfigEntry.from_dict(entry.to_dict()).parse_value(conf["values"].get(key))
raise KeyError(f"ConfigEntry {key} is invalid")
- @api_command("config/players/update")
- def update_player_config(self, player_id: str, update: ConfigUpdate) -> None:
- """Update PlayerConfig."""
+ @api_command("config/players/save")
+ def save_player_config(
+ self, player_id: str, values: dict[str, ConfigValueType]
+ ) -> PlayerConfig:
+ """Save/update PlayerConfig."""
config = self.get_player_config(player_id)
- changed_keys = config.update(update)
+ changed_keys = config.update(values)
if not changed_keys:
# no changes
data=config,
)
# signal update to the player manager
- try:
+ with suppress(PlayerUnavailableError):
player = self.mass.players.get(config.player_id)
player.enabled = config.enabled
self.mass.players.update(config.player_id, force_update=True)
- except PlayerUnavailableError:
- pass
# signal player provider that the config changed
- try:
+ with suppress(PlayerUnavailableError):
if provider := self.mass.get_provider(config.provider):
- assert isinstance(provider, PlayerProvider)
provider.on_player_config_changed(config, changed_keys)
- except PlayerUnavailableError:
- pass
-
- @api_command("config/players/create")
- async def create_player_config(
- self, provider_domain: str, config: PlayerConfig | None = None
- ) -> PlayerConfig:
- """Register a new Player(config) if the provider supports this."""
- provider: PlayerProvider = self.mass.get_provider(provider_domain)
- return await provider.create_player_config(config)
+ # return full player config (just in case)
+ return self.get_player_config(player_id)
@api_command("config/players/remove")
async def remove_player_config(self, player_id: str) -> None:
for conf in await self.get_provider_configs(provider_domain=provider_domain):
# return if there is already a config
return
- # config does not yet exist, create a default one
- default_config = await self._get_default_provider_config(provider_domain)
- conf_key = f"{CONF_PROVIDERS}/{default_config.instance_id}"
- self.set(conf_key, default_config.to_raw())
-
- async def _get_default_provider_config(self, provider_domain: str) -> ProviderConfig:
- """
- Return default/empty ProviderConfig.
-
- This is intended to be used as helper method to add a new provider,
- and it performs some quick sanity checks as well as handling the
- instance_id generation.
- """
- # lookup provider manifest
for prov in self.mass.get_available_providers():
if prov.domain == provider_domain:
manifest = prov
break
else:
raise KeyError(f"Unknown provider domain: {provider_domain}")
-
- # determine instance id based on previous configs
- existing = {
- x.instance_id for x in await self.get_provider_configs(provider_domain=provider_domain)
- }
-
- if existing and not manifest.multi_instance:
- raise ValueError(f"Provider {manifest.name} does not support multiple instances")
-
- if len(existing) == 0:
- instance_id = provider_domain
- name = manifest.name
- else:
- instance_id = f"{provider_domain}{len(existing)+1}"
- name = f"{manifest.name} {len(existing)+1}"
-
- # all checks passed, return a default config
- prov_mod = await get_provider_module(provider_domain)
- config_entries = await prov_mod.get_config_entries(self.mass, manifest)
- return ProviderConfig.parse(
- DEFAULT_PROVIDER_CONFIG_ENTRIES + config_entries,
+ config_entries = await self.get_provider_config_entries(provider_domain)
+ default_config: ProviderConfig = ProviderConfig.parse(
+ config_entries,
{
"type": manifest.type.value,
"domain": manifest.domain,
- "instance_id": instance_id,
- "name": name,
+ "instance_id": manifest.domain,
+ "name": manifest.name,
+ # note: this will only work for providers that do
+ # not have any required config entries or provide defaults
"values": {},
},
)
+ default_config.validate()
+ conf_key = f"{CONF_PROVIDERS}/{default_config.instance_id}"
+ self.set(conf_key, default_config.to_raw())
+
+ def save(self, immediate: bool = False) -> None:
+ """Schedule save of data to disk."""
+ if self._timer_handle is not None:
+ self._timer_handle.cancel()
+ self._timer_handle = None
+
+ if immediate:
+ self.mass.loop.create_task(self._async_save())
+ else:
+ # schedule the save for later
+ self._timer_handle = self.mass.loop.call_later(
+ DEFAULT_SAVE_DELAY, self.mass.create_task, self._async_save
+ )
+
+ def encrypt_string(self, str_value: str) -> str:
+ """Encrypt a (password)string with Fernet."""
+ if str_value.startswith(ENCRYPT_SUFFIX):
+ return str_value
+ return ENCRYPT_SUFFIX + self._fernet.encrypt(str_value.encode()).decode()
+
+ def decrypt_string(self, encrypted_str: str) -> str:
+ """Decrypt a (password)string with Fernet."""
+ if not encrypted_str:
+ return encrypted_str
+ if not encrypted_str.startswith(ENCRYPT_SUFFIX):
+ return encrypted_str
+ encrypted_str = encrypted_str.replace(ENCRYPT_SUFFIX, "")
+ try:
+ return self._fernet.decrypt(encrypted_str.encode()).decode()
+ except InvalidToken as err:
+ raise InvalidDataError("Password decryption failed") from err
async def _load(self) -> None:
"""Load data from persistent storage."""
LOGGER.debug("Loaded persistent settings from %s", filename)
LOGGER.debug("Started with empty storage: No persistent storage file found.")
- def save(self, immediate: bool = False) -> None:
- """Schedule save of data to disk."""
- if self._timer_handle is not None:
- self._timer_handle.cancel()
- self._timer_handle = None
-
- if immediate:
- self.mass.loop.create_task(self.async_save())
- else:
- # schedule the save for later
- self._timer_handle = self.mass.loop.call_later(
- DEFAULT_SAVE_DELAY, self.mass.create_task, self.async_save
- )
-
- async def async_save(self):
+ async def _async_save(self):
"""Save persistent data to disk."""
filename_backup = f"{self.filename}.backup"
# make backup before we write a new file
await _file.write(json_dumps(self._data, indent=True))
LOGGER.debug("Saved data to persistent storage")
- def encrypt_string(self, str_value: str) -> str:
- """Encrypt a (password)string with Fernet."""
- if str_value.startswith(ENCRYPT_SUFFIX):
- return str_value
- return ENCRYPT_SUFFIX + self._fernet.encrypt(str_value.encode()).decode()
+ async def _update_provider_config(
+ self, instance_id: str, values: dict[str, ConfigValueType]
+ ) -> ProviderConfig:
+ """Update ProviderConfig."""
+ config = await self.get_provider_config(instance_id)
+ changed_keys = config.update(values)
+ # validate the new config
+ config.validate()
+ available = prov.available if (prov := self.mass.get_provider(instance_id)) else False
+ if not changed_keys and (config.enabled == available):
+ # no changes
+ return config
+ # try to load the provider first to catch errors before we save it.
+ if config.enabled:
+ await self.mass.load_provider(config)
+ else:
+ await self.mass.unload_provider(config.instance_id)
+ # load succeeded, save new config
+ config.last_error = None
+ conf_key = f"{CONF_PROVIDERS}/{instance_id}"
+ self.set(conf_key, config.to_raw())
+ return config
- def decrypt_string(self, encrypted_str: str) -> str:
- """Decrypt a (password)string with Fernet."""
- if not encrypted_str:
- return encrypted_str
- if not encrypted_str.startswith(ENCRYPT_SUFFIX):
- return encrypted_str
- encrypted_str = encrypted_str.replace(ENCRYPT_SUFFIX, "")
- try:
- return self._fernet.decrypt(encrypted_str.encode()).decode()
- except InvalidToken as err:
- raise InvalidDataError("Password decryption failed") from err
+ async def _add_provider_config(
+ self,
+ provider_domain: str,
+ values: dict[str, ConfigValueType],
+ ) -> list[ConfigEntry] | ProviderConfig:
+ """
+ Add new Provider (instance).
+
+ params:
+ - provider_domain: domain of the provider for which to add an instance of.
+ - values: the raw values for config entries.
+
+ Returns: newly created ProviderConfig.
+ """
+ # lookup provider manifest and module
+ for prov in self.mass.get_available_providers():
+ if prov.domain == provider_domain:
+ manifest = prov
+ break
+ else:
+ raise KeyError(f"Unknown provider domain: {provider_domain}")
+ # create new provider config with given values
+ existing = {
+ x.instance_id for x in await self.get_provider_configs(provider_domain=provider_domain)
+ }
+ # determine instance id based on previous configs
+ if existing and not manifest.multi_instance:
+ raise ValueError(f"Provider {manifest.name} does not support multiple instances")
+ if len(existing) == 0:
+ instance_id = provider_domain
+ name = manifest.name
+ else:
+ instance_id = f"{provider_domain}{len(existing)+1}"
+ name = f"{manifest.name} {len(existing)+1}"
+ # all checks passed, create config object
+ config_entries = await self.get_provider_config_entries(
+ provider_domain=provider_domain, instance_id=instance_id, values=values
+ )
+ config: ProviderConfig = ProviderConfig.parse(
+ config_entries,
+ {
+ "type": manifest.type.value,
+ "domain": manifest.domain,
+ "instance_id": instance_id,
+ "name": name,
+ "values": values,
+ },
+ )
+ # validate the new config
+ config.validate()
+ # try to load the provider first to catch errors before we save it.
+ await self.mass.load_provider(config)
+ # the load was a success, store this config
+ conf_key = f"{CONF_PROVIDERS}/{config.instance_id}"
+ self.set(conf_key, config.to_raw())
+ return config
# execute the play_media command on the player(s)
player_prov = self.mass.players.get_player_provider(queue_id)
flow_mode = self.mass.config.get_player_config_value(queue.queue_id, CONF_FLOW_MODE)
- queue.flow_mode = flow_mode.value
+ queue.flow_mode = flow_mode
await player_prov.cmd_play_media(
queue_id,
queue_item=queue_item,
seek_position=seek_position,
fade_in=fade_in,
- flow_mode=flow_mode.value,
+ flow_mode=flow_mode,
)
# Interaction with player
try:
hide_group_childs = self.mass.config.get_player_config_value(
group_player.player_id, CONF_HIDE_GROUP_CHILDS
- ).value
+ )
except KeyError:
continue
if hide_group_childs == "always":
if output_format.is_pcm() or output_format == ContentType.WAV:
output_channels = self.mass.config.get_player_config_value(
player_id, CONF_OUTPUT_CHANNELS
- ).value
+ )
channels = 1 if output_channels != "stereo" else 2
output_format_str = (
f"x-wav;codec=pcm;rate={output_sample_rate};"
--- /dev/null
+"""Helper(s) to deal with authentication for (music) providers."""
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING
+
+from aiohttp.web import Request, Response
+
+from music_assistant.common.models.enums import EventType
+
+if TYPE_CHECKING:
+ from music_assistant.server import MusicAssistant
+
+
+class AuthenticationHelper:
+ """Context manager helper class for authentication with a forward and redirect URL."""
+
+ def __init__(self, mass: MusicAssistant, session_id: str):
+ """
+ Initialize the Authentication Helper.
+
+ Params:
+ - url: The URL the user needs to open for authentication.
+ - session_id: a unique id for this auth session.
+ """
+ self.mass = mass
+ self.session_id = session_id
+ self._callback_response: asyncio.Queue[dict[str, str]] = asyncio.Queue(1)
+
+ @property
+ def callback_url(self) -> str:
+ """Return the callback URL."""
+ return f"{self.mass.webserver.base_url}/callback/{self.session_id}"
+
+ async def __aenter__(self) -> AuthenticationHelper:
+ """Enter context manager."""
+ self.mass.webserver.register_route(
+ f"/callback/{self.session_id}", self._handle_callback, "GET"
+ )
+ return self
+
+ async def __aexit__(self, exc_type, exc_value, traceback) -> bool:
+ """Exit context manager."""
+ self.mass.webserver.unregister_route(f"/callback/{self.session_id}", "GET")
+
+ async def authenticate(self, auth_url: str, timeout: int = 60) -> dict[str, str]:
+ """Start the auth process and return any query params if received on the callback."""
+ # redirect the user in the frontend to the auth url
+ self.mass.signal_event(EventType.AUTH_SESSION, self.session_id, auth_url)
+ async with asyncio.timeout(timeout):
+ return await self._callback_response.get()
+
+ async def _handle_callback(self, request: Request) -> Response:
+ """Handle callback response."""
+ params = dict(request.query)
+ await self._callback_response.put(params)
+ return_html = """
+ <html>
+ <body onload="window.close();">
+ Authentication completed, you may now close this window.
+ </body>
+ </html>
+ """
+ return Response(body=return_html, headers={"content-type": "text/html"})
from typing import TYPE_CHECKING, Protocol
+from music_assistant.common.models.config_entries import ConfigValueType
+
from .metadata_provider import MetadataProvider
from .music_provider import MusicProvider
from .player_provider import PlayerProvider
@staticmethod
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
from abc import abstractmethod
from typing import TYPE_CHECKING
-from music_assistant.common.models.enums import ProviderFeature
from music_assistant.common.models.player import Player
from music_assistant.common.models.queue_item import QueueItem
def on_player_config_removed(self, player_id: str) -> None:
"""Call (by config manager) when the configuration of a player is removed."""
- async def create_player_config(self, config: PlayerConfig | None = None) -> PlayerConfig:
- """Handle CREATE_PLAYER flow for this player provider.
-
- Allows manually registering/creating a player,
- for example by manually entering an IP address etc.
-
- Called by the Config manager without a value to get the PlayerConfig to show in the UI.
- Called with PlayerConfig value with the submitted values.
- """
- # will only be called if the provider has the ADD_PLAYER feature set.
- if ProviderFeature.CREATE_PLAYER_CONFIG in self.supported_features:
- raise NotImplementedError
-
@abstractmethod
async def cmd_stop(self, player_id: str) -> None:
"""Send STOP command to given player.
import aiofiles
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType
from music_assistant.common.models.player import DeviceInfo, Player
from music_assistant.common.models.queue_item import QueueItem
PLAYER_CONFIG_ENTRIES = (
ConfigEntry(
- key="airplay_label",
- type=ConfigEntryType.LABEL,
+ key="airplay_header",
+ type=ConfigEntryType.DIVIDER,
label="Airplay specific settings",
description="Configure Airplay specific settings. "
"Note that changing any airplay specific setting, will reconnect all players.",
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
CONF_ENTRY_OUTPUT_CODEC,
ConfigEntry,
ConfigValueOption,
+ ConfigValueType,
)
from music_assistant.common.models.enums import (
ConfigEntryType,
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
) -> None:
"""Send PLAY MEDIA command to given player."""
castplayer = self.castplayers[player_id]
- output_codec = self.mass.config.get_player_config_value(player_id, CONF_OUTPUT_CODEC).value
+ output_codec = self.mass.config.get_player_config_value(player_id, CONF_OUTPUT_CODEC)
url = await self.mass.streams.resolve_stream_url(
queue_item=queue_item,
player_id=player_id,
event = asyncio.Event()
if use_alt_app := self.mass.config.get_player_config_value(
castplayer.player_id, CONF_ALT_APP
- ).value:
+ ):
app_id = pychromecast.config.APP_BUBBLEUPNP
else:
app_id = pychromecast.config.APP_MEDIA_RECEIVER
from async_upnp_client.search import async_search
from async_upnp_client.utils import CaseInsensitiveDict
-from music_assistant.common.models.config_entries import CONF_ENTRY_OUTPUT_CODEC, ConfigEntry
+from music_assistant.common.models.config_entries import (
+ CONF_ENTRY_OUTPUT_CODEC,
+ ConfigEntry,
+ ConfigValueType,
+)
from music_assistant.common.models.enums import ContentType, PlayerFeature, PlayerState, PlayerType
from music_assistant.common.models.errors import PlayerUnavailableError, QueueEmpty
from music_assistant.common.models.player import DeviceInfo, Player
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
# always clear queue (by sending stop) first
if dlna_player.device.can_stop:
await self.cmd_stop(player_id)
- output_codec = self.mass.config.get_player_config_value(player_id, CONF_OUTPUT_CODEC).value
+ output_codec = self.mass.config.get_player_config_value(player_id, CONF_OUTPUT_CODEC)
url = await self.mass.streams.resolve_stream_url(
queue_item=queue_item,
player_id=dlna_player.udn,
# send queue item to dlna queue
output_codec = self.mass.config.get_player_config_value(
dlna_player.player.player_id, CONF_OUTPUT_CODEC
- ).value
+ )
url = await self.mass.streams.resolve_stream_url(
queue_item=next_item,
player_id=dlna_player.udn,
import aiohttp.client_exceptions
from asyncio_throttle import Throttler
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ProviderFeature
from music_assistant.common.models.media_items import ImageType, MediaItemImage, MediaItemMetadata
from music_assistant.server.controllers.cache import use_cache
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
import aiofiles
from aiofiles.os import wrap
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType
from music_assistant.common.models.errors import SetupFailedError
from music_assistant.constants import CONF_PATH
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(key="path", type=ConfigEntryType.STRING, label="Path", default_value="/media"),
CONF_ENTRY_MISSING_ALBUM_ARTIST,
from typing import TYPE_CHECKING
from music_assistant.common.helpers.util import get_ip_from_host
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType
from music_assistant.common.models.errors import LoginFailed
from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(
key=CONF_HOST,
label="Mount options",
required=False,
advanced=True,
- default_value="file_mode=0775,dir_mode=0775,uid=0,gid=0",
+ default_value="noserverino,file_mode=0775,dir_mode=0775,uid=0,gid=0",
description="[optional] Any additional mount options you "
"want to pass to the mount command if needed for your particular setup.",
),
from music_assistant.common.helpers.json import json_dumps, json_loads
from music_assistant.common.helpers.util import select_free_port
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import PlayerState
from music_assistant.server.models.plugin import PluginProvider
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
from asyncio_throttle import Throttler
from music_assistant.common.helpers.util import create_sort_name
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ProviderFeature
from music_assistant.server.controllers.cache import use_cache
from music_assistant.server.helpers.compare import compare_strings
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
"""Plex musicprovider support for MusicAssistant."""
+from __future__ import annotations
+
import logging
from asyncio import TaskGroup
from collections.abc import AsyncGenerator, Callable, Coroutine
+import plexapi.exceptions
from aiohttp import ClientTimeout
from plexapi.audio import Album as PlexAlbum
from plexapi.audio import Artist as PlexArtist
from plexapi.media import AudioStream as PlexAudioStream
from plexapi.media import Media as PlexMedia
from plexapi.media import MediaPart as PlexMediaPart
-from plexapi.myplex import MyPlexAccount
+from plexapi.myplex import MyPlexAccount, MyPlexPinLogin
from plexapi.server import PlexServer
-from music_assistant.common.models.config_entries import ConfigEntry, ProviderConfig
+from music_assistant.common.models.config_entries import (
+ ConfigEntry,
+ ConfigValueOption,
+ ConfigValueType,
+ ProviderConfig,
+)
from music_assistant.common.models.enums import (
ConfigEntryType,
ContentType,
)
from music_assistant.common.models.provider import ProviderManifest
from music_assistant.server import MusicAssistant
+from music_assistant.server.helpers.auth import AuthenticationHelper
from music_assistant.server.helpers.tags import parse_tags
from music_assistant.server.models import ProviderInstanceType
from music_assistant.server.models.music_provider import MusicProvider
+from music_assistant.server.providers.plex.helpers import get_libraries
+CONF_ACTION_AUTH = "auth"
+CONF_ACTION_LIBRARY = "library"
CONF_AUTH_TOKEN = "token"
-CONF_SERVER_NAME = "server"
-CONF_LIBRARY_NAME = "library"
+CONF_LIBRARY_ID = "library_id"
async def setup(
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None, # noqa: ARG001
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # config flow auth action/step (authenticate button clicked)
+ if action == CONF_ACTION_AUTH:
+ async with AuthenticationHelper(mass, values["session_id"]) as auth_helper:
+ plex_auth = MyPlexPinLogin(headers={"X-Plex-Product": "Music Assistant"}, oauth=True)
+ auth_url = plex_auth.oauthUrl(auth_helper.callback_url)
+ await auth_helper.authenticate(auth_url)
+ if not plex_auth.checkLogin():
+ raise LoginFailed("Authentication to MyPlex failed")
+ # set the retrieved token on the values object to pass along
+ values[CONF_AUTH_TOKEN] = plex_auth.token
+
+ # 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.
+ conf_libraries = ConfigEntry(
+ key=CONF_LIBRARY_ID,
+ type=ConfigEntryType.STRING,
+ label="Library",
+ required=True,
+ description="The library to connect to (e.g. Music)",
+ depends_on=CONF_AUTH_TOKEN,
+ action=CONF_ACTION_LIBRARY,
+ action_label="Select Plex Music Library",
+ )
+ if action in (CONF_ACTION_LIBRARY, CONF_ACTION_AUTH):
+ token = mass.config.decrypt_string(values.get(CONF_AUTH_TOKEN))
+ if not (libraries := await get_libraries(mass, token)):
+ raise LoginFailed("Unable to retrieve Servers and/or Music Libraries")
+ conf_libraries.options = tuple(
+ # use the same value for both the value and the title
+ # until we find out what plex uses as stable identifiers
+ ConfigValueOption(
+ title=x,
+ value=x,
+ )
+ for x in libraries
+ )
+ # select first library as (default) value
+ conf_libraries.default_value = libraries[0]
+ conf_libraries.value = libraries[0]
+ # return the collected config entries
return (
- ConfigEntry(
- key=CONF_SERVER_NAME,
- type=ConfigEntryType.STRING,
- label="Server",
- required=True,
- description="The name of the server (as shown in the interface)",
- ),
- ConfigEntry(
- key=CONF_LIBRARY_NAME,
- type=ConfigEntryType.STRING,
- label="Library",
- required=True,
- description="The name of the library to connect to (e.g. Music)",
- ),
ConfigEntry(
key=CONF_AUTH_TOKEN,
type=ConfigEntryType.SECURE_STRING,
- label="Token",
- required=True,
- description="A token to connect to your plex.tv account.",
- help_link="https://support.plex.tv/articles/204059436-finding-an-authentication-token-x-plex-token/",
+ label="Authentication token for MyPlex.tv",
+ description="You need to link Music Assistant to your MyPlex account.",
+ action=CONF_ACTION_AUTH,
+ action_label="Authenticate on MyPlex.tv",
+ value=values.get(CONF_AUTH_TOKEN) if values else None,
),
+ conf_libraries,
)
"""Set up the music provider by connecting to the server."""
# silence urllib logger
logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
-
- def connect():
- plex_account = MyPlexAccount(token=self.config.get_value(CONF_AUTH_TOKEN))
- return plex_account.resource(self.config.get_value(CONF_SERVER_NAME)).connect(None, 10)
+ server_name, library_name = self.config.get_value(CONF_LIBRARY_ID).split(" / ", 1)
+
+ def connect() -> PlexServer:
+ try:
+ plex_account = MyPlexAccount(token=self.config.get_value(CONF_AUTH_TOKEN))
+ except plexapi.exceptions.BadRequest as err:
+ if "Invalid token" in str(err):
+ # token invalid, invaidate the config
+ self.mass.config.remove_provider_config_value(self.instance_id, CONF_AUTH_TOKEN)
+ raise LoginFailed("Authentication failed")
+ raise LoginFailed() from err
+ return plex_account.resource(server_name).connect(None, 10)
self._plex_server = await self._run_async(connect)
- self._plex_library = await self._run_async(
- self._plex_server.library.section, self.config.get_value(CONF_LIBRARY_NAME)
- )
+ self._plex_library = await self._run_async(self._plex_server.library.section, library_name)
@property
def supported_features(self) -> tuple[ProviderFeature, ...]:
--- /dev/null
+"""Several helpers/utils for the Plex Music Provider."""
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING
+
+import plexapi.exceptions
+from plexapi.library import LibrarySection as PlexLibrarySection
+from plexapi.library import MusicSection as PlexMusicSection
+from plexapi.myplex import MyPlexAccount
+from plexapi.server import PlexServer
+
+if TYPE_CHECKING:
+ from music_assistant.server import MusicAssistant
+
+
+async def get_libraries(mass: MusicAssistant, auth_token: str) -> set[str]:
+ """
+ Get all music libraries for all plex servers.
+
+ Returns a set of Library names in format 'servername / library name'
+ """
+ cache_key = "plex_libraries"
+
+ def _get_libraries():
+ # create a listing of available music libraries on all servers
+ all_libraries: list[str] = []
+ plex_account = MyPlexAccount(token=auth_token)
+ for server_resource in plex_account.resources():
+ try:
+ plex_server: PlexServer = server_resource.connect(None, 10)
+ except plexapi.exceptions.NotFound:
+ continue
+ for media_section in plex_server.library.sections():
+ media_section: PlexLibrarySection # noqa: PLW2901
+ if media_section.type != PlexMusicSection.TYPE:
+ continue
+ # TODO: figure out what plex uses as stable id and use that instead of names
+ all_libraries.append(f"{server_resource.name} / {media_section.title}")
+ return all_libraries
+
+ if cache := await mass.cache.get(cache_key, checksum=auth_token):
+ return cache
+
+ result = await asyncio.to_thread(_get_libraries)
+ # use short expiration for in-memory cache
+ await mass.cache.set(cache_key, result, checksum=auth_token, expiration=3600)
+ return result
from asyncio_throttle import Throttler
from music_assistant.common.helpers.util import parse_title_and_version, try_parse_int
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType, ProviderFeature
from music_assistant.common.models.errors import LoginFailed, MediaNotFoundError
from music_assistant.common.models.media_items import (
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(
key=CONF_USERNAME, type=ConfigEntryType.STRING, label="Username", required=True
from aioslimproto.const import EventType as SlimEventType
from aioslimproto.discovery import start_discovery
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import (
ConfigEntryType,
ContentType,
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
from soco.events_base import SubscriptionBase
from soco.groups import ZoneGroup
-from music_assistant.common.models.config_entries import CONF_ENTRY_OUTPUT_CODEC, ConfigEntry
+from music_assistant.common.models.config_entries import (
+ CONF_ENTRY_OUTPUT_CODEC,
+ ConfigEntry,
+ ConfigValueType,
+)
from music_assistant.common.models.enums import (
ContentType,
MediaType,
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
await asyncio.to_thread(sonos_player.soco.stop)
await asyncio.to_thread(sonos_player.soco.clear_queue)
- output_codec = self.mass.config.get_player_config_value(player_id, CONF_OUTPUT_CODEC).value
+ output_codec = self.mass.config.get_player_config_value(player_id, CONF_OUTPUT_CODEC)
radio_mode = (
flow_mode or not queue_item.duration or queue_item.media_type == MediaType.RADIO
)
# send queue item to sonos queue
output_codec = self.mass.config.get_player_config_value(
sonos_player.player_id, CONF_OUTPUT_CODEC
- ).value
+ )
is_radio = next_item.media_type != MediaType.TRACK
url = await self.mass.streams.resolve_stream_url(
queue_item=next_item,
from typing import TYPE_CHECKING
from music_assistant.common.helpers.util import parse_title_and_version
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType, ProviderFeature
from music_assistant.common.models.errors import InvalidDataError, LoginFailed
from music_assistant.common.models.media_items import (
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(
key=CONF_CLIENT_ID, type=ConfigEntryType.SECURE_STRING, label="Client ID", required=True
from asyncio_throttle import Throttler
from music_assistant.common.helpers.util import parse_title_and_version
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType, ProviderFeature
from music_assistant.common.models.errors import LoginFailed, MediaNotFoundError
from music_assistant.common.models.media_items import (
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(
key=CONF_USERNAME, type=ConfigEntryType.STRING, label="Username", required=True
try:
retries += 1
if not tokeninfo:
- tokeninfo = await asyncio.wait_for(self._get_token(), 5)
+ async with asyncio.timeout(5):
+ tokeninfo = await self._get_token()
if tokeninfo and not userinfo:
- userinfo = await asyncio.wait_for(self._get_data("me", tokeninfo=tokeninfo), 5)
+ async with asyncio.timeout(5):
+ userinfo = await self._get_data("me", tokeninfo=tokeninfo)
if tokeninfo and userinfo:
# we have all info we need!
break
import aiohttp.client_exceptions
from asyncio_throttle import Throttler
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ProviderFeature
from music_assistant.common.models.media_items import (
Album,
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
from asyncio_throttle import Throttler
from music_assistant.common.helpers.util import create_sort_name
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType, ProviderFeature
from music_assistant.common.models.errors import LoginFailed, MediaNotFoundError
from music_assistant.common.models.media_items import (
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(
key=CONF_USERNAME, type=ConfigEntryType.STRING, label="Username", required=True
from collections.abc import AsyncGenerator
from typing import TYPE_CHECKING
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ContentType, ImageType, MediaType
from music_assistant.common.models.media_items import (
Artist,
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
ServerInfoMessage,
SuccessResultMessage,
)
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.errors import InvalidCommand
from music_assistant.common.models.event import MassEvent
from music_assistant.constants import ROOT_LOGGER_NAME, __version__
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return tuple() # we do not have any config entries (yet)
from music_assistant.common.helpers.uri import create_uri
from music_assistant.common.helpers.util import create_sort_name
-from music_assistant.common.models.config_entries import ConfigEntry
+from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
from music_assistant.common.models.enums import ConfigEntryType, ProviderFeature
from music_assistant.common.models.errors import (
InvalidDataError,
async def get_config_entries(
- mass: MusicAssistant, manifest: ProviderManifest # noqa: ARG001
+ mass: MusicAssistant,
+ instance_id: str | None = None,
+ action: str | None = None,
+ values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
- """Return Config entries to setup this provider."""
+ """
+ Return Config entries to setup this provider.
+
+ instance_id: id of an existing provider instance (None if new instance setup).
+ action: [optional] action key called from config entries UI.
+ values: the (intermediate) raw values for config entries sent with the action.
+ """
+ # ruff: noqa: ARG001
return (
ConfigEntry(
key=CONF_USERNAME, type=ConfigEntryType.STRING, label="Username", required=True
task.add_done_callback(task_done_callback)
return task
+ def get_task(self, task_id: str) -> asyncio.Task | asyncio.Future:
+ """Get existing scheduled task."""
+ if existing := self._tracked_tasks.get(task_id):
+ # prevent duplicate tasks if task_id is given and already present
+ return existing
+ raise KeyError("Task does not exist")
+
def register_api_command(
self,
command: str,
# try to load the module
prov_mod = await get_provider_module(domain)
try:
- provider = await asyncio.wait_for(prov_mod.setup(self, prov_manifest, conf), 30)
+ async with asyncio.timeout(30):
+ provider = await prov_mod.setup(self, prov_manifest, conf)
except TimeoutError as err:
raise SetupFailedError(f"Provider {domain} did not load within 30 seconds") from err
# if we reach this point, the provider loaded successfully
"python-slugify==8.0.1",
"mashumaro==3.5.0",
"memory-tempfile==2.2.3",
- "music-assistant-frontend==20230404.0",
+ "music-assistant-frontend==20230414.0",
"pillow==9.5.0",
"unidecode==1.3.6",
"xmltodict==0.13.0",
git+https://github.com/pytube/pytube.git@refs/pull/1501/head
mashumaro==3.5.0
memory-tempfile==2.2.3
-music-assistant-frontend==20230404.0
+music-assistant-frontend==20230414.0
orjson==3.8.9
pillow==9.5.0
plexapi==4.13.4
projects / music-assistant-server.git / commitdiff