255 lines
9.0 KiB
Python
255 lines
9.0 KiB
Python
"""Config flow for My Music Library."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
import voluptuous as vol
|
|
from music_assistant_client import MusicAssistantClient
|
|
from music_assistant_client.exceptions import CannotConnect, InvalidServerVersion
|
|
from music_assistant_models.errors import AuthenticationFailed, AuthenticationRequired, MusicAssistantError
|
|
|
|
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import entity_registry as er
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.helpers.selector import (
|
|
SelectSelector,
|
|
SelectSelectorConfig,
|
|
SelectSelectorMode,
|
|
TextSelector,
|
|
TextSelectorConfig,
|
|
TextSelectorType,
|
|
)
|
|
|
|
from .const import (
|
|
CONF_DEBUG_MODE,
|
|
CONF_DEFAULT_PLAYER,
|
|
CONF_DEFAULT_TAB,
|
|
CONF_EXCLUDED_PLAYERS,
|
|
CONF_MA_TOKEN,
|
|
CONF_MA_URL,
|
|
CONFIG_ENTRY_VERSION,
|
|
DEFAULT_TAB,
|
|
DOMAIN,
|
|
NAME,
|
|
)
|
|
|
|
|
|
def _get_mml_players(hass: HomeAssistant) -> dict[str, str]:
|
|
"""Return {entity_id: friendly_name} for our own media_player entities."""
|
|
ent_reg = er.async_get(hass)
|
|
players: dict[str, str] = {}
|
|
for entity in ent_reg.entities.values():
|
|
if entity.platform == DOMAIN and entity.domain == "media_player" and not entity.disabled:
|
|
state = hass.states.get(entity.entity_id)
|
|
name = state.attributes.get("friendly_name", entity.entity_id) if state else entity.entity_id
|
|
players[entity.entity_id] = name
|
|
return players
|
|
|
|
|
|
def _get_all_players(hass: HomeAssistant) -> dict[str, str]:
|
|
"""Return {entity_id: friendly_name} for all non-unavailable media_player entities."""
|
|
players: dict[str, str] = {}
|
|
for state in hass.states.async_all("media_player"):
|
|
if state.state != "unavailable":
|
|
players[state.entity_id] = state.attributes.get("friendly_name", state.entity_id)
|
|
return dict(sorted(players.items(), key=lambda x: x[1].lower()))
|
|
|
|
|
|
def _validate_url(url: str) -> str | None:
|
|
"""Return None if valid, or an error key if invalid."""
|
|
try:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https") or not parsed.netloc:
|
|
return "invalid_url"
|
|
except Exception: # noqa: BLE001
|
|
return "invalid_url"
|
|
return None
|
|
|
|
|
|
async def _test_connection(hass: HomeAssistant, url: str, token: str) -> str | None:
|
|
"""Try connecting to the MA server. Return None on success, else an error key."""
|
|
session = async_get_clientsession(hass)
|
|
mass = MusicAssistantClient(url, session, token=token)
|
|
try:
|
|
await mass.connect()
|
|
except (AuthenticationRequired, AuthenticationFailed):
|
|
return "invalid_auth"
|
|
except InvalidServerVersion:
|
|
return "invalid_server_version"
|
|
except (CannotConnect, TimeoutError):
|
|
return "cannot_connect"
|
|
except MusicAssistantError:
|
|
return "cannot_connect"
|
|
else:
|
|
await mass.disconnect()
|
|
return None
|
|
|
|
|
|
class MyMusicLibraryConfigFlow(ConfigFlow, domain=DOMAIN):
|
|
"""Handle a config flow for My Music Library."""
|
|
|
|
VERSION = CONFIG_ENTRY_VERSION
|
|
|
|
@staticmethod
|
|
@callback
|
|
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
|
|
"""Return the options flow handler."""
|
|
return MyMusicLibraryOptionsFlow()
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Handle the initial step: Music Assistant server URL + long-lived token."""
|
|
await self.async_set_unique_id(DOMAIN)
|
|
self._abort_if_unique_id_configured()
|
|
|
|
errors: dict[str, str] = {}
|
|
|
|
if user_input is not None:
|
|
url = user_input[CONF_MA_URL].rstrip("/")
|
|
token = user_input[CONF_MA_TOKEN]
|
|
if url_error := _validate_url(url):
|
|
errors[CONF_MA_URL] = url_error
|
|
else:
|
|
error = await _test_connection(self.hass, url, token)
|
|
if error:
|
|
errors["base"] = error
|
|
else:
|
|
return self.async_create_entry(
|
|
title=NAME,
|
|
data={
|
|
CONF_MA_URL: url,
|
|
CONF_MA_TOKEN: token,
|
|
CONF_DEFAULT_TAB: user_input[CONF_DEFAULT_TAB],
|
|
},
|
|
)
|
|
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required(CONF_MA_URL): TextSelector(TextSelectorConfig(type=TextSelectorType.URL)),
|
|
vol.Required(CONF_MA_TOKEN): TextSelector(
|
|
TextSelectorConfig(type=TextSelectorType.PASSWORD)
|
|
),
|
|
vol.Optional(CONF_DEFAULT_TAB, default=DEFAULT_TAB): vol.In(
|
|
{
|
|
"player": "Player",
|
|
"search": "Search",
|
|
"library": "Library",
|
|
}
|
|
),
|
|
}
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=schema,
|
|
errors=errors,
|
|
description_placeholders={
|
|
"ma_url_hint": "e.g. http://homeassistant.local:8095",
|
|
"token_hint": (
|
|
"Generate a long-lived token in Music Assistant "
|
|
"(Settings → your user → API tokens) and paste it here."
|
|
),
|
|
},
|
|
)
|
|
|
|
async def async_step_reauth(
|
|
self, entry_data: dict[str, Any]
|
|
) -> ConfigFlowResult:
|
|
"""Handle reauth when the stored token is missing, revoked, or the URL changed."""
|
|
return await self.async_step_reauth_confirm()
|
|
|
|
async def async_step_reauth_confirm(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Ask for a fresh URL + token and update the existing entry."""
|
|
errors: dict[str, str] = {}
|
|
reauth_entry = self._get_reauth_entry()
|
|
|
|
if user_input is not None:
|
|
url = user_input[CONF_MA_URL].rstrip("/")
|
|
token = user_input[CONF_MA_TOKEN]
|
|
if url_error := _validate_url(url):
|
|
errors[CONF_MA_URL] = url_error
|
|
else:
|
|
error = await _test_connection(self.hass, url, token)
|
|
if error:
|
|
errors["base"] = error
|
|
else:
|
|
return self.async_update_reload_and_abort(
|
|
reauth_entry,
|
|
data={**reauth_entry.data, CONF_MA_URL: url, CONF_MA_TOKEN: token},
|
|
)
|
|
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Required(CONF_MA_URL, default=reauth_entry.data.get(CONF_MA_URL, "")): TextSelector(
|
|
TextSelectorConfig(type=TextSelectorType.URL)
|
|
),
|
|
vol.Required(CONF_MA_TOKEN): TextSelector(
|
|
TextSelectorConfig(type=TextSelectorType.PASSWORD)
|
|
),
|
|
}
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="reauth_confirm",
|
|
data_schema=schema,
|
|
errors=errors,
|
|
)
|
|
|
|
|
|
class MyMusicLibraryOptionsFlow(OptionsFlow):
|
|
"""Handle options for My Music Library (default player, excluded players, debug)."""
|
|
|
|
async def async_step_init(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Manage the options."""
|
|
if user_input is not None:
|
|
return self.async_create_entry(data=user_input)
|
|
|
|
all_players = _get_all_players(self.hass)
|
|
mml_players = _get_mml_players(self.hass)
|
|
|
|
current_excluded: list[str] = list(
|
|
self.config_entry.options.get(CONF_EXCLUDED_PLAYERS, [])
|
|
)
|
|
# Keep wildcard patterns as-is; drop stale exact entity IDs only.
|
|
current_excluded = [
|
|
p for p in current_excluded
|
|
if "*" in p or p in all_players
|
|
]
|
|
|
|
current_default_player = self.config_entry.options.get(CONF_DEFAULT_PLAYER, "")
|
|
if current_default_player not in mml_players:
|
|
current_default_player = ""
|
|
|
|
current_debug: bool = bool(
|
|
self.config_entry.options.get(CONF_DEBUG_MODE, False)
|
|
)
|
|
|
|
player_options = {
|
|
"": "Auto-detect (first available player)",
|
|
**mml_players,
|
|
}
|
|
|
|
schema = vol.Schema(
|
|
{
|
|
vol.Optional(CONF_DEFAULT_PLAYER, default=current_default_player): vol.In(player_options),
|
|
vol.Optional(CONF_EXCLUDED_PLAYERS, default=current_excluded): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=[{"value": k, "label": v} for k, v in all_players.items()],
|
|
multiple=True,
|
|
mode=SelectSelectorMode.LIST,
|
|
custom_value=True,
|
|
)
|
|
),
|
|
vol.Optional(CONF_DEBUG_MODE, default=current_debug): bool,
|
|
}
|
|
)
|
|
|
|
return self.async_show_form(step_id="init", data_schema=schema)
|