light.py

"""Lights on Zigbee Home Automation networks."""
from __future__ import annotations

from collections import Counter
from collections.abc import Callable
from datetime import timedelta
import functools
import itertools
import logging
import random
from typing import TYPE_CHECKING, Any, cast

from zigpy.zcl.clusters.general import Identify, LevelControl, OnOff
from zigpy.zcl.clusters.lighting import Color
from zigpy.zcl.foundation import Status

from homeassistant.components import light
from homeassistant.components.light import (
    ColorMode,
    LightEntityFeature,
    brightness_supported,
    filter_supported_color_modes,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    ATTR_SUPPORTED_FEATURES,
    STATE_ON,
    STATE_UNAVAILABLE,
    Platform,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, State, callback
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.dispatcher import (
    async_dispatcher_connect,
    async_dispatcher_send,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_call_later, async_track_time_interval

from .core import discovery, helpers
from .core.const import (
    CLUSTER_HANDLER_COLOR,
    CLUSTER_HANDLER_LEVEL,
    CLUSTER_HANDLER_ON_OFF,
    CONF_ALWAYS_PREFER_XY_COLOR_MODE,
    CONF_DEFAULT_LIGHT_TRANSITION,
    CONF_ENABLE_ENHANCED_LIGHT_TRANSITION,
    CONF_ENABLE_LIGHT_TRANSITIONING_FLAG,
    CONF_GROUP_MEMBERS_ASSUME_STATE,
    DATA_ZHA,
    SIGNAL_ADD_ENTITIES,
    SIGNAL_ATTR_UPDATED,
    SIGNAL_SET_LEVEL,
    ZHA_OPTIONS,
)
from .core.helpers import LogMixin, async_get_zha_config_value, get_zha_data
from .core.registries import ZHA_ENTITIES
from .entity import ZhaEntity, ZhaGroupEntity

if TYPE_CHECKING:
    from .core.device import ZHADevice

_LOGGER = logging.getLogger(__name__)

DEFAULT_ON_OFF_TRANSITION = 1  # most bulbs default to a 1-second turn on/off transition
DEFAULT_EXTRA_TRANSITION_DELAY_SHORT = 0.25
DEFAULT_EXTRA_TRANSITION_DELAY_LONG = 2.0
DEFAULT_LONG_TRANSITION_TIME = 10
DEFAULT_MIN_BRIGHTNESS = 2
ASSUME_UPDATE_GROUP_FROM_CHILD_DELAY = 0.05

FLASH_EFFECTS = {
    light.FLASH_SHORT: Identify.EffectIdentifier.Blink,
    light.FLASH_LONG: Identify.EffectIdentifier.Breathe,
}

STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, Platform.LIGHT)
GROUP_MATCH = functools.partial(ZHA_ENTITIES.group_match, Platform.LIGHT)
SIGNAL_LIGHT_GROUP_STATE_CHANGED = "zha_light_group_state_changed"
SIGNAL_LIGHT_GROUP_TRANSITION_START = "zha_light_group_transition_start"
SIGNAL_LIGHT_GROUP_TRANSITION_FINISHED = "zha_light_group_transition_finished"
SIGNAL_LIGHT_GROUP_ASSUME_GROUP_STATE = "zha_light_group_assume_group_state"
DEFAULT_MIN_TRANSITION_MANUFACTURERS = {"sengled"}

COLOR_MODES_GROUP_LIGHT = {ColorMode.COLOR_TEMP, ColorMode.XY}
SUPPORT_GROUP_LIGHT = (
    light.LightEntityFeature.EFFECT
    | light.LightEntityFeature.FLASH
    | light.LightEntityFeature.TRANSITION
)


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up the Zigbee Home Automation light from config entry."""
    zha_data = get_zha_data(hass)
    entities_to_create = zha_data.platforms[Platform.LIGHT]

    unsub = async_dispatcher_connect(
        hass,
        SIGNAL_ADD_ENTITIES,
        functools.partial(
            discovery.async_add_entities, async_add_entities, entities_to_create
        ),
    )
    config_entry.async_on_unload(unsub)


class BaseLight(LogMixin, light.LightEntity):
    """Operations common to all light entities."""

    _FORCE_ON = False
    _DEFAULT_MIN_TRANSITION_TIME: float = 0

    def __init__(self, *args, **kwargs):
        """Initialize the light."""
        self._zha_device: ZHADevice = None
        super().__init__(*args, **kwargs)
        self._attr_min_mireds: int | None = 153
        self._attr_max_mireds: int | None = 500
        self._attr_color_mode = ColorMode.UNKNOWN  # Set by subclasses
        self._attr_supported_features: int = 0
        self._attr_state: bool | None
        self._off_with_transition: bool = False
        self._off_brightness: int | None = None
        self._zha_config_transition = self._DEFAULT_MIN_TRANSITION_TIME
        self._zha_config_enhanced_light_transition: bool = False
        self._zha_config_enable_light_transitioning_flag: bool = True
        self._zha_config_always_prefer_xy_color_mode: bool = True
        self._on_off_cluster_handler = None
        self._level_cluster_handler = None
        self._color_cluster_handler = None
        self._identify_cluster_handler = None
        self._transitioning_individual: bool = False
        self._transitioning_group: bool = False
        self._transition_listener: Callable[[], None] | None = None

    async def async_will_remove_from_hass(self) -> None:
        """Disconnect entity object when removed."""
        self._async_unsub_transition_listener()
        await super().async_will_remove_from_hass()

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return state attributes."""
        attributes = {
            "off_with_transition": self._off_with_transition,
            "off_brightness": self._off_brightness,
        }
        return attributes

    @property
    def is_on(self) -> bool:
        """Return true if entity is on."""
        if self._attr_state is None:
            return False
        return self._attr_state

    @callback
    def set_level(self, value: int) -> None:
        """Set the brightness of this light between 0..254.

        brightness level 255 is a special value instructing the device to come
        on at `on_level` Zigbee attribute value, regardless of the last set
        level
        """
        if self.is_transitioning:
            self.debug(
                "received level %s while transitioning - skipping update",
                value,
            )
            return
        value = max(0, min(254, value))
        self._attr_brightness = value
        self.async_write_ha_state()

    async def async_turn_on(self, **kwargs: Any) -> None:
        """Turn the entity on."""
        transition = kwargs.get(light.ATTR_TRANSITION)
        duration = (
            transition if transition is not None else self._zha_config_transition
        ) or (
            # if 0 is passed in some devices still need the minimum default
            self._DEFAULT_MIN_TRANSITION_TIME
        )
        brightness = kwargs.get(light.ATTR_BRIGHTNESS)
        effect = kwargs.get(light.ATTR_EFFECT)
        flash = kwargs.get(light.ATTR_FLASH)
        temperature = kwargs.get(light.ATTR_COLOR_TEMP)
        xy_color = kwargs.get(light.ATTR_XY_COLOR)
        hs_color = kwargs.get(light.ATTR_HS_COLOR)

        execute_if_off_supported = (
            self._GROUP_SUPPORTS_EXECUTE_IF_OFF
            if isinstance(self, LightGroup)
            else self._color_cluster_handler
            and self._color_cluster_handler.execute_if_off_supported
        )

        set_transition_flag = (
            brightness_supported(self._attr_supported_color_modes)
            or temperature is not None
            or xy_color is not None
            or hs_color is not None
        ) and self._zha_config_enable_light_transitioning_flag
        transition_time = (
            (
                duration + DEFAULT_EXTRA_TRANSITION_DELAY_SHORT
                if (
                    (brightness is not None or transition is not None)
                    and brightness_supported(self._attr_supported_color_modes)
                    or (self._off_with_transition and self._off_brightness is not None)
                    or temperature is not None
                    or xy_color is not None
                    or hs_color is not None
                )
                else DEFAULT_ON_OFF_TRANSITION + DEFAULT_EXTRA_TRANSITION_DELAY_SHORT
            )
            if set_transition_flag
            else 0
        )

        # If we need to pause attribute report parsing, we'll do so here.
        # After successful calls, we later start a timer to unset the flag after
        # transition_time.
        # - On an error on the first move to level call, we unset the flag immediately
        #   if no previous timer is running.
        # - On an error on subsequent calls, we start the transition timer,
        #   as a brightness call might have come through.
        if set_transition_flag:
            self.async_transition_set_flag()

        # If the light is currently off but a turn_on call with a color/temperature is
        # sent, the light needs to be turned on first at a low brightness level where
        # the light is immediately transitioned to the correct color. Afterwards, the
        # transition is only from the low brightness to the new brightness.
        # Otherwise, the transition is from the color the light had before being turned
        # on to the new color. This can look especially bad with transitions longer than
        # a second. We do not want to do this for devices that need to be forced to use
        # the on command because we would end up with 4 commands sent:
        # move to level, on, color, move to level... We also will not set this
        # if the bulb is already in the desired color mode with the desired color
        # or color temperature.
        new_color_provided_while_off = (
            self._zha_config_enhanced_light_transition
            and not self._FORCE_ON
            and not self._attr_state
            and (
                (
                    temperature is not None
                    and (
                        self._attr_color_temp != temperature
                        or self._attr_color_mode != ColorMode.COLOR_TEMP
                    )
                )
                or (
                    xy_color is not None
                    and (
                        self._attr_xy_color != xy_color
                        or self._attr_color_mode != ColorMode.XY
                    )
                )
                or (
                    hs_color is not None
                    and (
                        self._attr_hs_color != hs_color
                        or self._attr_color_mode != ColorMode.HS
                    )
                )
            )
            and brightness_supported(self._attr_supported_color_modes)
            and not execute_if_off_supported
        )

        if (
            brightness is None
            and (self._off_with_transition or new_color_provided_while_off)
            and self._off_brightness is not None
        ):
            brightness = self._off_brightness

        if brightness is not None:
            level = min(254, brightness)
        else:
            level = self._attr_brightness or 254

        t_log = {}

        if new_color_provided_while_off:
            # If the light is currently off, we first need to turn it on at a low
            # brightness level with no transition.
            # After that, we set it to the desired color/temperature with no transition.
            result = await self._level_cluster_handler.move_to_level_with_on_off(
                level=DEFAULT_MIN_BRIGHTNESS,
                transition_time=int(10 * self._DEFAULT_MIN_TRANSITION_TIME),
            )
            t_log["move_to_level_with_on_off"] = result
            if result[1] is not Status.SUCCESS:
                # First 'move to level' call failed, so if the transitioning delay
                # isn't running from a previous call,
                # the flag can be unset immediately
                if set_transition_flag and not self._transition_listener:
                    self.async_transition_complete()
                self.debug("turned on: %s", t_log)
                return
            # Currently only setting it to "on", as the correct level state will
            # be set at the second move_to_level call
            self._attr_state = True

        if execute_if_off_supported:
            self.debug("handling color commands before turning on/level")
            if not await self.async_handle_color_commands(
                temperature,
                duration,  # duration is ignored by lights when off
                hs_color,
                xy_color,
                new_color_provided_while_off,
                t_log,
            ):
                # Color calls before on/level calls failed,
                # so if the transitioning delay isn't running from a previous call,
                # the flag can be unset immediately
                if set_transition_flag and not self._transition_listener:
                    self.async_transition_complete()
                self.debug("turned on: %s", t_log)
                return

        if (
            (brightness is not None or transition is not None)
            and not new_color_provided_while_off
            and brightness_supported(self._attr_supported_color_modes)
        ):
            result = await self._level_cluster_handler.move_to_level_with_on_off(
                level=level,
                transition_time=int(10 * duration),
            )
            t_log["move_to_level_with_on_off"] = result
            if result[1] is not Status.SUCCESS:
                # First 'move to level' call failed, so if the transitioning delay
                # isn't running from a previous call, the flag can be unset immediately
                if set_transition_flag and not self._transition_listener:
                    self.async_transition_complete()
                self.debug("turned on: %s", t_log)
                return
            self._attr_state = bool(level)
            if level:
                self._attr_brightness = level

        if (
            (brightness is None and transition is None)
            and not new_color_provided_while_off
            or (self._FORCE_ON and brightness != 0)
        ):
            # since FORCE_ON lights don't turn on with move_to_level_with_on_off,
            # we should call the on command on the on_off cluster
            # if brightness is not 0.
            result = await self._on_off_cluster_handler.on()
            t_log["on_off"] = result
            if result[1] is not Status.SUCCESS:
                # 'On' call failed, but as brightness may still transition
                # (for FORCE_ON lights), we start the timer to unset the flag after
                # the transition_time if necessary.
                self.async_transition_start_timer(transition_time)
                self.debug("turned on: %s", t_log)
                return
            self._attr_state = True

        if not execute_if_off_supported:
            self.debug("handling color commands after turning on/level")
            if not await self.async_handle_color_commands(
                temperature,
                duration,
                hs_color,
                xy_color,
                new_color_provided_while_off,
                t_log,
            ):
                # Color calls failed, but as brightness may still transition,
                # we start the timer to unset the flag
                self.async_transition_start_timer(transition_time)
                self.debug("turned on: %s", t_log)
                return

        if new_color_provided_while_off:
            # The light has the correct color, so we can now transition
            # it to the correct brightness level.
            result = await self._level_cluster_handler.move_to_level(
                level=level, transition_time=int(10 * duration)
            )
            t_log["move_to_level_if_color"] = result
            if result[1] is not Status.SUCCESS:
                self.debug("turned on: %s", t_log)
                return
            self._attr_state = bool(level)
            if level:
                self._attr_brightness = level

        # Our light is guaranteed to have just started the transitioning process
        # if necessary, so we start the delay for the transition (to stop parsing
        # attribute reports after the completed transition).
        self.async_transition_start_timer(transition_time)

        if effect == light.EFFECT_COLORLOOP:
            result = await self._color_cluster_handler.color_loop_set(
                update_flags=(
                    Color.ColorLoopUpdateFlags.Action
                    | Color.ColorLoopUpdateFlags.Direction
                    | Color.ColorLoopUpdateFlags.Time
                ),
                action=Color.ColorLoopAction.Activate_from_current_hue,
                direction=Color.ColorLoopDirection.Increment,
                time=transition if transition else 7,
                start_hue=0,
            )
            t_log["color_loop_set"] = result
            self._attr_effect = light.EFFECT_COLORLOOP
        elif (
            self._attr_effect == light.EFFECT_COLORLOOP
            and effect != light.EFFECT_COLORLOOP
        ):
            result = await self._color_cluster_handler.color_loop_set(
                update_flags=Color.ColorLoopUpdateFlags.Action,
                action=Color.ColorLoopAction.Deactivate,
                direction=Color.ColorLoopDirection.Decrement,
                time=0,
                start_hue=0,
            )
            t_log["color_loop_set"] = result
            self._attr_effect = None

        if flash is not None:
            result = await self._identify_cluster_handler.trigger_effect(
                effect_id=FLASH_EFFECTS[flash],
                effect_variant=Identify.EffectVariant.Default,
            )
            t_log["trigger_effect"] = result

        self._off_with_transition = False
        self._off_brightness = None
        self.debug("turned on: %s", t_log)
        self.async_write_ha_state()

    async def async_turn_off(self, **kwargs: Any) -> None:
        """Turn the entity off."""
        transition = kwargs.get(light.ATTR_TRANSITION)
        supports_level = brightness_supported(self._attr_supported_color_modes)

        transition_time = (
            transition or self._DEFAULT_MIN_TRANSITION_TIME
            if transition is not None
            else DEFAULT_ON_OFF_TRANSITION
        ) + DEFAULT_EXTRA_TRANSITION_DELAY_SHORT

        # Start pausing attribute report parsing
        if self._zha_config_enable_light_transitioning_flag:
            self.async_transition_set_flag()

        # is not none looks odd here, but it will override built in bulb
        # transition times if we pass 0 in here
        if transition is not None and supports_level:
            result = await self._level_cluster_handler.move_to_level_with_on_off(
                level=0,
                transition_time=int(
                    10 * (transition or self._DEFAULT_MIN_TRANSITION_TIME)
                ),
            )
        else:
            result = await self._on_off_cluster_handler.off()

        # Pause parsing attribute reports until transition is complete
        if self._zha_config_enable_light_transitioning_flag:
            self.async_transition_start_timer(transition_time)
        self.debug("turned off: %s", result)
        if result[1] is not Status.SUCCESS:
            return
        self._attr_state = False

        if supports_level and not self._off_with_transition:
            # store current brightness so that the next turn_on uses it:
            # when using "enhanced turn on"
            self._off_brightness = self._attr_brightness
            if transition is not None:
                # save for when calling turn_on without a brightness:
                # current_level is set to 1 after transitioning to level 0,
                # needed for correct state with light groups
                self._attr_brightness = 1
                self._off_with_transition = transition is not None

        self.async_write_ha_state()

    async def async_handle_color_commands(
        self,
        temperature,
        duration,
        hs_color,
        xy_color,
        new_color_provided_while_off,
        t_log,
    ):
        """Process ZCL color commands."""

        transition_time = (
            self._DEFAULT_MIN_TRANSITION_TIME
            if new_color_provided_while_off
            else duration
        )

        if temperature is not None:
            result = await self._color_cluster_handler.move_to_color_temp(
                color_temp_mireds=temperature,
                transition_time=int(10 * transition_time),
            )
            t_log["move_to_color_temp"] = result
            if result[1] is not Status.SUCCESS:
                return False
            self._attr_color_mode = ColorMode.COLOR_TEMP
            self._attr_color_temp = temperature
            self._attr_xy_color = None
            self._attr_hs_color = None

        if hs_color is not None:
            if (
                not isinstance(self, LightGroup)
                and self._color_cluster_handler.enhanced_hue_supported
            ):
                result = await self._color_cluster_handler.enhanced_move_to_hue_and_saturation(
                    enhanced_hue=int(hs_color[0] * 65535 / 360),
                    saturation=int(hs_color[1] * 2.54),
                    transition_time=int(10 * transition_time),
                )
                t_log["enhanced_move_to_hue_and_saturation"] = result
            else:
                result = await self._color_cluster_handler.move_to_hue_and_saturation(
                    hue=int(hs_color[0] * 254 / 360),
                    saturation=int(hs_color[1] * 2.54),
                    transition_time=int(10 * transition_time),
                )
                t_log["move_to_hue_and_saturation"] = result
            if result[1] is not Status.SUCCESS:
                return False
            self._attr_color_mode = ColorMode.HS
            self._attr_hs_color = hs_color
            self._attr_xy_color = None
            self._attr_color_temp = None
            xy_color = None  # don't set xy_color if it is also present

        if xy_color is not None:
            result = await self._color_cluster_handler.move_to_color(
                color_x=int(xy_color[0] * 65535),
                color_y=int(xy_color[1] * 65535),
                transition_time=int(10 * transition_time),
            )
            t_log["move_to_color"] = result
            if result[1] is not Status.SUCCESS:
                return False
            self._attr_color_mode = ColorMode.XY
            self._attr_xy_color = xy_color
            self._attr_color_temp = None
            self._attr_hs_color = None

        return True

    @property
    def is_transitioning(self) -> bool:
        """Return if the light is transitioning."""
        return self._transitioning_individual or self._transitioning_group

    @callback
    def async_transition_set_flag(self) -> None:
        """Set _transitioning to True."""
        self.debug("setting transitioning flag to True")
        self._transitioning_individual = True
        self._transitioning_group = False
        if isinstance(self, LightGroup):
            async_dispatcher_send(
                self.hass,
                SIGNAL_LIGHT_GROUP_TRANSITION_START,
                {"entity_ids": self._entity_ids},
            )
        self._async_unsub_transition_listener()

    @callback
    def async_transition_start_timer(self, transition_time) -> None:
        """Start a timer to unset _transitioning_individual after transition_time.

        If necessary.
        """
        if not transition_time:
            return
        # For longer transitions, we want to extend the timer a bit more
        if transition_time >= DEFAULT_LONG_TRANSITION_TIME:
            transition_time += DEFAULT_EXTRA_TRANSITION_DELAY_LONG
        self.debug("starting transitioning timer for %s", transition_time)
        self._transition_listener = async_call_later(
            self._zha_device.hass,
            transition_time,
            self.async_transition_complete,
        )

    @callback
    def _async_unsub_transition_listener(self) -> None:
        """Unsubscribe transition listener."""
        if self._transition_listener:
            self._transition_listener()
            self._transition_listener = None

    @callback
    def async_transition_complete(self, _=None) -> None:
        """Set _transitioning_individual to False and write HA state."""
        self.debug("transition complete - future attribute reports will write HA state")
        self._transitioning_individual = False
        self._async_unsub_transition_listener()
        self.async_write_ha_state()
        if isinstance(self, LightGroup):
            async_dispatcher_send(
                self.hass,
                SIGNAL_LIGHT_GROUP_TRANSITION_FINISHED,
                {"entity_ids": self._entity_ids},
            )
            if self._debounced_member_refresh is not None:
                self.debug("transition complete - refreshing group member states")
                assert self.platform.config_entry
                self.platform.config_entry.async_create_background_task(
                    self.hass,
                    self._debounced_member_refresh.async_call(),
                    "zha.light-refresh-debounced-member",
                )


@STRICT_MATCH(
    cluster_handler_names=CLUSTER_HANDLER_ON_OFF,
    aux_cluster_handlers={CLUSTER_HANDLER_COLOR, CLUSTER_HANDLER_LEVEL},
)
class Light(BaseLight, ZhaEntity):
    """Representation of a ZHA or ZLL light."""

    _attr_supported_color_modes: set[ColorMode]
    _attr_translation_key: str = "light"
    _REFRESH_INTERVAL = (45, 75)

    def __init__(
        self, unique_id, zha_device: ZHADevice, cluster_handlers, **kwargs
    ) -> None:
        """Initialize the ZHA light."""
        super().__init__(unique_id, zha_device, cluster_handlers, **kwargs)
        self._on_off_cluster_handler = self.cluster_handlers[CLUSTER_HANDLER_ON_OFF]
        self._attr_state = bool(self._on_off_cluster_handler.on_off)
        self._level_cluster_handler = self.cluster_handlers.get(CLUSTER_HANDLER_LEVEL)
        self._color_cluster_handler = self.cluster_handlers.get(CLUSTER_HANDLER_COLOR)
        self._identify_cluster_handler = zha_device.identify_ch
        if self._color_cluster_handler:
            self._attr_min_mireds: int = self._color_cluster_handler.min_mireds
            self._attr_max_mireds: int = self._color_cluster_handler.max_mireds
        self._cancel_refresh_handle: CALLBACK_TYPE | None = None
        effect_list = []

        self._zha_config_always_prefer_xy_color_mode = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_ALWAYS_PREFER_XY_COLOR_MODE,
            True,
        )

        self._attr_supported_color_modes = {ColorMode.ONOFF}
        if self._level_cluster_handler:
            self._attr_supported_color_modes.add(ColorMode.BRIGHTNESS)
            self._attr_supported_features |= light.LightEntityFeature.TRANSITION
            self._attr_brightness = self._level_cluster_handler.current_level

        if self._color_cluster_handler:
            if self._color_cluster_handler.color_temp_supported:
                self._attr_supported_color_modes.add(ColorMode.COLOR_TEMP)
                self._attr_color_temp = self._color_cluster_handler.color_temperature

            if self._color_cluster_handler.xy_supported and (
                self._zha_config_always_prefer_xy_color_mode
                or not self._color_cluster_handler.hs_supported
            ):
                self._attr_supported_color_modes.add(ColorMode.XY)
                curr_x = self._color_cluster_handler.current_x
                curr_y = self._color_cluster_handler.current_y
                if curr_x is not None and curr_y is not None:
                    self._attr_xy_color = (curr_x / 65535, curr_y / 65535)
                else:
                    self._attr_xy_color = (0, 0)

            if (
                self._color_cluster_handler.hs_supported
                and not self._zha_config_always_prefer_xy_color_mode
            ):
                self._attr_supported_color_modes.add(ColorMode.HS)
                if (
                    self._color_cluster_handler.enhanced_hue_supported
                    and self._color_cluster_handler.enhanced_current_hue is not None
                ):
                    curr_hue = (
                        self._color_cluster_handler.enhanced_current_hue * 65535 / 360
                    )
                elif self._color_cluster_handler.current_hue is not None:
                    curr_hue = self._color_cluster_handler.current_hue * 254 / 360
                else:
                    curr_hue = 0

                if (
                    curr_saturation := self._color_cluster_handler.current_saturation
                ) is None:
                    curr_saturation = 0

                self._attr_hs_color = (
                    int(curr_hue),
                    int(curr_saturation * 2.54),
                )

            if self._color_cluster_handler.color_loop_supported:
                self._attr_supported_features |= light.LightEntityFeature.EFFECT
                effect_list.append(light.EFFECT_COLORLOOP)
                if self._color_cluster_handler.color_loop_active == 1:
                    self._attr_effect = light.EFFECT_COLORLOOP
        self._attr_supported_color_modes = filter_supported_color_modes(
            self._attr_supported_color_modes
        )
        if len(self._attr_supported_color_modes) == 1:
            self._attr_color_mode = next(iter(self._attr_supported_color_modes))
        else:  # Light supports color_temp + hs, determine which mode the light is in
            assert self._color_cluster_handler
            if (
                self._color_cluster_handler.color_mode
                == Color.ColorMode.Color_temperature
            ):
                self._attr_color_mode = ColorMode.COLOR_TEMP
            else:
                self._attr_color_mode = ColorMode.XY

        if self._identify_cluster_handler:
            self._attr_supported_features |= light.LightEntityFeature.FLASH

        if effect_list:
            self._attr_effect_list = effect_list

        self._zha_config_transition = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_DEFAULT_LIGHT_TRANSITION,
            0,
        )
        self._zha_config_enhanced_light_transition = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_ENABLE_ENHANCED_LIGHT_TRANSITION,
            False,
        )
        self._zha_config_enable_light_transitioning_flag = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_ENABLE_LIGHT_TRANSITIONING_FLAG,
            True,
        )

    @callback
    def async_set_state(self, attr_id, attr_name, value):
        """Set the state."""
        if self.is_transitioning:
            self.debug(
                "received onoff %s while transitioning - skipping update",
                value,
            )
            return
        self._attr_state = bool(value)
        if value:
            self._off_with_transition = False
            self._off_brightness = None
        self.async_write_ha_state()

    async def async_added_to_hass(self) -> None:
        """Run when about to be added to hass."""
        await super().async_added_to_hass()
        self.async_accept_signal(
            self._on_off_cluster_handler, SIGNAL_ATTR_UPDATED, self.async_set_state
        )
        if self._level_cluster_handler:
            self.async_accept_signal(
                self._level_cluster_handler, SIGNAL_SET_LEVEL, self.set_level
            )
        refresh_interval = random.randint(*(x * 60 for x in self._REFRESH_INTERVAL))
        self._cancel_refresh_handle = async_track_time_interval(
            self.hass, self._refresh, timedelta(seconds=refresh_interval)
        )
        self.debug("started polling with refresh interval of %s", refresh_interval)
        self.async_accept_signal(
            None,
            SIGNAL_LIGHT_GROUP_STATE_CHANGED,
            self._maybe_force_refresh,
            signal_override=True,
        )

        @callback
        def transition_on(signal):
            """Handle a transition start event from a group."""
            if self.entity_id in signal["entity_ids"]:
                self.debug(
                    "group transition started - setting member transitioning flag"
                )
                self._transitioning_group = True

        self.async_accept_signal(
            None,
            SIGNAL_LIGHT_GROUP_TRANSITION_START,
            transition_on,
            signal_override=True,
        )

        @callback
        def transition_off(signal):
            """Handle a transition finished event from a group."""
            if self.entity_id in signal["entity_ids"]:
                self.debug(
                    "group transition completed - unsetting member transitioning flag"
                )
                self._transitioning_group = False

        self.async_accept_signal(
            None,
            SIGNAL_LIGHT_GROUP_TRANSITION_FINISHED,
            transition_off,
            signal_override=True,
        )

        self.async_accept_signal(
            None,
            SIGNAL_LIGHT_GROUP_ASSUME_GROUP_STATE,
            self._assume_group_state,
            signal_override=True,
        )

    async def async_will_remove_from_hass(self) -> None:
        """Disconnect entity object when removed."""
        assert self._cancel_refresh_handle
        self._cancel_refresh_handle()
        self._cancel_refresh_handle = None
        self.debug("stopped polling during device removal")
        await super().async_will_remove_from_hass()

    @callback
    def async_restore_last_state(self, last_state):
        """Restore previous state."""
        self._attr_state = last_state.state == STATE_ON
        if "brightness" in last_state.attributes:
            self._attr_brightness = last_state.attributes["brightness"]
        if "off_with_transition" in last_state.attributes:
            self._off_with_transition = last_state.attributes["off_with_transition"]
        if "off_brightness" in last_state.attributes:
            self._off_brightness = last_state.attributes["off_brightness"]
        if (color_mode := last_state.attributes.get("color_mode")) is not None:
            self._attr_color_mode = ColorMode(color_mode)
        if "color_temp" in last_state.attributes:
            self._attr_color_temp = last_state.attributes["color_temp"]
        if "xy_color" in last_state.attributes:
            self._attr_xy_color = last_state.attributes["xy_color"]
        if "hs_color" in last_state.attributes:
            self._attr_hs_color = last_state.attributes["hs_color"]
        if "effect" in last_state.attributes:
            self._attr_effect = last_state.attributes["effect"]

    async def async_get_state(self) -> None:
        """Attempt to retrieve the state from the light."""
        if not self._attr_available:
            return
        self.debug("polling current state")

        if self._on_off_cluster_handler:
            state = await self._on_off_cluster_handler.get_attribute_value(
                "on_off", from_cache=False
            )
            # check if transition started whilst waiting for polled state
            if self.is_transitioning:
                return

            if state is not None:
                self._attr_state = state
                if state:  # reset "off with transition" flag if the light is on
                    self._off_with_transition = False
                    self._off_brightness = None

        if self._level_cluster_handler:
            level = await self._level_cluster_handler.get_attribute_value(
                "current_level", from_cache=False
            )
            # check if transition started whilst waiting for polled state
            if self.is_transitioning:
                return
            if level is not None:
                self._attr_brightness = level

        if self._color_cluster_handler:
            attributes = [
                "color_mode",
                "current_x",
                "current_y",
            ]
            if (
                not self._zha_config_always_prefer_xy_color_mode
                and self._color_cluster_handler.enhanced_hue_supported
            ):
                attributes.append("enhanced_current_hue")
                attributes.append("current_saturation")
            if (
                self._color_cluster_handler.hs_supported
                and not self._color_cluster_handler.enhanced_hue_supported
                and not self._zha_config_always_prefer_xy_color_mode
            ):
                attributes.append("current_hue")
                attributes.append("current_saturation")
            if self._color_cluster_handler.color_temp_supported:
                attributes.append("color_temperature")
            if self._color_cluster_handler.color_loop_supported:
                attributes.append("color_loop_active")

            results = await self._color_cluster_handler.get_attributes(
                attributes, from_cache=False, only_cache=False
            )

            # although rare, a transition might have been started while we were waiting
            # for the polled attributes, so abort if we are transitioning,
            # as that state will not be accurate
            if self.is_transitioning:
                return

            if (color_mode := results.get("color_mode")) is not None:
                if color_mode == Color.ColorMode.Color_temperature:
                    self._attr_color_mode = ColorMode.COLOR_TEMP
                    color_temp = results.get("color_temperature")
                    if color_temp is not None and color_mode:
                        self._attr_color_temp = color_temp
                        self._attr_xy_color = None
                        self._attr_hs_color = None
                elif (
                    color_mode == Color.ColorMode.Hue_and_saturation
                    and not self._zha_config_always_prefer_xy_color_mode
                ):
                    self._attr_color_mode = ColorMode.HS
                    if self._color_cluster_handler.enhanced_hue_supported:
                        current_hue = results.get("enhanced_current_hue")
                    else:
                        current_hue = results.get("current_hue")
                    current_saturation = results.get("current_saturation")
                    if current_hue is not None and current_saturation is not None:
                        self._attr_hs_color = (
                            int(current_hue * 360 / 65535)
                            if self._color_cluster_handler.enhanced_hue_supported
                            else int(current_hue * 360 / 254),
                            int(current_saturation / 2.54),
                        )
                        self._attr_xy_color = None
                        self._attr_color_temp = None
                else:
                    self._attr_color_mode = ColorMode.XY
                    color_x = results.get("current_x")
                    color_y = results.get("current_y")
                    if color_x is not None and color_y is not None:
                        self._attr_xy_color = (color_x / 65535, color_y / 65535)
                        self._attr_color_temp = None
                        self._attr_hs_color = None

            color_loop_active = results.get("color_loop_active")
            if color_loop_active is not None:
                if color_loop_active == 1:
                    self._attr_effect = light.EFFECT_COLORLOOP
                else:
                    self._attr_effect = None

    async def async_update(self) -> None:
        """Update to the latest state."""
        if self.is_transitioning:
            self.debug("skipping async_update while transitioning")
            return
        await self.async_get_state()

    async def _refresh(self, time):
        """Call async_get_state at an interval."""
        if self.is_transitioning:
            self.debug("skipping _refresh while transitioning")
            return
        if self._zha_device.available and self.hass.data[DATA_ZHA].allow_polling:
            self.debug("polling for updated state")
            await self.async_get_state()
            self.async_write_ha_state()
        else:
            self.debug(
                "skipping polling for updated state, available: %s, allow polled requests: %s",
                self._zha_device.available,
                self.hass.data[DATA_ZHA].allow_polling,
            )

    async def _maybe_force_refresh(self, signal):
        """Force update the state if the signal contains the entity id for this entity."""
        if self.entity_id in signal["entity_ids"]:
            if self.is_transitioning:
                self.debug("skipping _maybe_force_refresh while transitioning")
                return
            if self._zha_device.available and self.hass.data[DATA_ZHA].allow_polling:
                self.debug("forcing polling for updated state")
                await self.async_get_state()
                self.async_write_ha_state()
            else:
                self.debug(
                    "skipping _maybe_force_refresh, available: %s, allow polled requests: %s",
                    self._zha_device.available,
                    self.hass.data[DATA_ZHA].allow_polling,
                )

    @callback
    def _assume_group_state(self, signal, update_params) -> None:
        """Handle an assume group state event from a group."""
        if self.entity_id in signal["entity_ids"] and self._attr_available:
            self.debug("member assuming group state with: %s", update_params)

            state = update_params["state"]
            brightness = update_params.get(light.ATTR_BRIGHTNESS)
            color_mode = update_params.get(light.ATTR_COLOR_MODE)
            color_temp = update_params.get(light.ATTR_COLOR_TEMP)
            xy_color = update_params.get(light.ATTR_XY_COLOR)
            hs_color = update_params.get(light.ATTR_HS_COLOR)
            effect = update_params.get(light.ATTR_EFFECT)

            supported_modes = self._attr_supported_color_modes

            # unset "off brightness" and "off with transition"
            # if group turned on this light
            if state and not self._attr_state:
                self._off_with_transition = False
                self._off_brightness = None

            # set "off brightness" and "off with transition"
            # if group turned off this light, and the light was not already off
            # (to not override _off_with_transition)
            elif (
                not state and self._attr_state and brightness_supported(supported_modes)
            ):
                # use individual brightness, instead of possibly averaged
                # brightness from group
                self._off_brightness = self._attr_brightness
                self._off_with_transition = update_params["off_with_transition"]

            # Note: If individual lights have off_with_transition set, but not the
            # group, and the group is then turned on without a level, individual lights
            # might fall back to brightness level 1.
            # Since all lights might need different brightness levels to be turned on,
            # we can't use one group call. And making individual calls when turning on
            # a ZHA group would cause a lot of traffic. In this case,
            # turn_on should either just be called with a level or individual turn_on
            # calls can be used.

            # state is always set (light.turn_on/light.turn_off)
            self._attr_state = state

            # before assuming a group state attribute, check if the attribute
            # was actually set in that call
            if brightness is not None and brightness_supported(supported_modes):
                self._attr_brightness = brightness
            if color_mode is not None and color_mode in supported_modes:
                self._attr_color_mode = color_mode
            if color_temp is not None and ColorMode.COLOR_TEMP in supported_modes:
                self._attr_color_temp = color_temp
            if xy_color is not None and ColorMode.XY in supported_modes:
                self._attr_xy_color = xy_color
            if hs_color is not None and ColorMode.HS in supported_modes:
                self._attr_hs_color = hs_color
            # the effect is always deactivated in async_turn_on if not provided
            if effect is None:
                self._attr_effect = None
            elif self._attr_effect_list and effect in self._attr_effect_list:
                self._attr_effect = effect

            self.async_write_ha_state()


