Simplify some multi pan code (#90135)

* Simplify some multi pan code

* Adjust ZHA config flow
This commit is contained in:
Erik Montnemery 2023-03-23 09:18:35 +01:00 committed by GitHub
parent b151923619
commit 6739542a5d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 34 additions and 22 deletions

View file

@ -244,6 +244,11 @@ HARDWARE_INTEGRATIONS = {
}
def hostname_from_addon_slug(addon_slug: str) -> str:
"""Return hostname of add-on."""
return addon_slug.replace("_", "-")
@callback
@bind_hass
def get_info(hass: HomeAssistant) -> dict[str, Any] | None:

View file

@ -15,6 +15,7 @@ from homeassistant.components.hassio import (
AddonInfo,
AddonManager,
AddonState,
hostname_from_addon_slug,
is_hassio,
)
from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN
@ -64,12 +65,13 @@ class SerialPortSettings:
flow_control: bool
def get_zigbee_socket(hass: HomeAssistant, addon_info: AddonInfo) -> str:
def get_zigbee_socket() -> str:
"""Return the zigbee socket.
Raises AddonError on error
"""
return f"socket://{addon_info.hostname}:9999"
hostname = hostname_from_addon_slug(SILABS_MULTIPROTOCOL_ADDON_SLUG)
return f"socket://{hostname}:9999"
class BaseMultiPanFlow(FlowHandler, ABC):
@ -290,7 +292,7 @@ class OptionsFlowHandler(BaseMultiPanFlow, config_entries.OptionsFlow):
"new_discovery_info": {
"name": self._zha_name(),
"port": {
"path": get_zigbee_socket(self.hass, addon_info),
"path": get_zigbee_socket(),
},
"radio_type": "ezsp",
},
@ -386,24 +388,22 @@ async def check_multi_pan_addon(hass: HomeAssistant) -> None:
raise HomeAssistantError
async def get_multi_pan_addon_info(
hass: HomeAssistant, device_path: str
) -> AddonInfo | None:
"""Return AddonInfo if the multi-PAN addon is using the given device.
async def multi_pan_addon_using_device(hass: HomeAssistant, device_path: str) -> bool:
"""Return True if the multi-PAN addon is using the given device.
Returns None if Hass.io is not loaded, the addon is not running or the addon is
Returns False if Hass.io is not loaded, the addon is not running or the addon is
connected to another device.
"""
if not is_hassio(hass):
return None
return False
addon_manager: AddonManager = get_addon_manager(hass)
addon_info: AddonInfo = await addon_manager.async_get_addon_info()
if addon_info.state != AddonState.RUNNING:
return None
return False
if addon_info.options["device"] != device_path:
return None
return False
return addon_info
return True

View file

@ -4,8 +4,8 @@ from __future__ import annotations
from homeassistant.components import usb
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
check_multi_pan_addon,
get_multi_pan_addon_info,
get_zigbee_socket,
multi_pan_addon_using_device,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
@ -33,9 +33,8 @@ async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None:
usb_dev = entry.data["device"]
dev_path = await hass.async_add_executor_job(usb.get_serial_by_id, usb_dev)
addon_info = await get_multi_pan_addon_info(hass, dev_path)
if not addon_info:
if not await multi_pan_addon_using_device(hass, dev_path):
usb_info = get_usb_service_info(entry)
await hass.config_entries.flow.async_init(
"zha",
@ -47,7 +46,7 @@ async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None:
hw_discovery_data = {
"name": "SkyConnect Multi-PAN",
"port": {
"path": get_zigbee_socket(hass, addon_info),
"path": get_zigbee_socket(),
},
"radio_type": "ezsp",
}

View file

@ -4,8 +4,8 @@ from __future__ import annotations
from homeassistant.components.hassio import get_os_info
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
check_multi_pan_addon,
get_multi_pan_addon_info,
get_zigbee_socket,
multi_pan_addon_using_device,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@ -31,15 +31,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
except HomeAssistantError as err:
raise ConfigEntryNotReady from err
addon_info = await get_multi_pan_addon_info(hass, RADIO_DEVICE)
if not addon_info:
if not await multi_pan_addon_using_device(hass, RADIO_DEVICE):
hw_discovery_data = ZHA_HW_DISCOVERY_DATA
else:
hw_discovery_data = {
"name": "Yellow Multi-PAN",
"port": {
"path": get_zigbee_socket(hass, addon_info),
"path": get_zigbee_socket(),
},
"radio_type": "ezsp",
}

View file

@ -101,7 +101,7 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]:
if addon_info is not None and addon_info.state != AddonState.NOT_INSTALLED:
addon_port = ListPortInfo(
device=silabs_multiprotocol_addon.get_zigbee_socket(hass, addon_info),
device=silabs_multiprotocol_addon.get_zigbee_socket(),
skip_link_detection=True,
)

View file

@ -14,6 +14,7 @@ from homeassistant.components.hassio import (
DOMAIN,
STORAGE_KEY,
async_get_addon_store_info,
hostname_from_addon_slug,
)
from homeassistant.components.hassio.handler import HassioAPIError
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
@ -871,3 +872,12 @@ async def test_get_store_addon_info(
data = await async_get_addon_store_info(hass, "test")
assert data["name"] == "bla"
assert aioclient_mock.call_count == 1
def test_hostname_from_addon_slug() -> None:
"""Test hostname_from_addon_slug."""
assert hostname_from_addon_slug("mqtt") == "mqtt"
assert (
hostname_from_addon_slug("core_silabs_multiprotocol")
== "core-silabs-multiprotocol"
)