Skip to content

Migrate ZHA to SerialPortSelector #169099

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
puddly merged 7 commits into from
Apr 27, 2026
Merged

Migrate ZHA to SerialPortSelector #169099

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0bd859d
Migrate ZHA to use the serial port selector
puddly Apr 24, 2026
6dab095
Drop unnecessary title
puddly Apr 24, 2026
71634c0
Clean up unnecessary tests
puddly Apr 24, 2026
90cdf25
Fix pre-commit
puddly Apr 24, 2026
bde7036
Drop remaining title
puddly Apr 24, 2026
d04ce13
Preserve titles for reconfigure flows but drop them everywhere else
puddly Apr 24, 2026
1d7c50d
Reinstate tests to check for empty title
puddly Apr 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 29 additions & 152 deletions homeassistant/components/zha/config_flow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
from enum import StrEnum
import json
import logging
import os
from typing import Any

import voluptuous as vol
Expand All @@ -20,17 +19,9 @@

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 (
SerialDevice,
USBDevice,
async_scan_serial_ports,
)
from homeassistant.config_entries import (
SOURCE_IGNORE,
SOURCE_ZEROCONF,
Expand All @@ -46,8 +37,11 @@
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.selector import (
FileSelector,
FileSelectorConfig,
SerialPortSelector,
)
from homeassistant.helpers.service_info.usb import UsbServiceInfo
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from homeassistant.util import dt as dt_util
Expand All @@ -64,7 +58,6 @@

_LOGGER = logging.getLogger(__name__)

CONF_MANUAL_PATH = "Enter Manually"
DECONZ_DOMAIN = "deconz"

# The ZHA config flow takes different branches depending on if you are migrating to a
Expand Down Expand Up @@ -107,12 +100,6 @@
extra=vol.ALLOW_EXTRA,
)

# USB devices to ignore in serial port selection (non-Zigbee devices)
# Format: (manufacturer, description)
IGNORED_USB_DEVICES = {
("Nabu Casa", "ZWA-2"),
}


class OptionsMigrationIntent(StrEnum):
"""Zigbee options flow intents."""
Expand All @@ -138,83 +125,12 @@ def _format_backup_choice(
return f"{dt_util.as_local(backup.backup_time).strftime('%c')} ({identifier})"


def _format_serial_port_choice(
serial_port: USBDevice | SerialDevice, resolved_paths: dict[str, str]
) -> str:
"""Format a serial port selector entry into a line of text."""
text = resolved_paths[serial_port.device]

if serial_port.description:
text += f" - {serial_port.description}"

if serial_port.serial_number:
text += f", s/n: {serial_port.serial_number}"

if serial_port.manufacturer:
text += f" - {serial_port.manufacturer}"

return text


async def list_serial_ports(hass: HomeAssistant) -> list[USBDevice | SerialDevice]:
"""List all serial ports, including the Yellow radio and the multi-PAN addon."""
ports: list[USBDevice | SerialDevice] = []
ports.extend(await async_scan_serial_ports(hass))

# 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 = SerialDevice(
device="/dev/ttyAMA1",
serial_number=None,
manufacturer="Nabu Casa",
description="Yellow Zigbee module",
)

ports = [p for p in ports if not p.device.startswith("/dev/ttyAMA")]
Comment thread
puddly marked this conversation as resolved.
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 = SerialDevice(
device=silabs_multiprotocol_addon.get_zigbee_socket(),
serial_number=None,
manufacturer="Nabu Casa",
description="Silicon Labs Multiprotocol add-on",
)

ports.append(addon_port)

# Filter out ignored USB devices
return [
port
for port in ports
if (port.manufacturer, port.description) not in IGNORED_USB_DEVICES
]


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."""
Expand Down Expand Up @@ -272,29 +188,9 @@ 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 = [_format_serial_port_choice(p, resolved_paths) 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
device_path = user_input[CONF_DEVICE_PATH]
self._radio_mgr.device_path = device_path

probe_result = await self._radio_mgr.detect_radio_type()
if probe_result == ProbeResult.WRONG_FIRMWARE_INSTALLED:
Expand All @@ -303,34 +199,17 @@ async def async_step_choose_serial_port(
description_placeholders={"repair_url": REPAIR_MY_URL},
)
if probe_result == ProbeResult.PROBING_FAILED:
# Did not autodetect anything, proceed to manual selection
# Did not autodetect anything, proceed to manual radio type
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()
Comment thread
puddly marked this conversation as resolved.

# Preselect 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

default_path = self._radio_mgr.device_path or vol.UNDEFINED
schema = vol.Schema(
{
vol.Required(CONF_DEVICE_PATH, default=default_port): vol.In(
list_of_ports
)
vol.Required(
CONF_DEVICE_PATH, default=default_path
): SerialPortSelector(),
}
)
return self.async_show_form(step_id="choose_serial_port", data_schema=schema)
Expand Down Expand Up @@ -368,7 +247,6 @@ async def async_step_manual_port_config(
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(
{
Expand Down Expand Up @@ -978,7 +856,11 @@ async def async_step_confirm(

return self.async_show_form(
step_id="confirm",
description_placeholders={CONF_NAME: self._title},
description_placeholders={
CONF_NAME: self.context.get("title_placeholders", {}).get(
CONF_NAME, self._radio_mgr.device_path or ""
)
},
)

async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResult:
Expand All @@ -1004,15 +886,17 @@ async def async_step_usb(self, discovery_info: UsbServiceInfo) -> ConfigFlowResu
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}
self.context["title_placeholders"] = {
CONF_NAME: description
or usb.human_readable_device_name(
dev_path,
serial_number,
manufacturer,
description,
vid,
pid,
)
}
return await self.async_step_confirm()

async def async_step_zeroconf(
Expand Down Expand Up @@ -1071,7 +955,6 @@ async def async_step_zeroconf(
)

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(
Expand Down Expand Up @@ -1104,7 +987,6 @@ async def async_step_hardware(
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
Expand All @@ -1125,7 +1007,6 @@ async def _async_create_radio_entry(self) -> ConfigFlowResult:
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",
Expand All @@ -1140,10 +1021,7 @@ async def _async_create_radio_entry(self) -> ConfigFlowResult:
)
await self.async_set_unique_id(unique_id)

return self.async_create_entry(
title=self._title,
data=data,
)
return self.async_create_entry(title="", data=data)
# This should never be reached
return self.async_abort(reason="single_instance_allowed")

Expand All @@ -1159,7 +1037,6 @@ def __init__(self, config_entry: ConfigEntry) -> None:
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
Expand Down
Loading
Loading