"""Config flow for Midea Air Appliance (Local) integration.""" from __future__ import annotations from ipaddress import IPv4Address, IPv4Network import logging from typing import Any from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow from homeassistant.const import ( ATTR_ID, ATTR_NAME, CONF_API_VERSION, CONF_BROADCAST_ADDRESS, CONF_DEVICES, CONF_DISCOVERY, CONF_ID, CONF_INCLUDE, CONF_IP_ADDRESS, CONF_NAME, CONF_PASSWORD, CONF_SCAN_INTERVAL, CONF_TOKEN, CONF_TTL, CONF_TYPE, CONF_UNIQUE_ID, CONF_USERNAME, ) from homeassistant.core import callback from homeassistant.data_entry_flow import FlowHandler, FlowResult from homeassistant.helpers import config_validation as cv import voluptuous as vol from midea_beautiful.cloud import MideaCloud from midea_beautiful.exceptions import ( AuthenticationError, CloudAuthenticationError, CloudError, MideaError, MideaNetworkError, ProtocolError, RetryLaterError, ) from midea_beautiful.lan import LanDevice from midea_beautiful.midea import ( APPLIANCE_TYPE_DEHUMIDIFIER, SUPPORTED_APPS, ) from custom_components.midea_dehumidifier_lan import Hub from custom_components.midea_dehumidifier_lan.const import ( NAME, CURRENT_CONFIG_VERSION, SUPPORTED_APPLIANCES, CONF_ADVANCED_SETTINGS, CONF_DEBUG, CONF_MOBILE_APP, CONF_TOKEN_KEY, DEFAULT_APP, DEFAULT_DISCOVERY_MODE, DEFAULT_PASSWORD, DEFAULT_SCAN_INTERVAL, DEFAULT_TTL, DEFAULT_USERNAME, DISCOVERY_CLOUD, DISCOVERY_IGNORE, DISCOVERY_LAN, DISCOVERY_MODE_LABELS, DISCOVERY_WAIT, DOMAIN, LOCAL_BROADCAST, UNKNOWN_IP, ) from custom_components.midea_dehumidifier_lan.util import ( MideaClient, RedactedConf, address_ok, supported_appliance, ) _LOGGER = logging.getLogger(__name__) def _appliance_schema( # pylint: disable=too-many-arguments name: str, address: str = UNKNOWN_IP, ttl: int = DEFAULT_TTL, token: str = "", token_key: str = "", discovery_mode=DISCOVERY_WAIT, ) -> vol.Schema: return vol.Schema( { vol.Optional(CONF_DISCOVERY, default=str(discovery_mode)): vol.In( DISCOVERY_MODE_LABELS ), vol.Optional( CONF_IP_ADDRESS, default=address or UNKNOWN_IP, ): cv.string, vol.Required(CONF_NAME, default=name): cv.string, vol.Required( CONF_TTL, msg="Test", default=ttl, description={"suffix": "minutes"}, ): cv.positive_int, vol.Optional(CONF_TOKEN, default=token or ""): cv.string, vol.Optional(CONF_TOKEN_KEY, default=token_key or ""): cv.string, } ) # pylint: disable=too-many-arguments def _advanced_settings_schema( username: str = "", password: str = "", app: str = DEFAULT_APP, broadcast_address: str = "", appliances: list[str] = None, debug: bool = False, ) -> vol.Schema: appliances = appliances or [APPLIANCE_TYPE_DEHUMIDIFIER] return vol.Schema( { vol.Required(CONF_USERNAME, default=username): cv.string, vol.Required(CONF_PASSWORD, default=password): cv.string, vol.Optional(CONF_MOBILE_APP, default=app): vol.In(SUPPORTED_APPS.keys()), vol.Optional(CONF_BROADCAST_ADDRESS, default=broadcast_address): cv.string, vol.Required( CONF_SCAN_INTERVAL, msg="Test", default=DEFAULT_SCAN_INTERVAL, description={"suffix": "minutes"}, ): cv.positive_int, vol.Required(CONF_INCLUDE, default=appliances): vol.All( cv.multi_select(SUPPORTED_APPLIANCES), vol.Length(min=1, msg="Must select at least one appliance category"), ), vol.Required(CONF_DEBUG, default=debug): bool, } ) def _reauth_schema( username: str, password: str, ) -> vol.Schema: return vol.Schema( { vol.Required(CONF_USERNAME, default=username): cv.string, vol.Required(CONF_PASSWORD, default=password): cv.string, } ) def _user_schema(username: str, password: str, app: str) -> vol.Schema: return vol.Schema( { vol.Required(CONF_USERNAME, default=username): cv.string, vol.Required(CONF_PASSWORD, default=password): cv.string, vol.Optional(CONF_MOBILE_APP, default=app): vol.In(SUPPORTED_APPS.keys()), vol.Required(CONF_ADVANCED_SETTINGS, default=False): bool, } ) # pylint: disable=too-many-instance-attributes class _MideaFlow(FlowHandler): """Base class for Midea data flows""" def __init__(self) -> None: super().__init__() self.appliance_idx = -1 self.appliances: list[LanDevice] = [] self._client: MideaClient | None = None self.cloud: MideaCloud | None = None # type: ignore self.conf = {} self.config_entry: ConfigEntry | None = None self.devices_conf: list[dict[str, Any]] = [] self.discovered_appliances: list[LanDevice | None] = [] self.error_cause: str = "" self.errors: dict[str, Any] = {} self.indexes_to_process = [] @property def client(self) -> MideaClient: """Returns instance of MideaClient.""" if not self._client: self._client = MideaClient(self.hass) return self._client def _process_exception(self: _MideaFlow, ex: Exception) -> None: if isinstance(ex, _FlowException): _LOGGER.warning( "Caught flow exception during appliance step %s", ex, exc_info=True ) self.error_cause = str(ex.cause) self.errors["base"] = ex.message elif isinstance(ex, CloudAuthenticationError): self.error_cause = f"{ex.error_code} - {ex.message}" self.errors["base"] = "invalid_auth" elif isinstance(ex, CloudError): self.error_cause = f"{ex.error_code} - {ex.message}" self.errors["base"] = "midea_client" elif isinstance(ex, RetryLaterError): self.error_cause = f"{ex.error_code} - {ex.message}" self.errors["base"] = "retry_later" elif isinstance(ex, MideaError): self.error_cause = f"{ex.message}" self.errors["base"] = "midea_client" else: raise ex def _connect_to_cloud(self: _MideaFlow, extra_conf: dict[str, Any] = None) -> None: """Validates that cloud credentials are valid""" cfg = self.conf | (extra_conf or {}) try: self.cloud = self.client.connect_to_cloud(cfg) except MideaError as ex: raise _FlowException("no_cloud", str(ex)) from ex def _validate_appliance( self: _MideaFlow, appliance: LanDevice, device_conf: dict ) -> LanDevice | None: """ Validates that appliance configuration is correct and matches physical device """ discovery_mode = device_conf.get(CONF_DISCOVERY, DEFAULT_DISCOVERY_MODE) if discovery_mode == DISCOVERY_IGNORE: _LOGGER.debug("Ignoring appliance %s", appliance) return None if discovery_mode == DISCOVERY_WAIT: _LOGGER.debug( "Attempt to discover appliance %s will be made later", appliance, ) return None try: if discovery_mode == DISCOVERY_CLOUD: discovered = self.client.appliance_state( appliance_id=appliance.appliance_id, cloud=self.cloud, use_cloud=True, ) else: # DISCOVERY_LAN ip_address = appliance.address if not address_ok(ip_address): raise _FlowException("invalid_ip_address", ip_address) try: IPv4Address(ip_address) except Exception as ex: _LOGGER.debug("Invalid appliance address %s: %s", ip_address, ex) raise _FlowException("invalid_ip_address", ip_address) from ex discovered = self.client.appliance_state( address=ip_address, cloud=self.cloud ) except ProtocolError as ex: raise _FlowException("connection_error", str(ex)) from ex except AuthenticationError as ex: raise _FlowException("invalid_auth", str(ex)) from ex except MideaNetworkError as ex: raise _FlowException("cannot_connect", str(ex)) from ex except MideaError as ex: raise _FlowException("not_discovered", str(ex)) from ex if discovered is None: raise _FlowException("not_discovered", appliance.address) return discovered async def _async_add_entry(self: _MideaFlow) -> FlowResult: supported_devices_conf = [] for i, appliance in enumerate(self.appliances): if not supported_appliance(self.conf, appliance): continue device_conf = self.devices_conf[i] if device_conf.get(CONF_DISCOVERY) != DISCOVERY_IGNORE: device_conf |= { CONF_API_VERSION: appliance.version, CONF_ID: appliance.appliance_id, CONF_IP_ADDRESS: ( appliance.address or device_conf[CONF_IP_ADDRESS] or UNKNOWN_IP ), CONF_NAME: appliance.name, CONF_TOKEN_KEY: appliance.key, CONF_TOKEN: appliance.token, CONF_TYPE: appliance.type, CONF_UNIQUE_ID: appliance.serial_number, } suggested_discovery = ( DISCOVERY_LAN if address_ok(device_conf[CONF_IP_ADDRESS]) else DISCOVERY_WAIT ) device_conf.get(CONF_DISCOVERY, suggested_discovery) supported_devices_conf.append(device_conf) self.devices_conf = supported_devices_conf self.conf[CONF_DEVICES] = self.devices_conf # Remove not used elements self.conf.pop(CONF_ADVANCED_SETTINGS, None) if self.config_entry: _LOGGER.debug("Updating configuration data %s", RedactedConf(self.conf)) self.hass.config_entries.async_update_entry( entry=self.config_entry, data=self.conf ) # Reload the config entry otherwise devices will remain unavailable self.hass.async_create_task( self.hass.config_entries.async_reload(self.config_entry.entry_id) ) if not self.devices_conf: _LOGGER.debug("No configured appliances %s", RedactedConf(self.conf)) return self.async_abort(reason="no_configured_devices") _LOGGER.debug("Creating configuration data %s", RedactedConf(self.conf)) return self.async_create_entry(title=NAME, data=self.conf) async def _async_step_appliance( # pylint: disable=too-many-locals self: _MideaFlow, step_id: str, user_input: dict[str, Any] | None = None, ) -> FlowResult: """Manage an appliances""" self.errors.clear() self.error_cause = "" appliance = self.appliances[self.appliance_idx] device_conf = self.devices_conf[self.appliance_idx] discovery_mode = device_conf.get(CONF_DISCOVERY, DEFAULT_DISCOVERY_MODE) ttl = device_conf.get(CONF_TTL, DEFAULT_TTL) ip_address = appliance.address or UNKNOWN_IP if user_input is not None: try: ip_address = user_input.get( CONF_IP_ADDRESS, device_conf.get(CONF_IP_ADDRESS, UNKNOWN_IP) ) self._check_ip_address_unique(ip_address) discovery_mode = user_input.get(CONF_DISCOVERY, discovery_mode) if discovery_mode not in [ DISCOVERY_WAIT, DISCOVERY_LAN, DISCOVERY_IGNORE, DISCOVERY_CLOUD, ]: discovery_mode = ( DISCOVERY_LAN if address_ok(ip_address) else DISCOVERY_CLOUD ) device_conf[CONF_DISCOVERY] = discovery_mode device_conf[CONF_TTL] = user_input.get(CONF_TTL, ttl) appliance.address = ip_address appliance.name = user_input.get(CONF_NAME, appliance.name) appliance.token = user_input.get(CONF_TOKEN, "") appliance.key = user_input.get(CONF_TOKEN_KEY, "") if not self.cloud: await self.hass.async_add_executor_job(self._connect_to_cloud) discovered = await self.hass.async_add_executor_job( self._validate_appliance, appliance, device_conf, ) self.discovered_appliances[self.appliance_idx] = discovered if not self.indexes_to_process: self._update_appliances_after_flow() return await self._async_add_entry() self.appliance_idx = self.indexes_to_process.pop(0) appliance = self.appliances[self.appliance_idx] device_conf = self.devices_conf[self.appliance_idx] ip_address = appliance.address or UNKNOWN_IP user_input = None discovery_mode = DEFAULT_DISCOVERY_MODE ttl = DEFAULT_TTL except Exception as ex: # pylint: disable=broad-except self._process_exception(ex) name = appliance.name extra = { "index": str(self.appliance_idx + 1), "count": str(len(self.appliances)), "serial_number": appliance.serial_number, } placeholders = self._placeholders(appliance, extra) schema_arg = { "name": name, "address": device_conf.get(CONF_IP_ADDRESS, ip_address), "token": device_conf.get(CONF_TOKEN, appliance.token), "token_key": device_conf.get(CONF_TOKEN_KEY, appliance.key), "ttl": device_conf.get(CONF_TTL, ttl), "discovery_mode": device_conf.get(CONF_DISCOVERY, discovery_mode), } schema = _appliance_schema(**schema_arg) return self.async_show_form( step_id=step_id, data_schema=schema, description_placeholders=placeholders, errors=self.errors, last_step=len(self.indexes_to_process) == 0, ) def _check_ip_address_unique(self, ip_address) -> None: if address_ok(ip_address): for i in range(self.appliance_idx): if ( self.devices_conf[i].get(CONF_IP_ADDRESS) == ip_address or ip_address == self.appliances[i].address ): raise _FlowException( "duplicate_ip_provided", self.appliances[i].name ) def _update_appliances_after_flow(self) -> None: for i, discovered in enumerate(self.discovered_appliances): if discovered: old_address = self.appliances[i].address self.appliances[i].update(discovered) if not discovered.address: self.appliances[i].address = old_address def _placeholders( self: _MideaFlow, appliance: LanDevice = None, extra: dict[str, str] = None ) -> dict[str, str]: extra = extra or {} placeholders = { "cause": self.error_cause or "", **extra, } if appliance: placeholders[ATTR_ID] = ( appliance.serial_number or f"{appliance.appliance_id} (Missing S/N)" ) placeholders[ATTR_NAME] = appliance.name return placeholders def _get_broadcast_addresses(user_input: dict[str, Any]) -> list[str]: address_entry = str(user_input.get(CONF_BROADCAST_ADDRESS, "")) addresses = [LOCAL_BROADCAST] specified_addresses = [ addr.strip() for addr in address_entry.split(",") if addr.strip() ] for addr in specified_addresses: _LOGGER.debug("Trying IPv4 %s", addr) try: IPv4Network(addr) addresses.append(addr) except ValueError as ex: raise _FlowException("invalid_ip_range", str(ex)) from ex except Exception as ex: _LOGGER.debug("Invalid IP address %s", addr, exc_info=True) raise _FlowException("invalid_ip_range", addr) from ex return addresses class _FlowException(Exception): def __init__(self, message, cause: str = None) -> None: super().__init__() self.message = message self.cause = cause # pylint: disable=too-many-instance-attributes class MideaConfigFlow(ConfigFlow, _MideaFlow, domain=DOMAIN): """Configuration flow for Midea dehumidifiers on local network uses discovery based on Midea cloud, so it first requires credentials for it. If some appliances are registered in the cloud, but not discovered, configuration flow will prompt for additional information. """ VERSION = CURRENT_CONFIG_VERSION def __init__(self) -> None: super().__init__() self.discovered_appliances: list[LanDevice | None] = [] self.appliances: list[LanDevice] = [] self.config_entry: ConfigEntry | None = None self.advanced_settings = False @staticmethod @callback def async_get_options_flow( config_entry: ConfigEntry, ) -> OptionsFlow: """Define the config flow to handle options.""" return MideaOptionsFlow(config_entry) def _connect_and_discover(self: MideaConfigFlow) -> None: """Validates that cloud credentials are valid and discovers local appliances""" self._connect_to_cloud() conf_addresses = self.conf.get(CONF_BROADCAST_ADDRESS, []) if isinstance(conf_addresses, str): conf_addresses = [conf_addresses] addresses = [ str(IPv4Network(addr).broadcast_address) for addr in conf_addresses ] self.appliances.clear() self.appliances += self.client.find_appliances(self.cloud, addresses) self.devices_conf = [{} for _ in self.appliances] async def _validate_discovery_phase( self, user_input: dict[str, Any] | None ) -> FlowResult: assert user_input is not None self.conf[CONF_USERNAME] = user_input[CONF_USERNAME] self.conf[CONF_PASSWORD] = user_input[CONF_PASSWORD] if self.advanced_settings: assert self.conf is not None self.conf[CONF_MOBILE_APP] = user_input.get(CONF_MOBILE_APP, DEFAULT_APP) self.conf[CONF_INCLUDE] = user_input[CONF_INCLUDE] self.conf[CONF_SCAN_INTERVAL] = user_input[CONF_SCAN_INTERVAL] self.conf[CONF_DEBUG] = user_input[CONF_DEBUG] self.conf[CONF_BROADCAST_ADDRESS] = _get_broadcast_addresses(user_input) else: self.conf[CONF_MOBILE_APP] = user_input.get(CONF_MOBILE_APP, DEFAULT_APP) if user_input.get(CONF_ADVANCED_SETTINGS): return await self.async_step_advanced_settings() self.conf[CONF_BROADCAST_ADDRESS] = [] self.conf[CONF_INCLUDE] = [APPLIANCE_TYPE_DEHUMIDIFIER] self.conf[CONF_SCAN_INTERVAL] = DEFAULT_SCAN_INTERVAL if self.conf.get(CONF_DEBUG, False): await self.client.async_debug_mode(True) await self.hass.async_add_executor_job(self._connect_and_discover) self.indexes_to_process = [ index for index, appliance in enumerate(self.appliances) if supported_appliance(self.conf, appliance) and not address_ok(appliance.address) ] if self.indexes_to_process: self.appliance_idx = self.indexes_to_process.pop(0) self.discovered_appliances = [None] * len(self.devices_conf) return await self.async_step_unreachable_appliance() return await self._async_add_entry() async def _do_validate(self, user_input: dict[str, Any]) -> FlowResult | None: try: return await self._validate_discovery_phase(user_input) except Exception as ex: # pylint: disable=broad-except self._process_exception(ex) return None async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> FlowResult: self.advanced_settings = False if self._async_current_entries(): return self.async_abort(reason="single_instance_allowed") self.errors.clear() self.error_cause = "" username = DEFAULT_USERNAME password = DEFAULT_PASSWORD app = DEFAULT_APP if user_input is not None: username = user_input.get(CONF_USERNAME, username) password = user_input.get(CONF_PASSWORD, password) app = user_input.get(CONF_MOBILE_APP, app) res = await self._do_validate(user_input) if res: return res return self.async_show_form( step_id="user", data_schema=_user_schema(username=username, password=password, app=app), description_placeholders=self._placeholders(), errors=self.errors, ) async def async_step_advanced_settings( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Step for managing advanced settings""" self.errors = {} self.error_cause = "" self.advanced_settings = True if user_input is not None: if res := await self._do_validate(user_input): return res else: user_input = {} username = user_input.get( CONF_USERNAME, self.conf.get(CONF_USERNAME, DEFAULT_USERNAME) ) password = user_input.get( CONF_PASSWORD, self.conf.get(CONF_PASSWORD, DEFAULT_PASSWORD) ) app = user_input.get(CONF_MOBILE_APP, DEFAULT_APP) broadcast_addresses = user_input.get( CONF_BROADCAST_ADDRESS, ",".join(self.conf.get(CONF_BROADCAST_ADDRESS, [])) ) return self.async_show_form( step_id="advanced_settings", data_schema=_advanced_settings_schema( username=username, password=password, app=app, broadcast_address=broadcast_addresses, ), description_placeholders=self._placeholders(), errors=self.errors, ) async def async_step_unreachable_appliance( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Manage the appliances that were not discovered automatically on LAN.""" return await self._async_step_appliance( step_id="unreachable_appliance", user_input=user_input, ) async def _async_add_entry(self) -> FlowResult: assert self.conf is not None self.config_entry = await self.async_set_unique_id(self.conf[CONF_USERNAME]) return await super()._async_add_entry() async def async_step_reauth(self, config) -> FlowResult: """Handle reauthorization request from Abode.""" self.conf = {**config} return await self.async_step_reauth_confirm() async def async_step_reauth_confirm( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Handle reauthorization flow.""" self.errors.clear() password = "" username = self.conf.get(CONF_USERNAME, "") app = self.conf.get(CONF_MOBILE_APP, DEFAULT_APP) if user_input is not None: extra_conf = { CONF_USERNAME: user_input.get(CONF_USERNAME, ""), CONF_PASSWORD: user_input.get(CONF_PASSWORD, ""), CONF_MOBILE_APP: user_input.get(CONF_MOBILE_APP, app), } try: await self.hass.async_add_executor_job( self._connect_to_cloud, extra_conf ) except Exception as ex: # pylint: disable=broad-except self._process_exception(ex) else: self.conf[CONF_USERNAME] = username self.conf[CONF_PASSWORD] = password self.conf[CONF_MOBILE_APP] = app return await self._async_add_entry() return self.async_show_form( step_id="reauth_confirm", data_schema=_reauth_schema( username=username, password=password, ), description_placeholders=self._placeholders(), errors=self.errors, ) class MideaOptionsFlow(OptionsFlow, _MideaFlow): """Handle Midea options flow.""" def __init__(self, config_entry: ConfigEntry) -> None: """Initialize Midea options flow.""" super().__init__() self.config_entry = config_entry self.conf = {**config_entry.data} self.devices_conf = self.conf.get(CONF_DEVICES, []) async def async_step_init( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Starts options flow""" self._build_appliance_list() return await self.async_step_appliance(user_input) async def async_step_appliance( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Options for an appliance""" return await self._async_step_appliance( step_id="appliance", user_input=user_input, ) def _build_appliance_list(self) -> None: assert self.config_entry hub: Hub = self.hass.data[DOMAIN][self.config_entry.entry_id] self.appliances.clear() self.devices_conf = self.conf[CONF_DEVICES] for device in self.devices_conf: for coord in hub.coordinators: if device[CONF_UNIQUE_ID] == coord.appliance.serial_number: self.appliances.append(coord.appliance) break else: appliance = LanDevice( appliance_id=device[CONF_ID], serial_number=device[CONF_UNIQUE_ID], appliance_type=device[CONF_TYPE], ) appliance.name = device[CONF_NAME] appliance.address = device.get(CONF_IP_ADDRESS, UNKNOWN_IP) self.appliances.append(appliance) self.indexes_to_process = list(range(len(self.appliances))) self.appliance_idx = self.indexes_to_process.pop(0) self.discovered_appliances = [None] * len(self.devices_conf)