esc/custom_components/easee_solar_charging/config_flow.py

from __future__ import annotations

import voluptuous as vol

from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, selector

from .const import (
    CONF_CHARGER_ID,
    CHARGING_PROFILE_ALL_SURPLUS,
    CHARGING_PROFILE_CONSERVATIVE,
    CONF_CHARGING_PROFILE,
    CONF_DEAD_BAND_RESYNC_MINUTES,
    CONF_DEVICE_ID,
    CONF_EV_CHARGING_SENSOR,
    CONF_HOUSE_LOAD_SENSOR,
    CONF_NOTIFY_TARGET,
    CONF_SITE_ID,
    CONF_SOLAR_PRODUCTION_SENSOR,
    DEFAULT_CHARGING_PROFILE,
    CONF_STOP_GRACE_MINUTES,
    DEFAULT_DEAD_BAND_RESYNC_MINUTES,
    DEFAULT_STOP_GRACE_MINUTES,
    DOMAIN,
    EASEE_DOMAIN,
)

_STOP_GRACE_SELECTOR = selector.NumberSelector(
    selector.NumberSelectorConfig(
        min=1, max=60, step=1, unit_of_measurement="min", mode=selector.NumberSelectorMode.BOX
    )
)

_DEAD_BAND_RESYNC_SELECTOR = selector.NumberSelector(
    selector.NumberSelectorConfig(
        min=1, max=60, step=1, unit_of_measurement="min", mode=selector.NumberSelectorMode.BOX
    )
)

_CHARGING_PROFILE_SELECTOR = selector.SelectSelector(
    selector.SelectSelectorConfig(
        options=[
            selector.SelectOptionDict(value=CHARGING_PROFILE_CONSERVATIVE, label="Conservative"),
            selector.SelectOptionDict(value=CHARGING_PROFILE_ALL_SURPLUS, label="All Surplus"),
        ],
        mode=selector.SelectSelectorMode.DROPDOWN,
    )
)

_POWER_SENSOR_SELECTOR = selector.EntitySelector(
    selector.EntitySelectorConfig(domain="sensor", device_class="power")
)

_STATIC_USER_FIELDS = {
    vol.Required(CONF_SITE_ID): str,
    vol.Required(CONF_SOLAR_PRODUCTION_SENSOR): _POWER_SENSOR_SELECTOR,
    vol.Required(CONF_HOUSE_LOAD_SENSOR): _POWER_SENSOR_SELECTOR,
    vol.Required(CONF_EV_CHARGING_SENSOR): _POWER_SENSOR_SELECTOR,
    vol.Required(CONF_DEVICE_ID): selector.DeviceSelector(
        selector.DeviceSelectorConfig(integration="easee")
    ),
}


def _notify_services(hass: HomeAssistant) -> list[str]:
    """Return sorted list of notify service names, excluding internal ones."""
    return sorted(
        svc
        for svc in hass.services.async_services().get("notify", {})
        if svc not in ("notify", "send_message", "reload")
    )


def _notify_selector(options: list[str]) -> selector.SelectSelector:
    return selector.SelectSelector(
        selector.SelectSelectorConfig(
            options=options,
            mode=selector.SelectSelectorMode.DROPDOWN,
            custom_value=True,  # allow typing a target not yet loaded
        )
    )


def _user_schema(hass: HomeAssistant) -> vol.Schema:
    return vol.Schema({
        **_STATIC_USER_FIELDS,
        vol.Optional(CONF_NOTIFY_TARGET): _notify_selector(_notify_services(hass)),
    })


def _options_schema(hass: HomeAssistant, current: dict) -> vol.Schema:
    return vol.Schema({
        vol.Required(
            CONF_CHARGING_PROFILE,
            default=current.get(CONF_CHARGING_PROFILE, DEFAULT_CHARGING_PROFILE),
        ): _CHARGING_PROFILE_SELECTOR,
        vol.Required(
            CONF_STOP_GRACE_MINUTES,
            default=current.get(CONF_STOP_GRACE_MINUTES, DEFAULT_STOP_GRACE_MINUTES),
        ): _STOP_GRACE_SELECTOR,
        vol.Required(
            CONF_DEAD_BAND_RESYNC_MINUTES,
            default=current.get(CONF_DEAD_BAND_RESYNC_MINUTES, DEFAULT_DEAD_BAND_RESYNC_MINUTES),
        ): _DEAD_BAND_RESYNC_SELECTOR,
        vol.Optional(CONF_NOTIFY_TARGET): _notify_selector(_notify_services(hass)),
    })


def _charger_serial_from_device(hass: HomeAssistant, device_id: str) -> str | None:
    """Extract the Easee charger serial number from the device registry entry."""
    registry = dr.async_get(hass)
    device = registry.async_get(device_id)
    if device is None:
        return None
    for domain, identifier in device.identifiers:
        if domain == EASEE_DOMAIN:
            return identifier
    return None


class EaseeSolarChargingConfigFlow(ConfigFlow, domain=DOMAIN):
    VERSION = 1

    @staticmethod
    @callback
    def async_get_options_flow(config_entry: ConfigEntry) -> EaseeSolarChargingOptionsFlow:
        return EaseeSolarChargingOptionsFlow()

    async def async_step_user(self, user_input: dict | None = None) -> ConfigFlowResult:
        errors: dict[str, str] = {}

        if user_input is not None:
            device_id = user_input[CONF_DEVICE_ID]
            charger_id = _charger_serial_from_device(self.hass, device_id)

            if charger_id is None:
                errors[CONF_DEVICE_ID] = "cannot_resolve_serial"
            else:
                await self.async_set_unique_id(charger_id)
                self._abort_if_unique_id_configured()

                notify_target = user_input.pop(CONF_NOTIFY_TARGET, None)
                initial_options = {CONF_NOTIFY_TARGET: notify_target} if notify_target else {}

                return self.async_create_entry(
                    title=f"Easee {charger_id}",
                    data={**user_input, CONF_CHARGER_ID: charger_id},
                    options=initial_options,
                )

        return self.async_show_form(
            step_id="user",
            data_schema=_user_schema(self.hass),
            errors=errors,
        )


class EaseeSolarChargingOptionsFlow(OptionsFlow):
    async def async_step_init(self, user_input: dict | None = None) -> ConfigFlowResult:
        if user_input is not None:
            return self.async_create_entry(data=user_input)

        return self.async_show_form(
            step_id="init",
            data_schema=self.add_suggested_values_to_schema(
                _options_schema(self.hass, self.config_entry.options),
                self.config_entry.options,
            ),
        )