@STRICT_MATCH(
    cluster_handler_names=CLUSTER_HANDLER_ON_OFF,
    aux_cluster_handlers={CLUSTER_HANDLER_COLOR, CLUSTER_HANDLER_LEVEL},
    manufacturers={"Philips", "Signify Netherlands B.V."},
)
class HueLight(Light):
    """Representation of a HUE light which does not report attributes."""

    _REFRESH_INTERVAL = (3, 5)


@STRICT_MATCH(
    cluster_handler_names=CLUSTER_HANDLER_ON_OFF,
    aux_cluster_handlers={CLUSTER_HANDLER_COLOR, CLUSTER_HANDLER_LEVEL},
    manufacturers={"Jasco", "Jasco Products", "Quotra-Vision", "eWeLight", "eWeLink"},
)
class ForceOnLight(Light):
    """Representation of a light which does not respect on/off for move_to_level_with_on_off commands."""

    _FORCE_ON = True


@STRICT_MATCH(
    cluster_handler_names=CLUSTER_HANDLER_ON_OFF,
    aux_cluster_handlers={CLUSTER_HANDLER_COLOR, CLUSTER_HANDLER_LEVEL},
    manufacturers=DEFAULT_MIN_TRANSITION_MANUFACTURERS,
)
class MinTransitionLight(Light):
    """Representation of a light which does not react to any "move to" calls with 0 as a transition."""

    # Transitions are counted in 1/10th of a second increments, so this is the smallest
    _DEFAULT_MIN_TRANSITION_TIME = 0.1


