ha-solaredge-optimizers/custom_components/solaredge_optimizers/config_flow.py

"""Config flow for SolarEdge Optimizers."""

from __future__ import annotations

from typing import Any, override

import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.hassio import HassioServiceInfo

from .api import (
    OptimizerSnapshot,
    SolarEdgeOptimizerApiClient,
    SolarEdgeOptimizerConnectionError,
    SolarEdgeOptimizerInvalidResponseError,
    normalize_base_url,
)
from .const import (
    CONF_SCAN_INTERVAL,
    CONFIG_ENTRY_VERSION,
    DEFAULT_PORT,
    DEFAULT_SCAN_INTERVAL,
    DOMAIN,
    LOGGER,
    MIN_SCAN_INTERVAL_SECONDS,
)


async def validate_input(
    hass: HomeAssistant, data: dict[str, Any]
) -> tuple[str, OptimizerSnapshot]:
    """Validate user input and return the normalized URL and snapshot."""
    url = normalize_base_url(data[CONF_URL])
    client = SolarEdgeOptimizerApiClient(url, async_get_clientsession(hass))
    return url, await client.async_get_optimizers()


def _schema(defaults: dict[str, Any]) -> vol.Schema:
    """Return the manual and reconfigure schema."""
    url_marker: vol.Marker
    if CONF_URL in defaults:
        url_marker = vol.Required(CONF_URL, default=defaults[CONF_URL])
    else:
        url_marker = vol.Required(CONF_URL)
    return vol.Schema(
        {
            url_marker: str,
            vol.Required(
                CONF_SCAN_INTERVAL,
                default=defaults.get(
                    CONF_SCAN_INTERVAL,
                    int(DEFAULT_SCAN_INTERVAL.total_seconds()),
                ),
            ): vol.All(vol.Coerce(int), vol.Range(min=MIN_SCAN_INTERVAL_SECONDS)),
        }
    )


class SolarEdgeOptimizersConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for SolarEdge Optimizers."""

    VERSION = CONFIG_ENTRY_VERSION

    def __init__(self) -> None:
        """Initialize the flow."""
        self._discovered_url: str | None = None
        self._discovered_name: str | None = None

    @override
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle manual setup."""
        return await self._async_handle_form("user", user_input)

    @override
    async def async_step_hassio(
        self, discovery_info: HassioServiceInfo
    ) -> ConfigFlowResult:
        """Handle discovery from the Home Assistant App."""
        host = discovery_info.config.get("host")
        port = discovery_info.config.get("port", DEFAULT_PORT)
        if not isinstance(host, str) or not host:
            return self.async_abort(reason="invalid_discovery")
        try:
            port = int(port)
            self._discovered_url = normalize_base_url(f"http://{host}:{port}")
        except TypeError, ValueError:
            return self.async_abort(reason="invalid_discovery")

        self._discovered_name = discovery_info.name
        self.context["title_placeholders"] = {"name": discovery_info.name}
        return await self.async_step_hassio_confirm()

    async def async_step_hassio_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Confirm App discovery."""
        if user_input is None:
            self._abort_if_unique_id_configured()
            return self.async_show_form(
                step_id="hassio_confirm",
                description_placeholders={"name": self._discovered_name or "App"},
            )

        assert self._discovered_url is not None
        data = {
            CONF_URL: self._discovered_url,
            CONF_SCAN_INTERVAL: int(DEFAULT_SCAN_INTERVAL.total_seconds()),
        }
        try:
            url, snapshot = await validate_input(self.hass, data)
        except SolarEdgeOptimizerConnectionError:
            return self.async_abort(reason="cannot_connect")
        except SolarEdgeOptimizerInvalidResponseError:
            return self.async_abort(reason="invalid_response")
        except Exception:  # noqa: BLE001
            LOGGER.exception("Unexpected exception during App discovery")
            return self.async_abort(reason="unknown")

        await self.async_set_unique_id(snapshot.site_id)
        self._abort_if_unique_id_configured(updates={CONF_URL: url})
        return self.async_create_entry(
            title=f"SolarEdge Site {snapshot.site_id}", data={**data, CONF_URL: url}
        )

    @override
    async def async_step_reconfigure(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Reconfigure the App connection."""
        entry = self._get_reconfigure_entry()
        if user_input is None:
            return self.async_show_form(
                step_id="reconfigure", data_schema=_schema(dict(entry.data))
            )

        result = await self._async_validate_form(user_input)
        if isinstance(result, dict):
            return self.async_show_form(
                step_id="reconfigure",
                data_schema=_schema(user_input),
                errors=result,
            )
        url, snapshot = result
        if entry.unique_id != snapshot.site_id:
            return self.async_show_form(
                step_id="reconfigure",
                data_schema=_schema(user_input),
                errors={"base": "site_id_mismatch"},
            )
        return self.async_update_reload_and_abort(
            entry,
            data_updates={**user_input, CONF_URL: url},
        )

    async def _async_handle_form(
        self, step_id: str, user_input: dict[str, Any] | None
    ) -> ConfigFlowResult:
        if user_input is None:
            return self.async_show_form(step_id=step_id, data_schema=_schema({}))

        result = await self._async_validate_form(user_input)
        if isinstance(result, dict):
            return self.async_show_form(
                step_id=step_id,
                data_schema=_schema(user_input),
                errors=result,
            )

        url, snapshot = result
        await self.async_set_unique_id(snapshot.site_id)
        self._abort_if_unique_id_configured(updates={CONF_URL: url})
        return self.async_create_entry(
            title=f"SolarEdge Site {snapshot.site_id}",
            data={**user_input, CONF_URL: url},
        )

    async def _async_validate_form(
        self, user_input: dict[str, Any]
    ) -> tuple[str, OptimizerSnapshot] | dict[str, str]:
        try:
            return await validate_input(self.hass, user_input)
        except ValueError:
            return {"base": "invalid_url"}
        except SolarEdgeOptimizerConnectionError:
            return {"base": "cannot_connect"}
        except SolarEdgeOptimizerInvalidResponseError:
            return {"base": "invalid_response"}
        except Exception:  # noqa: BLE001
            LOGGER.exception("Unexpected exception while validating configuration")
            return {"base": "unknown"}