Initial support for LIFX Ceiling SKY effect (#121820)

This commit is contained in:
Avi Miller 2024-07-13 12:16:16 +10:00 committed by GitHub
parent 162b734be7
commit 5f33e85b30
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 394 additions and 23 deletions

View file

@ -61,5 +61,6 @@ INFRARED_BRIGHTNESS_VALUES_MAP = {
}
DATA_LIFX_MANAGER = "lifx_manager"
LIFX_CEILING_PRODUCT_IDS = {176, 177}
_LOGGER = logging.getLogger(__package__)

View file

@ -6,7 +6,7 @@ import asyncio
from collections.abc import Callable
from datetime import timedelta
from enum import IntEnum
from functools import partial
from functools import cached_property, partial
from math import floor, log10
from typing import Any, cast
@ -15,6 +15,7 @@ from aiolifx.aiolifx import (
Message,
MultiZoneDirection,
MultiZoneEffectType,
TileEffectSkyType,
TileEffectType,
)
from aiolifx.connection import LIFXConnection
@ -70,9 +71,18 @@ class FirmwareEffect(IntEnum):
MOVE = 1
MORPH = 2
FLAME = 3
SKY = 5
class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
class SkyType(IntEnum):
"""Enumeration of sky types for SKY firmware effect."""
SUNRISE = 0
SUNSET = 1
CLOUDS = 2
class LIFXUpdateCoordinator(DataUpdateCoordinator[None]): # noqa: PLR0904
"""DataUpdateCoordinator to gather data for a specific lifx device."""
def __init__(
@ -128,14 +138,14 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
"""Return the current infrared brightness as a string."""
return infrared_brightness_value_to_option(self.device.infrared_brightness)
@property
@cached_property
def serial_number(self) -> str:
"""Return the internal mac address."""
return cast(
str, self.device.mac_addr
) # device.mac_addr is not the mac_address, its the serial number
@property
@cached_property
def mac_address(self) -> str:
"""Return the physical mac address."""
return get_real_mac_addr(
@ -149,6 +159,23 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
"""Return the label of the bulb."""
return cast(str, self.device.label)
@cached_property
def is_extended_multizone(self) -> bool:
"""Return true if this is a multizone device."""
return bool(lifx_features(self.device)["extended_multizone"])
@cached_property
def is_legacy_multizone(self) -> bool:
"""Return true if this is a legacy multizone device."""
return bool(
lifx_features(self.device)["multizone"] and not self.is_extended_multizone
)
@cached_property
def is_matrix(self) -> bool:
"""Return true if this is a matrix device."""
return bool(lifx_features(self.device)["matrix"])
async def diagnostics(self) -> dict[str, Any]:
"""Return diagnostic information about the device."""
features = lifx_features(self.device)
@ -269,17 +296,23 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
num_zones = self.get_number_of_zones()
features = lifx_features(self.device)
is_extended_multizone = features["extended_multizone"]
is_legacy_multizone = not is_extended_multizone and features["multizone"]
update_rssi = self._update_rssi
methods: list[Callable] = [self.device.get_color]
if update_rssi:
methods.append(self.device.get_wifiinfo)
if is_extended_multizone:
if self.is_matrix:
methods.extend(
[
self.device.get_tile_effect,
self.device.get_device_chain,
self.device.get64,
]
)
if self.is_extended_multizone:
methods.append(self.device.get_extended_color_zones)
elif is_legacy_multizone:
elif self.is_legacy_multizone:
methods.extend(self._async_build_color_zones_update_requests())
if is_extended_multizone or is_legacy_multizone:
if self.is_extended_multizone or self.is_legacy_multizone:
methods.append(self.device.get_multizone_effect)
if features["hev"]:
methods.append(self.device.get_hev_cycle)
@ -297,9 +330,9 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
# We always send the rssi request second
self._rssi = int(floor(10 * log10(responses[1].signal) + 0.5))
if is_extended_multizone or is_legacy_multizone:
if self.is_matrix or self.is_extended_multizone or self.is_legacy_multizone:
self.active_effect = FirmwareEffect[self.device.effect.get("effect", "OFF")]
if is_legacy_multizone and num_zones != self.get_number_of_zones():
if self.is_legacy_multizone and num_zones != self.get_number_of_zones():
# The number of zones has changed so we need
# to update the zones again. This happens rarely.
await self.async_get_color_zones()
@ -402,7 +435,7 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
power_on: bool = True,
) -> None:
"""Control the firmware-based Move effect on a multizone device."""
if lifx_features(self.device)["multizone"] is True:
if self.is_extended_multizone or self.is_legacy_multizone:
if power_on and self.device.power_level == 0:
await self.async_set_power(True, 0)
@ -422,27 +455,36 @@ class LIFXUpdateCoordinator(DataUpdateCoordinator[None]):
)
self.active_effect = FirmwareEffect[effect.upper()]
async def async_set_matrix_effect(
async def async_set_matrix_effect( # noqa: PLR0917
self,
effect: str,
palette: list[tuple[int, int, int, int]] | None = None,
speed: float = 3,
speed: float | None = None,
power_on: bool = True,
sky_type: str | None = None,
cloud_saturation_min: int | None = None,
cloud_saturation_max: int | None = None,
) -> None:
"""Control the firmware-based effects on a matrix device."""
if lifx_features(self.device)["matrix"] is True:
if self.is_matrix:
if power_on and self.device.power_level == 0:
await self.async_set_power(True, 0)
if palette is None:
palette = []
if sky_type is not None:
sky_type = TileEffectSkyType[sky_type.upper()].value
await async_execute_lifx(
partial(
self.device.set_tile_effect,
effect=TileEffectType[effect.upper()].value,
speed=speed,
palette=palette,
sky_type=sky_type,
cloud_saturation_min=cloud_saturation_min,
cloud_saturation_max=cloud_saturation_max,
)
)
self.active_effect = FirmwareEffect[effect.upper()]

View file

@ -7,6 +7,7 @@
"effect_move": "mdi:cube-send",
"effect_flame": "mdi:fire",
"effect_morph": "mdi:shape-outline",
"effect_sky": "mdi:clouds",
"effect_stop": "mdi:stop"
}
}

