zha ZCL color loop effect (#26549)

* Initial implementation of ZCL color loop effect
* Fix linter complaints
* Use const for action
* Reformat with Black
* Cleanup after review.
* Handle effect being None

homeassistant/components/zha/light.py

@@ -27,9 +27,15 @@
 
 DEFAULT_DURATION = 5
 
+CAPABILITIES_COLOR_LOOP = 0x4
 CAPABILITIES_COLOR_XY = 0x08
 CAPABILITIES_COLOR_TEMP = 0x10
 
+UPDATE_COLORLOOP_ACTION = 0x1
+UPDATE_COLORLOOP_DIRECTION = 0x2
+UPDATE_COLORLOOP_TIME = 0x4
+UPDATE_COLORLOOP_HUE = 0x8
+
 UNSUPPORTED_ATTRIBUTE = 0x86
 SCAN_INTERVAL = timedelta(minutes=60)
 PARALLEL_UPDATES = 5
@@ -85,6 +91,8 @@ def __init__(self, unique_id, zha_device, channels, **kwargs):
         self._color_temp = None
         self._hs_color = None
         self._brightness = None
+        self._effect_list = []
+        self._effect = None
         self._on_off_channel = self.cluster_channels.get(CHANNEL_ON_OFF)
         self._level_channel = self.cluster_channels.get(CHANNEL_LEVEL)
         self._color_channel = self.cluster_channels.get(CHANNEL_COLOR)
@@ -103,6 +111,10 @@ def __init__(self, unique_id, zha_device, channels, **kwargs):
                 self._supported_features |= light.SUPPORT_COLOR
                 self._hs_color = (0, 0)
 
+            if color_capabilities & CAPABILITIES_COLOR_LOOP:
+                self._supported_features |= light.SUPPORT_EFFECT
+                self._effect_list.append(light.EFFECT_COLORLOOP)
+
     @property
     def is_on(self) -> bool:
         """Return true if entity is on."""
@@ -141,6 +153,16 @@ def color_temp(self):
         """Return the CT color value in mireds."""
         return self._color_temp
 
+    @property
+    def effect_list(self):
+        """Return the list of supported effects."""
+        return self._effect_list
+
+    @property
+    def effect(self):
+        """Return the current effect."""
+        return self._effect
+
     @property
     def supported_features(self):
         """Flag supported features."""
@@ -173,12 +195,15 @@ def async_restore_last_state(self, last_state):
             self._color_temp = last_state.attributes["color_temp"]
         if "hs_color" in last_state.attributes:
             self._hs_color = last_state.attributes["hs_color"]
+        if "effect" in last_state.attributes:
+            self._effect = last_state.attributes["effect"]
 
     async def async_turn_on(self, **kwargs):
         """Turn the entity on."""
         transition = kwargs.get(light.ATTR_TRANSITION)
         duration = transition * 10 if transition else DEFAULT_DURATION
         brightness = kwargs.get(light.ATTR_BRIGHTNESS)
+        effect = kwargs.get(light.ATTR_EFFECT)
 
         t_log = {}
         if (
@@ -234,6 +259,36 @@ async def async_turn_on(self, **kwargs):
                 return
             self._hs_color = hs_color
 
+        if (
+            effect == light.EFFECT_COLORLOOP
+            and self.supported_features & light.SUPPORT_EFFECT
+        ):
+            result = await self._color_channel.color_loop_set(
+                UPDATE_COLORLOOP_ACTION
+                | UPDATE_COLORLOOP_DIRECTION
+                | UPDATE_COLORLOOP_TIME,
+                0x2,  # start from current hue
+                0x1,  # only support up
+                transition if transition else 7,  # transition
+                0,  # no hue
+            )
+            t_log["color_loop_set"] = result
+            self._effect = light.EFFECT_COLORLOOP
+        elif (
+            self._effect == light.EFFECT_COLORLOOP
+            and effect != light.EFFECT_COLORLOOP
+            and self.supported_features & light.SUPPORT_EFFECT
+        ):
+            result = await self._color_channel.color_loop_set(
+                UPDATE_COLORLOOP_ACTION,
+                0x0,
+                0x0,
+                0x0,
+                0x0,  # update action only, action off, no dir,time,hue
+            )
+            t_log["color_loop_set"] = result
+            self._effect = None
+
         self.debug("turned on: %s", t_log)
         self.async_schedule_update_ha_state()
 
@@ -292,6 +347,15 @@ async def async_get_state(self, from_cache=True):
                     self._hs_color = color_util.color_xy_to_hs(
                         float(color_x / 65535), float(color_y / 65535)
                     )
+            if (
+                color_capabilities is not None
+                and color_capabilities & CAPABILITIES_COLOR_LOOP
+            ):
+                color_loop_active = await self._color_channel.get_attribute_value(
+                    "color_loop_active", from_cache=from_cache
+                )
+                if color_loop_active is not None and color_loop_active == 1:
+                    self._effect = light.EFFECT_COLORLOOP
 
     async def refresh(self, time):
         """Call async_get_state at an interval."""