"""Config flow for ZHA.""" from __future__ import annotations from abc import abstractmethod import collections from contextlib import suppress import json from typing import Any import serial.tools.list_ports from serial.tools.list_ports_common import ListPortInfo import voluptuous as vol from zha.application.const import RadioType import zigpy.backups from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH from zigpy.exceptions import CannotWriteNetworkSettings, DestructiveWriteNetworkSettings from homeassistant.components import onboarding, usb from homeassistant.components.file_upload import process_uploaded_file from homeassistant.components.hassio import AddonError, AddonState from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware from homeassistant.config_entries import ( SOURCE_IGNORE, SOURCE_ZEROCONF, ConfigEntry, ConfigEntryBaseFlow, ConfigEntryState, ConfigFlow, ConfigFlowResult, OperationNotAllowed, OptionsFlow, ) from homeassistant.const import CONF_NAME from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import AbortFlow from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.hassio import is_hassio from homeassistant.helpers.selector import FileSelector, FileSelectorConfig from homeassistant.helpers.service_info.usb import UsbServiceInfo from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo from homeassistant.util import dt as dt_util from .const import CONF_BAUDRATE, CONF_FLOW_CONTROL, CONF_RADIO_TYPE, DOMAIN from .helpers import get_zha_gateway from .radio_manager import ( DEVICE_SCHEMA, HARDWARE_DISCOVERY_SCHEMA, RECOMMENDED_RADIOS, ProbeResult, ZhaRadioManager, ) CONF_MANUAL_PATH = "Enter Manually" DECONZ_DOMAIN = "deconz" # The ZHA config flow takes different branches depending on if you are migrating to a # new adapter via discovery or setting it up from scratch # For the fast path, we automatically migrate everything and restore the most recent backup MIGRATION_STRATEGY_RECOMMENDED = "migration_strategy_recommended" MIGRATION_STRATEGY_ADVANCED = "migration_strategy_advanced" # Similarly, setup follows the same approach: we create a new network SETUP_STRATEGY_RECOMMENDED = "setup_strategy_recommended" SETUP_STRATEGY_ADVANCED = "setup_strategy_advanced" # For the advanced paths, we allow users to pick how to form a network: form a brand new # network, use the settings currently on the stick, restore from a database backup, or # restore from a JSON backup FORMATION_STRATEGY = "formation_strategy" FORMATION_FORM_NEW_NETWORK = "form_new_network" FORMATION_FORM_INITIAL_NETWORK = "form_initial_network" FORMATION_REUSE_SETTINGS = "reuse_settings" FORMATION_CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup" FORMATION_UPLOAD_MANUAL_BACKUP = "upload_manual_backup" CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup" OVERWRITE_COORDINATOR_IEEE = "overwrite_coordinator_ieee" OPTIONS_INTENT_MIGRATE = "intent_migrate" OPTIONS_INTENT_RECONFIGURE = "intent_reconfigure" UPLOADED_BACKUP_FILE = "uploaded_backup_file" REPAIR_MY_URL = "https://my.home-assistant.io/redirect/repairs/" LEGACY_ZEROCONF_PORT = 6638 LEGACY_ZEROCONF_ESPHOME_API_PORT = 6053 ZEROCONF_SERVICE_TYPE = "_zigbee-coordinator._tcp.local." ZEROCONF_PROPERTIES_SCHEMA = vol.Schema( { vol.Required("radio_type"): vol.All(str, vol.In([t.name for t in RadioType])), vol.Required("serial_number"): str, }, extra=vol.ALLOW_EXTRA, ) def _format_backup_choice( backup: zigpy.backups.NetworkBackup, *, pan_ids: bool = True ) -> str: """Format network backup info into a short piece of text.""" if not pan_ids: return dt_util.as_local(backup.backup_time).strftime("%c") identifier = ( # PAN ID f"{str(backup.network_info.pan_id)[2:]}" # EPID f":{str(backup.network_info.extended_pan_id).replace(':', '')}" ).lower() return f"{dt_util.as_local(backup.backup_time).strftime('%c')} ({identifier})" async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]: """List all serial ports, including the Yellow radio and the multi-PAN addon.""" ports: list[ListPortInfo] = [] ports.extend(await hass.async_add_executor_job(serial.tools.list_ports.comports)) # Add useful info to the Yellow's serial port selection screen try: yellow_hardware.async_info(hass) except HomeAssistantError: pass else: # PySerial does not properly handle the Yellow's serial port with the CM5 # so we manually include it port = ListPortInfo(device="/dev/ttyAMA1", skip_link_detection=True) port.description = "Yellow Zigbee module" port.manufacturer = "Nabu Casa" ports = [p for p in ports if not p.device.startswith("/dev/ttyAMA")] ports.insert(0, port) if is_hassio(hass): # Present the multi-PAN addon as a setup option, if it's available multipan_manager = ( await silabs_multiprotocol_addon.get_multiprotocol_addon_manager(hass) ) try: addon_info = await multipan_manager.async_get_addon_info() except (AddonError, KeyError): addon_info = None if addon_info is not None and addon_info.state != AddonState.NOT_INSTALLED: addon_port = ListPortInfo( device=silabs_multiprotocol_addon.get_zigbee_socket(), skip_link_detection=True, ) addon_port.description = "Multiprotocol add-on" addon_port.manufacturer = "Nabu Casa" ports.append(addon_port) return ports class BaseZhaFlow(ConfigEntryBaseFlow): """Mixin for common ZHA flow steps and forms.""" _hass: HomeAssistant _title: str def __init__(self) -> None: """Initialize flow instance.""" super().__init__() self._hass = None # type: ignore[assignment] self._radio_mgr = ZhaRadioManager() @property def hass(self) -> HomeAssistant: """Return hass.""" return self._hass @hass.setter def hass(self, hass: HomeAssistant) -> None: """Set hass.""" self._hass = hass self._radio_mgr.hass = hass def _get_config_entry_data(self) -> dict[str, Any]: """Extract ZHA config entry data from the radio manager.""" assert self._radio_mgr.radio_type is not None assert self._radio_mgr.device_path is not None assert self._radio_mgr.device_settings is not None return { CONF_DEVICE: DEVICE_SCHEMA( { **self._radio_mgr.device_settings, CONF_DEVICE_PATH: self._radio_mgr.device_path, } ), CONF_RADIO_TYPE: self._radio_mgr.radio_type.name, } @abstractmethod async def _async_create_radio_entry(self) -> ConfigFlowResult: """Create a config entry with the current flow state.""" async def async_step_choose_serial_port( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Choose a serial port.""" ports = await list_serial_ports(self.hass) list_of_ports = [ f"{p}{', s/n: ' + p.serial_number if p.serial_number else ''}" + (f" - {p.manufacturer}" if p.manufacturer else "") for p in ports ] if not list_of_ports: return await self.async_step_manual_pick_radio_type() list_of_ports.append(CONF_MANUAL_PATH) if user_input is not None: user_selection = user_input[CONF_DEVICE_PATH] if user_selection == CONF_MANUAL_PATH: return await self.async_step_manual_pick_radio_type() port = ports[list_of_ports.index(user_selection)] self._radio_mgr.device_path = port.device probe_result = await self._radio_mgr.detect_radio_type() if probe_result == ProbeResult.WRONG_FIRMWARE_INSTALLED: return self.async_abort( reason="wrong_firmware_installed", description_placeholders={"repair_url": REPAIR_MY_URL}, ) if probe_result == ProbeResult.PROBING_FAILED: # Did not autodetect anything, proceed to manual selection return await self.async_step_manual_pick_radio_type() self._title = ( f"{port.description}{', s/n: ' + port.serial_number if port.serial_number else ''}" f" - {port.manufacturer}" if port.manufacturer else "" ) return await self.async_step_verify_radio() # Pre-select the currently configured port default_port: vol.Undefined | str = vol.UNDEFINED if self._radio_mgr.device_path is not None: for description, port in zip(list_of_ports, ports, strict=False): if port.device == self._radio_mgr.device_path: default_port = description break else: default_port = CONF_MANUAL_PATH schema = vol.Schema( { vol.Required(CONF_DEVICE_PATH, default=default_port): vol.In( list_of_ports ) } ) return self.async_show_form(step_id="choose_serial_port", data_schema=schema) async def async_step_manual_pick_radio_type( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Manually select the radio type.""" if user_input is not None: self._radio_mgr.radio_type = RadioType.get_by_description( user_input[CONF_RADIO_TYPE] ) return await self.async_step_manual_port_config() # Pre-select the current radio type default: vol.Undefined | str = vol.UNDEFINED if self._radio_mgr.radio_type is not None: default = self._radio_mgr.radio_type.description schema = { vol.Required(CONF_RADIO_TYPE, default=default): vol.In(RadioType.list()) } return self.async_show_form( step_id="manual_pick_radio_type", data_schema=vol.Schema(schema), ) async def async_step_manual_port_config( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Enter port settings specific for this type of radio.""" assert self._radio_mgr.radio_type is not None errors = {} if user_input is not None: self._title = user_input[CONF_DEVICE_PATH] self._radio_mgr.device_path = user_input[CONF_DEVICE_PATH] self._radio_mgr.device_settings = DEVICE_SCHEMA( { CONF_DEVICE_PATH: self._radio_mgr.device_path, CONF_BAUDRATE: user_input[CONF_BAUDRATE], # `None` shows up as the empty string in the frontend CONF_FLOW_CONTROL: ( user_input[CONF_FLOW_CONTROL] if user_input[CONF_FLOW_CONTROL] != "none" else None ), } ) if await self._radio_mgr.radio_type.controller.probe(user_input): return await self.async_step_verify_radio() errors["base"] = "cannot_connect" device_settings = self._radio_mgr.device_settings or {} return self.async_show_form( step_id="manual_port_config", data_schema=vol.Schema( { vol.Required( CONF_DEVICE_PATH, default=self._radio_mgr.device_path or vol.UNDEFINED, ): str, vol.Required( CONF_BAUDRATE, default=device_settings.get(CONF_BAUDRATE) or 115200, ): int, vol.Required( CONF_FLOW_CONTROL, default=device_settings.get(CONF_FLOW_CONTROL) or "none", ): vol.In(["hardware", "software", "none"]), } ), errors=errors, ) async def async_step_verify_radio( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Add a warning step to dissuade the use of deprecated radios.""" assert self._radio_mgr.radio_type is not None await self._radio_mgr.async_read_backups_from_database() # Skip this step if we are using a recommended radio if user_input is not None or self._radio_mgr.radio_type in RECOMMENDED_RADIOS: # ZHA disables the single instance check and will decide at runtime if we # are migrating or setting up from scratch if self.hass.config_entries.async_entries(DOMAIN, include_ignore=False): return await self.async_step_choose_migration_strategy() return await self.async_step_choose_setup_strategy() return self.async_show_form( step_id="verify_radio", description_placeholders={ CONF_NAME: self._radio_mgr.radio_type.description, "docs_recommended_adapters_url": ( "https://www.home-assistant.io/integrations/zha/#recommended-zigbee-radio-adapters-and-modules" ), }, ) async def async_step_choose_setup_strategy( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Choose how to set up the integration from scratch.""" # Allow onboarding for new users to just create a new network automatically if ( not onboarding.async_is_onboarded(self.hass) and not self.hass.config_entries.async_entries(DOMAIN, include_ignore=False) and not self._radio_mgr.backups ): return await self.async_step_setup_strategy_recommended() return self.async_show_menu( step_id="choose_setup_strategy", menu_options=[ SETUP_STRATEGY_RECOMMENDED, SETUP_STRATEGY_ADVANCED, ], ) async def async_step_setup_strategy_recommended( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Recommended setup strategy: form a brand-new network.""" return await self.async_step_form_new_network() async def async_step_setup_strategy_advanced( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Advanced setup strategy: let the user choose.""" return await self.async_step_choose_formation_strategy() async def async_step_choose_migration_strategy( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Choose how to deal with the current radio's settings during migration.""" return self.async_show_menu( step_id="choose_migration_strategy", menu_options=[ MIGRATION_STRATEGY_RECOMMENDED, MIGRATION_STRATEGY_ADVANCED, ], ) async def async_step_migration_strategy_recommended( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Recommended migration strategy: automatically migrate everything.""" # Assume the most recent backup is the correct one self._radio_mgr.chosen_backup = self._radio_mgr.backups[0] return await self.async_step_maybe_reset_old_radio() async def async_step_maybe_reset_old_radio( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Erase the old radio's network settings before migration.""" # Like in the options flow, pull the correct settings from the config entry config_entries = self.hass.config_entries.async_entries( DOMAIN, include_ignore=False ) if config_entries: assert len(config_entries) == 1 config_entry = config_entries[0] # Unload ZHA before connecting to the old adapter with suppress(OperationNotAllowed): await self.hass.config_entries.async_unload(config_entry.entry_id) # Create a radio manager to connect to the old stick to reset it temp_radio_mgr = ZhaRadioManager() temp_radio_mgr.hass = self.hass temp_radio_mgr.device_path = config_entry.data[CONF_DEVICE][ CONF_DEVICE_PATH ] temp_radio_mgr.device_settings = config_entry.data[CONF_DEVICE] temp_radio_mgr.radio_type = RadioType[config_entry.data[CONF_RADIO_TYPE]] await temp_radio_mgr.async_reset_adapter() return await self.async_step_maybe_confirm_ezsp_restore() async def async_step_migration_strategy_advanced( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Advanced migration strategy: let the user choose.""" return await self.async_step_choose_formation_strategy() async def async_step_choose_formation_strategy( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Choose how to deal with the current radio's settings.""" await self._radio_mgr.async_load_network_settings() strategies = [] # Check if we have any automatic backups *and* if the backups differ from # the current radio settings, if they exist (since restoring would be redundant) if self._radio_mgr.backups and ( self._radio_mgr.current_settings is None or any( not backup.is_compatible_with(self._radio_mgr.current_settings) for backup in self._radio_mgr.backups ) ): strategies.append(CHOOSE_AUTOMATIC_BACKUP) if self._radio_mgr.current_settings is not None: strategies.append(FORMATION_REUSE_SETTINGS) strategies.append(FORMATION_UPLOAD_MANUAL_BACKUP) # Do not show "erase network settings" if there are none to erase if self._radio_mgr.current_settings is None: strategies.append(FORMATION_FORM_INITIAL_NETWORK) else: strategies.append(FORMATION_FORM_NEW_NETWORK) # Automatically form a new network if we're onboarding with a brand new radio if not onboarding.async_is_onboarded(self.hass) and set(strategies) == { FORMATION_UPLOAD_MANUAL_BACKUP, FORMATION_FORM_INITIAL_NETWORK, }: return await self.async_step_form_initial_network() # Otherwise, let the user choose return self.async_show_menu( step_id="choose_formation_strategy", menu_options=strategies, ) async def async_step_reuse_settings( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Reuse the existing network settings on the stick.""" return await self._async_create_radio_entry() async def async_step_form_initial_network( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Form an initial network.""" # This step exists only for translations, it does nothing new return await self.async_step_form_new_network(user_input) async def async_step_form_new_network( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Form a brand-new network.""" await self._radio_mgr.async_form_network() return await self._async_create_radio_entry() def _parse_uploaded_backup( self, uploaded_file_id: str ) -> zigpy.backups.NetworkBackup: """Read and parse an uploaded backup JSON file.""" with process_uploaded_file(self.hass, uploaded_file_id) as file_path: contents = file_path.read_text() return zigpy.backups.NetworkBackup.from_dict(json.loads(contents)) async def async_step_upload_manual_backup( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Upload and restore a coordinator backup JSON file.""" errors = {} if user_input is not None: try: self._radio_mgr.chosen_backup = await self.hass.async_add_executor_job( self._parse_uploaded_backup, user_input[UPLOADED_BACKUP_FILE] ) except ValueError: errors["base"] = "invalid_backup_json" else: return await self.async_step_maybe_reset_old_radio() return self.async_show_form( step_id="upload_manual_backup", data_schema=vol.Schema( { vol.Required(UPLOADED_BACKUP_FILE): FileSelector( FileSelectorConfig(accept=".json,application/json") ) } ), errors=errors, ) async def async_step_choose_automatic_backup( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Choose an automatic backup.""" if self.show_advanced_options: # Always show the PAN IDs when in advanced mode choices = [ _format_backup_choice(backup, pan_ids=True) for backup in self._radio_mgr.backups ] else: # Only show the PAN IDs for multiple backups taken on the same day num_backups_on_date = collections.Counter( backup.backup_time.date() for backup in self._radio_mgr.backups ) choices = [ _format_backup_choice( backup, pan_ids=(num_backups_on_date[backup.backup_time.date()] > 1) ) for backup in self._radio_mgr.backups ] if user_input is not None: index = choices.index(user_input[CHOOSE_AUTOMATIC_BACKUP]) self._radio_mgr.chosen_backup = self._radio_mgr.backups[index] return await self.async_step_maybe_reset_old_radio() return self.async_show_form( step_id="choose_automatic_backup", data_schema=vol.Schema( { vol.Required(CHOOSE_AUTOMATIC_BACKUP, default=choices[0]): vol.In( choices ), } ), ) async def async_step_maybe_confirm_ezsp_restore( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Confirm restore for EZSP radios that require permanent IEEE writes.""" if user_input is not None: if user_input[OVERWRITE_COORDINATOR_IEEE]: # On confirmation, overwrite destructively try: await self._radio_mgr.restore_backup(overwrite_ieee=True) except CannotWriteNetworkSettings as exc: return self.async_abort( reason="cannot_restore_backup", description_placeholders={"error": str(exc)}, ) return await self._async_create_radio_entry() # On rejection, explain why we can't restore return self.async_abort(reason="cannot_restore_backup_no_ieee_confirm") # On first attempt, just try to restore nondestructively try: await self._radio_mgr.restore_backup() except DestructiveWriteNetworkSettings: # Restore cannot happen automatically, we need to ask for permission pass except CannotWriteNetworkSettings as exc: return self.async_abort( reason="cannot_restore_backup", description_placeholders={"error": str(exc)}, ) else: return await self._async_create_radio_entry() # If it fails, show the form return self.async_show_form( step_id="maybe_confirm_ezsp_restore", data_schema=vol.Schema( {vol.Required(OVERWRITE_COORDINATOR_IEEE, default=True): bool} ), ) class ZhaConfigFlowHandler(BaseZhaFlow, ConfigFlow, domain=DOMAIN): """Handle a config flow.""" VERSION = 4 async def _set_unique_id_and_update_ignored_flow( self, unique_id: str, device_path: str ) -> None: """Set the flow's unique ID and update the device path in an ignored flow.""" current_entry = await self.async_set_unique_id(unique_id) # Only update the current entry if it is an ignored discovery if current_entry and current_entry.source == SOURCE_IGNORE: self._abort_if_unique_id_configured( updates={ CONF_DEVICE: { **current_entry.data.get(CONF_DEVICE, {}), CONF_DEVICE_PATH: device_path, }, } ) @staticmethod @callback def async_get_options_flow( config_entry: ConfigEntry, ) -> OptionsFlow: """Create the options flow.""" return ZhaOptionsFlowHandler(config_entry) async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Handle a ZHA config flow start.""" return await self.async_step_choose_serial_port(user_input) async def async_step_confirm( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Confirm a discovery.""" self._set_confirm_only() zha_config_entries = self.hass.config_entries.async_entries( DOMAIN, include_ignore=False ) if self._radio_mgr.device_path is not None: # Ensure the radio manager device path is unique and will match ZHA's try: self._radio_mgr.device_path = await self.hass.async_add_executor_job( usb.get_serial_by_id, self._radio_mgr.device_path ) except OSError as error: raise AbortFlow( reason="cannot_resolve_path", description_placeholders={"path": self._radio_mgr.device_path}, ) from error # mDNS discovery can advertise the same adapter on multiple IPs or via a # hostname, which should be considered a duplicate current_device_paths = {self._radio_mgr.device_path} if self.source == SOURCE_ZEROCONF: discovery_info = self.init_data current_device_paths |= { f"socket://{ip}:{discovery_info.port}" for ip in discovery_info.ip_addresses } for entry in zha_config_entries: path = entry.data.get(CONF_DEVICE, {}).get(CONF_DEVICE_PATH) # Abort discovery if the device path is already configured if path is not None and path in current_device_paths: return self.async_abort(reason="single_instance_allowed") # Without confirmation, discovery can automatically progress into parts of the # config flow logic that interacts with hardware. if user_input is not None or ( not onboarding.async_is_onboarded(self.hass) and not zha_config_entries ): # Probe the radio type if we don't have one yet if self._radio_mgr.radio_type is None: probe_result = await self._radio_mgr.detect_radio_type() else: probe_result = ProbeResult.RADIO_TYPE_DETECTED if probe_result == ProbeResult.WRONG_FIRMWARE_INSTALLED: return self.async_abort( reason="wrong_firmware_installed", description_placeholders={"repair_url": REPAIR_MY_URL}, ) if probe_result == ProbeResult.PROBING_FAILED: # This path probably will not happen now that we have # more precise USB matching unless there is a problem # with the device return self.async_abort(reason="usb_probe_failed") if self._radio_mgr.device_settings is None: return await self.async_step_manual_port_config() return await self.async_step_verify_radio() return self.async_show_form( step_id="confirm", description_placeholders={CONF_NAME: self._title}, ) async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResult: """Handle usb discovery.""" vid = discovery_info.vid pid = discovery_info.pid serial_number = discovery_info.serial_number manufacturer = discovery_info.manufacturer description = discovery_info.description dev_path = discovery_info.device await self._set_unique_id_and_update_ignored_flow( unique_id=f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}", device_path=dev_path, ) # If they already have a discovery for deconz we ignore the usb discovery as # they probably want to use it there instead if self.hass.config_entries.flow.async_progress_by_handler(DECONZ_DOMAIN): return self.async_abort(reason="not_zha_device") for entry in self.hass.config_entries.async_entries(DECONZ_DOMAIN): if entry.source != SOURCE_IGNORE: return self.async_abort(reason="not_zha_device") self._radio_mgr.device_path = dev_path self._title = description or usb.human_readable_device_name( dev_path, serial_number, manufacturer, description, vid, pid, ) self.context["title_placeholders"] = {CONF_NAME: self._title} return await self.async_step_confirm() async def async_step_zeroconf( self, discovery_info: ZeroconfServiceInfo ) -> ConfigFlowResult: """Handle zeroconf discovery.""" # Transform legacy zeroconf discovery into the new format if discovery_info.type != ZEROCONF_SERVICE_TYPE: port = discovery_info.port or LEGACY_ZEROCONF_PORT name = discovery_info.name # Fix incorrect port for older TubesZB devices if "tube" in name and port == LEGACY_ZEROCONF_ESPHOME_API_PORT: port = LEGACY_ZEROCONF_PORT # Determine the radio type if "radio_type" in discovery_info.properties: radio_type = discovery_info.properties["radio_type"] elif "efr32" in name: radio_type = RadioType.ezsp.name elif "zigate" in name: radio_type = RadioType.zigate.name else: radio_type = RadioType.znp.name fallback_title = name.split("._", 1)[0] title = discovery_info.properties.get("name", fallback_title) discovery_info = ZeroconfServiceInfo( ip_address=discovery_info.ip_address, ip_addresses=discovery_info.ip_addresses, port=port, hostname=discovery_info.hostname, type=ZEROCONF_SERVICE_TYPE, name=f"{title}.{ZEROCONF_SERVICE_TYPE}", properties={ "radio_type": radio_type, # To maintain backwards compatibility "serial_number": discovery_info.hostname.removesuffix(".local."), }, ) try: discovery_props = ZEROCONF_PROPERTIES_SCHEMA(discovery_info.properties) except vol.Invalid: return self.async_abort(reason="invalid_zeroconf_data") radio_type = self._radio_mgr.parse_radio_type(discovery_props["radio_type"]) device_path = f"socket://{discovery_info.host}:{discovery_info.port}" title = discovery_info.name.removesuffix(f".{ZEROCONF_SERVICE_TYPE}") await self._set_unique_id_and_update_ignored_flow( unique_id=discovery_props["serial_number"], device_path=device_path, ) self.context["title_placeholders"] = {CONF_NAME: title} self._title = title self._radio_mgr.device_path = device_path self._radio_mgr.radio_type = radio_type self._radio_mgr.device_settings = DEVICE_SCHEMA( { CONF_DEVICE_PATH: device_path, CONF_BAUDRATE: 115200, CONF_FLOW_CONTROL: None, } ) return await self.async_step_confirm() async def async_step_hardware( self, data: dict[str, Any] | None = None ) -> ConfigFlowResult: """Handle hardware flow.""" try: discovery_data = HARDWARE_DISCOVERY_SCHEMA(data) except vol.Invalid: return self.async_abort(reason="invalid_hardware_data") name = discovery_data["name"] radio_type = self._radio_mgr.parse_radio_type(discovery_data["radio_type"]) device_settings = discovery_data["port"] device_path = device_settings[CONF_DEVICE_PATH] await self._set_unique_id_and_update_ignored_flow( unique_id=f"{name}_{radio_type.name}_{device_path}", device_path=device_path, ) self._title = name self._radio_mgr.radio_type = radio_type self._radio_mgr.device_path = device_path self._radio_mgr.device_settings = device_settings self.context["title_placeholders"] = {CONF_NAME: name} return await self.async_step_confirm() async def _async_create_radio_entry(self) -> ConfigFlowResult: """Create a config entry with the current flow state.""" # ZHA is still single instance only, even though we use discovery to allow for # migrating to a new radio zha_config_entries = self.hass.config_entries.async_entries( DOMAIN, include_ignore=False ) data = self._get_config_entry_data() if len(zha_config_entries) == 1: return self.async_update_reload_and_abort( entry=zha_config_entries[0], title=self._title, data=data, reload_even_if_entry_is_unchanged=True, reason="reconfigure_successful", ) if not zha_config_entries: return self.async_create_entry( title=self._title, data=data, ) # This should never be reached return self.async_abort(reason="single_instance_allowed") class ZhaOptionsFlowHandler(BaseZhaFlow, OptionsFlow): """Handle an options flow.""" def __init__(self, config_entry: ConfigEntry) -> None: """Initialize options flow.""" super().__init__() self._radio_mgr.device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH] self._radio_mgr.device_settings = config_entry.data[CONF_DEVICE] self._radio_mgr.radio_type = RadioType[config_entry.data[CONF_RADIO_TYPE]] self._title = config_entry.title async def async_step_init( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Launch the options flow.""" if user_input is not None: # Perform a backup first try: zha_gateway = get_zha_gateway(self.hass) except ValueError: pass else: # The backup itself will be stored in `zigbee.db`, which the radio # manager will read when the class is initialized application_controller = zha_gateway.application_controller await application_controller.backups.create_backup(load_devices=True) # Then unload the integration with suppress(OperationNotAllowed): # OperationNotAllowed: ZHA is not running await self.hass.config_entries.async_unload(self.config_entry.entry_id) return await self.async_step_prompt_migrate_or_reconfigure() return self.async_show_form(step_id="init") async def async_step_prompt_migrate_or_reconfigure( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Confirm if we are migrating adapters or just re-configuring.""" return self.async_show_menu( step_id="prompt_migrate_or_reconfigure", menu_options=[ OPTIONS_INTENT_RECONFIGURE, OPTIONS_INTENT_MIGRATE, ], ) async def async_step_intent_reconfigure( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Virtual step for when the user is reconfiguring the integration.""" return await self.async_step_choose_serial_port() async def async_step_intent_migrate( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Confirm the user wants to reset their current radio.""" if user_input is not None: await self._radio_mgr.async_reset_adapter() return await self.async_step_instruct_unplug() return self.async_show_form(step_id="intent_migrate") async def async_step_instruct_unplug( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Instruct the user to unplug the current radio, if possible.""" if user_input is not None: # Now that the old radio is gone, we can scan for serial ports again return await self.async_step_choose_serial_port() return self.async_show_form(step_id="instruct_unplug") async def _async_create_radio_entry(self): """Re-implementation of the base flow's final step to update the config.""" # Avoid creating both `.options` and `.data` by directly writing `data` here self.hass.config_entries.async_update_entry( entry=self.config_entry, data=self._get_config_entry_data(), options=self.config_entry.options, ) # Reload ZHA after we finish await self.hass.config_entries.async_setup(self.config_entry.entry_id) # Intentionally do not set `data` to avoid creating `options`, we set it above return self.async_create_entry(title=self._title, data={}) def async_remove(self): """Maybe reload ZHA if the flow is aborted.""" if self.config_entry.state not in ( ConfigEntryState.SETUP_ERROR, ConfigEntryState.NOT_LOADED, ): return self.hass.async_create_task( self.hass.config_entries.async_setup(self.config_entry.entry_id) )