config_flow.py

Go to the documentation of this file.

"""Config flow for TechnoVE."""
 
from typing import Any
 
from technove import Station as TechnoVEStation, TechnoVE, TechnoVEConnectionError
import voluptuous as vol
 
from homeassistant.components import onboarding, zeroconf
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_MAC
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import DOMAIN
 
 
class TechnoVEConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for TechnoVE."""
 
    VERSION = 1
    discovered_host: str
    discovered_station: TechnoVEStation
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initiated by the user."""
        errors = {}
        if user_input is not None:
            try:
                station = await self._async_get_station_async_get_station(user_input[CONF_HOST])
            except TechnoVEConnectionError:
                errors["base"] = "cannot_connect"
            else:
                await self.async_set_unique_idasync_set_unique_id(station.info.mac_address)
                self._abort_if_unique_id_configured_abort_if_unique_id_configured(
                    updates={CONF_HOST: user_input[CONF_HOST]}
                )
                return self.async_create_entryasync_create_entryasync_create_entry(
                    title=station.info.name,
                    data={
                        CONF_HOST: user_input[CONF_HOST],
                    },
                )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
            errors=errors,
        )
 
    async def async_step_zeroconf(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle zeroconf discovery."""
        # Abort quick if the device with provided mac is already configured
        if mac := discovery_info.properties.get(CONF_MAC):
            await self.async_set_unique_idasync_set_unique_id(mac)
            self._abort_if_unique_id_configured_abort_if_unique_id_configured(
                updates={CONF_HOST: discovery_info.host}
            )
 
        self.discovered_hostdiscovered_host = discovery_info.host
        try:
            self.discovered_stationdiscovered_station = await self._async_get_station_async_get_station(discovery_info.host)
        except TechnoVEConnectionError:
            return self.async_abortasync_abortasync_abort(reason="cannot_connect")
 
        await self.async_set_unique_idasync_set_unique_id(self.discovered_stationdiscovered_station.info.mac_address)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
 
        self.context.update(
            {
                "title_placeholders": {"name": self.discovered_stationdiscovered_station.info.name},
            }
        )
        return await self.async_step_zeroconf_confirmasync_step_zeroconf_confirm()
 
    async def async_step_zeroconf_confirm(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initiated by zeroconf."""
        if user_input is not None or not onboarding.async_is_onboarded(self.hass):
            return self.async_create_entryasync_create_entryasync_create_entry(
                title=self.discovered_stationdiscovered_station.info.name,
                data={
                    CONF_HOST: self.discovered_hostdiscovered_host,
                },
            )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="zeroconf_confirm",
            description_placeholders={"name": self.discovered_stationdiscovered_station.info.name},
        )
 
    async def _async_get_station(self, host: str) -> TechnoVEStation:
        """Get information from a TechnoVE station."""
        api = TechnoVE(host, session=async_get_clientsession(self.hass))
        return await api.update()