Initial support for LIFX Ceiling SKY effect (#121820)

homeassistant/components/lifx/const.py

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

homeassistant/components/lifx/coordinator.py

@@ -6,7 +6,7 @@
 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 @@
     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 @@ def current_infrared_brightness(self) -> str | 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 @@ def label(self) -> str:
         """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 @@ async def _async_update_data(self) -> 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 @@ async def _async_update_data(self) -> 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 @@ async def async_set_multizone_effect(
         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 @@ async def async_set_multizone_effect(
             )
             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()]

homeassistant/components/lifx/icons.json

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

homeassistant/components/lifx/light.py

@@ -36,6 +36,7 @@
     DATA_LIFX_MANAGER,
     DOMAIN,
     INFRARED_BRIGHTNESS,
+    LIFX_CEILING_PRODUCT_IDS,
 )
 from .coordinator import FirmwareEffect, LIFXUpdateCoordinator
 from .entity import LIFXEntity
@@ -45,6 +46,7 @@
     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,
+    ]

homeassistant/components/lifx/manager.py

@@ -41,9 +41,12 @@
 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,13 +55,15 @@
 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"
 
 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_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_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_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 @@ async def service_handler(service: ServiceCall) -> None:
             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 @@ async def start_effect(
             )
             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)