config_flow.py

Go to the documentation of this file.

"""Config flow for Lektrico Charging Station."""
 
from __future__ import annotations
 
from typing import Any
 
from lektricowifi import Device, DeviceConnectionError
import voluptuous as vol
 
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import (
    ATTR_HW_VERSION,
    ATTR_SERIAL_NUMBER,
    CONF_HOST,
    CONF_TYPE,
)
from homeassistant.core import callback
from homeassistant.helpers.httpx_client import get_async_client
 
from .const import DOMAIN
 
STEP_USER_DATA_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_HOST): str,
    }
)
 
 
class LektricoFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a Lektrico config flow."""
 
    VERSION = 1
 
    _host: str
    _name: str
    _serial_number: str
    _board_revision: str
    _device_type: str
 
    async def async_step_user(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initiated by the user."""
        errors = None
 
        if user_input is not None:
            self._host_host = user_input[CONF_HOST]
 
            # obtain serial number
            try:
                await self._get_lektrico_device_settings_and_treat_unique_id_get_lektrico_device_settings_and_treat_unique_id()
                return self._async_create_entry_async_create_entry()
            except DeviceConnectionError:
                errors = {CONF_HOST: "cannot_connect"}
 
        return self._async_show_setup_form_async_show_setup_form(user_input=user_input, errors=errors)
 
    @callback
    def _async_show_setup_form(
        self,
        user_input: dict[str, Any] | None = None,
        errors: dict[str, str] | None = None,
    ) -> ConfigFlowResult:
        """Show the setup form to the user."""
        if user_input is None:
            user_input = {}
 
        schema = self.add_suggested_values_to_schemaadd_suggested_values_to_schema(STEP_USER_DATA_SCHEMA, user_input)
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=schema,
            errors=errors or {},
        )
 
    @callback
    def _async_create_entry(self) -> ConfigFlowResult:
        return self.async_create_entryasync_create_entryasync_create_entry(
            title=self._name_name,
            data={
                CONF_HOST: self._host_host,
                ATTR_SERIAL_NUMBER: self._serial_number_serial_number,
                CONF_TYPE: self._device_type_device_type,
                ATTR_HW_VERSION: self._board_revision_board_revision,
            },
        )
 
    async def async_step_zeroconf(
        self, discovery_info: ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle zeroconf discovery."""
        self._host_host = discovery_info.host  # 192.168.100.11
 
        # read settings from the device
        try:
            await self._get_lektrico_device_settings_and_treat_unique_id_get_lektrico_device_settings_and_treat_unique_id()
        except DeviceConnectionError:
            return self.async_abortasync_abortasync_abort(reason="cannot_connect")
 
        self.context["title_placeholders"] = {
            "serial_number": self._serial_number_serial_number,
            "name": self._name_name,
        }
 
        return await self.async_step_confirmasync_step_confirm()
 
    async def _get_lektrico_device_settings_and_treat_unique_id(self) -> None:
        """Get device's serial number from a Lektrico device."""
        device = Device(
            _host=self._host_host,
            asyncClient=get_async_client(self.hass),
        )
 
        settings = await device.device_config()
        self._serial_number_serial_number = str(settings["serial_number"])
        self._device_type_device_type = settings["type"]
        self._board_revision_board_revision = settings["board_revision"]
        self._name_name = f"{settings["type"]}_{self._serial_number}"
 
        # Check if already configured
        # Set unique id
        await self.async_set_unique_idasync_set_unique_id(self._serial_number_serial_number, raise_on_progress=True)
        # Abort if already configured, but update the last-known host
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(
            updates={CONF_HOST: self._host_host}, reload_on_update=True
        )
 
    async def async_step_confirm(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Allow the user to confirm adding the device."""
 
        if user_input is not None:
            return self._async_create_entry_async_create_entry()
 
        self._set_confirm_only_set_confirm_only()
        return self.async_show_formasync_show_formasync_show_form(step_id="confirm")