109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlow
|
|
from homeassistant.core import callback
|
|
from homeassistant.helpers import device_registry as dr, selector
|
|
|
|
from .const import (
|
|
CONF_CHARGER_ID,
|
|
CONF_DEVICE_ID,
|
|
CONF_NOTIFY_TARGET,
|
|
CONF_SITE_ID,
|
|
CONF_SOLAR_EXCESS_SENSOR,
|
|
CONF_STOP_GRACE_MINUTES,
|
|
DEFAULT_STOP_GRACE_MINUTES,
|
|
DOMAIN,
|
|
EASEE_DOMAIN,
|
|
)
|
|
|
|
_NOTIFY_TARGET_SELECTOR = selector.TextSelector(
|
|
selector.TextSelectorConfig(type=selector.TextSelectorType.TEXT)
|
|
)
|
|
|
|
STEP_USER_DATA_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(CONF_SITE_ID): str,
|
|
vol.Required(CONF_SOLAR_EXCESS_SENSOR): selector.EntitySelector(
|
|
selector.EntitySelectorConfig(domain="sensor")
|
|
),
|
|
vol.Required(CONF_DEVICE_ID): selector.DeviceSelector(
|
|
selector.DeviceSelectorConfig(integration="easee")
|
|
),
|
|
vol.Optional(CONF_NOTIFY_TARGET): _NOTIFY_TARGET_SELECTOR,
|
|
}
|
|
)
|
|
|
|
OPTIONS_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(CONF_STOP_GRACE_MINUTES, default=DEFAULT_STOP_GRACE_MINUTES): selector.NumberSelector(
|
|
selector.NumberSelectorConfig(min=1, max=60, step=1, unit_of_measurement="min", mode=selector.NumberSelectorMode.BOX)
|
|
),
|
|
vol.Optional(CONF_NOTIFY_TARGET): _NOTIFY_TARGET_SELECTOR,
|
|
}
|
|
)
|
|
|
|
|
|
def _charger_serial_from_device(hass, 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=STEP_USER_DATA_SCHEMA,
|
|
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.config_entry.options,
|
|
),
|
|
)
|