Add support for USB discovery to ZHA (#54935)

* Add USB discovery support to ZHA

* dry

* dry

* Update homeassistant/components/zha/config_flow.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* black

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/zha/config_flow.py

@@ -27,6 +27,24 @@ SUPPORTED_PORT_SETTINGS = (
 )
 
 
+def _format_port_human_readable(
+    device: str,
+    serial_number: str | None,
+    manufacturer: str | None,
+    description: str | None,
+    vid: str | None,
+    pid: str | None,
+) -> str:
+    device_details = f"{device}, s/n: {serial_number or 'n/a'}"
+    manufacturer_details = f" - {manufacturer}" if manufacturer else ""
+    vendor_details = f" - {vid}:{pid}" if vid else ""
+    full_details = f"{device_details}{manufacturer_details}{vendor_details}"
+
+    if not description:
+        return full_details
+    return f"{description[:26]} - {full_details}"
+
+
 class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
     """Handle a config flow."""
 
@@ -36,6 +54,8 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         """Initialize flow instance."""
         self._device_path = None
         self._radio_type = None
+        self._auto_detected_data = None
+        self._title = None
 
     async def async_step_user(self, user_input=None):
         """Handle a zha config flow start."""
@@ -92,6 +112,70 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
             data_schema=vol.Schema(schema),
         )
 
+    async def async_step_usb(self, discovery_info: DiscoveryInfoType):
+        """Handle usb discovery."""
+        vid = discovery_info["vid"]
+        pid = discovery_info["pid"]
+        serial_number = discovery_info["serial_number"]
+        device = discovery_info["device"]
+        manufacturer = discovery_info["manufacturer"]
+        description = discovery_info["description"]
+        await self.async_set_unique_id(
+            f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}"
+        )
+        self._abort_if_unique_id_configured(
+            updates={
+                CONF_DEVICE: {CONF_DEVICE_PATH: self._device_path},
+            }
+        )
+        # Check if already configured
+        if self._async_current_entries():
+            return self.async_abort(reason="single_instance_allowed")
+
+        # If they already have a discovery for deconz
+        # we ignore the usb discovery as they probably
+        # want to use it there instead
+        for flow in self.hass.config_entries.flow.async_progress():
+            if flow["handler"] == "deconz":
+                return self.async_abort(reason="not_zha_device")
+
+        # The Nortek sticks are a special case since they
+        # have a Z-Wave and a Zigbee radio. We need to reject
+        # the Z-Wave radio.
+        if vid == "10C4" and pid == "8A2A" and "ZigBee" not in description:
+            return self.async_abort(reason="not_zha_device")
+
+        dev_path = await self.hass.async_add_executor_job(get_serial_by_id, device)
+        self._auto_detected_data = await detect_radios(dev_path)
+        if self._auto_detected_data is None:
+            return self.async_abort(reason="not_zha_device")
+        self._device_path = dev_path
+        self._title = _format_port_human_readable(
+            dev_path,
+            serial_number,
+            manufacturer,
+            description,
+            vid,
+            pid,
+        )
+        self._set_confirm_only()
+        self.context["title_placeholders"] = {CONF_NAME: self._title}
+        return await self.async_step_confirm()
+
+    async def async_step_confirm(self, user_input=None):
+        """Confirm a discovery."""
+        if user_input is not None:
+            return self.async_create_entry(
+                title=self._title,
+                data=self._auto_detected_data,
+            )
+
+        return self.async_show_form(
+            step_id="confirm",
+            description_placeholders={CONF_NAME: self._title},
+            data_schema=vol.Schema({}),
+        )
+
     async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType):
         """Handle zeroconf discovery."""
         # Hostname is format: livingroom.local.

homeassistant/components/zha/manifest.json

@@ -15,6 +15,12 @@
     "zigpy-zigate==0.7.3",
     "zigpy-znp==0.5.3"
   ],
