config_flow.py

Go to the documentation of this file.

"""Config flow to configure Xiaomi Aqara."""
 
import logging
from socket import gaierror
from typing import Any
 
import voluptuous as vol
from xiaomi_gateway import MULTICAST_PORT, XiaomiGateway, XiaomiGatewayDiscovery
 
from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_PORT, CONF_PROTOCOL
from homeassistant.core import callback
from homeassistant.helpers.device_registry import format_mac
 
from .const import (
    CONF_INTERFACE,
    CONF_KEY,
    CONF_SID,
    DEFAULT_DISCOVERY_RETRY,
    DOMAIN,
    ZEROCONF_ACPARTNER,
    ZEROCONF_GATEWAY,
)
 
_LOGGER = logging.getLogger(__name__)
 
DEFAULT_GATEWAY_NAME = "Xiaomi Aqara Gateway"
DEFAULT_INTERFACE = "any"
 
 
GATEWAY_CONFIG = vol.Schema(
    {vol.Optional(CONF_INTERFACE, default=DEFAULT_INTERFACE): str}
)
CONFIG_HOST = {
    vol.Optional(CONF_HOST): str,
    vol.Optional(CONF_MAC): str,
}
GATEWAY_CONFIG_HOST = GATEWAY_CONFIG.extend(CONFIG_HOST)
GATEWAY_SETTINGS = vol.Schema(
    {
        vol.Optional(CONF_KEY): vol.All(str, vol.Length(min=16, max=16)),
        vol.Optional(CONF_NAME, default=DEFAULT_GATEWAY_NAME): str,
    }
)
 
 
class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a Xiaomi Aqara config flow."""
 
    VERSION = 1
 
    selected_gateway: XiaomiGateway
    gateways: dict[str, XiaomiGateway]
 
    def __init__(self) -> None:
        """Initialize."""
        self.hosthost: str | None = None
        self.interfaceinterface = DEFAULT_INTERFACE
        self.sidsid: str | None = None
 
    @callback
    def async_show_form_step_user(self, errors):
        """Show the form belonging to the user step."""
        schema = GATEWAY_CONFIG
        if (self.hosthost is None and self.sidsid is None) or errors:
            schema = GATEWAY_CONFIG_HOST
 
        return self.async_show_formasync_show_formasync_show_form(step_id="user", data_schema=schema, errors=errors)
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user."""
        errors: dict[str, str] = {}
        if user_input is None:
            return self.async_show_form_step_userasync_show_form_step_user(errors)
 
        self.interfaceinterface = user_input[CONF_INTERFACE]
 
        # allow optional manual setting of host and mac
        if self.hosthost is None:
            self.hosthost = user_input.get(CONF_HOST)
        if self.sidsid is None:
            # format sid from mac_address
            if (mac_address := user_input.get(CONF_MAC)) is not None:
                self.sidsid = format_mac(mac_address).replace(":", "")
 
        # if host is already known by zeroconf discovery or manual optional settings
        if self.hosthost is not None and self.sidsid is not None:
            # Connect to Xiaomi Aqara Gateway
            self.selected_gatewayselected_gateway = await self.hass.async_add_executor_job(
                XiaomiGateway,
                self.hosthost,
                self.sidsid,
                None,
                DEFAULT_DISCOVERY_RETRY,
                self.interfaceinterface,
                MULTICAST_PORT,
                None,
            )
 
            if self.selected_gatewayselected_gateway.connection_error:
                errors[CONF_HOST] = "invalid_host"
            if self.selected_gatewayselected_gateway.mac_error:
                errors[CONF_MAC] = "invalid_mac"
            if errors:
                return self.async_show_form_step_userasync_show_form_step_user(errors)
 
            return await self.async_step_settingsasync_step_settings()
 
        # Discover Xiaomi Aqara Gateways in the network to get required SIDs.
        xiaomi = XiaomiGatewayDiscovery(self.interfaceinterface)
        try:
            await self.hass.async_add_executor_job(xiaomi.discover_gateways)
        except gaierror:
            errors[CONF_INTERFACE] = "invalid_interface"
            return self.async_show_form_step_userasync_show_form_step_user(errors)
 
        self.gatewaysgateways = xiaomi.gateways
 
        if len(self.gatewaysgateways) == 1:
            self.selected_gatewayselected_gateway = list(self.gatewaysgateways.values())[0]
            self.sidsid = self.selected_gatewayselected_gateway.sid
            return await self.async_step_settingsasync_step_settings()
        if len(self.gatewaysgateways) > 1:
            return await self.async_step_selectasync_step_select()
 
        errors["base"] = "discovery_error"
        return self.async_show_form_step_userasync_show_form_step_user(errors)
 
    async def async_step_select(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Handle multiple aqara gateways found."""
        errors: dict[str, str] = {}
        if user_input is not None:
            ip_adress = user_input["select_ip"]
            self.selected_gatewayselected_gateway = self.gatewaysgateways[ip_adress]
            self.sidsid = self.selected_gatewayselected_gateway.sid
            return await self.async_step_settingsasync_step_settings()
 
        select_schema = vol.Schema(
            {
                vol.Required("select_ip"): vol.In(
                    [gateway.ip_adress for gateway in self.gatewaysgateways.values()]
                )
            }
        )
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="select", data_schema=select_schema, errors=errors
        )
 
    async def async_step_zeroconf(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> ConfigFlowResult:
        """Handle zeroconf discovery."""
        name = discovery_info.name
        self.hosthost = discovery_info.host
        mac_address = discovery_info.properties.get("mac")
 
        if not name or not self.hosthost or not mac_address:
            return self.async_abortasync_abortasync_abort(reason="not_xiaomi_aqara")
 
        # Check if the discovered device is an xiaomi aqara gateway.
        if not (name.startswith((ZEROCONF_GATEWAY, ZEROCONF_ACPARTNER))):
            _LOGGER.debug(
                (
                    "Xiaomi device '%s' discovered with host %s, not identified as"
                    " xiaomi aqara gateway"
                ),
                name,
                self.hosthost,
            )
            return self.async_abortasync_abortasync_abort(reason="not_xiaomi_aqara")
 
        # format mac (include semicolns and make lowercase)
        mac_address = format_mac(mac_address)
 
        # format sid from mac_address
        self.sidsid = mac_address.replace(":", "")
 
        unique_id = mac_address
        await self.async_set_unique_idasync_set_unique_id(unique_id)
        self._abort_if_unique_id_configured_abort_if_unique_id_configured(
            {CONF_HOST: self.hosthost, CONF_MAC: mac_address}
        )
 
        self.context.update({"title_placeholders": {"name": self.hosthost}})
 
        return await self.async_step_userasync_step_userasync_step_user()
 
    async def async_step_settings(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Specify settings and connect aqara gateway."""
        errors = {}
        if user_input is not None:
            # get all required data
            name = user_input[CONF_NAME]
            key = user_input.get(CONF_KEY)
            ip_adress = self.selected_gatewayselected_gateway.ip_adress
            port = self.selected_gatewayselected_gateway.port
            protocol = self.selected_gatewayselected_gateway.proto
 
            if key is not None:
                # validate key by issuing stop ringtone playback command.
                self.selected_gatewayselected_gateway.key = key
                valid_key = self.selected_gatewayselected_gateway.write_to_hub(self.sidsid, mid=10000)
            else:
                valid_key = True
 
            if valid_key:
                # format_mac, for a gateway the sid equals the mac address
                mac_address = format_mac(self.sidsid)
 
                # set unique_id
                unique_id = mac_address
                await self.async_set_unique_idasync_set_unique_id(unique_id)
                self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
                return self.async_create_entryasync_create_entryasync_create_entry(
                    title=name,
                    data={
                        CONF_HOST: ip_adress,
                        CONF_PORT: port,
                        CONF_MAC: mac_address,
                        CONF_INTERFACE: self.interfaceinterface,
                        CONF_PROTOCOL: protocol,
                        CONF_KEY: key,
                        CONF_SID: self.sidsid,
                    },
                )
 
            errors[CONF_KEY] = "invalid_key"
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="settings", data_schema=GATEWAY_SETTINGS, errors=errors
        )