93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
"""Config flow for Linky Tariff integration."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import voluptuous as vol
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.core import callback
|
|
from homeassistant.data_entry_flow import FlowResult
|
|
from homeassistant.helpers import device_registry as dr
|
|
|
|
from .const import (
|
|
DOMAIN,
|
|
CONF_IEEE,
|
|
CONF_POLL_INTERVAL,
|
|
DEFAULT_POLL_INTERVAL,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
async def async_get_devices(hass):
|
|
"""Get ZHA devices."""
|
|
device_registry = dr.async_get(hass)
|
|
return [
|
|
entry
|
|
for entry in device_registry.devices.values()
|
|
if entry.via_device_id is not None and "zha" in entry.via_device_id
|
|
]
|
|
|
|
class LinkyTariffConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|
"""Handle a config flow for Linky Tariff."""
|
|
|
|
VERSION = 1
|
|
|
|
async def async_step_user(self, user_input=None):
|
|
"""Handle the initial step."""
|
|
errors = {}
|
|
devices = await async_get_devices(self.hass)
|
|
|
|
if not devices:
|
|
return self.async_abort(reason="no_zha_devices")
|
|
|
|
if user_input is not None:
|
|
await self.async_set_unique_id(user_input[CONF_IEEE])
|
|
self._abort_if_unique_id_configured()
|
|
return self.async_create_entry(title="Linky Tariff", data=user_input)
|
|
|
|
device_options = {
|
|
device.id: f"{device.name_by_user or device.name} ({device.id})"
|
|
for device in devices
|
|
}
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=vol.Schema({
|
|
vol.Required(CONF_IEEE): vol.In(device_options),
|
|
vol.Optional(
|
|
CONF_POLL_INTERVAL,
|
|
default=DEFAULT_POLL_INTERVAL
|
|
): int,
|
|
}),
|
|
errors=errors,
|
|
)
|
|
|
|
@staticmethod
|
|
@callback
|
|
def async_get_options_flow(config_entry):
|
|
"""Get the options flow for this handler."""
|
|
return LinkyTariffOptionsFlow(config_entry)
|
|
|
|
class LinkyTariffOptionsFlow(config_entries.OptionsFlow):
|
|
"""Handle options flow for Linky Tariff."""
|
|
|
|
def __init__(self, config_entry):
|
|
"""Initialize options flow."""
|
|
self.config_entry = config_entry
|
|
|
|
async def async_step_init(self, user_input=None):
|
|
"""Manage the options."""
|
|
if user_input is not None:
|
|
return self.async_create_entry(title="", data=user_input)
|
|
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=vol.Schema({
|
|
vol.Optional(
|
|
CONF_POLL_INTERVAL,
|
|
default=self.config_entry.options.get(
|
|
CONF_POLL_INTERVAL, DEFAULT_POLL_INTERVAL
|
|
),
|
|
): int,
|
|
}),
|
|
) |