core/homeassistant/components/saj/config_flow.py

"""Config flow for SAJ."""

import logging
from typing import TYPE_CHECKING, Any, override

import pysaj
import voluptuous as vol

from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
    CONF_HOST,
    CONF_NAME,
    CONF_PASSWORD,
    CONF_TYPE,
    CONF_USERNAME,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import ConfigType

from .const import CONNECTION_TYPES, DOMAIN, INTEGRATION_TITLE

_LOGGER = logging.getLogger(__name__)


class CannotConnect(HomeAssistantError):
    """Error to indicate we cannot connect."""


class InvalidAuth(HomeAssistantError):
    """Error to indicate invalid credentials (config flow)."""


class SAJConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for the SAJ Solar Inverter."""

    VERSION = 1

    def __init__(self) -> None:
        """Initialize flow."""
        self._host: str | None = None
        self._connection_type: str | None = None
        self._pending_username: str = ""
        self._pending_password: str = ""

    async def async_step_import(self, import_config: ConfigType) -> ConfigFlowResult:
        """Import a config entry from configuration.yaml (sensor platform)."""
        _LOGGER.warning("Importing SAJ from YAML is deprecated and will be removed")
        entry_input: dict[str, Any] = {
            CONF_HOST: import_config[CONF_HOST],
            CONF_TYPE: import_config.get(CONF_TYPE, CONNECTION_TYPES[0]),
            CONF_USERNAME: import_config.get(CONF_USERNAME),
            CONF_PASSWORD: import_config.get(CONF_PASSWORD),
        }
        try:
            serial_number = await self._async_validate_input(entry_input)
        except InvalidAuth:
            return self.async_abort(reason="invalid_auth")
        except CannotConnect:
            return self.async_abort(reason="cannot_connect")
        except Exception:
            _LOGGER.exception("Unexpected error importing SAJ from YAML")
            return self.async_abort(reason="unknown")

        data = {
            CONF_HOST: entry_input[CONF_HOST],
            CONF_TYPE: entry_input[CONF_TYPE],
            CONF_USERNAME: entry_input.get(CONF_USERNAME),
            CONF_PASSWORD: entry_input.get(CONF_PASSWORD),
        }
        title = (import_config.get(CONF_NAME) or "").strip() or INTEGRATION_TITLE

        await self.async_set_unique_id(serial_number)
        self._abort_if_unique_id_configured(updates=data)

        return self.async_create_entry(title=title, data=data)

    async def _async_validate_input(self, user_input: dict[str, Any]) -> str:
        """Validate the user input allows us to connect.

        Returns the device serial number (required after a successful read).
        """
        host = user_input[CONF_HOST]
        connection_type = user_input[CONF_TYPE]
        username = user_input.get(CONF_USERNAME)
        password = user_input.get(CONF_PASSWORD)

        wifi = connection_type == CONNECTION_TYPES[1]
        kwargs: dict[str, Any] = {}
        if wifi:
            kwargs["wifi"] = True
            if username:
                kwargs["username"] = username
            if password:
                kwargs["password"] = password

        async def _async_validate_connection() -> str:
            """Validate connection and get serial number."""
            saj = pysaj.SAJ(host, **kwargs)
            sensor_def = pysaj.Sensors(wifi)
            done = await saj.read(sensor_def)
            if not done:
                raise CannotConnect("Failed to read sensor data")

            serial_number = saj.serialnumber
            if not serial_number:
                raise CannotConnect("Device did not return a serial number")
            return serial_number

        try:
            serial_number = await _async_validate_connection()

        except pysaj.UnauthorizedException as err:
            # Only raise auth error for WiFi connections with wrong credentials
            if wifi:
                raise InvalidAuth("Invalid authentication") from err
            # For ethernet, this likely means wrong connection type - treat as connection error
            raise CannotConnect("Wrong connection type or cannot connect") from err
        except pysaj.UnexpectedResponseException as err:
            raise CannotConnect(f"Connection error: {err}") from err
        except CannotConnect, InvalidAuth:
            raise
        except Exception as err:
            raise CannotConnect(f"Connection failed: {err}") from err

        return serial_number

    @override
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow started by the user."""
        errors = {}

        if user_input is not None:
            host = user_input[CONF_HOST]
            connection_type = user_input.get(CONF_TYPE, CONNECTION_TYPES[0])

            # Store host/type for a possible WiFi credentials step
            self._host = host
            self._connection_type = connection_type
            self._pending_username = user_input.get(CONF_USERNAME) or ""
            self._pending_password = user_input.get(CONF_PASSWORD) or ""

            entry_input = {
                CONF_HOST: host,
                CONF_TYPE: connection_type,
                CONF_USERNAME: user_input.get(CONF_USERNAME),
                CONF_PASSWORD: user_input.get(CONF_PASSWORD),
            }

            # Ethernet ignores username/password; WiFi tries open access or creds from this step
            try:
                serial_number = await self._async_validate_input(entry_input)
            except InvalidAuth:
                # WiFi-only: device is SAJ but requires credentials
                return await self.async_step_device_credentials()
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected error during user flow")
                errors["base"] = "unknown"
            else:
                data = {
                    CONF_HOST: entry_input[CONF_HOST],
                    CONF_TYPE: entry_input[CONF_TYPE],
                    CONF_USERNAME: entry_input.get(CONF_USERNAME),
                    CONF_PASSWORD: entry_input.get(CONF_PASSWORD),
                }

                await self.async_set_unique_id(serial_number)
                self._abort_if_unique_id_configured(updates=data)

                return self.async_create_entry(title=INTEGRATION_TITLE, data=data)

        return self.async_show_form(
            step_id="user",
            data_schema=self._schema_user(),
            errors=errors or None,
        )

    def _schema_user(self) -> vol.Schema:
        """Define the schema for the user step."""
        return vol.Schema(
            {
                vol.Required(CONF_HOST): str,
                vol.Optional(CONF_TYPE, default=CONNECTION_TYPES[0]): vol.In(
                    CONNECTION_TYPES
                ),
                vol.Optional(CONF_USERNAME, default=""): str,
                vol.Optional(CONF_PASSWORD, default=""): str,
            }
        )

    async def async_step_device_credentials(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle device credentials step (only shown for WiFi connections)."""
        errors = {}

        if user_input is not None:
            if TYPE_CHECKING:
                assert self._host is not None
                assert self._connection_type is not None

            combined_input = {
                CONF_HOST: self._host,
                CONF_TYPE: self._connection_type,
                CONF_USERNAME: user_input.get(CONF_USERNAME),
                CONF_PASSWORD: user_input.get(CONF_PASSWORD),
            }

            try:
                serial_number = await self._async_validate_input(combined_input)
                data = {
                    CONF_HOST: combined_input[CONF_HOST],
                    CONF_TYPE: combined_input[CONF_TYPE],
                    CONF_USERNAME: combined_input.get(CONF_USERNAME),
                    CONF_PASSWORD: combined_input.get(CONF_PASSWORD),
                }
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except Exception:
                _LOGGER.exception("Unexpected error during device credentials flow")
                errors["base"] = "unknown"
            else:
                await self.async_set_unique_id(serial_number)
                self._abort_if_unique_id_configured(updates=data)

                return self.async_create_entry(title=INTEGRATION_TITLE, data=data)
        return self.async_show_form(
            step_id="device_credentials",
            data_schema=self._schema_device_credentials(),
            errors=errors or None,
        )

    def _schema_device_credentials(self) -> vol.Schema:
        """Define the schema for the device credentials step."""
        return vol.Schema(
            {
                vol.Optional(CONF_USERNAME, default=self._pending_username): str,
                vol.Optional(CONF_PASSWORD, default=self._pending_password): str,
            }
        )