From 8796eaec81370d1164823768255ff83764b372c2 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 20 Aug 2021 20:42:23 -0500 Subject: [PATCH] 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 * black Co-authored-by: Martin Hjelmare --- homeassistant/components/zha/config_flow.py | 84 +++++++++++ homeassistant/components/zha/manifest.json | 6 + homeassistant/components/zha/strings.json | 6 +- .../components/zha/translations/en.json | 4 + homeassistant/generated/usb.py | 23 ++- script/hassfest/usb.py | 2 +- tests/components/zha/test_config_flow.py | 139 +++++++++++++++++- 7 files changed, 260 insertions(+), 4 deletions(-) diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index 00aaf7c3625..a305c97d436 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/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. diff --git a/homeassistant/components/zha/manifest.json b/homeassistant/components/zha/manifest.json index 5200c0a8b31..57d926042db 100644 --- a/homeassistant/components/zha/manifest.json +++ b/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": [ { diff --git a/homeassistant/components/zha/strings.json b/homeassistant/components/zha/strings.json index 9abff4e83e2..4b5b429522f 100644 --- a/homeassistant/components/zha/strings.json +++ b/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": { diff --git a/homeassistant/components/zha/translations/en.json b/homeassistant/components/zha/translations/en.json index e13aca2cfb1..93d3c5f697a 100644 --- a/homeassistant/components/zha/translations/en.json +++ b/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" diff --git a/homeassistant/generated/usb.py b/homeassistant/generated/usb.py index d72cbc8c7a5..717640ce2f8 100644 --- a/homeassistant/generated/usb.py +++ b/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" + } +] diff --git a/script/hassfest/usb.py b/script/hassfest/usb.py index 927b87def98..49da04ee03f 100644 --- a/script/hassfest/usb.py +++ b/script/hassfest/usb.py @@ -13,7 +13,7 @@ To update, run python3 -m script.hassfest # fmt: off -USB = {} # type: ignore +USB = {} """.strip() diff --git a/tests/components/zha/test_config_flow.py b/tests/components/zha/test_config_flow.py index 16747980b15..c603e665912 100644 --- a/tests/components/zha/test_config_flow.py +++ b/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):