View file

@ -36,6 +36,7 @@ from .const import (
DATA_LIFX_MANAGER,
DOMAIN,
INFRARED_BRIGHTNESS,
LIFX_CEILING_PRODUCT_IDS,
)
from .coordinator import FirmwareEffect, LIFXUpdateCoordinator
from .entity import LIFXEntity
@ -45,6 +46,7 @@ from .manager import (
SERVICE_EFFECT_MORPH,
SERVICE_EFFECT_MOVE,
SERVICE_EFFECT_PULSE,
SERVICE_EFFECT_SKY,
SERVICE_EFFECT_STOP,
LIFXManager,
)
@ -97,7 +99,10 @@ async def async_setup_entry(
"set_hev_cycle_state",
)
if lifx_features(device)["matrix"]:
entity: LIFXLight = LIFXMatrix(coordinator, manager, entry)
if device.product in LIFX_CEILING_PRODUCT_IDS:
entity: LIFXLight = LIFXCeiling(coordinator, manager, entry)
else:
entity = LIFXMatrix(coordinator, manager, entry)
elif lifx_features(device)["extended_multizone"]:
entity = LIFXExtendedMultiZone(coordinator, manager, entry)
elif lifx_features(device)["multizone"]:
@ -499,3 +504,16 @@ class LIFXMatrix(LIFXColor):
SERVICE_EFFECT_MORPH,
SERVICE_EFFECT_STOP,
]
class LIFXCeiling(LIFXMatrix):
"""Representation of a LIFX Ceiling device."""
_attr_effect_list = [
SERVICE_EFFECT_COLORLOOP,
SERVICE_EFFECT_FLAME,
SERVICE_EFFECT_PULSE,
SERVICE_EFFECT_MORPH,
SERVICE_EFFECT_SKY,
SERVICE_EFFECT_STOP,
]