+  "usb": [
+   {"vid":"10C4","pid":"EA60"},
+   {"vid":"1CF1","pid":"0030"},
+   {"vid":"1A86","pid":"7523"},
+   {"vid":"10C4","pid":"8A2A"}
+  ],
   "codeowners": ["@dmulcahey", "@adminiuga"],
   "zeroconf": [
     {

homeassistant/components/zha/strings.json

@@ -7,6 +7,9 @@
         "data": { "path": "Serial Device Path" },
         "description": "Select serial port for Zigbee radio"
       },
+      "confirm": {
+        "description": "Do you want to setup {name}?"
+      },
       "pick_radio": {
         "data": { "radio_type": "Radio Type" },
         "title": "Radio Type",
@@ -26,7 +29,8 @@
       "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
     },
     "abort": {
-      "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]"
+      "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
+      "not_zha_device": "This device is not a zha device"
     }
   },
   "config_panel": {

homeassistant/components/zha/translations/en.json

@@ -1,6 +1,7 @@
 {
     "config": {
         "abort": {
+            "not_zha_device": "This device is not a zha device",
             "single_instance_allowed": "Already configured. Only a single configuration possible."
         },
         "error": {
@@ -8,6 +9,9 @@
         },
         "flow_title": "{name}",
         "step": {
+            "confirm": {
+                "description": "Do you want to setup {name}?"
+            },
             "pick_radio": {
                 "data": {
                     "radio_type": "Radio Type"

homeassistant/generated/usb.py

@@ -5,4 +5,25 @@ To update, run python3 -m script.hassfest
 
 # fmt: off
 
-USB = []  # type: ignore
+USB = [
+    {
+        "domain": "zha",
+        "vid": "10C4",
+        "pid": "EA60"
+    },
+    {
+        "domain": "zha",
+        "vid": "1CF1",
+        "pid": "0030"
+    },
+    {
+        "domain": "zha",
+        "vid": "1A86",
+        "pid": "7523"
+    },
+    {
+        "domain": "zha",
+        "vid": "10C4",
+        "pid": "8A2A"
+    }
+]

script/hassfest/usb.py

@@ -13,7 +13,7 @@ To update, run python3 -m script.hassfest
 
 # fmt: off
 
-USB = {}  # type: ignore
+USB = {}
 """.strip()
 
 

tests/components/zha/test_config_flow.py

@@ -8,9 +8,19 @@ import serial.tools.list_ports
 import zigpy.config
 
 from homeassistant import setup
+from homeassistant.components.ssdp import (
+    ATTR_SSDP_LOCATION,
+    ATTR_UPNP_MANUFACTURER_URL,
+    ATTR_UPNP_SERIAL,
+)
 from homeassistant.components.zha import config_flow
 from homeassistant.components.zha.core.const import CONF_RADIO_TYPE, DOMAIN, RadioType
-from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
+from homeassistant.config_entries import (
+    SOURCE_SSDP,
+    SOURCE_USB,
+    SOURCE_USER,
+    SOURCE_ZEROCONF,
+)
 from homeassistant.const import CONF_SOURCE
 from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
 
@@ -57,6 +67,133 @@ async def test_discovery(detect_mock, hass):
     }
 
 
+@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True)
+async def test_discovery_via_usb(detect_mock, hass):
+    """Test usb flow -- radio detected."""
+    discovery_info = {
+        "device": "/dev/ttyZIGBEE",
+        "pid": "AAAA",
+        "vid": "AAAA",
+        "serial_number": "1234",
+        "description": "zigbee radio",
+        "manufacturer": "test",
+    }
+    result = await hass.config_entries.flow.async_init(
+        "zha", context={"source": SOURCE_USB}, data=discovery_info
+    )
+    await hass.async_block_till_done()
+    assert result["type"] == RESULT_TYPE_FORM
+    assert result["step_id"] == "confirm"
+
+    with patch("homeassistant.components.zha.async_setup_entry"):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={}
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
+    assert "zigbee radio" in result2["title"]
+    assert result2["data"] == {
+        "device": {
+            "baudrate": 115200,
+            "flow_control": None,
+            "path": "/dev/ttyZIGBEE",
+        },
+        CONF_RADIO_TYPE: "znp",
+    }
+
+
+@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=False)
+async def test_discovery_via_usb_no_radio(detect_mock, hass):
+    """Test usb flow -- no radio detected."""
+    discovery_info = {
+        "device": "/dev/null",
+        "pid": "AAAA",
+        "vid": "AAAA",
+        "serial_number": "1234",
+        "description": "zigbee radio",
+        "manufacturer": "test",
+    }
+    result = await hass.config_entries.flow.async_init(
+        "zha", context={"source": SOURCE_USB}, data=discovery_info
+    )
+    await hass.async_block_till_done()
+    assert result["type"] == "abort"
+    assert result["reason"] == "not_zha_device"
+
+
+@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True)
+async def test_discovery_via_usb_rejects_nortek_zwave(detect_mock, hass):
+    """Test usb flow -- reject the nortek zwave radio."""
+    discovery_info = {
+        "device": "/dev/null",
+        "vid": "10C4",
+        "pid": "8A2A",
+        "serial_number": "612020FD",
+        "description": "HubZ Smart Home Controller - HubZ Z-Wave Com Port",
+        "manufacturer": "Silicon Labs",
+    }
+    result = await hass.config_entries.flow.async_init(
+        "zha", context={"source": SOURCE_USB}, data=discovery_info
+    )
+    await hass.async_block_till_done()
+    assert result["type"] == "abort"
+    assert result["reason"] == "not_zha_device"
+
+
+@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True)
+async def test_discovery_via_usb_already_setup(detect_mock, hass):
+    """Test usb flow -- already setup."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    MockConfigEntry(domain=DOMAIN, data={"usb_path": "/dev/ttyUSB1"}).add_to_hass(hass)
+
+    discovery_info = {
+        "device": "/dev/ttyZIGBEE",
+        "pid": "AAAA",
+        "vid": "AAAA",
+        "serial_number": "1234",
+        "description": "zigbee radio",
+        "manufacturer": "test",
+    }
+    result = await hass.config_entries.flow.async_init(
+        "zha", context={"source": SOURCE_USB}, data=discovery_info
+    )
+    await hass.async_block_till_done()
+
+    assert result["type"] == "abort"
+    assert result["reason"] == "single_instance_allowed"
+
+
+@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True)
+async def test_discovery_via_usb_deconz_already_discovered(detect_mock, hass):
+    """Test usb flow -- deconz discovered."""
+    result = await hass.config_entries.flow.async_init(
+        "deconz",
+        data={
+            ATTR_SSDP_LOCATION: "http://1.2.3.4:80/",
+            ATTR_UPNP_MANUFACTURER_URL: "http://www.dresden-elektronik.de",
+            ATTR_UPNP_SERIAL: "0000000000000000",
+        },
+        context={"source": SOURCE_SSDP},
+    )
+    await hass.async_block_till_done()
+    discovery_info = {
+        "device": "/dev/ttyZIGBEE",
+        "pid": "AAAA",
+        "vid": "AAAA",
+        "serial_number": "1234",
+        "description": "zigbee radio",
+        "manufacturer": "test",
+    }
+    result = await hass.config_entries.flow.async_init(
+        "zha", context={"source": SOURCE_USB}, data=discovery_info
+    )
+    await hass.async_block_till_done()
+
+    assert result["type"] == "abort"
+    assert result["reason"] == "not_zha_device"
+
+
 @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
 @patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True)
 async def test_discovery_already_setup(detect_mock, hass):