Flux led config flow (#56354)

Co-authored-by: Milan Meulemans <milan.meulemans@live.be>
Co-authored-by: Erik Montnemery <erik@montnemery.com>
Co-authored-by: J. Nick Koston <nick@koston.org>

.coveragerc

@@ -343,7 +343,6 @@ omit =
     homeassistant/components/flume/sensor.py
     homeassistant/components/flunearyou/__init__.py
     homeassistant/components/flunearyou/sensor.py
-    homeassistant/components/flux_led/light.py
     homeassistant/components/folder/sensor.py
     homeassistant/components/folder_watcher/*
     homeassistant/components/foobot/sensor.py

.strict-typing

@@ -41,6 +41,7 @@ homeassistant.components.energy.*
 homeassistant.components.fastdotcom.*
 homeassistant.components.fitbit.*
 homeassistant.components.flunearyou.*
+homeassistant.components.flux_led.*
 homeassistant.components.forecast_solar.*
 homeassistant.components.fritzbox.*
 homeassistant.components.frontend.*

CODEOWNERS

@@ -174,6 +174,7 @@ homeassistant/components/flo/* @dmulcahey
 homeassistant/components/flock/* @fabaff
 homeassistant/components/flume/* @ChrisMandich @bdraco
 homeassistant/components/flunearyou/* @bachya
+homeassistant/components/flux_led/* @icemanch
 homeassistant/components/forecast_solar/* @klaasnicolaas @frenck
 homeassistant/components/forked_daapd/* @uvjustin
 homeassistant/components/fortios/* @kimfrellsen

homeassistant/components/flux_led/__init__.py

@@ -1 +1,142 @@
-"""The flux_led component."""
+"""The Flux LED/MagicLight integration."""
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+from typing import Any, Final
+
+from flux_led import BulbScanner, WifiLedBulb
+
+from homeassistant import config_entries
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.debounce import Debouncer
+from homeassistant.helpers.event import async_track_time_interval
+from homeassistant.helpers.typing import ConfigType
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    DISCOVER_SCAN_TIMEOUT,
+    DOMAIN,
+    FLUX_LED_DISCOVERY,
+    FLUX_LED_EXCEPTIONS,
+    STARTUP_SCAN_TIMEOUT,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS: Final = ["light"]
+DISCOVERY_INTERVAL: Final = timedelta(minutes=15)
+REQUEST_REFRESH_DELAY: Final = 0.65
+
+
+async def async_wifi_bulb_for_host(hass: HomeAssistant, host: str) -> WifiLedBulb:
+    """Create a WifiLedBulb from a host."""
+    return await hass.async_add_executor_job(WifiLedBulb, host)
+
+
+async def async_discover_devices(
+    hass: HomeAssistant, timeout: int
+) -> list[dict[str, str]]:
+    """Discover flux led devices."""
+
+    def _scan_with_timeout() -> list[dict[str, str]]:
+        scanner = BulbScanner()
+        discovered: list[dict[str, str]] = scanner.scan(timeout=timeout)
+        return discovered
+
+    return await hass.async_add_executor_job(_scan_with_timeout)
+
+
+@callback
+def async_trigger_discovery(
+    hass: HomeAssistant,
+    discovered_devices: list[dict[str, Any]],
+) -> 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[DOMAIN] = {}
+    domain_data[FLUX_LED_DISCOVERY] = await async_discover_devices(
+        hass, STARTUP_SCAN_TIMEOUT
+    )
+
+    async def _async_discovery(*_: Any) -> None:
+        async_trigger_discovery(
+            hass, await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT)
+        )
+
+    async_trigger_discovery(hass, domain_data[FLUX_LED_DISCOVERY])
+    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STARTED, _async_discovery)
+    async_track_time_interval(hass, _async_discovery, DISCOVERY_INTERVAL)
+    return True
+
+
+async def async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Update listener."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Flux LED/MagicLight from a config entry."""
+
+    coordinator = FluxLedUpdateCoordinator(hass, entry.data[CONF_HOST])
+    await coordinator.async_config_entry_first_refresh()
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+    entry.async_on_unload(entry.add_update_listener(async_update_listener))
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+    return unload_ok
+
+
+class FluxLedUpdateCoordinator(DataUpdateCoordinator):
+    """DataUpdateCoordinator to gather data for a specific flux_led device."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        host: str,
+    ) -> None:
+        """Initialize DataUpdateCoordinator to gather data for specific device."""
+        self.host = host
+        self.device: WifiLedBulb | None = None
+        update_interval = timedelta(seconds=5)
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=host,
+            update_interval=update_interval,
+            # We don't want an immediate refresh since the device
+            # takes a moment to reflect the state change
+            request_refresh_debouncer=Debouncer(
+                hass, _LOGGER, cooldown=REQUEST_REFRESH_DELAY, immediate=False
+            ),
+        )
+
+    async def _async_update_data(self) -> None:
+        """Fetch all device and sensor data from api."""
+        try:
+            if not self.device:
+                self.device = await async_wifi_bulb_for_host(self.hass, self.host)
+            else:
+                await self.hass.async_add_executor_job(self.device.update_state)
+        except FLUX_LED_EXCEPTIONS as ex:
+            raise UpdateFailed(ex) from ex

homeassistant/components/flux_led/config_flow.py

@@ -0,0 +1,270 @@
+"""Config flow for Flux LED/MagicLight."""
+from __future__ import annotations
+
+import logging
+from typing import Any, Final
+
+from flux_led import WifiLedBulb
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS
+from homeassistant.const import CONF_HOST, CONF_MAC, CONF_MODE, CONF_NAME, CONF_PROTOCOL
+from homeassistant.core import callback
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import device_registry as dr
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.typing import DiscoveryInfoType
+
+from . import async_discover_devices, async_wifi_bulb_for_host
+from .const import (
+    CONF_CUSTOM_EFFECT_COLORS,
+    CONF_CUSTOM_EFFECT_SPEED_PCT,
+    CONF_CUSTOM_EFFECT_TRANSITION,
+    DEFAULT_EFFECT_SPEED,
+    DISCOVER_SCAN_TIMEOUT,
+    DOMAIN,
+    FLUX_HOST,
+    FLUX_LED_EXCEPTIONS,
+    FLUX_MAC,
+    FLUX_MODEL,
+    MODE_AUTO,
+    MODE_RGB,
+    MODE_RGBW,
+    MODE_WHITE,
+    TRANSITION_GRADUAL,
+    TRANSITION_JUMP,
+    TRANSITION_STROBE,
+)
+
+CONF_DEVICE: Final = "device"
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for FluxLED/MagicHome Integration."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._discovered_devices: dict[str, dict[str, Any]] = {}
+        self._discovered_device: dict[str, Any] = {}
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry: config_entries.ConfigEntry) -> OptionsFlow:
+        """Get the options flow for the Flux LED component."""
+        return OptionsFlow(config_entry)
+
+    async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
+        """Handle configuration via YAML import."""
+        _LOGGER.debug("Importing configuration from YAML for flux_led")
+        host = user_input[CONF_HOST]
+        self._async_abort_entries_match({CONF_HOST: host})
+        if mac := user_input[CONF_MAC]:
+            await self.async_set_unique_id(dr.format_mac(mac), raise_on_progress=False)
+            self._abort_if_unique_id_configured(updates={CONF_HOST: host})
+        return self.async_create_entry(
+            title=user_input[CONF_NAME],
+            data={
+                CONF_HOST: host,
+                CONF_NAME: user_input[CONF_NAME],
+                CONF_PROTOCOL: user_input.get(CONF_PROTOCOL),
+            },
+            options={
+                CONF_MODE: user_input[CONF_MODE],
+                CONF_CUSTOM_EFFECT_COLORS: user_input[CONF_CUSTOM_EFFECT_COLORS],
+                CONF_CUSTOM_EFFECT_SPEED_PCT: user_input[CONF_CUSTOM_EFFECT_SPEED_PCT],
+                CONF_CUSTOM_EFFECT_TRANSITION: user_input[
+                    CONF_CUSTOM_EFFECT_TRANSITION
+                ],
+            },
+        )
+
+    async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
+        """Handle discovery via dhcp."""
+        self._discovered_device = {
+            FLUX_HOST: discovery_info[IP_ADDRESS],
+            FLUX_MODEL: discovery_info[HOSTNAME],
+            FLUX_MAC: discovery_info[MAC_ADDRESS].replace(":", ""),
+        }
+        return await self._async_handle_discovery()
+
+    async def async_step_discovery(
+        self, discovery_info: DiscoveryInfoType
+    ) -> FlowResult:
+        """Handle discovery."""
+        self._discovered_device = discovery_info
+        return await self._async_handle_discovery()
+
+    async def _async_handle_discovery(self) -> FlowResult:
+        """Handle any discovery."""
+        device = self._discovered_device
+        mac = dr.format_mac(device[FLUX_MAC])
+        host = device[FLUX_HOST]
+        await self.async_set_unique_id(mac)
+        self._abort_if_unique_id_configured(updates={CONF_HOST: host})
+        for entry in self._async_current_entries(include_ignore=False):
+            if entry.data[CONF_HOST] == host and not entry.unique_id:
+                name = f"{device[FLUX_MODEL]} {device[FLUX_MAC]}"
+                self.hass.config_entries.async_update_entry(
+                    entry,
+                    data={**entry.data, CONF_NAME: name},
+                    title=name,
+                    unique_id=mac,
+                )
+                return self.async_abort(reason="already_configured")
+        self.context[CONF_HOST] = host
+        for progress in self._async_in_progress():
+            if progress.get("context", {}).get(CONF_HOST) == host:
+                return self.async_abort(reason="already_in_progress")
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_discovery_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Confirm discovery."""
+        if user_input is not None:
+            return self._async_create_entry_from_device(self._discovered_device)
+
+        self._set_confirm_only()
+        placeholders = self._discovered_device
+        self.context["title_placeholders"] = placeholders
+        return self.async_show_form(
+            step_id="discovery_confirm", description_placeholders=placeholders
+        )
+
+    @callback
+    def _async_create_entry_from_device(self, device: dict[str, Any]) -> FlowResult:
+        """Create a config entry from a device."""
+        self._async_abort_entries_match({CONF_HOST: device[FLUX_HOST]})
+        if device.get(FLUX_MAC):
+            name = f"{device[FLUX_MODEL]} {device[FLUX_MAC]}"
+        else:
+            name = device[FLUX_HOST]
+        return self.async_create_entry(
+            title=name,
+            data={
+                CONF_HOST: device[FLUX_HOST],
+                CONF_NAME: name,
+            },
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors = {}
+        if user_input is not None:
+            if not (host := user_input[CONF_HOST]):
+                return await self.async_step_pick_device()
+            try:
+                await self._async_try_connect(host)
+            except FLUX_LED_EXCEPTIONS:
+                errors["base"] = "cannot_connect"
+            else:
+                return self._async_create_entry_from_device(
+                    {FLUX_MAC: None, FLUX_MODEL: None, FLUX_HOST: host}
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema({vol.Optional(CONF_HOST, default=""): str}),
+            errors=errors,
+        )
+
+    async def async_step_pick_device(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the step to pick discovered device."""
+        if user_input is not None:
+            mac = user_input[CONF_DEVICE]
+            await self.async_set_unique_id(mac, raise_on_progress=False)
+            return self._async_create_entry_from_device(self._discovered_devices[mac])
+
+        current_unique_ids = self._async_current_ids()
+        current_hosts = {
+            entry.data[CONF_HOST]
+            for entry in self._async_current_entries(include_ignore=False)
+        }
+        discovered_devices = await async_discover_devices(
+            self.hass, DISCOVER_SCAN_TIMEOUT
+        )
+        self._discovered_devices = {
+            dr.format_mac(device[FLUX_MAC]): device for device in discovered_devices
+        }
+        devices_name = {
+            mac: f"{device[FLUX_MODEL]} {mac} ({device[FLUX_HOST]})"
+            for mac, device in self._discovered_devices.items()
+            if mac not in current_unique_ids and device[FLUX_HOST] not in current_hosts
+        }
+        # Check if there is at least one device
+        if not devices_name:
+            return self.async_abort(reason="no_devices_found")
+        return self.async_show_form(
+            step_id="pick_device",
+            data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(devices_name)}),
+        )
+
+    async def _async_try_connect(self, host: str) -> WifiLedBulb:
+        """Try to connect."""
+        self._async_abort_entries_match({CONF_HOST: host})
+        return await async_wifi_bulb_for_host(self.hass, host)
+
+
+class OptionsFlow(config_entries.OptionsFlow):
+    """Handle flux_led options."""
+
+    def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
+        """Initialize the flux_led options flow."""
+        self._config_entry = config_entry
+
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Configure the options."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            return self.async_create_entry(title="", data=user_input)
+
+        options = self._config_entry.options
+        options_schema = vol.Schema(
+            {
+                vol.Required(
+                    CONF_MODE, default=options.get(CONF_MODE, MODE_AUTO)
+                ): vol.All(
+                    cv.string,
+                    vol.In(
+                        [
+                            MODE_AUTO,
+                            MODE_RGBW,
+                            MODE_RGB,
+                            MODE_WHITE,
+                        ]
+                    ),
+                ),
+                vol.Optional(
+                    CONF_CUSTOM_EFFECT_COLORS,
+                    default=options.get(CONF_CUSTOM_EFFECT_COLORS, ""),
+                ): str,
+                vol.Optional(
+                    CONF_CUSTOM_EFFECT_SPEED_PCT,
+                    default=options.get(
+                        CONF_CUSTOM_EFFECT_SPEED_PCT, DEFAULT_EFFECT_SPEED
+                    ),
+                ): vol.All(vol.Coerce(int), vol.Range(min=1, max=100)),
+                vol.Optional(
+                    CONF_CUSTOM_EFFECT_TRANSITION,
+                    default=options.get(
+                        CONF_CUSTOM_EFFECT_TRANSITION, TRANSITION_GRADUAL
+                    ),
+                ): vol.In([TRANSITION_GRADUAL, TRANSITION_JUMP, TRANSITION_STROBE]),
+            }
+        )
+
+        return self.async_show_form(
+            step_id="init", data_schema=options_schema, errors=errors
+        )

homeassistant/components/flux_led/const.py

@@ -0,0 +1,51 @@
+"""Constants of the FluxLed/MagicHome Integration."""
+
+import socket
+from typing import Final
+
+DOMAIN: Final = "flux_led"
+
+API: Final = "flux_api"
+
+
+CONF_AUTOMATIC_ADD: Final = "automatic_add"
+DEFAULT_NETWORK_SCAN_INTERVAL: Final = 120
+DEFAULT_SCAN_INTERVAL: Final = 5
+DEFAULT_EFFECT_SPEED: Final = 50
+
+FLUX_LED_DISCOVERY: Final = "flux_led_discovery"
+
+FLUX_LED_EXCEPTIONS: Final = (socket.timeout, BrokenPipeError)
+
+STARTUP_SCAN_TIMEOUT: Final = 5
+DISCOVER_SCAN_TIMEOUT: Final = 10
+
+CONF_DEVICES: Final = "devices"
+CONF_CUSTOM_EFFECT: Final = "custom_effect"
+CONF_MODEL: Final = "model"
+
+MODE_AUTO: Final = "auto"
+MODE_RGB: Final = "rgb"
+MODE_RGBW: Final = "rgbw"
+
+
+# This mode enables white value to be controlled by brightness.
+# RGB value is ignored when this mode is specified.
+MODE_WHITE: Final = "w"
+
+TRANSITION_GRADUAL: Final = "gradual"
+TRANSITION_JUMP: Final = "jump"
+TRANSITION_STROBE: Final = "strobe"
+
+CONF_COLORS: Final = "colors"
+CONF_SPEED_PCT: Final = "speed_pct"
+CONF_TRANSITION: Final = "transition"
+
+
+CONF_CUSTOM_EFFECT_COLORS: Final = "custom_effect_colors"
+CONF_CUSTOM_EFFECT_SPEED_PCT: Final = "custom_effect_speed_pct"
+CONF_CUSTOM_EFFECT_TRANSITION: Final = "custom_effect_transition"
+
+FLUX_HOST: Final = "ipaddr"
+FLUX_MAC: Final = "id"
+FLUX_MODEL: Final = "model"

homeassistant/components/flux_led/light.py

@@ -1,10 +1,16 @@
-"""Support for Flux lights."""
+"""Support for FluxLED/MagicHome lights."""
+from __future__ import annotations
+
+import ast
+from functools import partial
 import logging
 import random
+from typing import Any, Final, cast
 
-from flux_led import BulbScanner, WifiLedBulb
+from flux_led import WifiLedBulb
 import voluptuous as vol
 
+from homeassistant import config_entries
 from homeassistant.components.light import (
     ATTR_BRIGHTNESS,
     ATTR_COLOR_TEMP,
@@ -21,56 +27,84 @@ from homeassistant.components.light import (
     SUPPORT_WHITE_VALUE,
     LightEntity,
 )
-from homeassistant.const import ATTR_MODE, CONF_DEVICES, CONF_NAME, CONF_PROTOCOL
+from homeassistant.const import (
+    ATTR_IDENTIFIERS,
+    ATTR_MANUFACTURER,
+    ATTR_MODE,
+    ATTR_MODEL,
+    ATTR_NAME,
+    CONF_DEVICES,
+    CONF_HOST,
+    CONF_MAC,
+    CONF_MODE,
+    CONF_NAME,
+    CONF_PROTOCOL,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_platform
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 import homeassistant.util.color as color_util
 
+from . import FluxLedUpdateCoordinator
+from .const import (
+    CONF_AUTOMATIC_ADD,
+    CONF_COLORS,
+    CONF_CUSTOM_EFFECT,
+    CONF_CUSTOM_EFFECT_COLORS,
+    CONF_CUSTOM_EFFECT_SPEED_PCT,
+    CONF_CUSTOM_EFFECT_TRANSITION,
+    CONF_SPEED_PCT,
+    CONF_TRANSITION,
+    DEFAULT_EFFECT_SPEED,
+    DOMAIN,
+    FLUX_HOST,
+    FLUX_LED_DISCOVERY,
+    FLUX_MAC,
+    MODE_AUTO,
+    MODE_RGB,
+    MODE_RGBW,
+    MODE_WHITE,
+    TRANSITION_GRADUAL,
+    TRANSITION_JUMP,
+    TRANSITION_STROBE,
+)
+
 _LOGGER = logging.getLogger(__name__)
 
-CONF_AUTOMATIC_ADD = "automatic_add"
-CONF_CUSTOM_EFFECT = "custom_effect"
-CONF_COLORS = "colors"
-CONF_SPEED_PCT = "speed_pct"
-CONF_TRANSITION = "transition"
+SUPPORT_FLUX_LED: Final = SUPPORT_BRIGHTNESS | SUPPORT_EFFECT | SUPPORT_COLOR
 
-DOMAIN = "flux_led"
-
-SUPPORT_FLUX_LED = SUPPORT_BRIGHTNESS | SUPPORT_EFFECT | SUPPORT_COLOR
-
-MODE_RGB = "rgb"
-MODE_RGBW = "rgbw"
-
-# This mode enables white value to be controlled by brightness.
-# RGB value is ignored when this mode is specified.
-MODE_WHITE = "w"
 
 # Constant color temp values for 2 flux_led special modes
 # Warm-white and Cool-white modes
-COLOR_TEMP_WARM_VS_COLD_WHITE_CUT_OFF = 285
+COLOR_TEMP_WARM_VS_COLD_WHITE_CUT_OFF: Final = 285
 
 # List of supported effects which aren't already declared in LIGHT
-EFFECT_RED_FADE = "red_fade"
-EFFECT_GREEN_FADE = "green_fade"
-EFFECT_BLUE_FADE = "blue_fade"
-EFFECT_YELLOW_FADE = "yellow_fade"
-EFFECT_CYAN_FADE = "cyan_fade"
-EFFECT_PURPLE_FADE = "purple_fade"
-EFFECT_WHITE_FADE = "white_fade"
-EFFECT_RED_GREEN_CROSS_FADE = "rg_cross_fade"
-EFFECT_RED_BLUE_CROSS_FADE = "rb_cross_fade"
-EFFECT_GREEN_BLUE_CROSS_FADE = "gb_cross_fade"
-EFFECT_COLORSTROBE = "colorstrobe"
-EFFECT_RED_STROBE = "red_strobe"
-EFFECT_GREEN_STROBE = "green_strobe"
-EFFECT_BLUE_STROBE = "blue_strobe"
-EFFECT_YELLOW_STROBE = "yellow_strobe"
-EFFECT_CYAN_STROBE = "cyan_strobe"
-EFFECT_PURPLE_STROBE = "purple_strobe"
-EFFECT_WHITE_STROBE = "white_strobe"
-EFFECT_COLORJUMP = "colorjump"
-EFFECT_CUSTOM = "custom"
+EFFECT_RED_FADE: Final = "red_fade"
+EFFECT_GREEN_FADE: Final = "green_fade"
+EFFECT_BLUE_FADE: Final = "blue_fade"
+EFFECT_YELLOW_FADE: Final = "yellow_fade"
+EFFECT_CYAN_FADE: Final = "cyan_fade"
+EFFECT_PURPLE_FADE: Final = "purple_fade"
+EFFECT_WHITE_FADE: Final = "white_fade"
+EFFECT_RED_GREEN_CROSS_FADE: Final = "rg_cross_fade"
+EFFECT_RED_BLUE_CROSS_FADE: Final = "rb_cross_fade"
+EFFECT_GREEN_BLUE_CROSS_FADE: Final = "gb_cross_fade"
+EFFECT_COLORSTROBE: Final = "colorstrobe"
+EFFECT_RED_STROBE: Final = "red_strobe"
+EFFECT_GREEN_STROBE: Final = "green_strobe"
+EFFECT_BLUE_STROBE: Final = "blue_strobe"
+EFFECT_YELLOW_STROBE: Final = "yellow_strobe"
+EFFECT_CYAN_STROBE: Final = "cyan_strobe"
+EFFECT_PURPLE_STROBE: Final = "purple_strobe"
+EFFECT_WHITE_STROBE: Final = "white_strobe"
+EFFECT_COLORJUMP: Final = "colorjump"
+EFFECT_CUSTOM: Final = "custom"
 
-EFFECT_MAP = {
+EFFECT_MAP: Final = {
     EFFECT_COLORLOOP: 0x25,
     EFFECT_RED_FADE: 0x26,
     EFFECT_GREEN_FADE: 0x27,
@@ -92,39 +126,36 @@ EFFECT_MAP = {
     EFFECT_WHITE_STROBE: 0x37,
     EFFECT_COLORJUMP: 0x38,
 }
-EFFECT_CUSTOM_CODE = 0x60
+EFFECT_ID_NAME: Final = {v: k for k, v in EFFECT_MAP.items()}
+EFFECT_CUSTOM_CODE: Final = 0x60
 
-TRANSITION_GRADUAL = "gradual"
-TRANSITION_JUMP = "jump"
-TRANSITION_STROBE = "strobe"
+WHITE_MODES: Final = {MODE_RGBW}
 
-FLUX_EFFECT_LIST = sorted(EFFECT_MAP) + [EFFECT_RANDOM]
+FLUX_EFFECT_LIST: Final = sorted(EFFECT_MAP) + [EFFECT_RANDOM]
 
-CUSTOM_EFFECT_SCHEMA = vol.Schema(
-    {
-        vol.Required(CONF_COLORS): vol.All(
-            cv.ensure_list,
-            vol.Length(min=1, max=16),
-            [
-                vol.All(
-                    vol.ExactSequence((cv.byte, cv.byte, cv.byte)), vol.Coerce(tuple)
-                )
-            ],
-        ),
-        vol.Optional(CONF_SPEED_PCT, default=50): vol.All(
-            vol.Range(min=0, max=100), vol.Coerce(int)
-        ),
-        vol.Optional(CONF_TRANSITION, default=TRANSITION_GRADUAL): vol.All(
-            cv.string, vol.In([TRANSITION_GRADUAL, TRANSITION_JUMP, TRANSITION_STROBE])
-        ),
-    }
-)
+SERVICE_CUSTOM_EFFECT: Final = "set_custom_effect"
 
-DEVICE_SCHEMA = vol.Schema(
+CUSTOM_EFFECT_DICT: Final = {
+    vol.Required(CONF_COLORS): vol.All(
+        cv.ensure_list,
+        vol.Length(min=1, max=16),
+        [vol.All(vol.ExactSequence((cv.byte, cv.byte, cv.byte)), vol.Coerce(tuple))],
+    ),
+    vol.Optional(CONF_SPEED_PCT, default=50): vol.All(
+        vol.Range(min=0, max=100), vol.Coerce(int)
+    ),
+    vol.Optional(CONF_TRANSITION, default=TRANSITION_GRADUAL): vol.All(
+        cv.string, vol.In([TRANSITION_GRADUAL, TRANSITION_JUMP, TRANSITION_STROBE])
+    ),
+}
+
+CUSTOM_EFFECT_SCHEMA: Final = vol.Schema(CUSTOM_EFFECT_DICT)
+
+DEVICE_SCHEMA: Final = vol.Schema(
     {
         vol.Optional(CONF_NAME): cv.string,
-        vol.Optional(ATTR_MODE, default=MODE_RGBW): vol.All(
-            cv.string, vol.In([MODE_RGBW, MODE_RGB, MODE_WHITE])
+        vol.Optional(ATTR_MODE, default=MODE_AUTO): vol.All(
+            cv.string, vol.In([MODE_AUTO, MODE_RGBW, MODE_RGB, MODE_WHITE])
         ),
         vol.Optional(CONF_PROTOCOL): vol.All(cv.string, vol.In(["ledenet"])),
         vol.Optional(CONF_CUSTOM_EFFECT): CUSTOM_EFFECT_SCHEMA,
@@ -139,160 +170,206 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
 )
 
 
-def setup_platform(hass, config, add_entities, discovery_info=None):
+async def async_setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    async_add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> bool:
+    """Set up the flux led platform."""
+    domain_data = hass.data[DOMAIN]
+    discovered_mac_by_host = {
+        device[FLUX_HOST]: device[FLUX_MAC]
+        for device in domain_data[FLUX_LED_DISCOVERY]
+    }
+    for host, device_config in config.get(CONF_DEVICES, {}).items():
+        _LOGGER.warning(
+            "Configuring flux_led via yaml is deprecated; the configuration for"
+            " %s has been migrated to a config entry and can be safely removed",
+            host,
+        )
+        custom_effects = device_config.get(CONF_CUSTOM_EFFECT, {})
+        hass.async_create_task(
+            hass.config_entries.flow.async_init(
+                DOMAIN,
+                context={"source": config_entries.SOURCE_IMPORT},
+                data={
+                    CONF_HOST: host,
+                    CONF_MAC: discovered_mac_by_host.get(host),
+                    CONF_NAME: device_config[CONF_NAME],
+                    CONF_PROTOCOL: device_config.get(CONF_PROTOCOL),
+                    CONF_MODE: device_config.get(ATTR_MODE, MODE_AUTO),
+                    CONF_CUSTOM_EFFECT_COLORS: str(custom_effects.get(CONF_COLORS)),
+                    CONF_CUSTOM_EFFECT_SPEED_PCT: custom_effects.get(
+                        CONF_SPEED_PCT, DEFAULT_EFFECT_SPEED
+                    ),
+                    CONF_CUSTOM_EFFECT_TRANSITION: custom_effects.get(
+                        CONF_TRANSITION, TRANSITION_GRADUAL
+                    ),
+                },
+            )
+        )
+    return True
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: config_entries.ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
     """Set up the Flux lights."""
-    lights = []
-    light_ips = []
+    coordinator: FluxLedUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
 
-    for ipaddr, device_config in config.get(CONF_DEVICES, {}).items():
-        device = {}
-        device["name"] = device_config[CONF_NAME]
-        device["ipaddr"] = ipaddr
-        device[CONF_PROTOCOL] = device_config.get(CONF_PROTOCOL)
-        device[ATTR_MODE] = device_config[ATTR_MODE]
-        device[CONF_CUSTOM_EFFECT] = device_config.get(CONF_CUSTOM_EFFECT)
-        light = FluxLight(device)
-        lights.append(light)
-        light_ips.append(ipaddr)
+    platform = entity_platform.async_get_current_platform()
+    platform.async_register_entity_service(
+        SERVICE_CUSTOM_EFFECT,
+        CUSTOM_EFFECT_DICT,
+        "set_custom_effect",
+    )
+    options = entry.options
 
-    if not config.get(CONF_AUTOMATIC_ADD, False):
-        add_entities(lights, True)
-        return
+    try:
+        custom_effect_colors = ast.literal_eval(
+            options.get(CONF_CUSTOM_EFFECT_COLORS) or "[]"
+        )
+    except (ValueError, TypeError, SyntaxError, MemoryError) as ex:
+        _LOGGER.warning(
+            "Could not parse custom effect colors for %s: %s", entry.unique_id, ex
+        )
+        custom_effect_colors = []
 
-    # Find the bulbs on the LAN
-    scanner = BulbScanner()
-    scanner.scan(timeout=10)
-    for device in scanner.getBulbInfo():
-        ipaddr = device["ipaddr"]
-        if ipaddr in light_ips:
-            continue
-        device["name"] = f"{device['id']} {ipaddr}"
-        device[ATTR_MODE] = None
-        device[CONF_PROTOCOL] = None
-        device[CONF_CUSTOM_EFFECT] = None
-        light = FluxLight(device)
-        lights.append(light)
-
-    add_entities(lights, True)
+    async_add_entities(
+        [
+            FluxLight(
+                coordinator,
+                entry.unique_id,
+                entry.data[CONF_NAME],
+                options.get(CONF_MODE) or MODE_AUTO,
+                list(custom_effect_colors),
+                options.get(CONF_CUSTOM_EFFECT_SPEED_PCT, DEFAULT_EFFECT_SPEED),
+                options.get(CONF_CUSTOM_EFFECT_TRANSITION, TRANSITION_GRADUAL),
+            )
+        ]
+    )
 
 
-class FluxLight(LightEntity):
+class FluxLight(CoordinatorEntity, LightEntity):
     """Representation of a Flux light."""
 
-    def __init__(self, device):
+    coordinator: FluxLedUpdateCoordinator
+
+    def __init__(
+        self,
+        coordinator: FluxLedUpdateCoordinator,
+        unique_id: str | None,
+        name: str,
+        mode: str,
+        custom_effect_colors: list[tuple[int, int, int]],
+        custom_effect_speed_pct: int,
+        custom_effect_transition: str,
+    ) -> None:
         """Initialize the light."""
-        self._name = device["name"]
-        self._ipaddr = device["ipaddr"]
-        self._protocol = device[CONF_PROTOCOL]
-        self._mode = device[ATTR_MODE]
-        self._custom_effect = device[CONF_CUSTOM_EFFECT]
-        self._bulb = None
-        self._error_reported = False
-
-    def _connect(self):
-        """Connect to Flux light."""
-
-        self._bulb = WifiLedBulb(self._ipaddr, timeout=5)
-        if self._protocol:
-            self._bulb.setProtocol(self._protocol)
-
-        # After bulb object is created the status is updated. We can
-        # now set the correct mode if it was not explicitly defined.
-        if not self._mode:
-            if self._bulb.rgbwcapable:
-                self._mode = MODE_RGBW
-            else:
-                self._mode = MODE_RGB
-
-    def _disconnect(self):
-        """Disconnect from Flux light."""
-        self._bulb = None
+        super().__init__(coordinator)
+        self._bulb: WifiLedBulb = coordinator.device
+        self._name = name
+        self._unique_id = unique_id
+        self._ip_address = coordinator.host
+        self._mode = mode
+        self._custom_effect_colors = custom_effect_colors
+        self._custom_effect_speed_pct = custom_effect_speed_pct
+        self._custom_effect_transition = custom_effect_transition
 
     @property
-    def available(self) -> bool:
-        """Return True if entity is available."""
-        return self._bulb is not None
+    def unique_id(self) -> str | None:
+        """Return the unique ID of the light."""
+        return self._unique_id
 
     @property
-    def name(self):
-        """Return the name of the device if any."""
+    def name(self) -> str:
+        """Return the name of the device."""
         return self._name
 
     @property
-    def is_on(self):
+    def is_on(self) -> bool:
         """Return true if device is on."""
-        return self._bulb.isOn()
+        return cast(bool, self._bulb.is_on)
 
     @property
-    def brightness(self):
+    def brightness(self) -> int:
         """Return the brightness of this light between 0..255."""
         if self._mode == MODE_WHITE:
             return self.white_value
-
-        return self._bulb.brightness
+        return cast(int, self._bulb.brightness)
 
     @property
-    def hs_color(self):
+    def hs_color(self) -> tuple[float, float] | None:
         """Return the color property."""
         return color_util.color_RGB_to_hs(*self._bulb.getRgb())
 
     @property
-    def supported_features(self):
+    def supported_features(self) -> int:
         """Flag supported features."""
-        if self._mode == MODE_RGBW:
-            return SUPPORT_FLUX_LED | SUPPORT_WHITE_VALUE | SUPPORT_COLOR_TEMP
-
         if self._mode == MODE_WHITE:
             return SUPPORT_BRIGHTNESS
-
+        if self._mode in WHITE_MODES:
+            return SUPPORT_FLUX_LED | SUPPORT_WHITE_VALUE | SUPPORT_COLOR_TEMP
         return SUPPORT_FLUX_LED
 
     @property
-    def white_value(self):
+    def white_value(self) -> int:
         """Return the white value of this light between 0..255."""
-        return self._bulb.getRgbw()[3]
+        return cast(int, self._bulb.getRgbw()[3])
 
     @property
-    def effect_list(self):
+    def effect_list(self) -> list[str]:
         """Return the list of supported effects."""
-        if self._custom_effect:
+        if self._custom_effect_colors:
             return FLUX_EFFECT_LIST + [EFFECT_CUSTOM]
-
         return FLUX_EFFECT_LIST
 
     @property
-    def effect(self):
+    def effect(self) -> str | None:
         """Return the current effect."""
-        current_mode = self._bulb.raw_state[3]
-
-        if current_mode == EFFECT_CUSTOM_CODE:
+        if (current_mode := self._bulb.raw_state[3]) == EFFECT_CUSTOM_CODE:
             return EFFECT_CUSTOM
+        return EFFECT_ID_NAME.get(current_mode)
 
-        for effect, code in EFFECT_MAP.items():
-            if current_mode == code:
-                return effect
+    @property
+    def extra_state_attributes(self) -> dict[str, str]:
+        """Return the attributes."""
+        return {
+            "ip_address": self._ip_address,
+        }
 
-        return None
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return the device information."""
+        assert self._unique_id is not None
+        return {
+            ATTR_IDENTIFIERS: {(DOMAIN, self._unique_id)},
+            ATTR_NAME: self._name,
+            ATTR_MANUFACTURER: "FluxLED/Magic Home",
+            ATTR_MODEL: "LED Lights",
+        }
 
-    def turn_on(self, **kwargs):
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn the specified or all lights on."""
+        await self.hass.async_add_executor_job(partial(self._turn_on, **kwargs))
+        await self.coordinator.async_request_refresh()
+
+    def _turn_on(self, **kwargs: Any) -> None:
         """Turn the specified or all lights on."""
         if not self.is_on:
             self._bulb.turnOn()
 
-        hs_color = kwargs.get(ATTR_HS_COLOR)
-
-        if hs_color:
-            rgb = color_util.color_hs_to_RGB(*hs_color)
+        if hs_color := kwargs.get(ATTR_HS_COLOR):
+            rgb: tuple[int, int, int] | None = color_util.color_hs_to_RGB(*hs_color)
         else:
             rgb = None
 
         brightness = kwargs.get(ATTR_BRIGHTNESS)
-        effect = kwargs.get(ATTR_EFFECT)
-        white = kwargs.get(ATTR_WHITE_VALUE)
-        color_temp = kwargs.get(ATTR_COLOR_TEMP)
-
         # handle special modes
-        if color_temp is not None:
+        if (color_temp := kwargs.get(ATTR_COLOR_TEMP)) is not None:
             if brightness is None:
                 brightness = self.brightness
             if color_temp > COLOR_TEMP_WARM_VS_COLD_WHITE_CUT_OFF:
@@ -301,6 +378,8 @@ class FluxLight(LightEntity):
                 self._bulb.setRgbw(w2=brightness)
             return
 
+        white = kwargs.get(ATTR_WHITE_VALUE)
+        effect = kwargs.get(ATTR_EFFECT)
         # Show warning if effect set with rgb, brightness, or white level
         if effect and (brightness or white or rgb):
             _LOGGER.warning(
@@ -315,12 +394,13 @@ class FluxLight(LightEntity):
             )
             return
 
+        # Custom effect
         if effect == EFFECT_CUSTOM:
-            if self._custom_effect:
+            if self._custom_effect_colors:
                 self._bulb.setCustomPattern(
-                    self._custom_effect[CONF_COLORS],
-                    self._custom_effect[CONF_SPEED_PCT],
-                    self._custom_effect[CONF_TRANSITION],
+                    self._custom_effect_colors,
+                    self._custom_effect_speed_pct,
+                    self._custom_effect_transition,
                 )
             return
 
@@ -333,42 +413,58 @@ class FluxLight(LightEntity):
         if brightness is None:
             brightness = self.brightness
 
+        # handle W only mode (use brightness instead of white value)
+        if self._mode == MODE_WHITE:
+            self._bulb.setRgbw(0, 0, 0, w=brightness)
+            return
+
+        if white is None and self._mode in WHITE_MODES:
+            white = self.white_value
+
         # Preserve color on brightness/white level change
         if rgb is None:
             rgb = self._bulb.getRgb()
 
-        if white is None and self._mode == MODE_RGBW:
-            white = self.white_value
-
-        # handle W only mode (use brightness instead of white value)
-        if self._mode == MODE_WHITE:
-            self._bulb.setRgbw(0, 0, 0, w=brightness)
-
         # handle RGBW mode
-        elif self._mode == MODE_RGBW:
+        if self._mode == MODE_RGBW:
             self._bulb.setRgbw(*tuple(rgb), w=white, brightness=brightness)
+            return
 
         # handle RGB mode
-        else:
-            self._bulb.setRgb(*tuple(rgb), brightness=brightness)
+        self._bulb.setRgb(*tuple(rgb), brightness=brightness)
 
-    def turn_off(self, **kwargs):
+    def set_custom_effect(
+        self, colors: list[tuple[int, int, int]], speed_pct: int, transition: str
+    ) -> None:
+        """Set a custom effect on the bulb."""
+        self._bulb.setCustomPattern(
+            colors,
+            speed_pct,
+            transition,
+        )
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
         """Turn the specified or all lights off."""
-        self._bulb.turnOff()
+        await self.hass.async_add_executor_job(self._bulb.turnOff)
+        await self.coordinator.async_request_refresh()
 
-    def update(self):
-        """Synchronize state with bulb."""
-        if not self.available:
-            try:
-                self._connect()
-                self._error_reported = False
-            except OSError:
-                self._disconnect()
-                if not self._error_reported:
-                    _LOGGER.warning(
-                        "Failed to connect to bulb %s, %s", self._ipaddr, self._name
-                    )
-                    self._error_reported = True
-                return
+    async def async_added_to_hass(self) -> None:
+        """When entity is added to hass."""
+        await super().async_added_to_hass()
+        if self._mode and self._mode != MODE_AUTO:
+            return
 
-        self._bulb.update_state(retry=2)
+        if self._bulb.mode == "ww":
+            self._mode = MODE_WHITE
+        elif self._bulb.rgbwcapable:
+            self._mode = MODE_RGBW
+        else:
+            self._mode = MODE_RGB
+        _LOGGER.debug(
+            "Detected mode for %s (%s) with raw_state=%s rgbwcapable=%s is %s",
+            self._name,
+            self.unique_id,
+            self._bulb.raw_state,
+            self._bulb.rgbwcapable,
+            self._mode,
+        )

homeassistant/components/flux_led/manifest.json

@@ -1,8 +1,25 @@
 {
   "domain": "flux_led",
-  "name": "Flux LED/MagicLight",
+  "name": "Flux LED/MagicHome",
+  "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/flux_led",
   "requirements": ["flux_led==0.22"],
-  "codeowners": [],
-  "iot_class": "local_polling"
+  "codeowners": ["@icemanch"],
+  "iot_class": "local_polling",
+  "dhcp": [
+  {
+    "macaddress": "18B905*",
+    "hostname": "[ba][lk]*"
+  },
+  {
+    "macaddress": "249494*",
+    "hostname": "[ba][lk]*"
+  },
+  {
+    "macaddress": "B4E842*",
+    "hostname": "[ba][lk]*"
+  }
+  ]
 }
+
+

homeassistant/components/flux_led/services.yaml

@@ -0,0 +1,38 @@
+set_custom_effect:
+  description: Set a custom light effect.
+  target:
+    entity:
+      integration: flux_led
+      domain: light
+  fields:
+    colors:
+      description: List of colors for the custom effect (RGB). (Max 16 Colors)
+      example: |
+        - [255,0,0]
+        - [0,255,0]
+        - [0,0,255]
+      required: true
+      selector:
+        object:
+    speed_pct:
+      description: Effect speed for the custom effect (0-100).
+      example: 80
+      default: 50
+      required: false
+      selector:
+        number:
+          min: 1
+          step: 1
+          max: 100
+          unit_of_measurement: "%"
+    transition:
+      description: Effect transition.
+      example: 'jump'
+      default: 'gradual'
+      required: false
+      selector:
+        select:
+          options:
+            - "gradual"
+            - "jump"
+            - "strobe"

homeassistant/components/flux_led/strings.json

@@ -0,0 +1,36 @@
+{
+  "config": {
+    "flow_title": "{model} {id} ({ipaddr})",
+    "step": {
+      "user": {
+        "description": "If you leave the host empty, discovery will be used to find devices.",
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]"
+        }
+      },
+      "discovery_confirm": {
+        "description": "Do you want to setup {model} {id} ({ipaddr})?"
+      }      
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
+    },    
+    "abort": {
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
+    }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "data": {
+          "mode": "The chosen brightness mode.",
+          "custom_effect_colors": "Custom Effect: List of 1 to 16 [R,G,B] colors. Example: [255,0,255],[60,128,0]",
+          "custom_effect_speed_pct": "Custom Effect: Speed in percents for the effect that switch colors.",
+          "custom_effect_transition": "Custom Effect: Type of transition between the colors."
+        }
+      }
+    }
+  }
+}

homeassistant/components/flux_led/translations/en.json

@@ -0,0 +1,35 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Device is already configured",
+            "no_devices_found": "No devices found on the network"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect"
+        },
+        "flow_title": "{model} {id} ({ipaddr})",
+        "step": {
+            "discovery_confirm": {
+                "description": "Do you want to setup {model} {id} ({ipaddr})?"
+            },
+            "user": {
+                "data": {
+                    "host": "Host"
+                },
+                "description": "If you leave the host empty, discovery will be used to find devices."
+            }
+        }
+    },
+    "options": {
+        "step": {
+            "init": {
+                "data": {
+                    "custom_effect_colors": "Custom Effect: List of 1 to 16 [R,G,B] colors. Example: [255,0,255],[60,128,0]",
+                    "custom_effect_speed_pct": "Custom Effect: Speed in percents for the effect that switch colors.",
+                    "custom_effect_transition": "Custom Effect: Type of transition between the colors.",
+                    "mode": "The chosen brightness mode."
+                }
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -87,6 +87,7 @@ FLOWS = [
     "flo",
     "flume",
     "flunearyou",
+    "flux_led",
     "forecast_solar",
     "forked_daapd",
     "foscam",

homeassistant/generated/dhcp.py

@@ -71,6 +71,21 @@ DHCP = [
         "domain": "flume",
         "hostname": "flume-gw-*"
     },
+    {
+        "domain": "flux_led",
+        "macaddress": "18B905*",
+        "hostname": "[ba][lk]*"
+    },
+    {
+        "domain": "flux_led",
+        "macaddress": "249494*",
+        "hostname": "[ba][lk]*"
+    },
+    {
+        "domain": "flux_led",
+        "macaddress": "B4E842*",
+        "hostname": "[ba][lk]*"
+    },
     {
         "domain": "goalzero",
         "hostname": "yeti*"

mypy.ini

@@ -462,6 +462,17 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.flux_led.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.forecast_solar.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_test_all.txt

@@ -377,6 +377,9 @@ fjaraskupan==1.0.1
 # homeassistant.components.flipr
 flipr-api==1.4.1
 
+# homeassistant.components.flux_led
+flux_led==0.22
+
 # homeassistant.components.homekit
 fnvhash==0.1.0
 

tests/components/flux_led/__init.py__

@@ -0,0 +1 @@
+"""Tests for the flux_led integration."""

tests/components/flux_led/__init__.py

@@ -0,0 +1,57 @@
+"""Tests for the flux_led integration."""
+from __future__ import annotations
+
+import socket
+from unittest.mock import MagicMock, patch
+
+from flux_led import WifiLedBulb
+
+from homeassistant.components.dhcp import (
+    HOSTNAME as DHCP_HOSTNAME,
+    IP_ADDRESS as DHCP_IP_ADDRESS,
+    MAC_ADDRESS as DHCP_MAC_ADDRESS,
+)
+from homeassistant.components.flux_led.const import FLUX_HOST, FLUX_MAC, FLUX_MODEL
+
+MODULE = "homeassistant.components.flux_led"
+MODULE_CONFIG_FLOW = "homeassistant.components.flux_led.config_flow"
+IP_ADDRESS = "127.0.0.1"
+MODEL = "AZ120444"
+MAC_ADDRESS = "aa:bb:cc:dd:ee:ff"
+FLUX_MAC_ADDRESS = "aabbccddeeff"
+
+DEFAULT_ENTRY_TITLE = f"{MODEL} {FLUX_MAC_ADDRESS}"
+
+DHCP_DISCOVERY = {
+    DHCP_HOSTNAME: MODEL,
+    DHCP_IP_ADDRESS: IP_ADDRESS,
+    DHCP_MAC_ADDRESS: MAC_ADDRESS,
+}
+FLUX_DISCOVERY = {FLUX_HOST: IP_ADDRESS, FLUX_MODEL: MODEL, FLUX_MAC: FLUX_MAC_ADDRESS}
+
+
+def _mocked_bulb() -> WifiLedBulb:
+    bulb = MagicMock(auto_spec=WifiLedBulb)
+    bulb.getRgb = MagicMock(return_value=[255, 0, 0])
+    bulb.getRgbw = MagicMock(return_value=[255, 0, 0, 50])
+    bulb.brightness = 128
+    bulb.rgbwcapable = True
+    return bulb
+
+
+def _patch_discovery(device=None, no_device=False):
+    def _discovery(*args, **kwargs):
+        if no_device:
+            return []
+        return [FLUX_DISCOVERY]
+
+    return patch("homeassistant.components.flux_led.BulbScanner.scan", new=_discovery)
+
+
+def _patch_wifibulb(device=None, no_device=False):
+    def _wifi_led_bulb(*args, **kwargs):
+        if no_device:
+            raise socket.timeout
+        return device if device else _mocked_bulb()
+
+    return patch("homeassistant.components.flux_led.WifiLedBulb", new=_wifi_led_bulb)

tests/components/flux_led/test_config_flow.py

@@ -0,0 +1,456 @@
+"""Define tests for the Flux LED/Magic Home config flow."""
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant import config_entries, setup
+from homeassistant.components.flux_led.const import (
+    CONF_CUSTOM_EFFECT_COLORS,
+    CONF_CUSTOM_EFFECT_SPEED_PCT,
+    CONF_CUSTOM_EFFECT_TRANSITION,
+    DOMAIN,
+    MODE_AUTO,
+    MODE_RGB,
+    TRANSITION_JUMP,
+    TRANSITION_STROBE,
+)
+from homeassistant.const import (
+    CONF_DEVICE,
+    CONF_HOST,
+    CONF_MAC,
+    CONF_MODE,
+    CONF_NAME,
+    CONF_PROTOCOL,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import RESULT_TYPE_ABORT, RESULT_TYPE_FORM
+
+from . import (
+    DEFAULT_ENTRY_TITLE,
+    DHCP_DISCOVERY,
+    DHCP_HOSTNAME,
+    DHCP_IP_ADDRESS,
+    DHCP_MAC_ADDRESS,
+    FLUX_DISCOVERY,
+    IP_ADDRESS,
+    MAC_ADDRESS,
+    MODULE,
+    _patch_discovery,
+    _patch_wifibulb,
+)
+
+from tests.common import MockConfigEntry
+
+
+async def test_discovery(hass: HomeAssistant):
+    """Test setting up discovery."""
+    with _patch_discovery(), _patch_wifibulb():
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+        await hass.async_block_till_done()
+        assert result["type"] == "form"
+        assert result["step_id"] == "user"
+        assert not result["errors"]
+
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+        assert result2["type"] == "form"
+        assert result2["step_id"] == "pick_device"
+        assert not result2["errors"]
+
+        # test we can try again
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+        assert result["type"] == "form"
+        assert result["step_id"] == "user"
+        assert not result["errors"]
+
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+        assert result2["type"] == "form"
+        assert result2["step_id"] == "pick_device"
+        assert not result2["errors"]
+
+    with _patch_discovery(), _patch_wifibulb(), patch(
+        f"{MODULE}.async_setup", return_value=True
+    ) as mock_setup, patch(
+        f"{MODULE}.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DEVICE: MAC_ADDRESS},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "create_entry"
+    assert result3["title"] == DEFAULT_ENTRY_TITLE
+    assert result3["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
+    mock_setup.assert_called_once()
+    mock_setup_entry.assert_called_once()
+
+    # ignore configured devices
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert not result["errors"]
+
+    with _patch_discovery(), _patch_wifibulb():
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "abort"
+    assert result2["reason"] == "no_devices_found"
+
+
+async def test_discovery_with_existing_device_present(hass: HomeAssistant):
+    """Test setting up discovery."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN, data={CONF_HOST: "127.0.0.2"}, unique_id="dd:dd:dd:dd:dd:dd"
+    )
+    config_entry.add_to_hass(hass)
+
+    with _patch_discovery(), _patch_wifibulb(no_device=True):
+        await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert not result["errors"]
+
+    with _patch_discovery(), _patch_wifibulb():
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "pick_device"
+    assert not result2["errors"]
+
+    # Now abort and make sure we can start over
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert not result["errors"]
+
+    with _patch_discovery(), _patch_wifibulb():
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "pick_device"
+    assert not result2["errors"]
+
+    with _patch_discovery(), _patch_wifibulb(), patch(
+        f"{MODULE}.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_DEVICE: MAC_ADDRESS}
+        )
+        assert result3["type"] == "create_entry"
+        assert result3["title"] == DEFAULT_ENTRY_TITLE
+        assert result3["data"] == {
+            CONF_HOST: IP_ADDRESS,
+            CONF_NAME: DEFAULT_ENTRY_TITLE,
+        }
+        await hass.async_block_till_done()
+
+    mock_setup_entry.assert_called_once()
+
+    # ignore configured devices
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert not result["errors"]
+
+    with _patch_discovery(), _patch_wifibulb():
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "abort"
+    assert result2["reason"] == "no_devices_found"
+
+
+async def test_discovery_no_device(hass: HomeAssistant):
+    """Test discovery without device."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with _patch_discovery(no_device=True), _patch_wifibulb():
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "abort"
+    assert result2["reason"] == "no_devices_found"
+
+
+async def test_import(hass: HomeAssistant):
+    """Test import from yaml."""
+    config = {
+        CONF_HOST: IP_ADDRESS,
+        CONF_MAC: MAC_ADDRESS,
+        CONF_NAME: "floor lamp",
+        CONF_PROTOCOL: "ledenet",
+        CONF_MODE: MODE_RGB,
+        CONF_CUSTOM_EFFECT_COLORS: "[255,0,0], [0,0,255]",
+        CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
+        CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_STROBE,
+    }
+
+    # Success
+    with _patch_discovery(), _patch_wifibulb(), patch(
+        f"{MODULE}.async_setup", return_value=True
+    ) as mock_setup, patch(
+        f"{MODULE}.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == "floor lamp"
+    assert result["data"] == {
+        CONF_HOST: IP_ADDRESS,
+        CONF_NAME: "floor lamp",
+        CONF_PROTOCOL: "ledenet",
+    }
+    assert result["options"] == {
+        CONF_MODE: MODE_RGB,
+        CONF_CUSTOM_EFFECT_COLORS: "[255,0,0], [0,0,255]",
+        CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
+        CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_STROBE,
+    }
+    mock_setup.assert_called_once()
+    mock_setup_entry.assert_called_once()
+
+    # Duplicate
+    with _patch_discovery(), _patch_wifibulb():
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=config
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "abort"
+    assert result["reason"] == "already_configured"
+
+
+async def test_manual(hass: HomeAssistant):
+    """Test manually setup."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert not result["errors"]
+
+    # Cannot connect (timeout)
+    with _patch_discovery(no_device=True), _patch_wifibulb(no_device=True):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_HOST: IP_ADDRESS}
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "user"
+    assert result2["errors"] == {"base": "cannot_connect"}
+
+    # Success
+    with _patch_discovery(), _patch_wifibulb(), patch(
+        f"{MODULE}.async_setup", return_value=True
+    ), patch(f"{MODULE}.async_setup_entry", return_value=True):
+        result4 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_HOST: IP_ADDRESS}
+        )
+        await hass.async_block_till_done()
+    assert result4["type"] == "create_entry"
+    assert result4["title"] == IP_ADDRESS
+    assert result4["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: IP_ADDRESS}
+
+    # Duplicate
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    with _patch_discovery(no_device=True), _patch_wifibulb(no_device=True):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_HOST: IP_ADDRESS}
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "abort"
+    assert result2["reason"] == "already_configured"
+
+
+async def test_manual_no_discovery_data(hass: HomeAssistant):
+    """Test manually setup without discovery data."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert not result["errors"]
+
+    with _patch_discovery(no_device=True), _patch_wifibulb(), patch(
+        f"{MODULE}.async_setup", return_value=True
+    ), patch(f"{MODULE}.async_setup_entry", return_value=True):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {CONF_HOST: IP_ADDRESS}
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == "create_entry"
+    assert result["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: IP_ADDRESS}
+
+
+async def test_discovered_by_discovery_and_dhcp(hass):
+    """Test we get the form with discovery and abort for dhcp source when we get both."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with _patch_discovery(), _patch_wifibulb():
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_DISCOVERY},
+            data=FLUX_DISCOVERY,
+        )
+        await hass.async_block_till_done()
+    assert result["type"] == RESULT_TYPE_FORM
+    assert result["errors"] is None
+
+    with _patch_discovery(), _patch_wifibulb():
+        result2 = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_DHCP},
+            data=DHCP_DISCOVERY,
+        )
+        await hass.async_block_till_done()
+    assert result2["type"] == RESULT_TYPE_ABORT
+    assert result2["reason"] == "already_in_progress"
+
+    with _patch_discovery(), _patch_wifibulb():
+        result3 = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_DHCP},
+            data={
+                DHCP_HOSTNAME: "any",
+                DHCP_IP_ADDRESS: IP_ADDRESS,
+                DHCP_MAC_ADDRESS: "00:00:00:00:00:00",
+            },
+        )
+        await hass.async_block_till_done()
+    assert result3["type"] == RESULT_TYPE_ABORT
+    assert result3["reason"] == "already_in_progress"
+
+
+@pytest.mark.parametrize(
+    "source, data",
+    [
+        (config_entries.SOURCE_DHCP, DHCP_DISCOVERY),
+        (config_entries.SOURCE_DISCOVERY, FLUX_DISCOVERY),
+    ],
+)
+async def test_discovered_by_dhcp_or_discovery(hass, source, data):
+    """Test we can setup when discovered from dhcp or discovery."""
+
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with _patch_discovery(), _patch_wifibulb():
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": source}, data=data
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == RESULT_TYPE_FORM
+    assert result["errors"] is None
+
+    with _patch_discovery(), _patch_wifibulb(), patch(
+        f"{MODULE}.async_setup", return_value=True
+    ) as mock_async_setup, patch(
+        f"{MODULE}.async_setup_entry", return_value=True
+    ) as mock_async_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+        await hass.async_block_till_done()
+
+    assert result2["type"] == "create_entry"
+    assert result2["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
+    assert mock_async_setup.called
+    assert mock_async_setup_entry.called
+
+
+@pytest.mark.parametrize(
+    "source, data",
+    [
+        (config_entries.SOURCE_DHCP, DHCP_DISCOVERY),
+        (config_entries.SOURCE_DISCOVERY, FLUX_DISCOVERY),
+    ],
+)
+async def test_discovered_by_dhcp_or_discovery_adds_missing_unique_id(
+    hass, source, data
+):
+    """Test we can setup when discovered from dhcp or discovery."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: IP_ADDRESS})
+    config_entry.add_to_hass(hass)
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with _patch_discovery(), _patch_wifibulb():
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": source}, data=data
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == RESULT_TYPE_ABORT
+    assert result["reason"] == "already_configured"
+
+    assert config_entry.unique_id == MAC_ADDRESS
+
+
+async def test_options(hass: HomeAssistant):
+    """Test options flow."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        options={
+            CONF_MODE: MODE_RGB,
+            CONF_CUSTOM_EFFECT_COLORS: "[255,0,0], [0,0,255]",
+            CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
+            CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_STROBE,
+        },
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+
+    with _patch_discovery(), _patch_wifibulb():
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    result = await hass.config_entries.options.async_init(config_entry.entry_id)
+    assert result["type"] == "form"
+    assert result["step_id"] == "init"
+
+    user_input = {
+        CONF_MODE: MODE_AUTO,
+        CONF_CUSTOM_EFFECT_COLORS: "[0,0,255], [255,0,0]",
+        CONF_CUSTOM_EFFECT_SPEED_PCT: 50,
+        CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_JUMP,
+    }
+    with _patch_discovery(), _patch_wifibulb():
+        result2 = await hass.config_entries.options.async_configure(
+            result["flow_id"], user_input
+        )
+        await hass.async_block_till_done()
+    assert result2["type"] == "create_entry"
+    assert result2["data"] == user_input
+    assert result2["data"] == config_entry.options
+    assert hass.states.get("light.az120444_aabbccddeeff") is not None

tests/components/flux_led/test_init.py

@@ -0,0 +1,58 @@
+"""Tests for the flux_led component."""
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from homeassistant.components import flux_led
+from homeassistant.components.flux_led.const import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+from homeassistant.util.dt import utcnow
+
+from . import FLUX_DISCOVERY, IP_ADDRESS, MAC_ADDRESS, _patch_discovery, _patch_wifibulb
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+
+async def test_configuring_flux_led_causes_discovery(hass: HomeAssistant) -> None:
+    """Test that specifying empty config does discovery."""
+    with patch("homeassistant.components.flux_led.BulbScanner.scan") 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
+        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
+        await hass.async_block_till_done()
+        assert len(discover.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
+
+
+async def test_config_entry_reload(hass: HomeAssistant) -> None:
+    """Test that a config entry can be reloaded."""
+    config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=MAC_ADDRESS)
+    config_entry.add_to_hass(hass)
+    with _patch_discovery(), _patch_wifibulb():
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+        assert config_entry.state == ConfigEntryState.LOADED
+        await hass.config_entries.async_unload(config_entry.entry_id)
+        await hass.async_block_till_done()
+        assert config_entry.state == ConfigEntryState.NOT_LOADED
+
+
+async def test_config_entry_retry(hass: HomeAssistant) -> None:
+    """Test that a config entry can be retried."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS
+    )
+    config_entry.add_to_hass(hass)
+    with _patch_discovery(no_device=True), _patch_wifibulb(no_device=True):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+        assert config_entry.state == ConfigEntryState.SETUP_RETRY

tests/components/flux_led/test_light.py

@@ -0,0 +1,654 @@
+"""Tests for light platform."""
+from datetime import timedelta
+
+import pytest
+
+from homeassistant.components import flux_led
+from homeassistant.components.flux_led.const import (
+    CONF_COLORS,
+    CONF_CUSTOM_EFFECT,
+    CONF_CUSTOM_EFFECT_COLORS,
+    CONF_CUSTOM_EFFECT_SPEED_PCT,
+    CONF_CUSTOM_EFFECT_TRANSITION,
+    CONF_DEVICES,
+    CONF_SPEED_PCT,
+    CONF_TRANSITION,
+    DOMAIN,
+    MODE_AUTO,
+    TRANSITION_JUMP,
+)
+from homeassistant.components.flux_led.light import EFFECT_CUSTOM_CODE, FLUX_EFFECT_LIST
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_COLOR_MODE,
+    ATTR_COLOR_TEMP,
+    ATTR_EFFECT,
+    ATTR_EFFECT_LIST,
+    ATTR_HS_COLOR,
+    ATTR_SUPPORTED_COLOR_MODES,
+    DOMAIN as LIGHT_DOMAIN,
+)
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    CONF_HOST,
+    CONF_MODE,
+    CONF_NAME,
+    CONF_PLATFORM,
+    CONF_PROTOCOL,
+    STATE_OFF,
+    STATE_ON,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.setup import async_setup_component
+from homeassistant.util.dt import utcnow
+
+from . import (
+    DEFAULT_ENTRY_TITLE,
+    IP_ADDRESS,
+    MAC_ADDRESS,
+    _mocked_bulb,
+    _patch_discovery,
+    _patch_wifibulb,
+)
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+
+async def test_light_unique_id(hass: HomeAssistant) -> None:
+    """Test a light unique id."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+    entity_registry = er.async_get(hass)
+    assert entity_registry.async_get(entity_id).unique_id == MAC_ADDRESS
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+
+
+async def test_light_no_unique_id(hass: HomeAssistant) -> None:
+    """Test a light without a unique id."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN, data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    with _patch_discovery(no_device=True), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+    entity_registry = er.async_get(hass)
+    assert entity_registry.async_get(entity_id) is None
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+
+
+async def test_rgb_light(hass: HomeAssistant) -> None:
+    """Test an rgb light."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    bulb.rgbwcapable = False
+    bulb.protocol = None
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "hs"
+    assert attributes[ATTR_EFFECT_LIST] == FLUX_EFFECT_LIST
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["hs"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOff.assert_called_once()
+
+    bulb.is_on = False
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOn.assert_called_once()
+    bulb.turnOn.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
+        blocking=True,
+    )
+    bulb.setRgb.assert_called_with(255, 0, 0, brightness=100)
+    bulb.setRgb.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_HS_COLOR: (10, 30)},
+        blocking=True,
+    )
+    bulb.setRgb.assert_called_with(255, 191, 178, brightness=128)
+    bulb.setRgb.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
+        blocking=True,
+    )
+    bulb.setRgb.assert_called_once()
+    bulb.setRgb.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
+        blocking=True,
+    )
+    bulb.setPresetPattern.assert_called_with(43, 50)
+    bulb.setPresetPattern.reset_mock()
+
+
+async def test_rgbw_light(hass: HomeAssistant) -> None:
+    """Test an rgbw light."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "rgbw"
+    assert attributes[ATTR_EFFECT_LIST] == FLUX_EFFECT_LIST
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["color_temp", "hs", "rgbw"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOff.assert_called_once()
+
+    bulb.is_on = False
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOn.assert_called_once()
+    bulb.turnOn.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(255, 0, 0, w=50, brightness=100)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 150},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(w2=128)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 290},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(w=128)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_HS_COLOR: (10, 30)},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(255, 191, 178, w=50, brightness=128)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
+        blocking=True,
+    )
+    bulb.setRgb.assert_called_once()
+    bulb.setRgb.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
+        blocking=True,
+    )
+    bulb.setPresetPattern.assert_called_with(43, 50)
+    bulb.setPresetPattern.reset_mock()
+
+
+async def test_rgbcw_light(hass: HomeAssistant) -> None:
+    """Test an rgbcw light."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    bulb.raw_state = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
+    bulb.raw_state[9] = 1
+    bulb.raw_state[11] = 2
+
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "rgbw"
+    assert attributes[ATTR_EFFECT_LIST] == FLUX_EFFECT_LIST
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["color_temp", "hs", "rgbw"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOff.assert_called_once()
+
+    bulb.is_on = False
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOn.assert_called_once()
+    bulb.turnOn.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(255, 0, 0, w=50, brightness=100)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 150},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(w2=128)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 290},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(w=128)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_HS_COLOR: (10, 30)},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(255, 191, 178, w=50, brightness=128)
+    bulb.setRgbw.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
+        blocking=True,
+    )
+    bulb.setRgb.assert_called_once()
+    bulb.setRgb.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
+        blocking=True,
+    )
+    bulb.setPresetPattern.assert_called_with(43, 50)
+    bulb.setPresetPattern.reset_mock()
+
+
+async def test_white_light(hass: HomeAssistant) -> None:
+    """Test a white light."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    bulb.mode = "ww"
+    bulb.protocol = None
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 50
+    assert attributes[ATTR_COLOR_MODE] == "brightness"
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["brightness"]
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOff.assert_called_once()
+
+    bulb.is_on = False
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOn.assert_called_once()
+    bulb.turnOn.reset_mock()
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
+        blocking=True,
+    )
+    bulb.setRgbw.assert_called_with(0, 0, 0, w=100)
+    bulb.setRgbw.reset_mock()
+
+
+async def test_rgb_light_custom_effects(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test an rgb light with a custom effect."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+        options={
+            CONF_MODE: MODE_AUTO,
+            CONF_CUSTOM_EFFECT_COLORS: "[0,0,255], [255,0,0]",
+            CONF_CUSTOM_EFFECT_SPEED_PCT: 88,
+            CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_JUMP,
+        },
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "rgbw"
+    assert attributes[ATTR_EFFECT_LIST] == [*FLUX_EFFECT_LIST, "custom"]
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["color_temp", "hs", "rgbw"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOff.assert_called_once()
+
+    bulb.is_on = False
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "custom"},
+        blocking=True,
+    )
+    bulb.setCustomPattern.assert_called_with([[0, 0, 255], [255, 0, 0]], 88, "jump")
+    bulb.setCustomPattern.reset_mock()
+    bulb.raw_state = [0, 0, 0, EFFECT_CUSTOM_CODE, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
+    bulb.is_on = True
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=20))
+    await hass.async_block_till_done()
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_EFFECT] == "custom"
+
+    caplog.clear()
+    await hass.services.async_call(
+        LIGHT_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 55, ATTR_EFFECT: "custom"},
+        blocking=True,
+    )
+    bulb.setCustomPattern.assert_called_with([[0, 0, 255], [255, 0, 0]], 88, "jump")
+    bulb.setCustomPattern.reset_mock()
+    bulb.raw_state = [0, 0, 0, EFFECT_CUSTOM_CODE, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
+    bulb.is_on = True
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=20))
+    await hass.async_block_till_done()
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_EFFECT] == "custom"
+    assert "RGB, brightness and white level are ignored when" in caplog.text
+
+
+async def test_rgb_light_custom_effects_invalid_colors(hass: HomeAssistant) -> None:
+    """Test an rgb light with a invalid effect."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+        options={
+            CONF_MODE: MODE_AUTO,
+            CONF_CUSTOM_EFFECT_COLORS: ":: CANNOT BE PARSED ::",
+            CONF_CUSTOM_EFFECT_SPEED_PCT: 88,
+            CONF_CUSTOM_EFFECT_TRANSITION: TRANSITION_JUMP,
+        },
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "rgbw"
+    assert attributes[ATTR_EFFECT_LIST] == FLUX_EFFECT_LIST
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["color_temp", "hs", "rgbw"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+
+async def test_rgb_light_custom_effect_via_service(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test an rgb light with a custom effect set via the service."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "rgbw"
+    assert attributes[ATTR_EFFECT_LIST] == [*FLUX_EFFECT_LIST]
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["color_temp", "hs", "rgbw"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+    await hass.services.async_call(
+        LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
+    )
+    bulb.turnOff.assert_called_once()
+
+    bulb.is_on = False
+    async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == STATE_OFF
+
+    await hass.services.async_call(
+        DOMAIN,
+        "set_custom_effect",
+        {
+            ATTR_ENTITY_ID: entity_id,
+            CONF_COLORS: [[0, 0, 255], [255, 0, 0]],
+            CONF_SPEED_PCT: 30,
+            CONF_TRANSITION: "jump",
+        },
+        blocking=True,
+    )
+    bulb.setCustomPattern.assert_called_with([(0, 0, 255), (255, 0, 0)], 30, "jump")
+    bulb.setCustomPattern.reset_mock()
+
+
+async def test_rgbw_detection_without_protocol(hass: HomeAssistant) -> None:
+    """Test an rgbw detection without protocol."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE},
+        unique_id=MAC_ADDRESS,
+    )
+    config_entry.add_to_hass(hass)
+    bulb = _mocked_bulb()
+    bulb.protocol = None
+    bulb.rgbwprotocol = None
+    bulb.rgbwcapable = True
+    with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
+        await async_setup_component(hass, flux_led.DOMAIN, {flux_led.DOMAIN: {}})
+        await hass.async_block_till_done()
+
+    entity_id = "light.az120444_aabbccddeeff"
+
+    state = hass.states.get(entity_id)
+    assert state.state == STATE_ON
+    attributes = state.attributes
+    assert attributes[ATTR_BRIGHTNESS] == 128
+    assert attributes[ATTR_COLOR_MODE] == "rgbw"
+    assert attributes[ATTR_EFFECT_LIST] == FLUX_EFFECT_LIST
+    assert attributes[ATTR_SUPPORTED_COLOR_MODES] == ["color_temp", "hs", "rgbw"]
+    assert attributes[ATTR_HS_COLOR] == (0, 100)
+
+
+async def test_migrate_from_yaml(hass: HomeAssistant) -> None:
+    """Test migrate from yaml."""
+    config = {
+        LIGHT_DOMAIN: [
+            {
+                CONF_PLATFORM: DOMAIN,
+                CONF_DEVICES: {
+                    IP_ADDRESS: {
+                        CONF_NAME: "flux_lamppost",
+                        CONF_PROTOCOL: "ledenet",
+                        CONF_CUSTOM_EFFECT: {
+                            CONF_SPEED_PCT: 30,
+                            CONF_TRANSITION: "strobe",
+                            CONF_COLORS: [[255, 0, 0], [255, 255, 0], [0, 255, 0]],
+                        },
+                    }
+                },
+            }
+        ],
+    }
+    with _patch_discovery(), _patch_wifibulb():
+        await async_setup_component(hass, LIGHT_DOMAIN, config)
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+
+    entries = hass.config_entries.async_entries(DOMAIN)
+    assert entries
+
+    migrated_entry = None
+    for entry in entries:
+        if entry.unique_id == MAC_ADDRESS:
+            migrated_entry = entry
+            break
+
+    assert migrated_entry is not None
+    assert migrated_entry.data == {
+        CONF_HOST: IP_ADDRESS,
+        CONF_NAME: "flux_lamppost",
+        CONF_PROTOCOL: "ledenet",
+    }
+    assert migrated_entry.options == {
+        CONF_MODE: "auto",
+        CONF_CUSTOM_EFFECT_COLORS: "[(255, 0, 0), (255, 255, 0), (0, 255, 0)]",
+        CONF_CUSTOM_EFFECT_SPEED_PCT: 30,
+        CONF_CUSTOM_EFFECT_TRANSITION: "strobe",
+    }