config_flow.py

Go to the documentation of this file.

"""Config flow to configure ZWaveMe integration."""
 
from __future__ import annotations
 
import logging
 
from url_normalize import url_normalize
import voluptuous as vol
 
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_TOKEN, CONF_URL
 
from . import helpers
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
class ZWaveMeConfigFlow(ConfigFlow, domain=DOMAIN):
    """ZWaveMe integration config flow."""
 
    def __init__(self) -> None:
        """Initialize flow."""
        self.urlurl: str | None = None
        self.tokentoken: str | None = None
        self.uuiduuid: str | None = None
 
    async def async_step_user(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user or started with zeroconf."""
        errors = {}
        placeholders = {
            "local_token": "/112f7a4a-0051-cc2b-3b61-1898181b9950",
            "find_token": (
                "0481effe8a5c6f757b455babb678dc0e764feae279/112f7a4a-0051"
                "-cc2b-3b61-1898181b9950"
            ),
            "local_url": "ws://192.168.1.39:8083",
            "add_on_url": "ws://127.0.0.1:8083",
            "find_url": "wss://find.z-wave.me",
            "remote_url": "wss://87.250.250.242:8083",
        }
        if self.urlurl is None:
            schema = vol.Schema(
                {
                    vol.Required(CONF_URL): str,
                    vol.Required(CONF_TOKEN): str,
                }
            )
        else:
            schema = vol.Schema(
                {
                    vol.Required(CONF_TOKEN): str,
                }
            )
 
        if user_input is not None:
            if self.urlurl is None:
                self.urlurl = user_input[CONF_URL]
 
            self.tokentoken = user_input[CONF_TOKEN]
            if not self.urlurl.startswith(("ws://", "wss://")):
                self.urlurl = f"ws://{self.url}"
            self.urlurl = url_normalize(self.urlurl, default_scheme="ws")
            assert self.urlurl
            if self.uuiduuid is None:
                self.uuiduuid = await helpers.get_uuid(self.urlurl, self.tokentoken)
                if self.uuiduuid is not None:
                    await self.async_set_unique_idasync_set_unique_id(self.uuiduuid, raise_on_progress=False)
                    self._abort_if_unique_id_configured_abort_if_unique_id_configured()
                else:
                    errors["base"] = "no_valid_uuid_set"
 
            if not errors:
                return self.async_create_entryasync_create_entryasync_create_entry(
                    title=self.urlurl,
                    data={CONF_URL: self.urlurl, CONF_TOKEN: self.tokentoken},
                )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user",
            description_placeholders=placeholders,
            data_schema=schema,
            errors=errors,
        )
 
    async def async_step_zeroconf(
        self, discovery_info: ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle a discovered Z-Wave accessory - get url to pass into user step.
 
        This flow is triggered by the discovery component.
        """
        self.urlurl = discovery_info.host
        self.uuiduuid = await helpers.get_uuid(self.urlurl)
        if self.uuiduuid is None:
            return self.async_abortasync_abortasync_abort(reason="no_valid_uuid_set")
 
        await self.async_set_unique_idasync_set_unique_id(self.uuiduuid)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured()
        return await self.async_step_userasync_step_userasync_step_user()