Add network support to flux_led discovery

homeassistant/components/flux_led/__init__.py

@@ -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 @@
     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 @@ def _async_state_changed(*_: Any) -> None:
         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
 
 

homeassistant/components/flux_led/config_flow.py

@@ -16,13 +16,7 @@
 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 @@
     TRANSITION_JUMP,
     TRANSITION_STROBE,
 )
+from .discovery import (
+    async_discover_device,
+    async_discover_devices,
+    async_name_from_discovery,
+)
 
 CONF_DEVICE: Final = "device"
 

homeassistant/components/flux_led/const.py

@@ -39,7 +39,6 @@
 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,

homeassistant/components/flux_led/discovery.py

@@ -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},
+            )
+        )

homeassistant/components/flux_led/manifest.json

@@ -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.5"],
   "quality_scale": "platinum",

tests/components/flux_led/__init__.py

@@ -168,10 +168,10 @@ async def _discovery(*args, **kwargs):
     @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

tests/components/flux_led/conftest.py

@@ -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 @@
 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