jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Deduplicate multiprotocol addon helper (#90102)

Browse Source 
* Deduplicate multiprotocol addon helper

* Clarify
This commit is contained in:
Erik Montnemery 2023-03-22 20:20:42 +01:00 committed by GitHub
parent 4c98495fe0
commit 1ea3312ed4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 75 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py
View File

@ -26,6 +26,7 @@ from homeassistant.data_entry_flow import (
FlowManager,
FlowResult,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.singleton import singleton
from .const import LOGGER, SILABS_MULTIPROTOCOL_ADDON_SLUG
@ -356,3 +357,53 @@ class OptionsFlowHandler(BaseMultiPanFlow, config_entries.OptionsFlow):
if user_input is None:
return self.async_show_form(step_id="addon_installed_other_device")
return self.async_create_entry(title="", data={})
async def check_multi_pan_addon(hass: HomeAssistant) -> None:
"""Check the multi-PAN addon state, and start it if installed but not started.
Does nothing if Hass.io is not loaded.
Raises on error or if the add-on is installed but not started.
"""
if not is_hassio(hass):
return
addon_manager: AddonManager = get_addon_manager(hass)
try:
addon_info: AddonInfo = await addon_manager.async_get_addon_info()
except AddonError as err:
_LOGGER.error(err)
raise HomeAssistantError from err
# Request the addon to start if it's not started
# addon_manager.async_start_addon returns as soon as the start request has been sent
# and does not wait for the addon to be started, so we raise below
if addon_info.state == AddonState.NOT_RUNNING:
await addon_manager.async_start_addon()
if addon_info.state not in (AddonState.NOT_INSTALLED, AddonState.RUNNING):
_LOGGER.debug("Multi pan addon installed and in state %s", addon_info.state)
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.
Returns None 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
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
if addon_info.options["device"] != device_path:
return None
return addon_info

71
homeassistant/components/homeassistant_sky_connect/__init__.py
View File

@ -1,75 +1,19 @@
"""The Home Assistant SkyConnect integration."""
from __future__ import annotations
import logging
from homeassistant.components import usb
from homeassistant.components.hassio import (
AddonError,
AddonInfo,
AddonManager,
AddonState,
is_hassio,
)
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
get_addon_manager,
check_multi_pan_addon,
get_multi_pan_addon_info,
get_zigbee_socket,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from .const import DOMAIN
from .util import get_usb_service_info
_LOGGER = logging.getLogger(__name__)
async def _wait_multi_pan_addon(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Wait for multi-PAN info to be available."""
if not is_hassio(hass):
return
addon_manager: AddonManager = get_addon_manager(hass)
try:
addon_info: AddonInfo = await addon_manager.async_get_addon_info()
except AddonError as err:
_LOGGER.error(err)
raise ConfigEntryNotReady from err
# Start the addon if it's not started
if addon_info.state == AddonState.NOT_RUNNING:
await addon_manager.async_start_addon()
if addon_info.state not in (AddonState.NOT_INSTALLED, AddonState.RUNNING):
_LOGGER.debug(
"Multi pan addon in state %s, delaying yellow config entry setup",
addon_info.state,
)
raise ConfigEntryNotReady
async def _multi_pan_addon_info(
hass: HomeAssistant, entry: ConfigEntry
) -> AddonInfo | None:
"""Return AddonInfo if the multi-PAN addon is enabled for our SkyConnect."""
if not is_hassio(hass):
return None
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
usb_dev = entry.data["device"]
dev_path = await hass.async_add_executor_job(usb.get_serial_by_id, usb_dev)
if addon_info.options["device"] != dev_path:
return None
return addon_info
async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Finish Home Assistant SkyConnect config entry setup."""
@ -87,7 +31,9 @@ async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None:
hass.async_create_task(hass.config_entries.async_remove(entry.entry_id))
return
addon_info = await _multi_pan_addon_info(hass, entry)
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:
usb_info = get_usb_service_info(entry)
@ -115,7 +61,10 @@ async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a Home Assistant SkyConnect config entry."""
await _wait_multi_pan_addon(hass, entry)
try:
await check_multi_pan_addon(hass)
except HomeAssistantError as err:
raise ConfigEntryNotReady from err
@callback
def async_usb_scan_done() -> None:

55
homeassistant/components/homeassistant_yellow/__init__.py
View File

@ -1,58 +1,18 @@
"""The Home Assistant Yellow integration."""
from __future__ import annotations
import logging
from homeassistant.components.hassio import (
AddonError,
AddonInfo,
AddonManager,
AddonState,
get_os_info,
)
from homeassistant.components.hassio import get_os_info
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
get_addon_manager,
check_multi_pan_addon,
get_multi_pan_addon_info,
get_zigbee_socket,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from .const import RADIO_DEVICE, ZHA_HW_DISCOVERY_DATA
_LOGGER = logging.getLogger(__name__)
async def _multi_pan_addon_info(
hass: HomeAssistant, entry: ConfigEntry
) -> AddonInfo | None:
"""Return AddonInfo if the multi-PAN addon is enabled for the Yellow's radio."""
addon_manager: AddonManager = get_addon_manager(hass)
try:
addon_info: AddonInfo = await addon_manager.async_get_addon_info()
except AddonError as err:
_LOGGER.error(err)
raise ConfigEntryNotReady from err
# Start the addon if it's not started
if addon_info.state == AddonState.NOT_RUNNING:
await addon_manager.async_start_addon()
if addon_info.state not in (AddonState.NOT_INSTALLED, AddonState.RUNNING):
_LOGGER.debug(
"Multi pan addon in state %s, delaying yellow config entry setup",
addon_info.state,
)
raise ConfigEntryNotReady
if addon_info.state == AddonState.NOT_INSTALLED:
return None
if addon_info.options["device"] != RADIO_DEVICE:
return None
return addon_info
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a Home Assistant Yellow config entry."""
@ -66,7 +26,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.async_create_task(hass.config_entries.async_remove(entry.entry_id))
return False
addon_info = await _multi_pan_addon_info(hass, entry)
try:
await check_multi_pan_addon(hass)
except HomeAssistantError as err:
raise ConfigEntryNotReady from err
addon_info = await get_multi_pan_addon_info(hass, RADIO_DEVICE)
if not addon_info:
hw_discovery_data = ZHA_HW_DISCOVERY_DATA

8
tests/components/homeassistant_sky_connect/test_init.py
View File

@ -172,7 +172,7 @@ async def test_setup_zha_multipan(
) as mock_is_plugged_in, patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
), patch(
"homeassistant.components.homeassistant_sky_connect.is_hassio",
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -226,7 +226,7 @@ async def test_setup_zha_multipan_other_device(
) as mock_is_plugged_in, patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
), patch(
"homeassistant.components.homeassistant_sky_connect.is_hassio",
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -304,7 +304,7 @@ async def test_setup_entry_addon_info_fails(
), patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
), patch(
"homeassistant.components.homeassistant_sky_connect.is_hassio",
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
):
assert not await hass.config_entries.async_setup(config_entry.entry_id)
@ -333,7 +333,7 @@ async def test_setup_entry_addon_not_running(
), patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
), patch(
"homeassistant.components.homeassistant_sky_connect.is_hassio",
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
):
assert not await hass.config_entries.async_setup(config_entry.entry_id)