Convert flux_led to use asyncio (#57440)
J. Nick Koston
GitHub
parent
5c91d8d379
commit
e148939b78
|
|
@ -5,13 +5,17 @@ from datetime import timedelta
|
|||
import logging
|
||||
from typing import Any, Final
|
||||
|
||||
from flux_led import BulbScanner, WifiLedBulb
|
||||
from flux_led.aio import AIOWifiLedBulb
|
||||
from flux_led.aioscanner import AIOBulbScanner
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STARTED
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STARTED
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.debounce import Debouncer
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
from homeassistant.helpers.event import async_track_time_interval
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
|
@ -19,8 +23,12 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
|
|||
from .const import (
|
||||
DISCOVER_SCAN_TIMEOUT,
|
||||
DOMAIN,
|
||||
FLUX_HOST,
|
||||
FLUX_LED_DISCOVERY,
|
||||
FLUX_LED_EXCEPTIONS,
|
||||
FLUX_MAC,
|
||||
FLUX_MODEL,
|
||||
SIGNAL_STATE_UPDATED,
|
||||
STARTUP_SCAN_TIMEOUT,
|
||||
)
|
||||
|
||||
|
|
@ -31,22 +39,52 @@ DISCOVERY_INTERVAL: Final = timedelta(minutes=15)
|
|||
REQUEST_REFRESH_DELAY: Final = 1.5
|
||||
|
||||
|
||||
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)
|
||||
@callback
|
||||
def async_wifi_bulb_for_host(host: str) -> AIOWifiLedBulb:
|
||||
"""Create a AIOWifiLedBulb from a host."""
|
||||
return AIOWifiLedBulb(host)
|
||||
|
||||
|
||||
@callback
|
||||
def async_update_entry_from_discovery(
|
||||
hass: HomeAssistant, entry: config_entries.ConfigEntry, device: dict[str, Any]
|
||||
) -> None:
|
||||
"""Update a config entry from a flux_led discovery."""
|
||||
name = f"{device[FLUX_MODEL]} {device[FLUX_MAC]}"
|
||||
hass.config_entries.async_update_entry(
|
||||
entry,
|
||||
data={**entry.data, CONF_NAME: name},
|
||||
title=name,
|
||||
unique_id=dr.format_mac(device[FLUX_MAC]),
|
||||
)
|
||||
|
||||
|
||||
async def async_discover_devices(
|
||||
hass: HomeAssistant, timeout: int
|
||||
hass: HomeAssistant, timeout: int, address: str | None = None
|
||||
) -> 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)
|
||||
scanner = AIOBulbScanner()
|
||||
try:
|
||||
discovered: list[dict[str, str]] = await scanner.async_scan(
|
||||
timeout=timeout, address=address
|
||||
)
|
||||
except OSError as ex:
|
||||
_LOGGER.debug("Scanning failed with error: %s", ex)
|
||||
return []
|
||||
else:
|
||||
return discovered
|
||||
|
||||
return await hass.async_add_executor_job(_scan_with_timeout)
|
||||
|
||||
async def async_discover_device(
|
||||
hass: HomeAssistant, host: str
|
||||
) -> dict[str, str] | None:
|
||||
"""Direct discovery at a single ip instead of broadcast."""
|
||||
# If we are missing the unique_id we should be able to fetch it
|
||||
# from the device by doing a directed discovery at the host only
|
||||
for device in await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT, host):
|
||||
if device[FLUX_HOST] == host:
|
||||
return device
|
||||
return None
|
||||
|
||||
|
||||
@callback
|
||||
|
|
@ -90,9 +128,26 @@ async def async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None
|
|||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Flux LED/MagicLight from a config entry."""
|
||||
host = entry.data[CONF_HOST]
|
||||
if not entry.unique_id:
|
||||
if discovery := await async_discover_device(hass, host):
|
||||
async_update_entry_from_discovery(hass, entry, discovery)
|
||||
|
||||
coordinator = FluxLedUpdateCoordinator(hass, entry.data[CONF_HOST])
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
device: AIOWifiLedBulb = async_wifi_bulb_for_host(host)
|
||||
signal = SIGNAL_STATE_UPDATED.format(device.ipaddr)
|
||||
|
||||
@callback
|
||||
def _async_state_changed(*_: Any) -> None:
|
||||
_LOGGER.debug("%s: Device state updated: %s", device.ipaddr, device.raw_state)
|
||||
async_dispatcher_send(hass, signal)
|
||||
|
||||
try:
|
||||
await device.async_setup(_async_state_changed)
|
||||
except FLUX_LED_EXCEPTIONS as ex:
|
||||
raise ConfigEntryNotReady(
|
||||
str(ex) or f"Timed out trying to connect to {device.ipaddr}"
|
||||
) from ex
|
||||
coordinator = FluxLedUpdateCoordinator(hass, device)
|
||||
hass.data[DOMAIN][entry.entry_id] = coordinator
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
entry.async_on_unload(entry.add_update_listener(async_update_listener))
|
||||
|
|
@ -103,7 +158,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
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)
|
||||
coordinator = hass.data[DOMAIN].pop(entry.entry_id)
|
||||
await coordinator.device.async_stop()
|
||||
return unload_ok
|
||||
|
||||
|
||||
|
|
@ -113,17 +169,15 @@ class FluxLedUpdateCoordinator(DataUpdateCoordinator):
|
|||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
host: str,
|
||||
device: AIOWifiLedBulb,
|
||||
) -> None:
|
||||
"""Initialize DataUpdateCoordinator to gather data for specific device."""
|
||||
self.host = host
|
||||
self.device: WifiLedBulb | None = None
|
||||
update_interval = timedelta(seconds=5)
|
||||
self.device = device
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=host,
|
||||
update_interval=update_interval,
|
||||
name=self.device.ipaddr,
|
||||
update_interval=timedelta(seconds=5),
|
||||
# We don't want an immediate refresh since the device
|
||||
# takes a moment to reflect the state change
|
||||
request_refresh_debouncer=Debouncer(
|
||||
|
|
@ -134,12 +188,6 @@ class FluxLedUpdateCoordinator(DataUpdateCoordinator):
|
|||
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)
|
||||
await self.device.async_update()
|
||||
except FLUX_LED_EXCEPTIONS as ex:
|
||||
raise UpdateFailed(ex) from ex
|
||||
|
||||
if not self.device.raw_state:
|
||||
raise UpdateFailed("The device failed to update")
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ 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
|
||||
|
|
@ -15,7 +14,12 @@ from homeassistant.data_entry_flow import FlowResult
|
|||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.typing import DiscoveryInfoType
|
||||
|
||||
from . import async_discover_devices, async_wifi_bulb_for_host
|
||||
from . import (
|
||||
async_discover_device,
|
||||
async_discover_devices,
|
||||
async_update_entry_from_discovery,
|
||||
async_wifi_bulb_for_host,
|
||||
)
|
||||
from .const import (
|
||||
CONF_CUSTOM_EFFECT_COLORS,
|
||||
CONF_CUSTOM_EFFECT_SPEED_PCT,
|
||||
|
|
@ -34,7 +38,6 @@ from .const import (
|
|||
|
||||
CONF_DEVICE: Final = "device"
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
|
|
@ -104,13 +107,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
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,
|
||||
)
|
||||
async_update_entry_from_discovery(self.hass, entry, device)
|
||||
return self.async_abort(reason="already_configured")
|
||||
self.context[CONF_HOST] = host
|
||||
for progress in self._async_in_progress():
|
||||
|
|
@ -157,13 +154,16 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
if not (host := user_input[CONF_HOST]):
|
||||
return await self.async_step_pick_device()
|
||||
try:
|
||||
await self._async_try_connect(host)
|
||||
device = 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}
|
||||
)
|
||||
if device[FLUX_MAC]:
|
||||
await self.async_set_unique_id(
|
||||
dr.format_mac(device[FLUX_MAC]), raise_on_progress=False
|
||||
)
|
||||
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
|
||||
return self._async_create_entry_from_device(device)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
|
|
@ -204,10 +204,17 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(devices_name)}),
|
||||
)
|
||||
|
||||
async def _async_try_connect(self, host: str) -> WifiLedBulb:
|
||||
async def _async_try_connect(self, host: str) -> dict[str, Any]:
|
||||
"""Try to connect."""
|
||||
self._async_abort_entries_match({CONF_HOST: host})
|
||||
return await async_wifi_bulb_for_host(self.hass, host)
|
||||
if device := await async_discover_device(self.hass, host):
|
||||
return device
|
||||
bulb = async_wifi_bulb_for_host(host)
|
||||
try:
|
||||
await bulb.async_setup(lambda: None)
|
||||
finally:
|
||||
await bulb.async_stop()
|
||||
return {FLUX_MAC: None, FLUX_MODEL: None, FLUX_HOST: host}
|
||||
|
||||
|
||||
class OptionsFlow(config_entries.OptionsFlow):
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""Constants of the FluxLed/MagicHome Integration."""
|
||||
|
||||
import asyncio
|
||||
import socket
|
||||
from typing import Final
|
||||
|
||||
|
|
@ -7,6 +8,7 @@ DOMAIN: Final = "flux_led"
|
|||
|
||||
API: Final = "flux_api"
|
||||
|
||||
SIGNAL_STATE_UPDATED = "flux_led_{}_state_updated"
|
||||
|
||||
CONF_AUTOMATIC_ADD: Final = "automatic_add"
|
||||
DEFAULT_NETWORK_SCAN_INTERVAL: Final = 120
|
||||
|
|
@ -15,7 +17,12 @@ DEFAULT_EFFECT_SPEED: Final = 50
|
|||
|
||||
FLUX_LED_DISCOVERY: Final = "flux_led_discovery"
|
||||
|
||||
FLUX_LED_EXCEPTIONS: Final = (socket.timeout, BrokenPipeError)
|
||||
FLUX_LED_EXCEPTIONS: Final = (
|
||||
asyncio.TimeoutError,
|
||||
socket.error,
|
||||
RuntimeError,
|
||||
BrokenPipeError,
|
||||
)
|
||||
|
||||
STARTUP_SCAN_TIMEOUT: Final = 5
|
||||
DISCOVER_SCAN_TIMEOUT: Final = 10
|
||||
|
|
|
|||
|
|
@ -2,12 +2,11 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from functools import partial
|
||||
import logging
|
||||
import random
|
||||
from typing import Any, Final, cast
|
||||
|
||||
from flux_led import WifiLedBulb
|
||||
from flux_led.aiodevice import AIOWifiLedBulb
|
||||
from flux_led.const import (
|
||||
COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT,
|
||||
COLOR_MODE_DIM as FLUX_COLOR_MODE_DIM,
|
||||
|
|
@ -15,7 +14,6 @@ from flux_led.const import (
|
|||
COLOR_MODE_RGBW as FLUX_COLOR_MODE_RGBW,
|
||||
COLOR_MODE_RGBWW as FLUX_COLOR_MODE_RGBWW,
|
||||
)
|
||||
from flux_led.device import MAX_TEMP, MIN_TEMP
|
||||
from flux_led.utils import (
|
||||
color_temp_to_white_levels,
|
||||
rgbcw_brightness,
|
||||
|
|
@ -61,13 +59,16 @@ from homeassistant.const import (
|
|||
CONF_NAME,
|
||||
CONF_PROTOCOL,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import device_registry as dr, entity_platform
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
from homeassistant.util.color import (
|
||||
color_hs_to_RGB,
|
||||
color_RGB_to_hs,
|
||||
color_temperature_kelvin_to_mired,
|
||||
color_temperature_mired_to_kelvin,
|
||||
)
|
||||
|
|
@ -91,6 +92,7 @@ from .const import (
|
|||
MODE_RGB,
|
||||
MODE_RGBW,
|
||||
MODE_WHITE,
|
||||
SIGNAL_STATE_UPDATED,
|
||||
TRANSITION_GRADUAL,
|
||||
TRANSITION_JUMP,
|
||||
TRANSITION_STROBE,
|
||||
|
|
@ -254,7 +256,7 @@ async def async_setup_entry(
|
|||
platform.async_register_entity_service(
|
||||
SERVICE_CUSTOM_EFFECT,
|
||||
CUSTOM_EFFECT_DICT,
|
||||
"set_custom_effect",
|
||||
"async_set_custom_effect",
|
||||
)
|
||||
options = entry.options
|
||||
|
||||
|
|
@ -298,17 +300,18 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
) -> None:
|
||||
"""Initialize the light."""
|
||||
super().__init__(coordinator)
|
||||
self._bulb: WifiLedBulb = coordinator.device
|
||||
self._device: AIOWifiLedBulb = coordinator.device
|
||||
self._responding = True
|
||||
self._attr_name = name
|
||||
self._attr_unique_id = unique_id
|
||||
self._attr_supported_features = SUPPORT_FLUX_LED
|
||||
self._attr_min_mireds = (
|
||||
color_temperature_kelvin_to_mired(MAX_TEMP) + 1
|
||||
color_temperature_kelvin_to_mired(self._device.max_temp) + 1
|
||||
) # for rounding
|
||||
self._attr_max_mireds = color_temperature_kelvin_to_mired(MIN_TEMP)
|
||||
self._attr_max_mireds = color_temperature_kelvin_to_mired(self._device.min_temp)
|
||||
self._attr_supported_color_modes = {
|
||||
FLUX_COLOR_MODE_TO_HASS.get(mode, COLOR_MODE_ONOFF)
|
||||
for mode in self._bulb.color_modes
|
||||
for mode in self._device.color_modes
|
||||
}
|
||||
self._attr_effect_list = FLUX_EFFECT_LIST
|
||||
if custom_effect_colors:
|
||||
|
|
@ -317,82 +320,83 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
self._custom_effect_speed_pct = custom_effect_speed_pct
|
||||
self._custom_effect_transition = custom_effect_transition
|
||||
if self.unique_id:
|
||||
old_protocol = self._bulb.protocol == "LEDENET_ORIGINAL"
|
||||
raw_state = self._bulb.raw_state
|
||||
self._attr_device_info = {
|
||||
"connections": {(dr.CONNECTION_NETWORK_MAC, self.unique_id)},
|
||||
ATTR_MODEL: f"0x{self._bulb.model_num:02X}",
|
||||
ATTR_MODEL: f"0x{self._device.model_num:02X}",
|
||||
ATTR_NAME: self.name,
|
||||
ATTR_SW_VERSION: "1" if old_protocol else str(raw_state.version_number),
|
||||
ATTR_SW_VERSION: str(self._device.version_num),
|
||||
ATTR_MANUFACTURER: "FluxLED/Magic Home",
|
||||
}
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if device is on."""
|
||||
return cast(bool, self._bulb.is_on)
|
||||
return cast(bool, self._device.is_on)
|
||||
|
||||
@property
|
||||
def brightness(self) -> int:
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
return cast(int, self._bulb.brightness)
|
||||
return cast(int, self._device.brightness)
|
||||
|
||||
@property
|
||||
def color_temp(self) -> int:
|
||||
"""Return the kelvin value of this light in mired."""
|
||||
return color_temperature_kelvin_to_mired(self._bulb.getWhiteTemperature()[0])
|
||||
return color_temperature_kelvin_to_mired(self._device.color_temp)
|
||||
|
||||
@property
|
||||
def rgb_color(self) -> tuple[int, int, int]:
|
||||
"""Return the rgb color value."""
|
||||
rgb: tuple[int, int, int] = self._bulb.rgb
|
||||
# Note that we call color_RGB_to_hs and not color_RGB_to_hsv
|
||||
# to get the unscaled value since this is what the frontend wants
|
||||
# https://github.com/home-assistant/frontend/blob/e797c017614797bb11671496d6bd65863de22063/src/dialogs/more-info/controls/more-info-light.ts#L263
|
||||
rgb: tuple[int, int, int] = color_hs_to_RGB(*color_RGB_to_hs(*self._device.rgb))
|
||||
return rgb
|
||||
|
||||
@property
|
||||
def rgbw_color(self) -> tuple[int, int, int, int]:
|
||||
"""Return the rgbw color value."""
|
||||
rgbw: tuple[int, int, int, int] = self._bulb.rgbw
|
||||
rgbw: tuple[int, int, int, int] = self._device.rgbw
|
||||
return rgbw
|
||||
|
||||
@property
|
||||
def rgbww_color(self) -> tuple[int, int, int, int, int]:
|
||||
"""Return the rgbww aka rgbcw color value."""
|
||||
rgbcw: tuple[int, int, int, int, int] = self._bulb.rgbcw
|
||||
rgbcw: tuple[int, int, int, int, int] = self._device.rgbcw
|
||||
return rgbcw
|
||||
|
||||
@property
|
||||
def rgbwc_color(self) -> tuple[int, int, int, int, int]:
|
||||
"""Return the rgbwc color value."""
|
||||
rgbwc: tuple[int, int, int, int, int] = self._bulb.rgbww
|
||||
rgbwc: tuple[int, int, int, int, int] = self._device.rgbww
|
||||
return rgbwc
|
||||
|
||||
@property
|
||||
def color_mode(self) -> str:
|
||||
"""Return the color mode of the light."""
|
||||
return FLUX_COLOR_MODE_TO_HASS.get(self._bulb.color_mode, COLOR_MODE_ONOFF)
|
||||
return FLUX_COLOR_MODE_TO_HASS.get(self._device.color_mode, COLOR_MODE_ONOFF)
|
||||
|
||||
@property
|
||||
def effect(self) -> str | None:
|
||||
"""Return the current effect."""
|
||||
if (current_mode := self._bulb.raw_state.preset_pattern) == EFFECT_CUSTOM_CODE:
|
||||
if (current_mode := self._device.preset_pattern_num) == EFFECT_CUSTOM_CODE:
|
||||
return EFFECT_CUSTOM
|
||||
return EFFECT_ID_NAME.get(current_mode)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, str]:
|
||||
"""Return the attributes."""
|
||||
return {"ip_address": self._bulb.ipaddr}
|
||||
return {"ip_address": self._device.ipaddr}
|
||||
|
||||
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._async_turn_on(**kwargs)
|
||||
self.async_write_ha_state()
|
||||
await self.coordinator.async_request_refresh()
|
||||
|
||||
def _turn_on(self, **kwargs: Any) -> None:
|
||||
async def _async_turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the specified or all lights on."""
|
||||
if not self.is_on:
|
||||
self._bulb.turnOn()
|
||||
await self._device.async_turn_on()
|
||||
if not kwargs:
|
||||
return
|
||||
|
||||
|
|
@ -404,21 +408,23 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
color_temp_mired = kwargs[ATTR_COLOR_TEMP]
|
||||
color_temp_kelvin = color_temperature_mired_to_kelvin(color_temp_mired)
|
||||
if self.color_mode != COLOR_MODE_RGBWW:
|
||||
self._bulb.setWhiteTemperature(color_temp_kelvin, brightness)
|
||||
await self._device.async_set_white_temp(color_temp_kelvin, brightness)
|
||||
return
|
||||
|
||||
# When switching to color temp from RGBWW mode,
|
||||
# we do not want the overall brightness, we only
|
||||
# want the brightness of the white channels
|
||||
brightness = kwargs.get(
|
||||
ATTR_BRIGHTNESS, self._bulb.getWhiteTemperature()[1]
|
||||
ATTR_BRIGHTNESS, self._device.getWhiteTemperature()[1]
|
||||
)
|
||||
cold, warm = color_temp_to_white_levels(color_temp_kelvin, brightness)
|
||||
self._bulb.set_levels(r=0, b=0, g=0, w=warm, w2=cold)
|
||||
await self._device.async_set_levels(r=0, b=0, g=0, w=warm, w2=cold)
|
||||
return
|
||||
# Handle switch to HS Color Mode
|
||||
# Handle switch to RGB Color Mode
|
||||
if ATTR_RGB_COLOR in kwargs:
|
||||
self._bulb.set_levels(*kwargs[ATTR_RGB_COLOR], brightness=brightness)
|
||||
await self._device.async_set_levels(
|
||||
*kwargs[ATTR_RGB_COLOR], brightness=brightness
|
||||
)
|
||||
return
|
||||
# Handle switch to RGBW Color Mode
|
||||
if ATTR_RGBW_COLOR in kwargs:
|
||||
|
|
@ -426,7 +432,7 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
rgbw = rgbw_brightness(kwargs[ATTR_RGBW_COLOR], brightness)
|
||||
else:
|
||||
rgbw = kwargs[ATTR_RGBW_COLOR]
|
||||
self._bulb.set_levels(*rgbw)
|
||||
await self._device.async_set_levels(*rgbw)
|
||||
return
|
||||
# Handle switch to RGBWW Color Mode
|
||||
if ATTR_RGBWW_COLOR in kwargs:
|
||||
|
|
@ -434,17 +440,17 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
rgbcw = rgbcw_brightness(kwargs[ATTR_RGBWW_COLOR], brightness)
|
||||
else:
|
||||
rgbcw = kwargs[ATTR_RGBWW_COLOR]
|
||||
self._bulb.set_levels(*rgbcw_to_rgbwc(rgbcw))
|
||||
await self._device.async_set_levels(*rgbcw_to_rgbwc(rgbcw))
|
||||
return
|
||||
# Handle switch to White Color Mode
|
||||
if ATTR_WHITE in kwargs:
|
||||
self._bulb.set_levels(w=kwargs[ATTR_WHITE])
|
||||
await self._device.async_set_levels(w=kwargs[ATTR_WHITE])
|
||||
return
|
||||
if ATTR_EFFECT in kwargs:
|
||||
effect = kwargs[ATTR_EFFECT]
|
||||
# Random color effect
|
||||
if effect == EFFECT_RANDOM:
|
||||
self._bulb.set_levels(
|
||||
await self._device.async_set_levels(
|
||||
random.randint(0, 255),
|
||||
random.randint(0, 255),
|
||||
random.randint(0, 255),
|
||||
|
|
@ -453,7 +459,7 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
# Custom effect
|
||||
if effect == EFFECT_CUSTOM:
|
||||
if self._custom_effect_colors:
|
||||
self._bulb.setCustomPattern(
|
||||
await self._device.async_set_custom_pattern(
|
||||
self._custom_effect_colors,
|
||||
self._custom_effect_speed_pct,
|
||||
self._custom_effect_transition,
|
||||
|
|
@ -461,39 +467,41 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
return
|
||||
# Effect selection
|
||||
if effect in EFFECT_MAP:
|
||||
self._bulb.setPresetPattern(EFFECT_MAP[effect], DEFAULT_EFFECT_SPEED)
|
||||
await self._device.async_set_preset_pattern(
|
||||
EFFECT_MAP[effect], DEFAULT_EFFECT_SPEED
|
||||
)
|
||||
return
|
||||
raise ValueError(f"Unknown effect {effect}")
|
||||
# Handle brightness adjustment in CCT Color Mode
|
||||
if self.color_mode == COLOR_MODE_COLOR_TEMP:
|
||||
self._bulb.setWhiteTemperature(
|
||||
self._bulb.getWhiteTemperature()[0], brightness
|
||||
)
|
||||
await self._device.async_set_white_temp(self._device.color_temp, brightness)
|
||||
return
|
||||
# Handle brightness adjustment in RGB Color Mode
|
||||
if self.color_mode == COLOR_MODE_RGB:
|
||||
self._bulb.set_levels(*self.rgb_color, brightness=brightness)
|
||||
await self._device.async_set_levels(*self.rgb_color, brightness=brightness)
|
||||
return
|
||||
# Handle brightness adjustment in RGBW Color Mode
|
||||
if self.color_mode == COLOR_MODE_RGBW:
|
||||
self._bulb.set_levels(*rgbw_brightness(self.rgbw_color, brightness))
|
||||
await self._device.async_set_levels(
|
||||
*rgbw_brightness(self.rgbw_color, brightness)
|
||||
)
|
||||
return
|
||||
# Handle brightness adjustment in RGBWW Color Mode
|
||||
if self.color_mode == COLOR_MODE_RGBWW:
|
||||
rgbwc = self.rgbwc_color
|
||||
self._bulb.set_levels(*rgbww_brightness(rgbwc, brightness))
|
||||
await self._device.async_set_levels(*rgbww_brightness(rgbwc, brightness))
|
||||
return
|
||||
# Handle White Color Mode and Brightness Only Color Mode
|
||||
if self.color_mode in (COLOR_MODE_WHITE, COLOR_MODE_BRIGHTNESS):
|
||||
self._bulb.set_levels(w=brightness)
|
||||
await self._device.async_set_levels(w=brightness)
|
||||
return
|
||||
raise ValueError(f"Unsupported color mode {self.color_mode}")
|
||||
|
||||
def set_custom_effect(
|
||||
async def async_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(
|
||||
await self._device.async_set_custom_pattern(
|
||||
colors,
|
||||
speed_pct,
|
||||
transition,
|
||||
|
|
@ -501,6 +509,24 @@ class FluxLight(CoordinatorEntity, LightEntity):
|
|||
|
||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn the specified or all lights off."""
|
||||
await self.hass.async_add_executor_job(self._bulb.turnOff)
|
||||
await self._device.async_turn_off()
|
||||
self.async_write_ha_state()
|
||||
await self.coordinator.async_request_refresh()
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle updated data from the coordinator."""
|
||||
if self.coordinator.last_update_success != self._responding:
|
||||
self.async_write_ha_state()
|
||||
self._responding = self.coordinator.last_update_success
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Handle entity which will be added."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass,
|
||||
SIGNAL_STATE_UPDATED.format(self._device.ipaddr),
|
||||
self.async_write_ha_state,
|
||||
)
|
||||
)
|
||||
await super().async_added_to_hass()
|
||||
|
|
|
|||
|
|
@ -1 +0,0 @@
|
|||
"""Tests for the flux_led integration."""
|
||||
|
|
@ -1,10 +1,11 @@
|
|||
"""Tests for the flux_led integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import socket
|
||||
from unittest.mock import MagicMock, patch
|
||||
import asyncio
|
||||
from typing import Callable
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from flux_led import WifiLedBulb
|
||||
from flux_led.aio import AIOWifiLedBulb
|
||||
from flux_led.const import (
|
||||
COLOR_MODE_CCT as FLUX_COLOR_MODE_CCT,
|
||||
COLOR_MODE_RGB as FLUX_COLOR_MODE_RGB,
|
||||
|
|
@ -17,6 +18,7 @@ from homeassistant.components.dhcp import (
|
|||
MAC_ADDRESS as DHCP_MAC_ADDRESS,
|
||||
)
|
||||
from homeassistant.components.flux_led.const import FLUX_HOST, FLUX_MAC, FLUX_MODEL
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
MODULE = "homeassistant.components.flux_led"
|
||||
MODULE_CONFIG_FLOW = "homeassistant.components.flux_led.config_flow"
|
||||
|
|
@ -35,8 +37,23 @@ DHCP_DISCOVERY = {
|
|||
FLUX_DISCOVERY = {FLUX_HOST: IP_ADDRESS, FLUX_MODEL: MODEL, FLUX_MAC: FLUX_MAC_ADDRESS}
|
||||
|
||||
|
||||
def _mocked_bulb() -> WifiLedBulb:
|
||||
bulb = MagicMock(auto_spec=WifiLedBulb)
|
||||
def _mocked_bulb() -> AIOWifiLedBulb:
|
||||
bulb = MagicMock(auto_spec=AIOWifiLedBulb)
|
||||
|
||||
async def _save_setup_callback(callback: Callable) -> None:
|
||||
bulb.data_receive_callback = callback
|
||||
|
||||
bulb.async_setup = AsyncMock(side_effect=_save_setup_callback)
|
||||
bulb.async_set_custom_pattern = AsyncMock()
|
||||
bulb.async_set_preset_pattern = AsyncMock()
|
||||
bulb.async_set_white_temp = AsyncMock()
|
||||
bulb.async_stop = AsyncMock()
|
||||
bulb.async_update = AsyncMock()
|
||||
bulb.async_turn_off = AsyncMock()
|
||||
bulb.async_turn_on = AsyncMock()
|
||||
bulb.async_set_levels = AsyncMock()
|
||||
bulb.min_temp = 2700
|
||||
bulb.max_temp = 6500
|
||||
bulb.getRgb = MagicMock(return_value=[255, 0, 0])
|
||||
bulb.getRgbw = MagicMock(return_value=[255, 0, 0, 50])
|
||||
bulb.getRgbww = MagicMock(return_value=[255, 0, 0, 50, 0])
|
||||
|
|
@ -45,9 +62,11 @@ def _mocked_bulb() -> WifiLedBulb:
|
|||
bulb.rgbw = (255, 0, 0, 50)
|
||||
bulb.rgbww = (255, 0, 0, 50, 0)
|
||||
bulb.rgbcw = (255, 0, 0, 0, 50)
|
||||
bulb.color_temp = 2700
|
||||
bulb.getWhiteTemperature = MagicMock(return_value=(2700, 128))
|
||||
bulb.brightness = 128
|
||||
bulb.model_num = 0x35
|
||||
bulb.version_num = 8
|
||||
bulb.rgbwcapable = True
|
||||
bulb.color_modes = {FLUX_COLOR_MODE_RGB, FLUX_COLOR_MODE_CCT}
|
||||
bulb.color_mode = FLUX_COLOR_MODE_RGB
|
||||
|
|
@ -57,19 +76,39 @@ def _mocked_bulb() -> WifiLedBulb:
|
|||
return bulb
|
||||
|
||||
|
||||
async def async_mock_bulb_turn_off(hass: HomeAssistant, bulb: AIOWifiLedBulb) -> None:
|
||||
"""Mock the bulb being off."""
|
||||
bulb.is_on = False
|
||||
bulb.raw_state._replace(power_state=0x24)
|
||||
bulb.data_receive_callback()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def async_mock_bulb_turn_on(hass: HomeAssistant, bulb: AIOWifiLedBulb) -> None:
|
||||
"""Mock the bulb being on."""
|
||||
bulb.is_on = True
|
||||
bulb.raw_state._replace(power_state=0x23)
|
||||
bulb.data_receive_callback()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
def _patch_discovery(device=None, no_device=False):
|
||||
def _discovery(*args, **kwargs):
|
||||
async def _discovery(*args, **kwargs):
|
||||
if no_device:
|
||||
return []
|
||||
raise OSError
|
||||
return [FLUX_DISCOVERY]
|
||||
|
||||
return patch("homeassistant.components.flux_led.BulbScanner.scan", new=_discovery)
|
||||
return patch(
|
||||
"homeassistant.components.flux_led.AIOBulbScanner.async_scan", new=_discovery
|
||||
)
|
||||
|
||||
|
||||
def _patch_wifibulb(device=None, no_device=False):
|
||||
def _wifi_led_bulb(*args, **kwargs):
|
||||
bulb = _mocked_bulb()
|
||||
if no_device:
|
||||
raise socket.timeout
|
||||
bulb.async_setup = AsyncMock(side_effect=asyncio.TimeoutError)
|
||||
return bulb
|
||||
return device if device else _mocked_bulb()
|
||||
|
||||
return patch("homeassistant.components.flux_led.WifiLedBulb", new=_wifi_led_bulb)
|
||||
return patch("homeassistant.components.flux_led.AIOWifiLedBulb", new=_wifi_led_bulb)
|
||||
|
|
|
|||
|
|
@ -247,7 +247,7 @@ async def test_import(hass: HomeAssistant):
|
|||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_manual(hass: HomeAssistant):
|
||||
async def test_manual_working_discovery(hass: HomeAssistant):
|
||||
"""Test manually setup."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
|
|
@ -276,8 +276,8 @@ async def test_manual(hass: HomeAssistant):
|
|||
)
|
||||
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}
|
||||
assert result4["title"] == DEFAULT_ENTRY_TITLE
|
||||
assert result4["data"] == {CONF_HOST: IP_ADDRESS, CONF_NAME: DEFAULT_ENTRY_TITLE}
|
||||
|
||||
# Duplicate
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
|
|
|
|||
|
|
@ -6,16 +6,16 @@ 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.const import CONF_HOST, CONF_NAME, EVENT_HOMEASSISTANT_STARTED
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from . import (
|
||||
DEFAULT_ENTRY_TITLE,
|
||||
FLUX_DISCOVERY,
|
||||
IP_ADDRESS,
|
||||
MAC_ADDRESS,
|
||||
_mocked_bulb,
|
||||
_patch_discovery,
|
||||
_patch_wifibulb,
|
||||
)
|
||||
|
|
@ -25,7 +25,9 @@ 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:
|
||||
with patch(
|
||||
"homeassistant.components.flux_led.AIOBulbScanner.async_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()
|
||||
|
|
@ -65,15 +67,26 @@ async def test_config_entry_retry(hass: HomeAssistant) -> None:
|
|||
assert config_entry.state == ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_config_entry_retry_when_state_missing(hass: HomeAssistant) -> None:
|
||||
"""Test that a config entry is retried when state is missing."""
|
||||
async def test_config_entry_fills_unique_id_with_directed_discovery(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test that the unique id is added if its missing via directed (not broadcast) discovery."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=MAC_ADDRESS
|
||||
domain=DOMAIN, data={CONF_HOST: IP_ADDRESS}, unique_id=None
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
bulb = _mocked_bulb()
|
||||
bulb.raw_state = None
|
||||
with _patch_discovery(device=bulb), _patch_wifibulb(device=bulb):
|
||||
|
||||
async def _discovery(self, *args, address=None, **kwargs):
|
||||
# Only return discovery results when doing directed discovery
|
||||
return [FLUX_DISCOVERY] if address == IP_ADDRESS else []
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.flux_led.AIOBulbScanner.async_scan", new=_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.SETUP_RETRY
|
||||
assert config_entry.state == ConfigEntryState.LOADED
|
||||
|
||||
assert config_entry.unique_id == MAC_ADDRESS
|
||||
assert config_entry.data[CONF_NAME] == DEFAULT_ENTRY_TITLE
|
||||
assert config_entry.title == DEFAULT_ENTRY_TITLE
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
"""Tests for light platform."""
|
||||
from datetime import timedelta
|
||||
from unittest.mock import Mock
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
from flux_led.const import (
|
||||
COLOR_MODE_ADDRESSABLE as FLUX_COLOR_MODE_ADDRESSABLE,
|
||||
|
|
@ -50,6 +50,7 @@ from homeassistant.const import (
|
|||
CONF_PROTOCOL,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
|
|
@ -63,6 +64,8 @@ from . import (
|
|||
_mocked_bulb,
|
||||
_patch_discovery,
|
||||
_patch_wifibulb,
|
||||
async_mock_bulb_turn_off,
|
||||
async_mock_bulb_turn_on,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
|
@ -88,6 +91,40 @@ async def test_light_unique_id(hass: HomeAssistant) -> None:
|
|||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_light_goes_unavailable_and_recovers(hass: HomeAssistant) -> None:
|
||||
"""Test a light goes unavailable and then recovers."""
|
||||
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
|
||||
|
||||
now = utcnow()
|
||||
bulb.async_update = AsyncMock(side_effect=RuntimeError)
|
||||
for i in range(10, 50, 10):
|
||||
async_fire_time_changed(hass, now + timedelta(seconds=i))
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
bulb.async_update = AsyncMock()
|
||||
for i in range(60, 100, 10):
|
||||
async_fire_time_changed(hass, now + timedelta(seconds=i))
|
||||
await hass.async_block_till_done()
|
||||
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(
|
||||
|
|
@ -120,6 +157,7 @@ async def test_light_device_registry(
|
|||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
bulb = _mocked_bulb()
|
||||
bulb.version_num = sw_version
|
||||
bulb.protocol = protocol
|
||||
bulb.raw_state = bulb.raw_state._replace(model_num=model, version_number=sw_version)
|
||||
bulb.model_num = model
|
||||
|
|
@ -166,18 +204,16 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
|
||||
bulb.is_on = False
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
|
||||
await hass.async_block_till_done()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
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()
|
||||
bulb.async_turn_on.assert_called_once()
|
||||
bulb.async_turn_on.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -185,8 +221,8 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 0, 0, brightness=100)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 0, 0, brightness=100)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -194,8 +230,8 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_HS_COLOR: (10, 30)},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 191, 178, brightness=128)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 191, 178, brightness=128)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -203,8 +239,8 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_once()
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_once()
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -212,8 +248,8 @@ async def test_rgb_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setPresetPattern.assert_called_with(43, 50)
|
||||
bulb.setPresetPattern.reset_mock()
|
||||
bulb.async_set_preset_pattern.assert_called_with(43, 50)
|
||||
bulb.async_set_preset_pattern.reset_mock()
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
await hass.services.async_call(
|
||||
|
|
@ -254,18 +290,16 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
|
||||
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()
|
||||
bulb.async_turn_on.assert_called_once()
|
||||
bulb.async_turn_on.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -273,8 +307,8 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 0, 0, brightness=100)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 0, 0, brightness=100)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -282,8 +316,8 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_HS_COLOR: (10, 30)},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 191, 178, brightness=128)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 191, 178, brightness=128)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -291,8 +325,8 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_once()
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_once()
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -300,17 +334,16 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setPresetPattern.assert_called_with(43, 50)
|
||||
bulb.setPresetPattern.reset_mock()
|
||||
|
||||
bulb.is_on = True
|
||||
bulb.async_set_preset_pattern.assert_called_with(43, 50)
|
||||
bulb.async_set_preset_pattern.reset_mock()
|
||||
bulb.color_mode = FLUX_COLOR_MODE_CCT
|
||||
bulb.getWhiteTemperature = Mock(return_value=(5000, 128))
|
||||
bulb.color_temp = 5000
|
||||
|
||||
bulb.raw_state = bulb.raw_state._replace(
|
||||
red=0, green=0, blue=0, warm_white=1, cool_white=2
|
||||
)
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=60))
|
||||
await hass.async_block_till_done()
|
||||
await async_mock_bulb_turn_on(hass, bulb)
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.state == STATE_ON
|
||||
attributes = state.attributes
|
||||
|
|
@ -325,8 +358,8 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 370},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setWhiteTemperature.assert_called_with(2702, 128)
|
||||
bulb.setWhiteTemperature.reset_mock()
|
||||
bulb.async_set_white_temp.assert_called_with(2702, 128)
|
||||
bulb.async_set_white_temp.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -334,8 +367,8 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 255},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setWhiteTemperature.assert_called_with(5000, 255)
|
||||
bulb.setWhiteTemperature.reset_mock()
|
||||
bulb.async_set_white_temp.assert_called_with(5000, 255)
|
||||
bulb.async_set_white_temp.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -343,8 +376,8 @@ async def test_rgb_cct_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 128},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setWhiteTemperature.assert_called_with(5000, 128)
|
||||
bulb.setWhiteTemperature.reset_mock()
|
||||
bulb.async_set_white_temp.assert_called_with(5000, 128)
|
||||
bulb.async_set_white_temp.reset_mock()
|
||||
|
||||
|
||||
async def test_rgbw_light(hass: HomeAssistant) -> None:
|
||||
|
|
@ -376,18 +409,16 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
|
||||
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()
|
||||
bulb.async_turn_on.assert_called_once()
|
||||
bulb.async_turn_on.reset_mock()
|
||||
bulb.is_on = True
|
||||
|
||||
await hass.services.async_call(
|
||||
|
|
@ -396,8 +427,8 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(168, 0, 0, 33)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(168, 0, 0, 33)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
|
@ -411,8 +442,8 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(128, 128, 128, 128)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(128, 128, 128, 128)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -420,8 +451,8 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_RGBW_COLOR: (255, 255, 255, 255)},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 255, 255, 255)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 255, 255, 255)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -429,8 +460,8 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_RGBW_COLOR: (255, 191, 178, 0)},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 191, 178, 0)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 191, 178, 0)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -438,8 +469,8 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_once()
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_once()
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -447,8 +478,8 @@ async def test_rgbw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setPresetPattern.assert_called_with(43, 50)
|
||||
bulb.setPresetPattern.reset_mock()
|
||||
bulb.async_set_preset_pattern.assert_called_with(43, 50)
|
||||
bulb.async_set_preset_pattern.reset_mock()
|
||||
|
||||
|
||||
async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
||||
|
|
@ -481,18 +512,16 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
|
||||
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()
|
||||
bulb.async_turn_on.assert_called_once()
|
||||
bulb.async_turn_on.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -500,8 +529,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(250, 0, 0, 49, 0)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(250, 0, 0, 49, 0)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
bulb.is_on = True
|
||||
|
||||
await hass.services.async_call(
|
||||
|
|
@ -514,8 +543,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(192, 192, 192, 192, 0)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(192, 192, 192, 192, 0)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -523,8 +552,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_RGBWW_COLOR: (255, 255, 255, 255, 50)},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 255, 255, 50, 255)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 255, 255, 50, 255)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -532,8 +561,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 154},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(r=0, b=0, g=0, w=0, w2=127)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(r=0, b=0, g=0, w=0, w2=127)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -541,8 +570,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 154, ATTR_BRIGHTNESS: 255},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(r=0, b=0, g=0, w=0, w2=255)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(r=0, b=0, g=0, w=0, w2=255)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -550,8 +579,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_COLOR_TEMP: 290},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(r=0, b=0, g=0, w=102, w2=25)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(r=0, b=0, g=0, w=102, w2=25)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -559,8 +588,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_RGBWW_COLOR: (255, 191, 178, 0, 0)},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(255, 191, 178, 0, 0)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(255, 191, 178, 0, 0)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -568,8 +597,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "random"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_once()
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_once()
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -577,8 +606,8 @@ async def test_rgbcw_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_EFFECT: "purple_fade"},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setPresetPattern.assert_called_with(43, 50)
|
||||
bulb.setPresetPattern.reset_mock()
|
||||
bulb.async_set_preset_pattern.assert_called_with(43, 50)
|
||||
bulb.async_set_preset_pattern.reset_mock()
|
||||
|
||||
|
||||
async def test_white_light(hass: HomeAssistant) -> None:
|
||||
|
|
@ -610,18 +639,16 @@ async def test_white_light(hass: HomeAssistant) -> None:
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
|
||||
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()
|
||||
bulb.async_turn_on.assert_called_once()
|
||||
bulb.async_turn_on.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -629,8 +656,8 @@ async def test_white_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 100},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(w=100)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(w=100)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN,
|
||||
|
|
@ -638,8 +665,8 @@ async def test_white_light(hass: HomeAssistant) -> None:
|
|||
{ATTR_ENTITY_ID: entity_id, ATTR_WHITE: 100},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.set_levels.assert_called_with(w=100)
|
||||
bulb.set_levels.reset_mock()
|
||||
bulb.async_set_levels.assert_called_with(w=100)
|
||||
bulb.async_set_levels.reset_mock()
|
||||
|
||||
|
||||
async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
|
||||
|
|
@ -677,10 +704,8 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
|
|||
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))
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_OFF
|
||||
|
||||
|
|
@ -690,12 +715,13 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
|
|||
{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 = bulb.raw_state._replace(preset_pattern=EFFECT_CUSTOM_CODE)
|
||||
bulb.is_on = True
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=20))
|
||||
await hass.async_block_till_done()
|
||||
bulb.async_set_custom_pattern.assert_called_with(
|
||||
[[0, 0, 255], [255, 0, 0]], 88, "jump"
|
||||
)
|
||||
bulb.async_set_custom_pattern.reset_mock()
|
||||
bulb.preset_pattern_num = EFFECT_CUSTOM_CODE
|
||||
await async_mock_bulb_turn_on(hass, bulb)
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.state == STATE_ON
|
||||
attributes = state.attributes
|
||||
|
|
@ -707,12 +733,13 @@ async def test_rgb_light_custom_effects(hass: HomeAssistant) -> None:
|
|||
{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 = bulb.raw_state._replace(preset_pattern=EFFECT_CUSTOM_CODE)
|
||||
bulb.is_on = True
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=20))
|
||||
await hass.async_block_till_done()
|
||||
bulb.async_set_custom_pattern.assert_called_with(
|
||||
[[0, 0, 255], [255, 0, 0]], 88, "jump"
|
||||
)
|
||||
bulb.async_set_custom_pattern.reset_mock()
|
||||
bulb.preset_pattern_num = EFFECT_CUSTOM_CODE
|
||||
await async_mock_bulb_turn_on(hass, bulb)
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
assert state.state == STATE_ON
|
||||
attributes = state.attributes
|
||||
|
|
@ -783,11 +810,9 @@ async def test_rgb_light_custom_effect_via_service(
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
|
||||
bulb.is_on = False
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
|
||||
await hass.async_block_till_done()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
assert hass.states.get(entity_id).state == STATE_OFF
|
||||
|
||||
await hass.services.async_call(
|
||||
|
|
@ -801,8 +826,10 @@ async def test_rgb_light_custom_effect_via_service(
|
|||
},
|
||||
blocking=True,
|
||||
)
|
||||
bulb.setCustomPattern.assert_called_with([(0, 0, 255), (255, 0, 0)], 30, "jump")
|
||||
bulb.setCustomPattern.reset_mock()
|
||||
bulb.async_set_custom_pattern.assert_called_with(
|
||||
[(0, 0, 255), (255, 0, 0)], 30, "jump"
|
||||
)
|
||||
bulb.async_set_custom_pattern.reset_mock()
|
||||
|
||||
|
||||
async def test_migrate_from_yaml(hass: HomeAssistant) -> None:
|
||||
|
|
@ -882,19 +909,17 @@ async def test_addressable_light(hass: HomeAssistant) -> None:
|
|||
await hass.services.async_call(
|
||||
LIGHT_DOMAIN, "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
|
||||
)
|
||||
bulb.turnOff.assert_called_once()
|
||||
bulb.async_turn_off.assert_called_once()
|
||||
|
||||
bulb.is_on = False
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=10))
|
||||
await hass.async_block_till_done()
|
||||
await async_mock_bulb_turn_off(hass, bulb)
|
||||
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()
|
||||
bulb.is_on = True
|
||||
bulb.async_turn_on.assert_called_once()
|
||||
bulb.async_turn_on.reset_mock()
|
||||
await async_mock_bulb_turn_on(hass, bulb)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
await hass.services.async_call(
|
||||
|
|
|
|||
Loading…
Reference in New Issue