Add ZiGate device on automatic integration USB and ZEROCONF (#68577)

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Fairesoimeme 2022-05-09 01:27:09 +02:00 committed by GitHub
parent 9a5e0281db
commit 1d63c2069e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 148 additions and 6 deletions

View File

@ -164,9 +164,15 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle zeroconf discovery."""
# Hostname is format: livingroom.local.
local_name = discovery_info.hostname[:-1]
radio_type = discovery_info.properties.get("radio_type") or local_name
node_name = local_name[: -len(".local")]
host = discovery_info.host
device_path = f"socket://{host}:6638"
if local_name.startswith("tube") or "efr32" in local_name:
# This is hard coded to work with legacy devices
port = 6638
else:
port = discovery_info.port
device_path = f"socket://{host}:{port}"
if current_entry := await self.async_set_unique_id(node_name):
self._abort_if_unique_id_configured(
@ -187,9 +193,12 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}
self._device_path = device_path
self._radio_type = (
RadioType.ezsp.name if "efr32" in local_name else RadioType.znp.name
)
if "efr32" in radio_type:
self._radio_type = RadioType.ezsp.name
elif "zigate" in radio_type:
self._radio_type = RadioType.zigate.name
else:
self._radio_type = RadioType.znp.name
return await self.async_step_port_config()

View File

@ -57,6 +57,18 @@
"description": "*zigbee*",
"known_devices": ["Nortek HUSBZB-1"]
},
{
"vid": "0403",
"pid": "6015",
"description": "*zigate*",
"known_devices": ["ZiGate+"]
},
{
"vid": "10C4",
"pid": "EA60",
"description": "*zigate*",
"known_devices": ["ZiGate"]
},
{
"vid": "10C4",
"pid": "8B34",
@ -69,6 +81,10 @@
{
"type": "_esphomelib._tcp.local.",
"name": "tube*"
},
{
"type": "_zigate-zigbee-gateway._tcp.local.",
"name": "*zigate*"
}
],
"after_dependencies": ["usb", "zeroconf"],

View File

@ -77,6 +77,18 @@ USB = [
"pid": "8A2A",
"description": "*zigbee*"
},
{
"domain": "zha",
"vid": "0403",
"pid": "6015",
"description": "*zigate*"
},
{
"domain": "zha",
"vid": "10C4",
"pid": "EA60",
"description": "*zigate*"
},
{
"domain": "zha",
"vid": "10C4",

View File

@ -381,6 +381,12 @@ ZEROCONF = {
"domain": "kodi"
}
],
"_zigate-zigbee-gateway._tcp.local.": [
{
"domain": "zha",
"name": "*zigate*"
}
],
"_zwave-js-server._tcp.local.": [
{
"domain": "zwave_js"

View File

@ -52,8 +52,8 @@ async def test_discovery(detect_mock, hass):
service_info = zeroconf.ZeroconfServiceInfo(
host="192.168.1.200",
addresses=["192.168.1.200"],
hostname="_tube_zb_gw._tcp.local.",
name="mock_name",
hostname="tube._tube_zb_gw._tcp.local.",
name="tube",
port=6053,
properties={"name": "tube_123456"},
type="mock_type",
@ -77,6 +77,68 @@ async def test_discovery(detect_mock, hass):
}
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
@patch("zigpy_zigate.zigbee.application.ControllerApplication.probe")
async def test_zigate_via_zeroconf(probe_mock, hass):
"""Test zeroconf flow -- zigate radio detected."""
service_info = zeroconf.ZeroconfServiceInfo(
host="192.168.1.200",
addresses=["192.168.1.200"],
hostname="_zigate-zigbee-gateway._tcp.local.",
name="any",
port=1234,
properties={"radio_type": "zigate"},
type="mock_type",
)
flow = await hass.config_entries.flow.async_init(
"zha", context={"source": SOURCE_ZEROCONF}, data=service_info
)
result = await hass.config_entries.flow.async_configure(
flow["flow_id"], user_input={}
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "socket://192.168.1.200:1234"
assert result["data"] == {
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://192.168.1.200:1234",
},
CONF_RADIO_TYPE: "zigate",
}
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
@patch("bellows.zigbee.application.ControllerApplication.probe", return_value=True)
async def test_efr32_via_zeroconf(probe_mock, hass):
"""Test zeroconf flow -- efr32 radio detected."""
service_info = zeroconf.ZeroconfServiceInfo(
host="192.168.1.200",
addresses=["192.168.1.200"],
hostname="efr32._esphomelib._tcp.local.",
name="efr32",
port=1234,
properties={},
type="mock_type",
)
flow = await hass.config_entries.flow.async_init(
"zha", context={"source": SOURCE_ZEROCONF}, data=service_info
)
result = await hass.config_entries.flow.async_configure(
flow["flow_id"], user_input={"baudrate": 115200}
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "socket://192.168.1.200:6638"
assert result["data"] == {
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://192.168.1.200:6638",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: "software",
},
CONF_RADIO_TYPE: "ezsp",
}
@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_via_zeroconf_ip_change(detect_mock, hass):
@ -183,6 +245,43 @@ async def test_discovery_via_usb(detect_mock, hass):
}
@patch("zigpy_zigate.zigbee.application.ControllerApplication.probe")
async def test_zigate_discovery_via_usb(detect_mock, hass):
"""Test zigate usb flow -- radio detected."""
discovery_info = usb.UsbServiceInfo(
device="/dev/ttyZIGBEE",
pid="0403",
vid="6015",
serial_number="1234",
description="zigate 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 (
"zigate radio - /dev/ttyZIGBEE, s/n: 1234 - test - 6015:0403"
in result2["title"]
)
assert result2["data"] == {
"device": {
"path": "/dev/ttyZIGBEE",
},
CONF_RADIO_TYPE: "zigate",
}
@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."""