Fix non-unique ZHA serial port paths and migrate USB integration to always list unique paths (#155019)

Author: puddly

homeassistant/components/usb/utils.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Sequence
+import dataclasses
 import fnmatch
 import os
 
@@ -29,15 +30,6 @@ def usb_device_from_port(port: ListPortInfo) -> USBDevice:
 
 def scan_serial_ports() -> Sequence[USBDevice]:
     """Scan serial ports for USB devices."""
-    return [
-        usb_device_from_port(port)
-        for port in comports()
-        if port.vid is not None or port.pid is not None
-    ]
-
-
-def usb_device_from_path(device_path: str) -> USBDevice | None:
-    """Get USB device info from a device path."""
 
     # Scan all symlinks first
     by_id = "/dev/serial/by-id"
@@ -46,23 +38,30 @@ def usb_device_from_path(device_path: str) -> USBDevice | None:
         for path in (entry.path for entry in os.scandir(by_id) if entry.is_symlink()):
             realpath_to_by_id[os.path.realpath(path)] = path
 
-    # Then compare the actual path to each serial port's
+    serial_ports = []
+
+    for port in comports():
+        if port.vid is not None or port.pid is not None:
+            usb_device = usb_device_from_port(port)
+            device_path = realpath_to_by_id.get(port.device, port.device)
+
+            if device_path != port.device:
+                # Prefer the unique /dev/serial/by-id/ path if it exists
+                usb_device = dataclasses.replace(usb_device, device=device_path)
+
+            serial_ports.append(usb_device)
+
+    return serial_ports
+
+
+def usb_device_from_path(device_path: str) -> USBDevice | None:
+    """Get USB device info from a device path."""
+
     device_path_real = os.path.realpath(device_path)
 
     for device in scan_serial_ports():
-        normalized_path = realpath_to_by_id.get(device.device, device.device)
-        if (
-            normalized_path == device_path
-            or os.path.realpath(device.device) == device_path_real
-        ):
-            return USBDevice(
-                device=normalized_path,
-                vid=device.vid,
-                pid=device.pid,
-                serial_number=device.serial_number,
-                manufacturer=device.manufacturer,
-                description=device.description,
-            )
+        if os.path.realpath(device.device) == device_path_real:
+            return device
 
     return None
 

homeassistant/components/zha/__init__.py

@@ -17,6 +17,7 @@
     async_notify_firmware_info,
     async_register_firmware_info_provider,
 )
+from homeassistant.components.usb import usb_device_from_path
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     CONF_TYPE,
@@ -134,6 +135,21 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
 
     Will automatically load components to support devices found on the network.
     """
+
+    # Try to perform an in-place migration if we detect that the device path can be made
+    # unique
+    device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
+    usb_device = await hass.async_add_executor_job(usb_device_from_path, device_path)
+
+    if usb_device is not None and device_path != usb_device.device:
+        _LOGGER.info(
+            "Migrating ZHA device path from %s to %s", device_path, usb_device.device
+        )
+        new_data = {**config_entry.data}
+        new_data[CONF_DEVICE][CONF_DEVICE_PATH] = usb_device.device
+        hass.config_entries.async_update_entry(config_entry, data=new_data)
+        device_path = usb_device.device
+
     ha_zha_data: HAZHAData = get_zha_data(hass)
     ha_zha_data.config_entry = config_entry
     zha_lib_data: ZHAData = create_zha_config(hass, ha_zha_data)
@@ -163,7 +179,6 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
     _LOGGER.debug("Trigger cache: %s", zha_lib_data.device_trigger_cache)
 
     # Check if firmware update is in progress for this device
-    device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
     _raise_if_port_in_use(hass, device_path)
 
     try:

homeassistant/components/zha/config_flow.py

@@ -7,10 +7,9 @@
 from contextlib import suppress
 from enum import StrEnum
 import json
+import os
 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
@@ -25,6 +24,7 @@
     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,
@@ -124,10 +124,10 @@ def _format_backup_choice(
     return f"{dt_util.as_local(backup.backup_time).strftime('%c')} ({identifier})"
 
 
-async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]:
+async def list_serial_ports(hass: HomeAssistant) -> list[USBDevice]:
     """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))
+    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:
@@ -137,9 +137,14 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]:
     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"
+        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)
@@ -156,13 +161,15 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]:
             addon_info = None
 
         if addon_info is not None and addon_info.state != AddonState.NOT_INSTALLED:
-            addon_port = ListPortInfo(
+            addon_port = USBDevice(
                 device=silabs_multiprotocol_addon.get_zigbee_socket(),
-                skip_link_detection=True,
+                vid="ffff",  # This is technically not a USB device
+                pid="ffff",
+                serial_number=None,
+                manufacturer="Nabu Casa",
+                description="Silicon Labs Multiprotocol add-on",
             )
 
-            addon_port.description = "Multiprotocol add-on"
-            addon_port.manufacturer = "Nabu Casa"
             ports.append(addon_port)
 
     return ports
@@ -218,8 +225,15 @@ async def async_step_choose_serial_port(
     ) -> 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"{p}{', s/n: ' + p.serial_number if p.serial_number else ''}"
+            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
         ]