core/homeassistant/components/zha/config_flow.py

"""Config flow for ZHA."""

from __future__ import annotations

from abc import abstractmethod
import asyncio
import collections
from contextlib import suppress
from enum import StrEnum
import json
import os
from typing import Any

import voluptuous as vol
from zha.application.const import RadioType
import zigpy.backups
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH, CONF_NWK_TX_POWER
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_hardware.firmware_config_flow import (
    ZigbeeFlowStrategy,
)
from homeassistant.components.homeassistant_yellow import hardware as yellow_hardware
from homeassistant.components.usb import USBDevice, scan_serial_ports
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, progress_step
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_config_entry_unique_id, 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"

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,
)


class OptionsMigrationIntent(StrEnum):
    """Zigbee options flow intents."""

    MIGRATE = "intent_migrate"
    RECONFIGURE = "intent_reconfigure"


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[USBDevice]:
    """List all serial ports, including the Yellow radio and the multi-PAN addon."""
    ports: list[USBDevice] = []
    ports.extend(await hass.async_add_executor_job(scan_serial_ports))

    # 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 = USBDevice(
            device="/dev/ttyAMA1",
            vid="ffff",  # This is technically not a USB device
            pid="ffff",
            serial_number=None,
            manufacturer="Nabu Casa",
            description="Yellow Zigbee module",
        )

        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 = USBDevice(
                device=silabs_multiprotocol_addon.get_zigbee_socket(),
                vid="ffff",  # This is technically not a USB device
                pid="ffff",
                serial_number=None,
                manufacturer="Nabu Casa",
                description="Silicon Labs Multiprotocol add-on",
            )

            ports.append(addon_port)

    return ports


