Add network support to flux_led discovery (#61132)

This commit is contained in:
J. Nick Koston 2021-12-18 23:55:57 -06:00 committed by GitHub
parent a40549c1b9
commit a4c101b021
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 185 additions and 87 deletions

View File

@ -1,15 +1,13 @@
"""The Flux LED/MagicLight integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any, Final
from flux_led import DeviceType
from flux_led.aio import AIOWifiLedBulb
from flux_led.aioscanner import AIOBulbScanner
from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
from flux_led.const import ATTR_ID
from flux_led.scanner import FluxLEDDiscovery
from homeassistant import config_entries
@ -33,11 +31,16 @@ from .const import (
DISCOVER_SCAN_TIMEOUT,
DOMAIN,
FLUX_LED_DISCOVERY,
FLUX_LED_DISCOVERY_LOCK,
FLUX_LED_EXCEPTIONS,
SIGNAL_STATE_UPDATED,
STARTUP_SCAN_TIMEOUT,
)
from .discovery import (
async_discover_device,
async_discover_devices,
async_name_from_discovery,
async_trigger_discovery,
)
_LOGGER = logging.getLogger(__name__)
@ -55,18 +58,6 @@ def async_wifi_bulb_for_host(host: str) -> AIOWifiLedBulb:
return AIOWifiLedBulb(host)
@callback
def async_name_from_discovery(device: FluxLEDDiscovery) -> str:
"""Convert a flux_led discovery to a human readable name."""
mac_address = device[ATTR_ID]
if mac_address is None:
return device[ATTR_IPADDR]
short_mac = mac_address[-6:]
if device[ATTR_MODEL_DESCRIPTION]:
return f"{device[ATTR_MODEL_DESCRIPTION]} {short_mac}"
return f"{device[ATTR_MODEL]} {short_mac}"
@callback
def async_update_entry_from_discovery(
hass: HomeAssistant, entry: config_entries.ConfigEntry, device: FluxLEDDiscovery
@ -83,52 +74,6 @@ def async_update_entry_from_discovery(
)
async def async_discover_devices(
hass: HomeAssistant, timeout: int, address: str | None = None
) -> list[FluxLEDDiscovery]:
"""Discover flux led devices."""
domain_data = hass.data.setdefault(DOMAIN, {})
if FLUX_LED_DISCOVERY_LOCK not in domain_data:
domain_data[FLUX_LED_DISCOVERY_LOCK] = asyncio.Lock()
async with domain_data[FLUX_LED_DISCOVERY_LOCK]:
scanner = AIOBulbScanner()
try:
discovered = await scanner.async_scan(timeout=timeout, address=address)
except OSError as ex:
_LOGGER.debug("Scanning failed with error: %s", ex)
return []
else:
return discovered
async def async_discover_device(
hass: HomeAssistant, host: str
) -> FluxLEDDiscovery | None:
"""Direct discovery at a single ip instead of broadcast."""
# If we are missing the unique_id we should be able to fetch it
# from the device by doing a directed discovery at the host only
for device in await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT, host):
if device[ATTR_IPADDR] == host:
return device
return None
@callback
def async_trigger_discovery(
hass: HomeAssistant,
discovered_devices: list[FluxLEDDiscovery],
) -> None:
"""Trigger config flows for discovered devices."""
for device in discovered_devices:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DISCOVERY},
data={**device},
)
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the flux_led component."""
domain_data = hass.data.setdefault(DOMAIN, {})
@ -173,11 +118,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
raise ConfigEntryNotReady(
str(ex) or f"Timed out trying to connect to {device.ipaddr}"
) from ex
coordinator = FluxLedUpdateCoordinator(hass, device)
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(
entry, PLATFORMS_BY_TYPE[device.device_type]
)
platforms = PLATFORMS_BY_TYPE[device.device_type]
hass.config_entries.async_setup_platforms(entry, platforms)
entry.async_on_unload(entry.add_update_listener(async_update_listener))
return True
@ -188,8 +133,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device: AIOWifiLedBulb = hass.data[DOMAIN][entry.entry_id].device
platforms = PLATFORMS_BY_TYPE[device.device_type]
if unload_ok := await hass.config_entries.async_unload_platforms(entry, platforms):
coordinator = hass.data[DOMAIN].pop(entry.entry_id)
await coordinator.device.async_stop()
del hass.data[DOMAIN][entry.entry_id]
await device.async_stop()
return unload_ok

View File

@ -16,13 +16,7 @@ from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import DiscoveryInfoType
from . import (
async_discover_device,
async_discover_devices,
async_name_from_discovery,
async_update_entry_from_discovery,
async_wifi_bulb_for_host,
)
from . import async_update_entry_from_discovery, async_wifi_bulb_for_host
from .const import (
CONF_CUSTOM_EFFECT_COLORS,
CONF_CUSTOM_EFFECT_SPEED_PCT,
@ -35,6 +29,11 @@ from .const import (
TRANSITION_JUMP,
TRANSITION_STROBE,
)
from .discovery import (
async_discover_device,
async_discover_devices,
async_name_from_discovery,
)
CONF_DEVICE: Final = "device"

View File

@ -39,7 +39,6 @@ DEFAULT_SCAN_INTERVAL: Final = 5
DEFAULT_EFFECT_SPEED: Final = 50
FLUX_LED_DISCOVERY: Final = "flux_led_discovery"
FLUX_LED_DISCOVERY_LOCK: Final = "flux_led_discovery_lock"
FLUX_LED_EXCEPTIONS: Final = (
asyncio.TimeoutError,

View File

@ -0,0 +1,91 @@
"""The Flux LED/MagicLight integration discovery."""
from __future__ import annotations
import asyncio
import logging
from flux_led.aioscanner import AIOBulbScanner
from flux_led.const import ATTR_ID, ATTR_IPADDR, ATTR_MODEL, ATTR_MODEL_DESCRIPTION
from flux_led.scanner import FluxLEDDiscovery
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.core import HomeAssistant, callback
from .const import DISCOVER_SCAN_TIMEOUT, DOMAIN
_LOGGER = logging.getLogger(__name__)
@callback
def async_name_from_discovery(device: FluxLEDDiscovery) -> str:
"""Convert a flux_led discovery to a human readable name."""
mac_address = device[ATTR_ID]
if mac_address is None:
return device[ATTR_IPADDR]
short_mac = mac_address[-6:]
if device[ATTR_MODEL_DESCRIPTION]:
return f"{device[ATTR_MODEL_DESCRIPTION]} {short_mac}"
return f"{device[ATTR_MODEL]} {short_mac}"
async def async_discover_devices(
hass: HomeAssistant, timeout: int, address: str | None = None
) -> list[FluxLEDDiscovery]:
"""Discover flux led devices."""
if address:
targets = [address]
else:
targets = [
str(address)
for address in await network.async_get_ipv4_broadcast_addresses(hass)
]
scanner = AIOBulbScanner()
for idx, discovered in enumerate(
await asyncio.gather(
*[
scanner.async_scan(timeout=timeout, address=address)
for address in targets
],
return_exceptions=True,
)
):
if isinstance(discovered, Exception):
_LOGGER.debug("Scanning %s failed with error: %s", targets[idx], discovered)
continue
if not address:
return scanner.getBulbInfo()
return [
device for device in scanner.getBulbInfo() if device[ATTR_IPADDR] == address
]
async def async_discover_device(
hass: HomeAssistant, host: str
) -> FluxLEDDiscovery | None:
"""Direct discovery at a single ip instead of broadcast."""
# If we are missing the unique_id we should be able to fetch it
# from the device by doing a directed discovery at the host only
for device in await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT, host):
if device[ATTR_IPADDR] == host:
return device
return None
@callback
def async_trigger_discovery(
hass: HomeAssistant,
discovered_devices: list[FluxLEDDiscovery],
) -> None:
"""Trigger config flows for discovered devices."""
for device in discovered_devices:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DISCOVERY},
data={**device},
)
)