View file

@ -41,9 +41,12 @@ SERVICE_EFFECT_FLAME = "effect_flame"
SERVICE_EFFECT_MORPH = "effect_morph"
SERVICE_EFFECT_MOVE = "effect_move"
SERVICE_EFFECT_PULSE = "effect_pulse"
SERVICE_EFFECT_SKY = "effect_sky"
SERVICE_EFFECT_STOP = "effect_stop"
ATTR_CHANGE = "change"
ATTR_CLOUD_SATURATION_MIN = "cloud_saturation_min"
ATTR_CLOUD_SATURATION_MAX = "cloud_saturation_max"
ATTR_CYCLES = "cycles"
ATTR_DIRECTION = "direction"
ATTR_PALETTE = "palette"
@ -52,6 +55,7 @@ ATTR_POWER_OFF = "power_off"
ATTR_POWER_ON = "power_on"
ATTR_SATURATION_MAX = "saturation_max"
ATTR_SATURATION_MIN = "saturation_min"
ATTR_SKY_TYPE = "sky_type"
ATTR_SPEED = "speed"
ATTR_SPREAD = "spread"
@ -59,6 +63,7 @@ EFFECT_FLAME = "FLAME"
EFFECT_MORPH = "MORPH"
EFFECT_MOVE = "MOVE"
EFFECT_OFF = "OFF"
EFFECT_SKY = "SKY"
EFFECT_FLAME_DEFAULT_SPEED = 3
@ -72,6 +77,13 @@ EFFECT_MOVE_DIRECTION_LEFT = "left"
EFFECT_MOVE_DIRECTIONS = [EFFECT_MOVE_DIRECTION_LEFT, EFFECT_MOVE_DIRECTION_RIGHT]
EFFECT_SKY_DEFAULT_SPEED = 50
EFFECT_SKY_DEFAULT_SKY_TYPE = "Clouds"
EFFECT_SKY_DEFAULT_CLOUD_SATURATION_MIN = 50
EFFECT_SKY_DEFAULT_CLOUD_SATURATION_MAX = 180
EFFECT_SKY_SKY_TYPES = ["Sunrise", "Sunset", "Clouds"]
PULSE_MODE_BLINK = "blink"
PULSE_MODE_BREATHE = "breathe"
PULSE_MODE_PING = "ping"
@ -137,13 +149,6 @@ LIFX_EFFECT_COLORLOOP_SCHEMA = cv.make_entity_service_schema(
LIFX_EFFECT_STOP_SCHEMA = cv.make_entity_service_schema({})
SERVICES = (
SERVICE_EFFECT_STOP,
SERVICE_EFFECT_PULSE,
SERVICE_EFFECT_MOVE,
SERVICE_EFFECT_COLORLOOP,
)
LIFX_EFFECT_FLAME_SCHEMA = cv.make_entity_service_schema(
{
**LIFX_EFFECT_SCHEMA,
@ -185,6 +190,28 @@ LIFX_EFFECT_MOVE_SCHEMA = cv.make_entity_service_schema(
}
)
LIFX_EFFECT_SKY_SCHEMA = cv.make_entity_service_schema(
{
**LIFX_EFFECT_SCHEMA,
ATTR_SPEED: vol.All(vol.Coerce(int), vol.Clamp(min=1, max=86400)),
ATTR_SKY_TYPE: vol.In(EFFECT_SKY_SKY_TYPES),
ATTR_CLOUD_SATURATION_MIN: vol.All(vol.Coerce(int), vol.Clamp(min=0, max=255)),
ATTR_CLOUD_SATURATION_MAX: vol.All(vol.Coerce(int), vol.Clamp(min=0, max=255)),
ATTR_PALETTE: vol.All(cv.ensure_list, [HSBK_SCHEMA]),
}
)
SERVICES = (
SERVICE_EFFECT_COLORLOOP,
SERVICE_EFFECT_FLAME,
SERVICE_EFFECT_MORPH,
SERVICE_EFFECT_MOVE,
SERVICE_EFFECT_PULSE,
SERVICE_EFFECT_SKY,
SERVICE_EFFECT_STOP,
)
class LIFXManager:
"""Representation of all known LIFX entities."""
@ -261,6 +288,13 @@ class LIFXManager:
schema=LIFX_EFFECT_MOVE_SCHEMA,
)
self.hass.services.async_register(
DOMAIN,
SERVICE_EFFECT_SKY,
service_handler,
schema=LIFX_EFFECT_SKY_SCHEMA,
)
self.hass.services.async_register(
DOMAIN,
SERVICE_EFFECT_STOP,
@ -375,6 +409,39 @@ class LIFXManager:
)
await self.effects_conductor.start(effect, bulbs)
elif service == SERVICE_EFFECT_SKY:
palette = kwargs.get(ATTR_PALETTE, None)
if palette is not None:
theme = Theme()
for hsbk in palette:
theme.add_hsbk(hsbk[0], hsbk[1], hsbk[2], hsbk[3])
speed = kwargs.get(ATTR_SPEED, EFFECT_SKY_DEFAULT_SPEED)
sky_type = kwargs.get(ATTR_SKY_TYPE, EFFECT_SKY_DEFAULT_SKY_TYPE)
cloud_saturation_min = kwargs.get(
ATTR_CLOUD_SATURATION_MIN,
EFFECT_SKY_DEFAULT_CLOUD_SATURATION_MIN,
)
cloud_saturation_max = kwargs.get(
ATTR_CLOUD_SATURATION_MAX,
EFFECT_SKY_DEFAULT_CLOUD_SATURATION_MAX,
)
await asyncio.gather(
*(
coordinator.async_set_matrix_effect(
effect=EFFECT_SKY,
speed=speed,
sky_type=sky_type,
cloud_saturation_min=cloud_saturation_min,
cloud_saturation_max=cloud_saturation_max,
palette=theme.colors,
)
for coordinator in coordinators
)
)
elif service == SERVICE_EFFECT_STOP:
await self.effects_conductor.stop(bulbs)

View file

@ -281,6 +281,58 @@ effect_morph:
default: true
selector:
boolean:
effect_sky:
target:
entity:
integration: lifx
domain: light
fields:
power_on:
default: true
selector:
boolean:
speed:
default: 50
example: 50
selector:
number:
min: 1
max: 86400
step: 1
unit_of_measurement: seconds
sky_type:
default: "Clouds"
example: "Clouds"
selector:
select:
options:
- "Clouds"
- "Sunrise"
- "Sunset"
cloud_saturation_min:
default: 50
example: 50
selector:
number:
min: 0
max: 255
cloud_saturation_max:
default: 180
example: 180
selector:
number:
min: 0
max: 255
palette:
example:
- "[200, 1, 1, 3500]"
- "[241, 1, 0.01, 3500]"
- "[189, 1, 0.08, 3500]"
- "[40, 1, 1, 3500]"
- "[40, 0.5, 1, 3500]"
- "[40, 0, 1, 6500]"
selector:
object:
effect_stop:
target:
entity:

View file

@ -220,6 +220,36 @@
}
}
},
"effect_sky": {
"name": "Sky effect",
"description": "Starts the firmware-based Sky effect on LIFX Ceiling.",
"fields": {
"speed": {
"name": "Speed",
"description": "How long the Sunrise and Sunset sky types will take to complete. For the Cloud sky type, it is the speed of the clouds across the device."
},
"sky_type": {
"name": "Sky type",
"description": "The style of sky that will be animated by the effect."
},
"cloud_saturation_min": {
"name": "Cloud saturation Minimum",
"description": "Minimum cloud saturation."
},
"cloud_saturation_max": {
"name": "Cloud Saturation maximum",
"description": "Maximum cloud saturation."
},
"palette": {
"name": "Palette",
"description": "List of 1 to 6 colors as hue (0-360), saturation (0-100), brightness (0-100) and kelvin (1500-9000) values to use for this effect."
},
"power_on": {
"name": "Power on",
"description": "[%key:component::lifx::services::effect_move::fields::power_on::description%]"
}
}
},
"effect_stop": {
"name": "Stop effect",
"description": "Stops a running effect."

View file

@ -172,6 +172,19 @@ def _mocked_tile() -> Light:
bulb.effect = {"effect": "OFF"}
bulb.get_tile_effect = MockLifxCommand(bulb)
bulb.set_tile_effect = MockLifxCommand(bulb)
bulb.get64 = MockLifxCommand(bulb)
bulb.get_device_chain = MockLifxCommand(bulb)
return bulb
def _mocked_ceiling() -> Light:
bulb = _mocked_bulb()
bulb.product = 176 # LIFX Ceiling
bulb.effect = {"effect": "OFF"}
bulb.get_tile_effect = MockLifxCommand(bulb)
bulb.set_tile_effect = MockLifxCommand(bulb)
bulb.get64 = MockLifxCommand(bulb)
bulb.get_device_chain = MockLifxCommand(bulb)
return bulb

View file

@ -11,15 +11,19 @@ from homeassistant.components.lifx import DOMAIN
from homeassistant.components.lifx.const import ATTR_POWER
from homeassistant.components.lifx.light import ATTR_INFRARED, ATTR_ZONES
from homeassistant.components.lifx.manager import (
ATTR_CLOUD_SATURATION_MAX,
ATTR_CLOUD_SATURATION_MIN,
ATTR_DIRECTION,
ATTR_PALETTE,
ATTR_SATURATION_MAX,
ATTR_SATURATION_MIN,
ATTR_SKY_TYPE,
ATTR_SPEED,
ATTR_THEME,
SERVICE_EFFECT_COLORLOOP,
SERVICE_EFFECT_MORPH,
SERVICE_EFFECT_MOVE,
SERVICE_EFFECT_SKY,
)
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -62,6 +66,7 @@ from . import (
_mocked_brightness_bulb,
_mocked_bulb,
_mocked_bulb_new_firmware,
_mocked_ceiling,
_mocked_clean_bulb,
_mocked_light_strip,
_mocked_tile,
@ -691,6 +696,7 @@ async def test_matrix_flame_morph_effects(hass: HomeAssistant) -> None:
entity_id = "light.my_bulb"
# FLAME effect test
await hass.services.async_call(
LIGHT_DOMAIN,
"turn_on",
@ -707,11 +713,15 @@ async def test_matrix_flame_morph_effects(hass: HomeAssistant) -> None:
"effect": 3,
"speed": 3,
"palette": [],
"sky_type": None,
"cloud_saturation_min": None,
"cloud_saturation_max": None,
}
bulb.get_tile_effect.reset_mock()
bulb.set_tile_effect.reset_mock()
bulb.set_power.reset_mock()
# MORPH effect tests
bulb.power_level = 0
await hass.services.async_call(
DOMAIN,
@ -750,6 +760,9 @@ async def test_matrix_flame_morph_effects(hass: HomeAssistant) -> None:
(8920, 65535, 32768, 3500),
(10558, 65535, 32768, 3500),
],
"sky_type": None,
"cloud_saturation_min": None,
"cloud_saturation_max": None,
}
bulb.get_tile_effect.reset_mock()
bulb.set_tile_effect.reset_mock()
@ -808,6 +821,140 @@ async def test_matrix_flame_morph_effects(hass: HomeAssistant) -> None:
(43690, 65535, 65535, 3500),
(54613, 65535, 65535, 3500),
],
"sky_type": None,
"cloud_saturation_min": None,
"cloud_saturation_max": None,
}
bulb.get_tile_effect.reset_mock()
bulb.set_tile_effect.reset_mock()
bulb.set_power.reset_mock()
@pytest.mark.usefixtures("mock_discovery")
async def test_sky_effect(hass: HomeAssistant) -> None:
"""Test the firmware sky effect on a ceiling device."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: "127.0.0.1"}, unique_id=SERIAL
)
config_entry.add_to_hass(hass)
bulb = _mocked_ceiling()
bulb.power_level = 0
bulb.color = [65535, 65535, 65535, 65535]
with (
_patch_discovery(device=bulb),
_patch_config_flow_try_connect(device=bulb),
_patch_device(device=bulb),
):
await async_setup_component(hass, lifx.DOMAIN, {lifx.DOMAIN: {}})
await hass.async_block_till_done()
entity_id = "light.my_bulb"
# SKY effect test
bulb.power_level = 0
await hass.services.async_call(
DOMAIN,
SERVICE_EFFECT_SKY,
{
ATTR_ENTITY_ID: entity_id,
ATTR_PALETTE: [],
ATTR_SKY_TYPE: "Clouds",
ATTR_CLOUD_SATURATION_MAX: 180,
ATTR_CLOUD_SATURATION_MIN: 50,
},
blocking=True,
)
bulb.power_level = 65535
bulb.effect = {
"effect": "SKY",
"palette": [],
"sky_type": 2,
"cloud_saturation_min": 50,
"cloud_saturation_max": 180,
}
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=30))
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert len(bulb.set_power.calls) == 1
assert len(bulb.set_tile_effect.calls) == 1
call_dict = bulb.set_tile_effect.calls[0][1]
call_dict.pop("callb")
assert call_dict == {
"effect": 5,
"speed": 50,
"palette": [],
"sky_type": 2,
"cloud_saturation_min": 50,
"cloud_saturation_max": 180,
}
bulb.get_tile_effect.reset_mock()
bulb.set_tile_effect.reset_mock()
bulb.set_power.reset_mock()
bulb.power_level = 0
await hass.services.async_call(
DOMAIN,
SERVICE_EFFECT_SKY,
{
ATTR_ENTITY_ID: entity_id,
ATTR_PALETTE: [
(200, 100, 1, 3500),
(241, 100, 1, 3500),
(189, 100, 8, 3500),
(40, 100, 100, 3500),
(40, 50, 100, 3500),
(0, 0, 100, 6500),
],
ATTR_SKY_TYPE: "Sunrise",
ATTR_CLOUD_SATURATION_MAX: 180,
ATTR_CLOUD_SATURATION_MIN: 50,
},
blocking=True,
)
bulb.power_level = 65535
bulb.effect = {
"effect": "SKY",
"palette": [
(200, 100, 1, 3500),
(241, 100, 1, 3500),
(189, 100, 8, 3500),
(40, 100, 100, 3500),
(40, 50, 100, 3500),
(0, 0, 100, 6500),
],
"sky_type": 0,
"cloud_saturation_min": 50,
"cloud_saturation_max": 180,
}
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=30))
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert len(bulb.set_power.calls) == 1
assert len(bulb.set_tile_effect.calls) == 1
call_dict = bulb.set_tile_effect.calls[0][1]
call_dict.pop("callb")
assert call_dict == {
"effect": 5,
"speed": 50,
"palette": [
(36408, 65535, 65535, 3500),
(43872, 65535, 65535, 3500),
(34406, 65535, 5243, 3500),
(7281, 65535, 65535, 3500),
(7281, 32768, 65535, 3500),
(0, 0, 65535, 6500),
],
"sky_type": 0,
"cloud_saturation_min": 50,
"cloud_saturation_max": 180,
}
bulb.get_tile_effect.reset_mock()
bulb.set_tile_effect.reset_mock()