class BaseZhaFlow(ConfigEntryBaseFlow):
    """Mixin for common ZHA flow steps and forms."""

    _flow_strategy: ZigbeeFlowStrategy | None = None
    _overwrite_ieee_during_restore: bool = False
    _hass: HomeAssistant
    _title: str

    def __init__(self) -> None:
        """Initialize flow instance."""
        super().__init__()

        self._hass = None  # type: ignore[assignment]
        self._radio_mgr = ZhaRadioManager()
        self._restore_backup_task: asyncio.Task[None] | None = None
        self._extra_network_config: dict[str, Any] = {}

    @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)

        # The full `/dev/serial/by-id/` path is too verbose to show
        resolved_paths = {
            p.device: await self.hass.async_add_executor_job(os.path.realpath, p.device)
            for p in ports
        }

        list_of_ports = [
            f"{resolved_paths[p.device]} - {p.description}{', 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(
                self._radio_mgr.device_settings
            ):
                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."""
        if self._flow_strategy == ZigbeeFlowStrategy.RECOMMENDED:
            # Fast path: automatically form a new network
            return await self.async_step_setup_strategy_recommended()
        if self._flow_strategy == ZigbeeFlowStrategy.ADVANCED:
            # Advanced path: let the user choose
            return await self.async_step_setup_strategy_advanced()

        # 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."""
        if self._flow_strategy == ZigbeeFlowStrategy.RECOMMENDED:
            # Fast path: automatically migrate everything
            return await self.async_step_migration_strategy_recommended()
        if self._flow_strategy == ZigbeeFlowStrategy.ADVANCED:
            # Advanced path: let the user choose
            return await self.async_step_migration_strategy_advanced()
        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()

    @progress_step()
    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]]

            try:
                await temp_radio_mgr.async_reset_adapter()
            except HomeAssistantError:
                # Old adapter not found or cannot connect, show prompt to plug back in
                return await self.async_step_plug_in_old_radio()

        return await self.async_step_restore_backup()

    async def async_step_plug_in_old_radio(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Prompt user to plug in the old radio if connection fails."""
        config_entries = self.hass.config_entries.async_entries(
            DOMAIN, include_ignore=False
        )

        # Unless the user removes the config entry whilst we try to reset the old radio
        # for a few seconds and then also unplugs it, we will basically never hit this
        if not config_entries:
            return await self.async_step_restore_backup()

        config_entry = config_entries[0]
        old_device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]

        return self.async_show_menu(
            step_id="plug_in_old_radio",
            menu_options=["retry_old_radio", "skip_reset_old_radio"],
            description_placeholders={"device_path": old_device_path},
        )

    async def async_step_retry_old_radio(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Retry connecting to the old radio."""
        return await self.async_step_maybe_reset_old_radio()

    async def async_step_skip_reset_old_radio(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Skip resetting the old radio and continue with migration."""
        return await self.async_step_restore_backup()

    async def async_step_pre_plug_in_new_radio(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Strip user_input before showing "plug in new radio" form."""
        # This step is necessary to prevent `user_input` from being passed through
        return await self.async_step_plug_in_new_radio()

    async def async_step_plug_in_new_radio(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Prompt user to plug in the new radio if connection fails."""
        if user_input is not None:
            # User confirmed, retry now
            return await self.async_step_restore_backup()

        assert self._radio_mgr.device_path is not None

        return self.async_show_form(
            step_id="plug_in_new_radio",
            description_placeholders={"device_path": self._radio_mgr.device_path},
        )

    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)

    @progress_step()
    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(config=self._extra_network_config)

        # Load the newly formed network settings to get the network info
        await self._radio_mgr.async_load_network_settings()
        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_restore_backup(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Restore network backup to new radio."""
        if self._restore_backup_task is None:
            self._restore_backup_task = self.hass.async_create_task(
                self._radio_mgr.restore_backup(
                    overwrite_ieee=self._overwrite_ieee_during_restore
                ),
                "Restore backup",
            )

        if not self._restore_backup_task.done():
            return self.async_show_progress(
                step_id="restore_backup",
                progress_action="restore_backup",
                progress_task=self._restore_backup_task,
            )

        try:
            await self._restore_backup_task
        except DestructiveWriteNetworkSettings:
            # If we cannot restore without overwriting the IEEE, ask for confirmation
            return self.async_show_progress_done(
                next_step_id="pre_confirm_ezsp_ieee_overwrite"
            )
        except HomeAssistantError:
            # User unplugged the new adapter, allow retry
            return self.async_show_progress_done(next_step_id="pre_plug_in_new_radio")
        except CannotWriteNetworkSettings as exc:
            return self.async_abort(
                reason="cannot_restore_backup",
                description_placeholders={"error": str(exc)},
            )
        finally:
            self._restore_backup_task = None

        # Otherwise, proceed to entry creation
        return self.async_show_progress_done(next_step_id="create_entry")

    async def async_step_pre_confirm_ezsp_ieee_overwrite(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Strip user_input before showing confirmation form."""
        # This step is necessary to prevent `user_input` from being passed through
        return await self.async_step_confirm_ezsp_ieee_overwrite()

    async def async_step_confirm_ezsp_ieee_overwrite(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Show confirmation form for EZSP IEEE address overwrite."""
        if user_input is None:
            return self.async_show_form(
                step_id="confirm_ezsp_ieee_overwrite",
                data_schema=vol.Schema(
                    {vol.Required(OVERWRITE_COORDINATOR_IEEE, default=True): bool}
                ),
            )

        if not user_input[OVERWRITE_COORDINATOR_IEEE]:
            return self.async_abort(reason="cannot_restore_backup_no_ieee_confirm")

        self._overwrite_ieee_during_restore = True
        return await self.async_step_restore_backup()

    async def async_step_create_entry(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Create the config entry after successful setup/migration."""

        # This step only exists so that we can create entries from other steps
        return await self._async_create_radio_entry()


class ZhaConfigFlowHandler(BaseZhaFlow, ConfigFlow, domain=DOMAIN):
    """Handle a config flow."""

    VERSION = 5

    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.
        # Ignore Zeroconf discoveries during onboarding, as they may be in use already.
        if user_input is not None or (
            not onboarding.async_is_onboarded(self.hass)
            and not zha_config_entries
            and self.source != SOURCE_ZEROCONF
        ):
            # 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]
        self._flow_strategy = discovery_data.get("flow_strategy")

        if "tx_power" in discovery_data:
            self._extra_network_config[CONF_NWK_TX_POWER] = discovery_data["tx_power"]

        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:
            # Load network settings from the radio to get the EPID
            await self._radio_mgr.async_load_network_settings()
            assert self._radio_mgr.current_settings is not None

            unique_id = get_config_entry_unique_id(
                self._radio_mgr.current_settings.network_info
            )
            await self.async_set_unique_id(unique_id)

            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."""

    _migration_intent: OptionsMigrationIntent

    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=[
                OptionsMigrationIntent.RECONFIGURE,
                OptionsMigrationIntent.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."""
        self._migration_intent = OptionsMigrationIntent.RECONFIGURE
        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."""
        self._migration_intent = OptionsMigrationIntent.MIGRATE
        return await self.async_step_choose_serial_port()

    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."""

        # If we are reconfiguring, the old radio will not be available
        if self._migration_intent is OptionsMigrationIntent.RECONFIGURE:
            return await self.async_step_restore_backup()

        return await super().async_step_maybe_reset_old_radio(user_input)

    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)

        return self.async_abort(reason="reconfigure_successful")

    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)
        )