View File

@ -2,6 +2,7 @@
"domain": "flux_led",
"name": "Magic Home",
"config_flow": true,
"dependencies": ["network"],
"documentation": "https://www.home-assistant.io/integrations/flux_led",
"requirements": ["flux_led==0.26.15"],
"quality_scale": "platinum",

View File

@ -168,10 +168,10 @@ def _patch_discovery(device=None, no_device=False):
@contextmanager
def _patcher():
with patch(
"homeassistant.components.flux_led.AIOBulbScanner.async_scan",
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan",
new=_discovery,
), patch(
"homeassistant.components.flux_led.AIOBulbScanner.getBulbInfo",
"homeassistant.components.flux_led.discovery.AIOBulbScanner.getBulbInfo",
return_value=[] if no_device else [device or FLUX_DISCOVERY],
):
yield

View File

@ -1,5 +1,7 @@
"""Tests for the flux_led integration."""
from unittest.mock import patch
import pytest
from tests.common import mock_device_registry
@ -9,3 +11,23 @@ from tests.common import mock_device_registry
def device_reg_fixture(hass):
"""Return an empty, loaded, registry."""
return mock_device_registry(hass)
@pytest.fixture
def mock_single_broadcast_address():
"""Mock network's async_async_get_ipv4_broadcast_addresses."""
with patch(
"homeassistant.components.network.async_get_ipv4_broadcast_addresses",
return_value={"10.255.255.255"},
):
yield
@pytest.fixture
def mock_multiple_broadcast_addresses():
"""Mock network's async_async_get_ipv4_broadcast_addresses to return multiple addresses."""
with patch(
"homeassistant.components.network.async_get_ipv4_broadcast_addresses",
return_value={"10.255.255.255", "192.168.0.255"},
):
yield

