From a95c6b10f7fb483b21f472cc6d39d7dcf6f280c0 Mon Sep 17 00:00:00 2001 From: icemanch Date: Sat, 2 Oct 2021 13:19:36 -0400 Subject: [PATCH] Flux led config flow (#56354) Co-authored-by: Milan Meulemans Co-authored-by: Erik Montnemery Co-authored-by: J. Nick Koston --- .coveragerc | 1 - .strict-typing | 1 + CODEOWNERS | 1 + homeassistant/components/flux_led/__init__.py | 143 +++- .../components/flux_led/config_flow.py | 270 ++++++++ homeassistant/components/flux_led/const.py | 51 ++ homeassistant/components/flux_led/light.py | 488 +++++++------ .../components/flux_led/manifest.json | 23 +- .../components/flux_led/services.yaml | 38 + .../components/flux_led/strings.json | 36 + .../components/flux_led/translations/en.json | 35 + homeassistant/generated/config_flows.py | 1 + homeassistant/generated/dhcp.py | 15 + mypy.ini | 11 + requirements_test_all.txt | 3 + tests/components/flux_led/__init.py__ | 1 + tests/components/flux_led/__init__.py | 57 ++ tests/components/flux_led/test_config_flow.py | 456 ++++++++++++ tests/components/flux_led/test_init.py | 58 ++ tests/components/flux_led/test_light.py | 654 ++++++++++++++++++ 20 files changed, 2142 insertions(+), 201 deletions(-) create mode 100644 homeassistant/components/flux_led/config_flow.py create mode 100644 homeassistant/components/flux_led/const.py create mode 100644 homeassistant/components/flux_led/services.yaml create mode 100644 homeassistant/components/flux_led/strings.json create mode 100644 homeassistant/components/flux_led/translations/en.json create mode 100644 tests/components/flux_led/__init.py__ create mode 100644 tests/components/flux_led/__init__.py create mode 100644 tests/components/flux_led/test_config_flow.py create mode 100644 tests/components/flux_led/test_init.py create mode 100644 tests/components/flux_led/test_light.py diff --git a/.coveragerc b/.coveragerc index a5d7eec9115..1a716ef4e5c 100644 --- a/.coveragerc +++ b/.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 diff --git a/.strict-typing b/.strict-typing index 50841c49f2f..8e957faabf0 100644 --- a/.strict-typing +++ b/.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.* diff --git a/CODEOWNERS b/CODEOWNERS index 70fb56f73f4..58225f4e36a 100644 --- a/CODEOWNERS +++ b/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 diff --git a/homeassistant/components/flux_led/__init__.py b/homeassistant/components/flux_led/__init__.py index 572d6e3c983..df7334a8ebc 100644 --- a/homeassistant/components/flux_led/__init__.py +++ b/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 diff --git a/homeassistant/components/flux_led/config_flow.py b/homeassistant/components/flux_led/config_flow.py new file mode 100644 index 00000000000..206c8f91433 --- /dev/null +++ b/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 + ) diff --git a/homeassistant/components/flux_led/const.py b/homeassistant/components/flux_led/const.py new file mode 100644 index 00000000000..6b2c4a8dace --- /dev/null +++ b/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" diff --git a/homeassistant/components/flux_led/light.py b/homeassistant/components/flux_led/light.py index 2f8d2cc5536..581d5fbaab6 100644 --- a/homeassistant/components/flux_led/light.py +++ b/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, + ) diff --git a/homeassistant/components/flux_led/manifest.json b/homeassistant/components/flux_led/manifest.json index 0c6d8ae8db1..279ea05e3d2 100644 --- a/homeassistant/components/flux_led/manifest.json +++ b/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]*" + } + ] } + + diff --git a/homeassistant/components/flux_led/services.yaml b/homeassistant/components/flux_led/services.yaml new file mode 100644 index 00000000000..f1dae55560b --- /dev/null +++ b/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" diff --git a/homeassistant/components/flux_led/strings.json b/homeassistant/components/flux_led/strings.json new file mode 100644 index 00000000000..f311f559589 --- /dev/null +++ b/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." + } + } + } + } +} diff --git a/homeassistant/components/flux_led/translations/en.json b/homeassistant/components/flux_led/translations/en.json new file mode 100644 index 00000000000..3445e3e6764 --- /dev/null +++ b/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." + } + } + } + } +} \ No newline at end of file diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 0bd6edaf146..2a04ec39478 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -87,6 +87,7 @@ FLOWS = [ "flo", "flume", "flunearyou", + "flux_led", "forecast_solar", "forked_daapd", "foscam", diff --git a/homeassistant/generated/dhcp.py b/homeassistant/generated/dhcp.py index 34b0a468fc1..de74a41cff4 100644 --- a/homeassistant/generated/dhcp.py +++ b/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*" diff --git a/mypy.ini b/mypy.ini index afaf7dc6c21..8f9e49702fc 100644 --- a/mypy.ini +++ b/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 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 1a3c20ffb65..61cd8bd909a 100644 --- a/requirements_test_all.txt +++ b/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 diff --git a/tests/components/flux_led/__init.py__ b/tests/components/flux_led/__init.py__ new file mode 100644 index 00000000000..57af0b3751a --- /dev/null +++ b/tests/components/flux_led/__init.py__ @@ -0,0 +1 @@ +"""Tests for the flux_led integration.""" \ No newline at end of file diff --git a/tests/components/flux_led/__init__.py b/tests/components/flux_led/__init__.py new file mode 100644 index 00000000000..49b9db491f3 --- /dev/null +++ b/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) diff --git a/tests/components/flux_led/test_config_flow.py b/tests/components/flux_led/test_config_flow.py new file mode 100644 index 00000000000..1c239108f41 --- /dev/null +++ b/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 diff --git a/tests/components/flux_led/test_init.py b/tests/components/flux_led/test_init.py new file mode 100644 index 00000000000..7b01088d1f6 --- /dev/null +++ b/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 diff --git a/tests/components/flux_led/test_light.py b/tests/components/flux_led/test_light.py new file mode 100644 index 00000000000..c2e98158b0b --- /dev/null +++ b/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", + }