Initial commit for Emby Media Player HAOS HACS integration

custom_components/emby_player/config_flow.py

@@ -0,0 +1,230 @@
+"""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",
+                        )
+                    ),
+                }
+            ),
+        )