adaptive-lighting/custom_components/adaptive_lighting/adaptation_utils.py

"""Utility functions for adaptation commands."""

import logging
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from enum import IntFlag, auto
from typing import Any

from homeassistant.components.light import (
    ATTR_BRIGHTNESS,
    ATTR_BRIGHTNESS_PCT,
    ATTR_BRIGHTNESS_STEP,
    ATTR_BRIGHTNESS_STEP_PCT,
    ATTR_COLOR_NAME,
    ATTR_COLOR_TEMP_KELVIN,
    ATTR_EFFECT,
    ATTR_FLASH,
    ATTR_HS_COLOR,
    ATTR_RGB_COLOR,
    ATTR_RGBW_COLOR,
    ATTR_RGBWW_COLOR,
    ATTR_TRANSITION,
    ATTR_XY_COLOR,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import Context, HomeAssistant, State

_LOGGER = logging.getLogger(__name__)

COLOR_ATTRS = {  # Should ATTR_PROFILE be in here?
    ATTR_COLOR_NAME,
    ATTR_COLOR_TEMP_KELVIN,
    ATTR_HS_COLOR,
    ATTR_RGB_COLOR,
    ATTR_XY_COLOR,
    ATTR_RGBW_COLOR,
    ATTR_RGBWW_COLOR,
}


BRIGHTNESS_ATTRS = {
    ATTR_BRIGHTNESS,
    ATTR_BRIGHTNESS_PCT,
    ATTR_BRIGHTNESS_STEP,
    ATTR_BRIGHTNESS_STEP_PCT,
}

ServiceData = dict[str, Any]


class LightControlAttributes(IntFlag):
    """Attributes of lights that the adaptation engine can control."""

    NONE = 0
    BRIGHTNESS = auto()
    COLOR = auto()

    ALL = BRIGHTNESS | COLOR

    def __str__(self) -> str:
        """Return a string representation of the attributes."""
        if self == LightControlAttributes.NONE:
            return "NONE"

        return "|".join(
            member.name
            for member in type(self)
            if member is not LightControlAttributes.NONE
            and member in self
            and member.name is not None
        )

    def has_any(self) -> bool:
        """Determine whether any attribute is selected."""
        return self != LightControlAttributes.NONE

    def has_none(self) -> bool:
        """Determine whether no attribute is selected."""
        return self == LightControlAttributes.NONE

    def has_all(self) -> bool:
        """Determine whether all attributes are selected."""
        return (self & LightControlAttributes.ALL) == LightControlAttributes.ALL


def _split_service_call_data(service_data: ServiceData) -> list[ServiceData]:
    """Splits the service data by the adapted attributes.

    i.e., into separate data items for brightness and color.
    """
    common_attrs = {ATTR_ENTITY_ID}
    common_data = {k: service_data[k] for k in common_attrs if k in service_data}

    attributes_split_sequence = [BRIGHTNESS_ATTRS, COLOR_ATTRS]
    service_datas: list[dict[str, Any]] = []

    for attributes in attributes_split_sequence:
        split_data = {
            attribute: service_data[attribute]
            for attribute in attributes
            if service_data.get(attribute)
        }
        if split_data:
            service_datas.append(common_data | split_data)

    # Distribute the transition duration across all service calls
    if service_datas and (transition := service_data.get(ATTR_TRANSITION)) is not None:
        transition /= len(service_datas)

        for _service_data in service_datas:
            _service_data[ATTR_TRANSITION] = transition

    return service_datas


def _remove_redundant_attributes(
    service_data: ServiceData,
    state: State,
) -> ServiceData:
    """Filter service data by removing attributes that already equal the given state.

    Removes all attributes from service call data whose values are already present
    in the target entity's state.
    """
    attributes: dict[str, Any] = dict(state.attributes)
    return {
        k: v
        for k, v in service_data.items()
        if k not in attributes or v != attributes[k]
    }


def _has_relevant_service_data_attributes(service_data: ServiceData) -> bool:
    """Determines whether the service data justifies an adaptation service call.

    A service call is not justified for data which does not contain any entries that
    change relevant attributes of an adapting entity, e.g., brightness or color.
    """
    common_attrs = {ATTR_ENTITY_ID, ATTR_TRANSITION}

    return any(attr not in common_attrs for attr in service_data)


async def _create_service_call_data_iterator(
    hass: HomeAssistant,
    service_datas: list[ServiceData],
    filter_by_state: bool,
) -> AsyncGenerator[ServiceData]:
    """Enumerates and filters a list of service datas on the fly.

    If filtering is enabled, every service data is filtered by the current state of
    the related entity and only returned if it contains relevant data that justifies
    a service call.
    The main advantage of this generator over a list is that it applies the filter
    at the time when the service data is read instead of up front. This gives greater
    flexibility because entity states can change while the items are iterated.
    """
    for service_data in service_datas:
        if filter_by_state and (entity_id := service_data.get(ATTR_ENTITY_ID)):
            current_entity_state = hass.states.get(entity_id)

            # Filter data to remove attributes that equal the current state
            if current_entity_state is not None:
                service_data = _remove_redundant_attributes(  # noqa: PLW2901
                    service_data,
                    state=current_entity_state,
                )

            # Emit service data if it still contains relevant attributes (else try next)
            if _has_relevant_service_data_attributes(service_data):
                yield service_data
        else:
            yield service_data


@dataclass
class AdaptationData:
    """Holds all data required to execute an adaptation."""

    entity_id: str
    context: Context
    sleep_time: float
    service_call_datas: AsyncGenerator[ServiceData]
    force: bool
    max_length: int
    attributes: LightControlAttributes
    initial_sleep: bool = False

    async def next_service_call_data(self) -> ServiceData | None:
        """Return data for the next service call, or none if no more data exists."""
        return await anext(self.service_call_datas, None)

    def __str__(self) -> str:
        """Return a string representation of the data."""
        return (
            f"{self.__class__.__name__}("
            f"entity_id={self.entity_id}, "
            f"context_id={self.context.id}, "
            f"sleep_time={self.sleep_time}, "
            f"force={self.force}, "
            f"max_length={self.max_length}, "
            f"attributes={self.attributes}, "
            f"initial_sleep={self.initial_sleep}"
            ")"
        )


class NoColorOrBrightnessInServiceDataError(Exception):
    """Exception raised when no color or brightness attributes are found in service data."""


def _identify_light_control_attributes(
    service_data: ServiceData,
) -> LightControlAttributes:
    """Extract the 'which' attribute from the service data."""
    has_brightness = ATTR_BRIGHTNESS in service_data
    has_color = any(attr in service_data for attr in COLOR_ATTRS)

    parameters = LightControlAttributes.NONE

    if has_brightness:
        parameters |= LightControlAttributes.BRIGHTNESS
    if has_color:
        parameters |= LightControlAttributes.COLOR

    if parameters == LightControlAttributes.NONE:
        msg = f"Invalid service_data, no brightness or color attributes found: {service_data=}"
        raise NoColorOrBrightnessInServiceDataError(msg)

    return parameters


def prepare_adaptation_data(
    hass: HomeAssistant,
    entity_id: str,
    context: Context,
    transition: float | None,
    split_delay: float,
    service_data: ServiceData,
    split: bool,
    filter_by_state: bool,
    force: bool,
) -> AdaptationData:
    """Prepares a data object carrying all data required to execute an adaptation."""
    _LOGGER.debug(
        "Preparing adaptation data for %s with service data %s",
        entity_id,
        service_data,
    )
    service_datas = _split_service_call_data(service_data) if split else [service_data]

    service_datas_length = len(service_datas)

    if transition is not None:
        transition_duration_per_data = transition / max(1, service_datas_length)
        sleep_time = transition_duration_per_data + split_delay
    else:
        sleep_time = split_delay

    service_data_iterator = _create_service_call_data_iterator(
        hass,
        service_datas,
        filter_by_state,
    )

    attributes = _identify_light_control_attributes(service_data)

    return AdaptationData(
        entity_id=entity_id,
        context=context,
        sleep_time=sleep_time,
        service_call_datas=service_data_iterator,
        force=force,
        max_length=service_datas_length,
        attributes=attributes,
    )


def manual_control_event_attribute_to_flags(
    manual_control_attribute: bool | str,
) -> LightControlAttributes:
    """Convert manual control event data to light control attributes."""
    if isinstance(manual_control_attribute, bool) and manual_control_attribute:
        return LightControlAttributes.ALL
    if manual_control_attribute == "brightness":
        return LightControlAttributes.BRIGHTNESS
    if manual_control_attribute == "color":
        return LightControlAttributes.COLOR
    return LightControlAttributes.NONE


def has_brightness_attribute(
    service_data: ServiceData,
) -> bool:
    """Determine whether the service data contains brightness attributes."""
    return any(attr in BRIGHTNESS_ATTRS for attr in service_data)


def has_color_attribute(
    service_data: ServiceData,
) -> bool:
    """Determine whether the service data contains color attributes."""
    return any(attr in COLOR_ATTRS for attr in service_data)


def has_effect_attribute(
    service_data: ServiceData,
) -> bool:
    """Determine whether the service data contains effect attributes."""
    return ATTR_FLASH in service_data or ATTR_EFFECT in service_data


def get_light_control_attributes(
    service_data: ServiceData,
) -> LightControlAttributes:
    """Get the light control attributes affected by the service call data."""
    parameters = LightControlAttributes.NONE

    if has_brightness_attribute(service_data):
        parameters |= LightControlAttributes.BRIGHTNESS

    if has_color_attribute(service_data):
        parameters |= LightControlAttributes.COLOR

    if has_effect_attribute(service_data):
        parameters |= LightControlAttributes.BRIGHTNESS
        parameters |= LightControlAttributes.COLOR

    return parameters