@GROUP_MATCH()
class LightGroup(BaseLight, ZhaGroupEntity):
    """Representation of a light group."""

    _attr_translation_key: str = "light_group"

    def __init__(
        self,
        entity_ids: list[str],
        unique_id: str,
        group_id: int,
        zha_device: ZHADevice,
        **kwargs: Any,
    ) -> None:
        """Initialize a light group."""
        super().__init__(entity_ids, unique_id, group_id, zha_device, **kwargs)
        group = self.zha_device.gateway.get_group(self._group_id)

        self._GROUP_SUPPORTS_EXECUTE_IF_OFF = True

        for member in group.members:
            # Ensure we do not send group commands that violate the minimum transition
            # time of any members.
            if member.device.manufacturer in DEFAULT_MIN_TRANSITION_MANUFACTURERS:
                self._DEFAULT_MIN_TRANSITION_TIME = (
                    MinTransitionLight._DEFAULT_MIN_TRANSITION_TIME
                )

            # Check all group members to see if they support execute_if_off.
            # If at least one member has a color cluster and doesn't support it,
            # it's not used.
            for endpoint in member.device._endpoints.values():
                for cluster_handler in endpoint.all_cluster_handlers.values():
                    if (
                        cluster_handler.name == CLUSTER_HANDLER_COLOR
                        and not cluster_handler.execute_if_off_supported
                    ):
                        self._GROUP_SUPPORTS_EXECUTE_IF_OFF = False
                        break

        self._on_off_cluster_handler = group.endpoint[OnOff.cluster_id]
        self._level_cluster_handler = group.endpoint[LevelControl.cluster_id]
        self._color_cluster_handler = group.endpoint[Color.cluster_id]
        self._identify_cluster_handler = group.endpoint[Identify.cluster_id]
        self._debounced_member_refresh: Debouncer | None = None
        self._zha_config_transition = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_DEFAULT_LIGHT_TRANSITION,
            0,
        )
        self._zha_config_enable_light_transitioning_flag = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_ENABLE_LIGHT_TRANSITIONING_FLAG,
            True,
        )
        self._zha_config_always_prefer_xy_color_mode = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_ALWAYS_PREFER_XY_COLOR_MODE,
            True,
        )
        self._zha_config_group_members_assume_state = async_get_zha_config_value(
            zha_device.gateway.config_entry,
            ZHA_OPTIONS,
            CONF_GROUP_MEMBERS_ASSUME_STATE,
            True,
        )
        if self._zha_config_group_members_assume_state:
            self._update_group_from_child_delay = ASSUME_UPDATE_GROUP_FROM_CHILD_DELAY
        self._zha_config_enhanced_light_transition = False
        self._attr_color_mode = None

    # remove this when all ZHA platforms and base entities are updated
    @property
    def available(self) -> bool:
        """Return entity availability."""
        return self._attr_available

    async def async_added_to_hass(self) -> None:
        """Run when about to be added to hass."""
        await super().async_added_to_hass()
        if self._debounced_member_refresh is None:
            force_refresh_debouncer = Debouncer(
                self.hass,
                _LOGGER,
                cooldown=3,
                immediate=True,
                function=self._force_member_updates,
            )
            self._debounced_member_refresh = force_refresh_debouncer
            self.async_on_remove(force_refresh_debouncer.async_cancel)

    async def async_turn_on(self, **kwargs: Any) -> None:
        """Turn the entity on."""
        # "off with transition" and "off brightness" will get overridden when
        # turning on the group, but they are needed for setting the assumed
        # member state correctly, so save them here
        off_brightness = self._off_brightness if self._off_with_transition else None
        await super().async_turn_on(**kwargs)
        if self._zha_config_group_members_assume_state:
            self._send_member_assume_state_event(True, kwargs, off_brightness)
        if self.is_transitioning:  # when transitioning, state is refreshed at the end
            return
        if self._debounced_member_refresh:
            await self._debounced_member_refresh.async_call()

    async def async_turn_off(self, **kwargs: Any) -> None:
        """Turn the entity off."""
        await super().async_turn_off(**kwargs)
        if self._zha_config_group_members_assume_state:
            self._send_member_assume_state_event(False, kwargs)
        if self.is_transitioning:
            return
        if self._debounced_member_refresh:
            await self._debounced_member_refresh.async_call()

    async def async_update(self) -> None:
        """Query all members and determine the light group state."""
        self.debug("updating group state")
        all_states = [self.hass.states.get(x) for x in self._entity_ids]
        states: list[State] = list(filter(None, all_states))
        on_states = [state for state in states if state.state == STATE_ON]

        self._attr_state = len(on_states) > 0

        # reset "off with transition" flag if any member is on
        if self._attr_state:
            self._off_with_transition = False
            self._off_brightness = None

        self._attr_available = any(state.state != STATE_UNAVAILABLE for state in states)

        self._attr_brightness = helpers.reduce_attribute(
            on_states, light.ATTR_BRIGHTNESS
        )

        self._attr_xy_color = helpers.reduce_attribute(
            on_states, light.ATTR_XY_COLOR, reduce=helpers.mean_tuple
        )

        if not self._zha_config_always_prefer_xy_color_mode:
            self._attr_hs_color = helpers.reduce_attribute(
                on_states, light.ATTR_HS_COLOR, reduce=helpers.mean_tuple
            )

        self._attr_color_temp = helpers.reduce_attribute(
            on_states, light.ATTR_COLOR_TEMP
        )
        self._attr_min_mireds = helpers.reduce_attribute(
            states, light.ATTR_MIN_MIREDS, default=153, reduce=min
        )
        self._attr_max_mireds = helpers.reduce_attribute(
            states, light.ATTR_MAX_MIREDS, default=500, reduce=max
        )

        self._attr_effect_list = None
        all_effect_lists = list(
            helpers.find_state_attributes(states, light.ATTR_EFFECT_LIST)
        )
        if all_effect_lists:
            # Merge all effects from all effect_lists with a union merge.
            self._attr_effect_list = list(set().union(*all_effect_lists))

        self._attr_effect = None
        all_effects = list(helpers.find_state_attributes(on_states, light.ATTR_EFFECT))
        if all_effects:
            # Report the most common effect.
            effects_count = Counter(itertools.chain(all_effects))
            self._attr_effect = effects_count.most_common(1)[0][0]

        self._attr_color_mode = None
        all_color_modes = list(
            helpers.find_state_attributes(on_states, light.ATTR_COLOR_MODE)
        )
        if all_color_modes:
            # Report the most common color mode, select brightness and onoff last
            color_mode_count = Counter(itertools.chain(all_color_modes))
            if ColorMode.ONOFF in color_mode_count:
                color_mode_count[ColorMode.ONOFF] = -1
            if ColorMode.BRIGHTNESS in color_mode_count:
                color_mode_count[ColorMode.BRIGHTNESS] = 0
            self._attr_color_mode = color_mode_count.most_common(1)[0][0]
            if self._attr_color_mode == ColorMode.HS and (
                color_mode_count[ColorMode.HS] != len(self._group.members)
                or self._zha_config_always_prefer_xy_color_mode
            ):  # switch to XY if all members do not support HS
                self._attr_color_mode = ColorMode.XY

        self._attr_supported_color_modes = None
        all_supported_color_modes = list(
            helpers.find_state_attributes(states, light.ATTR_SUPPORTED_COLOR_MODES)
        )
        if all_supported_color_modes:
            # Merge all color modes.
            self._attr_supported_color_modes = cast(
                set[str], set().union(*all_supported_color_modes)
            )

        self._attr_supported_features = LightEntityFeature(0)
        for support in helpers.find_state_attributes(states, ATTR_SUPPORTED_FEATURES):
            # Merge supported features by emulating support for every feature
            # we find.
            self._attr_supported_features |= support
        # Bitwise-and the supported features with the GroupedLight's features
        # so that we don't break in the future when a new feature is added.
        self._attr_supported_features &= SUPPORT_GROUP_LIGHT

    async def _force_member_updates(self) -> None:
        """Force the update of member entities to ensure the states are correct for bulbs that don't report their state."""
        async_dispatcher_send(
            self.hass,
            SIGNAL_LIGHT_GROUP_STATE_CHANGED,
            {"entity_ids": self._entity_ids},
        )

    def _send_member_assume_state_event(
        self, state, service_kwargs, off_brightness=None
    ) -> None:
        """Send an assume event to all members of the group."""
        update_params = {
            "state": state,
            "off_with_transition": self._off_with_transition,
        }

        # check if the parameters were actually updated
        # in the service call before updating members
        if light.ATTR_BRIGHTNESS in service_kwargs:  # or off brightness
            update_params[light.ATTR_BRIGHTNESS] = self._attr_brightness
        elif off_brightness is not None:
            # if we turn on the group light with "off brightness",
            # pass that to the members
            update_params[light.ATTR_BRIGHTNESS] = off_brightness

        if light.ATTR_COLOR_TEMP in service_kwargs:
            update_params[light.ATTR_COLOR_MODE] = self._attr_color_mode
            update_params[light.ATTR_COLOR_TEMP] = self._attr_color_temp

        if light.ATTR_XY_COLOR in service_kwargs:
            update_params[light.ATTR_COLOR_MODE] = self._attr_color_mode
            update_params[light.ATTR_XY_COLOR] = self._attr_xy_color

        if light.ATTR_HS_COLOR in service_kwargs:
            update_params[light.ATTR_COLOR_MODE] = self._attr_color_mode
            update_params[light.ATTR_HS_COLOR] = self._attr_hs_color

        if light.ATTR_EFFECT in service_kwargs:
            update_params[light.ATTR_EFFECT] = self._attr_effect

        async_dispatcher_send(
            self.hass,
            SIGNAL_LIGHT_GROUP_ASSUME_GROUP_STATE,
            {"entity_ids": self._entity_ids},
            update_params,
        )