haos-hacs-emby-media-player/custom_components/emby_player/config_flow.py

"""Config flow for Emby Media Player integration."""

from __future__ import annotations

import logging
from typing import Any

import voluptuous as vol

from homeassistant.config_entries import (
    ConfigEntry,
    ConfigFlow,
    ConfigFlowResult,
    OptionsFlow,
)
from homeassistant.core import callback
from homeassistant.helpers.selector import (
    NumberSelector,
    NumberSelectorConfig,
    NumberSelectorMode,
    SelectSelector,
    SelectSelectorConfig,
    SelectSelectorMode,
    TextSelector,
    TextSelectorConfig,
    TextSelectorType,
)

from .api import EmbyApiClient, EmbyAuthenticationError, EmbyConnectionError
from .const import (
    CONF_API_KEY,
    CONF_HOST,
    CONF_PORT,
    CONF_SCAN_INTERVAL,
    CONF_SSL,
    CONF_USER_ID,
    DEFAULT_PORT,
    DEFAULT_SCAN_INTERVAL,
    DEFAULT_SSL,
    DOMAIN,
)

_LOGGER = logging.getLogger(__name__)


class EmbyConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Emby Media Player."""

    VERSION = 1

    def __init__(self) -> None:
        """Initialize the config flow."""
        self._host: str | None = None
        self._port: int = DEFAULT_PORT
        self._api_key: str | None = None
        self._ssl: bool = DEFAULT_SSL
        self._users: list[dict[str, Any]] = []
        self._server_info: dict[str, Any] = {}

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step - server connection."""
        errors: dict[str, str] = {}

        if user_input is not None:
            self._host = user_input[CONF_HOST].strip()
            self._port = int(user_input.get(CONF_PORT, DEFAULT_PORT))
            self._api_key = user_input[CONF_API_KEY].strip()
            self._ssl = user_input.get(CONF_SSL, DEFAULT_SSL)

            _LOGGER.debug(
                "Testing connection to %s:%s (SSL: %s)",
                self._host,
                self._port,
                self._ssl,
            )

            # Test connection
            api = EmbyApiClient(
                host=self._host,
                port=self._port,
                api_key=self._api_key,
                ssl=self._ssl,
            )

            try:
                self._server_info = await api.test_connection()
                self._users = await api.get_users()
                await api.close()

                if not self._users:
                    errors["base"] = "no_users"
                else:
                    return await self.async_step_user_select()

            except EmbyAuthenticationError:
                errors["base"] = "invalid_auth"
            except EmbyConnectionError:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            finally:
                await api.close()

        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_HOST): TextSelector(
                        TextSelectorConfig(type=TextSelectorType.TEXT)
                    ),
                    vol.Optional(CONF_PORT, default=DEFAULT_PORT): NumberSelector(
                        NumberSelectorConfig(
                            min=1,
                            max=65535,
                            mode=NumberSelectorMode.BOX,
                        )
                    ),
                    vol.Required(CONF_API_KEY): TextSelector(
                        TextSelectorConfig(type=TextSelectorType.PASSWORD)
                    ),
                    vol.Optional(CONF_SSL, default=DEFAULT_SSL): bool,
                }
            ),
            errors=errors,
        )

    async def async_step_user_select(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle user selection step."""
        errors: dict[str, str] = {}

        if user_input is not None:
            user_id = user_input[CONF_USER_ID]

            # Find user name
            user_name = next(
                (u["Name"] for u in self._users if u["Id"] == user_id),
                "Unknown",
            )

            # Create unique ID based on server ID and user
            server_id = self._server_info.get("Id", self._host)
            await self.async_set_unique_id(f"{server_id}_{user_id}")
            self._abort_if_unique_id_configured()

            server_name = self._server_info.get("ServerName", self._host)

            return self.async_create_entry(
                title=f"{server_name} ({user_name})",
                data={
                    CONF_HOST: self._host,
                    CONF_PORT: self._port,
                    CONF_API_KEY: self._api_key,
                    CONF_SSL: self._ssl,
                    CONF_USER_ID: user_id,
                },
                options={
                    CONF_SCAN_INTERVAL: DEFAULT_SCAN_INTERVAL,
                },
            )

        # Build user selection options
        user_options = [
            {"value": user["Id"], "label": user["Name"]} for user in self._users
        ]

        return self.async_show_form(
            step_id="user_select",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_USER_ID): SelectSelector(
                        SelectSelectorConfig(
                            options=user_options,
                            mode=SelectSelectorMode.DROPDOWN,
                        )
                    ),
                }
            ),
            errors=errors,
        )

    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: ConfigEntry,
    ) -> EmbyOptionsFlow:
        """Get the options flow for this handler."""
        return EmbyOptionsFlow(config_entry)


class EmbyOptionsFlow(OptionsFlow):
    """Handle options flow for Emby Media Player."""

    def __init__(self, config_entry: ConfigEntry) -> None:
        """Initialize options flow."""
        self._config_entry = config_entry

    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(title="", data=user_input)

        current_interval = self._config_entry.options.get(
            CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
        )

        return self.async_show_form(
            step_id="init",
            data_schema=vol.Schema(
                {
                    vol.Optional(
                        CONF_SCAN_INTERVAL, default=current_interval
                    ): NumberSelector(
                        NumberSelectorConfig(
                            min=5,
                            max=60,
                            step=1,
                            mode=NumberSelectorMode.SLIDER,
                            unit_of_measurement="seconds",
                        )
                    ),
                }
            ),
        )