Add zwave_js usb port selection (#76385)

homeassistant/components/zwave_js/config_flow.py

@@ -8,6 +8,7 @@ from typing import Any
 
 import aiohttp
 from async_timeout import timeout
+from serial.tools import list_ports
 import voluptuous as vol
 from zwave_js_server.version import VersionInfo, get_server_version
 
@@ -119,6 +120,30 @@ async def async_get_version_info(hass: HomeAssistant, ws_address: str) -> Versio
     return version_info
 
 
+def get_usb_ports() -> dict[str, str]:
+    """Return a dict of USB ports and their friendly names."""
+    ports = list_ports.comports()
+    port_descriptions = {}
+    for port in ports:
+        usb_device = usb.usb_device_from_port(port)
+        dev_path = usb.get_serial_by_id(usb_device.device)
+        human_name = usb.human_readable_device_name(
+            dev_path,
+            usb_device.serial_number,
+            usb_device.manufacturer,
+            usb_device.description,
+            usb_device.vid,
+            usb_device.pid,
+        )
+        port_descriptions[dev_path] = human_name
+    return port_descriptions
+
+
+async def async_get_usb_ports(hass: HomeAssistant) -> dict[str, str]:
+    """Return a dict of USB ports and their friendly names."""
+    return await hass.async_add_executor_job(get_usb_ports)
+
+
 class BaseZwaveJSFlow(FlowHandler):
     """Represent the base config flow for Z-Wave JS."""
 
@@ -402,7 +427,9 @@ class ConfigFlow(BaseZwaveJSFlow, config_entries.ConfigFlow, domain=DOMAIN):
             vid,
             pid,
         )
-        self.context["title_placeholders"] = {CONF_NAME: self._title}
+        self.context["title_placeholders"] = {
+            CONF_NAME: self._title.split(" - ")[0].strip()
+        }
         return await self.async_step_usb_confirm()
 
     async def async_step_usb_confirm(
@@ -579,7 +606,11 @@ class ConfigFlow(BaseZwaveJSFlow, config_entries.ConfigFlow, domain=DOMAIN):
         }
 
         if not self._usb_discovery:
-            schema = {vol.Required(CONF_USB_PATH, default=usb_path): str, **schema}
+            ports = await async_get_usb_ports(self.hass)
+            schema = {
+                vol.Required(CONF_USB_PATH, default=usb_path): vol.In(ports),
+                **schema,
+            }
 
         data_schema = vol.Schema(schema)
 
@@ -801,9 +832,11 @@ class OptionsFlowHandler(BaseZwaveJSFlow, config_entries.OptionsFlow):
         log_level = addon_config.get(CONF_ADDON_LOG_LEVEL, "info")
         emulate_hardware = addon_config.get(CONF_ADDON_EMULATE_HARDWARE, False)
 