View File

@ -3,6 +3,8 @@ from __future__ import annotations
from unittest.mock import patch
from flux_led.aioscanner import AIOBulbScanner
from flux_led.scanner import FluxLEDDiscovery
import pytest
from homeassistant.components import flux_led
@ -26,23 +28,50 @@ from . import (
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.usefixtures("mock_single_broadcast_address")
async def test_configuring_flux_led_causes_discovery(hass: HomeAssistant) -> None:
"""Test that specifying empty config does discovery."""
with patch(
"homeassistant.components.flux_led.AIOBulbScanner.async_scan"
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan"
) as scan, patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.getBulbInfo"
) as discover:
discover.return_value = [FLUX_DISCOVERY]
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
assert len(discover.mock_calls) == 1
assert len(scan.mock_calls) == 1
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert len(discover.mock_calls) == 2
assert len(scan.mock_calls) == 2
async_fire_time_changed(hass, utcnow() + flux_led.DISCOVERY_INTERVAL)
await hass.async_block_till_done()
assert len(discover.mock_calls) == 3
assert len(scan.mock_calls) == 3
@pytest.mark.usefixtures("mock_multiple_broadcast_addresses")
async def test_configuring_flux_led_causes_discovery_multiple_addresses(
hass: HomeAssistant,
) -> None:
"""Test that specifying empty config does discovery."""
with patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.async_scan"
) as scan, patch(
"homeassistant.components.flux_led.discovery.AIOBulbScanner.getBulbInfo"
) as discover:
discover.return_value = [FLUX_DISCOVERY]
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()
assert len(scan.mock_calls) == 2
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert len(scan.mock_calls) == 4
async_fire_time_changed(hass, utcnow() + flux_led.DISCOVERY_INTERVAL)
await hass.async_block_till_done()
assert len(scan.mock_calls) == 6
async def test_config_entry_reload(hass: HomeAssistant) -> None:
@ -78,20 +107,32 @@ async def test_config_entry_retry(hass: HomeAssistant) -> None:
],
)
async def test_config_entry_fills_unique_id_with_directed_discovery(
hass: HomeAssistant, discovery: dict[str, str], title: str
hass: HomeAssistant, discovery: FluxLEDDiscovery, title: str
) -> None:
"""Test that the unique id is added if its missing via directed (not broadcast) discovery."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=None
domain=DOMAIN, data={CONF_NAME: "bogus", CONF_HOST: IP_ADDRESS}, unique_id=None
)
config_entry.add_to_hass(hass)
assert config_entry.unique_id is None
async def _discovery(self, *args, address=None, **kwargs):
# Only return discovery results when doing directed discovery
return [discovery] if address == IP_ADDRESS else []
class MockBulbScanner(AIOBulbScanner):
def __init__(self) -> None:
self._last_address: str | None = None
super().__init__()
async def async_scan(
self, timeout: int = 10, address: str | None = None
) -> list[FluxLEDDiscovery]:
self._last_address = address
return [discovery] if address == IP_ADDRESS else []
def getBulbInfo(self) -> FluxLEDDiscovery:
return [discovery] if self._last_address == IP_ADDRESS else []
with patch(
"homeassistant.components.flux_led.AIOBulbScanner.async_scan", new=_discovery
"homeassistant.components.flux_led.discovery.AIOBulbScanner",
return_value=MockBulbScanner(),
), _patch_wifibulb():
await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
await hass.async_block_till_done()