adaptive-lighting/custom_components/adaptive_lighting/config_flow.py

"""Config flow for Adaptive Lighting integration."""

import logging
from typing import Any

import voluptuous as vol
from homeassistant import config_entries, data_entry_flow
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers.selector import EntitySelector, EntitySelectorConfig

from .const import (  # pylint: disable=unused-import
    CONF_LIGHTS,
    DOMAIN,
    EXTRA_VALIDATION,
    NONE_STR,
    SECTION_MISC,
    SECTION_SLEEP,
    SECTION_SUNRISE,
    SECTION_SUNSET,
    SECTION_TRANSITION,
    VALIDATION_TUPLES,
)
from .switch import validate

_LOGGER = logging.getLogger(__name__)

OPTIONS_FLOW_DESCRIPTION_PLACEHOLDERS = {
    "webapp_url": "https://basnijholt.github.io/adaptive-lighting",
    "docs_url": "https://github.com/basnijholt/adaptive-lighting#readme",
}


class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Adaptive Lighting."""

    VERSION = 1

    source_options: dict[str, Any] | None = None

    async def async_step_user(self, user_input: dict[str, Any] | None = None):
        """Handle the initial step."""
        if user_input is None and self._async_current_entries():
            return await self.async_step_menu()
        return await self.async_step_wait_for_name(user_input)

    async def async_step_menu(self, user_input: dict[str, Any] | None = None):
        """Handle the menu step."""
        if user_input is not None:
            if user_input["action"] != "new":
                entry_id = user_input["action"]
                entry = self.hass.config_entries.async_get_entry(entry_id)
                if entry:
                    self.source_options = dict(entry.options)
            return await self.async_step_wait_for_name()

        entries = self._async_current_entries()
        options = {"new": "Create new instance"}
        for entry in entries:
            options[entry.entry_id] = f"Duplicate '{entry.title}'"

        return self.async_show_form(
            step_id="menu",
            data_schema=vol.Schema(
                {vol.Required("action", default="new"): vol.In(options)},
            ),
        )

    async def async_step_wait_for_name(self, user_input: dict[str, Any] | None = None):
        """Handle the name step."""
        errors: dict[str, str] = {}

        if user_input is not None:
            await self.async_set_unique_id(user_input[CONF_NAME])
            self._abort_if_unique_id_configured()
            options = self.source_options
            return self.async_create_entry(
                title=user_input[CONF_NAME],
                data=user_input,
                options=options,
            )

        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema({vol.Required(CONF_NAME): str}),
            errors=errors,
        )

    async def async_step_import(self, user_input: dict[str, Any] | None = None):
        """Handle configuration by YAML file."""
        if user_input is None:
            return self.async_abort(reason="no_data")

        await self.async_set_unique_id(user_input[CONF_NAME])
        # Keep a list of switches that are configured via YAML
        data = self.hass.data.setdefault(DOMAIN, {})
        data.setdefault("__yaml__", set()).add(self.unique_id)

        for entry in self._async_current_entries():
            if entry.unique_id == self.unique_id:
                self.hass.config_entries.async_update_entry(entry, data=user_input)
                self._abort_if_unique_id_configured()

        return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)

    @staticmethod
    @callback
    def async_get_options_flow(
        config_entry: config_entries.ConfigEntry,  # noqa: ARG004
    ) -> "OptionsFlowHandler":
        """Get the options flow for this handler."""
        return OptionsFlowHandler()


def validate_options(user_input: dict[str, Any], errors: dict[str, str]) -> None:
    """Validate the options in the OptionsFlow.

    This is an extra validation step because the validators
    in `EXTRA_VALIDATION` cannot be serialized to json.
    """
    for key, (_validate, _) in EXTRA_VALIDATION.items():
        # these are unserializable validators
        value = user_input.get(key)
        try:
            if value is not None and value != NONE_STR:
                _validate(value)
        except vol.Invalid:
            _LOGGER.exception("Configuration option %s=%s is incorrect", key, value)
            errors["base"] = "option_error"


class OptionsFlowHandler(config_entries.OptionsFlow):
    """Handle a option flow for Adaptive Lighting."""

    SECTION_MAP: dict[str, set[str]] = {
        "sleep": SECTION_SLEEP,
        "sunrise": SECTION_SUNRISE,
        "sunset": SECTION_SUNSET,
        "transition": SECTION_TRANSITION,
        "misc": SECTION_MISC,
    }

    # Reverse lookup, avoids looping through sections for each(!) option
    OPTION_TO_SECTION: dict[str, str] = {
        name: section for section, names in SECTION_MAP.items() for name in names
    }

    def _flatten_section_input(self, user_input: dict[str, Any]) -> dict[str, Any]:
        """Flatten section input by merging nested section dicts into top level."""
        flat_input: dict[str, Any] = {}
        for key, value in user_input.items():
            if key in self.SECTION_MAP and isinstance(value, dict):
                flat_input.update(value)
            else:
                flat_input[key] = value
        return flat_input

    async def async_step_init(self, user_input: dict[str, Any] | None = None):
        """Handle options flow with collapsible sections."""
        conf = self.config_entry
        data = validate(conf)
        form_data = {**conf.data, **conf.options}
        if conf.source == config_entries.SOURCE_IMPORT:
            return self.async_show_form(
                step_id="init",
                data_schema=None,
                description_placeholders=OPTIONS_FLOW_DESCRIPTION_PLACEHOLDERS,
            )
        errors: dict[str, str] = {}
        if user_input is not None:
            flat_input = self._flatten_section_input(user_input)
            validate_options(flat_input, errors)
            if not errors:
                return self.async_create_entry(title="", data=flat_input)
            data.update(flat_input)
            form_data.update(flat_input)

        # Validate that all configured lights still exist
        all_lights = set(self.hass.states.async_entity_ids("light"))
        for configured_light in data[CONF_LIGHTS]:
            if configured_light not in all_lights:
                errors[CONF_LIGHTS] = "entity_missing"
                _LOGGER.error(
                    "%s: light entity %s is configured, but was not found",
                    data[CONF_NAME],
                    configured_light,
                )

        to_replace: dict[str, Any] = {
            CONF_LIGHTS: EntitySelector(
                EntitySelectorConfig(
                    domain="light",
                    multiple=True,
                ),
            ),
        }

        full_schema: dict[vol.Marker, Any] = {}

        section_schemas: dict[str, dict[vol.Marker, Any]] = {
            name: {} for name in self.SECTION_MAP
        }

        for name, default, validation in VALIDATION_TUPLES:
            key = vol.Optional(name, default=form_data.get(name, default))
            field_schema = to_replace.get(name, validation)

            section_name = self.OPTION_TO_SECTION.get(name)
            if section_name:
                # Option is part of section, put into section schema
                section_schemas[section_name][key] = field_schema
            else:
                # Option is NOT part of section, place at top level
                full_schema[key] = field_schema

        # Create collapsible sections for each section set
        for section_name in self.SECTION_MAP:
            if section_schemas[section_name]:
                full_schema[vol.Required(section_name)] = data_entry_flow.section(
                    vol.Schema(section_schemas[section_name]),
                    data_entry_flow.SectionConfig(collapsed=True),
                )

        return self.async_show_form(
            step_id="init",
            data_schema=vol.Schema(full_schema),
            errors=errors,
            description_placeholders=OPTIONS_FLOW_DESCRIPTION_PLACEHOLDERS,
        )