+        ports = await async_get_usb_ports(self.hass)
+
         data_schema = vol.Schema(
             {
-                vol.Required(CONF_USB_PATH, default=usb_path): str,
+                vol.Required(CONF_USB_PATH, default=usb_path): vol.In(ports),
                 vol.Optional(CONF_S0_LEGACY_KEY, default=s0_legacy_key): str,
                 vol.Optional(
                     CONF_S2_ACCESS_CONTROL_KEY, default=s2_access_control_key

homeassistant/components/zwave_js/manifest.json

@@ -3,7 +3,7 @@
   "name": "Z-Wave",
   "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/zwave_js",
-  "requirements": ["zwave-js-server-python==0.40.0"],
+  "requirements": ["pyserial==3.5", "zwave-js-server-python==0.40.0"],
   "codeowners": ["@home-assistant/z-wave"],
   "dependencies": ["usb", "http", "websocket_api"],
   "iot_class": "local_push",

requirements_all.txt

@@ -1815,6 +1815,7 @@ pyserial-asyncio==0.6
 # homeassistant.components.crownstone
 # homeassistant.components.usb
 # homeassistant.components.zha
+# homeassistant.components.zwave_js
 pyserial==3.5
 
 # homeassistant.components.sesame

requirements_test_all.txt

@@ -1250,6 +1250,7 @@ pyserial-asyncio==0.6
 # homeassistant.components.crownstone
 # homeassistant.components.usb
 # homeassistant.components.zha
+# homeassistant.components.zwave_js
 pyserial==3.5
 
 # homeassistant.components.sia

tests/components/zwave_js/test_config_flow.py

@@ -1,9 +1,12 @@
 """Test the Z-Wave JS config flow."""
 import asyncio
-from unittest.mock import DEFAULT, call, patch
+from collections.abc import Generator
+from copy import copy
+from unittest.mock import DEFAULT, MagicMock, call, patch
 
 import aiohttp
 import pytest
+from serial.tools.list_ports_common import ListPortInfo
 from zwave_js_server.version import VersionInfo
 
 from homeassistant import config_entries
@@ -134,6 +137,45 @@ def mock_addon_setup_time():
         yield addon_setup_time
 
 
+@pytest.fixture(name="serial_port")
+def serial_port_fixture() -> ListPortInfo:
+    """Return a mock serial port."""
+    port = ListPortInfo("/test", skip_link_detection=True)
+    port.serial_number = "1234"
+    port.manufacturer = "Virtual serial port"
+    port.device = "/test"
+    port.description = "Some serial port"
+    port.pid = 9876
+    port.vid = 5678
+
+    return port
+
+
+@pytest.fixture(name="mock_list_ports", autouse=True)
+def mock_list_ports_fixture(serial_port) -> Generator[MagicMock, None, None]:
+    """Mock list ports."""
+    with patch(
+        "homeassistant.components.zwave_js.config_flow.list_ports.comports"
+    ) as mock_list_ports:
+        another_port = copy(serial_port)
+        another_port.device = "/new"
+        another_port.description = "New serial port"
+        another_port.serial_number = "5678"
+        another_port.pid = 8765
+        mock_list_ports.return_value = [serial_port, another_port]
+        yield mock_list_ports
+
+
+@pytest.fixture(name="mock_usb_serial_by_id", autouse=True)
+def mock_usb_serial_by_id_fixture() -> Generator[MagicMock, None, None]:
+    """Mock usb serial by id."""
+    with patch(
+        "homeassistant.components.zwave_js.config_flow.usb.get_serial_by_id"
+    ) as mock_usb_serial_by_id:
+        mock_usb_serial_by_id.side_effect = lambda x: x
+        yield mock_usb_serial_by_id
+
+
 async def test_manual(hass):
     """Test we create an entry with manual step."""
 
@@ -1397,7 +1439,7 @@ async def test_addon_installed_already_configured(
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         {
-            "usb_path": "/test_new",
+            "usb_path": "/new",
             "s0_legacy_key": "new123",
             "s2_access_control_key": "new456",
             "s2_authenticated_key": "new789",
@@ -1410,7 +1452,7 @@ async def test_addon_installed_already_configured(
         "core_zwave_js",
         {
             "options": {
-                "device": "/test_new",
+                "device": "/new",
                 "s0_legacy_key": "new123",
                 "s2_access_control_key": "new456",
                 "s2_authenticated_key": "new789",
@@ -1430,7 +1472,7 @@ async def test_addon_installed_already_configured(
     assert result["type"] == "abort"
     assert result["reason"] == "already_configured"
     assert entry.data["url"] == "ws://host1:3001"
-    assert entry.data["usb_path"] == "/test_new"
+    assert entry.data["usb_path"] == "/new"
     assert entry.data["s0_legacy_key"] == "new123"
     assert entry.data["s2_access_control_key"] == "new456"
     assert entry.data["s2_authenticated_key"] == "new789"
@@ -2380,8 +2422,10 @@ async def test_import_addon_installed(
     set_addon_options,
     start_addon,
     get_addon_discovery_info,
+    serial_port,
 ):
     """Test import step while add-on already installed on Supervisor."""
+    serial_port.device = "/test/imported"
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": config_entries.SOURCE_IMPORT},