Merge branch 'master' into dishwasher-footering

custom_components/adaptive_lighting/__init__.py

@@ -2,15 +2,15 @@
 import logging
 from typing import Any
 
+import homeassistant.helpers.config_validation as cv
+import voluptuous as vol
 from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
 from homeassistant.const import CONF_SOURCE
 from homeassistant.core import HomeAssistant
-import homeassistant.helpers.config_validation as cv
-import voluptuous as vol
 
 from .const import (
     _DOMAIN_SCHEMA,
-    ATTR_TURN_ON_OFF_LISTENER,
+    ATTR_ADAPTIVE_LIGHTING_MANAGER,
     CONF_NAME,
     DOMAIN,
     UNDO_UPDATE_LISTENER,
@@ -35,20 +35,21 @@ def _all_unique_names(value):
 )
 
 
-async def reload_configuration_yaml(event: dict, hass: HomeAssistant):
+async def reload_configuration_yaml(event: dict, hass: HomeAssistant):  # noqa: ARG001
     """Reload configuration.yaml."""
     await hass.services.async_call("homeassistant", "check_config", {})
 
 
 async def async_setup(hass: HomeAssistant, config: dict[str, Any]):
     """Import integration from config."""
-
     if DOMAIN in config:
         for entry in config[DOMAIN]:
             hass.async_create_task(
                 hass.config_entries.flow.async_init(
-                    DOMAIN, context={CONF_SOURCE: SOURCE_IMPORT}, data=entry
-                )
+                    DOMAIN,
+                    context={CONF_SOURCE: SOURCE_IMPORT},
+                    data=entry,
+                ),
             )
     return True
 
@@ -65,7 +66,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
     data[config_entry.entry_id] = {UNDO_UPDATE_LISTENER: undo_listener}
     for platform in PLATFORMS:
         hass.async_create_task(
-            hass.config_entries.async_forward_entry_setup(config_entry, platform)
+            hass.config_entries.async_forward_entry_setup(config_entry, platform),
         )
 
     return True
@@ -79,17 +80,18 @@ async def async_update_options(hass, config_entry: ConfigEntry):
 async def async_unload_entry(hass, config_entry: ConfigEntry) -> bool:
     """Unload a config entry."""
     unload_ok = await hass.config_entries.async_forward_entry_unload(
-        config_entry, "switch"
+        config_entry,
+        "switch",
     )
     data = hass.data[DOMAIN]
     data[config_entry.entry_id][UNDO_UPDATE_LISTENER]()
     if unload_ok:
         data.pop(config_entry.entry_id)
 
-    if len(data) == 1 and ATTR_TURN_ON_OFF_LISTENER in data:
+    if len(data) == 1 and ATTR_ADAPTIVE_LIGHTING_MANAGER in data:
         # no more config_entries
-        turn_on_off_listener = data.pop(ATTR_TURN_ON_OFF_LISTENER)
-        turn_on_off_listener.disable()
+        manager = data.pop(ATTR_ADAPTIVE_LIGHTING_MANAGER)
+        manager.disable()
 
     if not data:
         hass.data.pop(DOMAIN)

custom_components/adaptive_lighting/_docs_helpers.py

@@ -1,9 +1,9 @@
 from typing import Any
 
-from homeassistant.helpers import selector
 import homeassistant.helpers.config_validation as cv
 import pandas as pd
 import voluptuous as vol
+from homeassistant.helpers import selector
 
 from .const import (
     DOCS,
@@ -23,38 +23,37 @@ def _format_voluptuous_instance(instance):
     for validator in instance.validators:
         if isinstance(validator, vol.Coerce):
             coerce_type = validator.type.__name__
-        elif isinstance(validator, (vol.Clamp, vol.Range)):
+        elif isinstance(validator, vol.Clamp | vol.Range):
             min_val = validator.min
             max_val = validator.max
 
     if min_val is not None and max_val is not None:
         return f"`{coerce_type}` {min_val}-{max_val}"
-    elif min_val is not None:
+    if min_val is not None:
         return f"`{coerce_type} > {min_val}`"
-    elif max_val is not None:
+    if max_val is not None:
         return f"`{coerce_type} < {max_val}`"
-    else:
-        return f"`{coerce_type}`"
+    return f"`{coerce_type}`"
 
 
-def _type_to_str(type_: Any) -> str:
+def _type_to_str(type_: Any) -> str:  # noqa: PLR0911
     """Convert a (voluptuous) type to a string."""
     if type_ == cv.entity_ids:
         return "list of `entity_id`s"
-    elif type_ in (bool, int, float, str):
+    if type_ in (bool, int, float, str):
         return f"`{type_.__name__}`"
-    elif type_ == cv.boolean:
+    if type_ == cv.boolean:
         return "bool"
-    elif isinstance(type_, vol.All):
+    if isinstance(type_, vol.All):
         return _format_voluptuous_instance(type_)
-    elif isinstance(type_, vol.In):
+    if isinstance(type_, vol.In):
         return f"one of `{type_.container}`"
-    elif isinstance(type_, selector.SelectSelector):
+    if isinstance(type_, selector.SelectSelector):
         return f"one of `{type_.config['options']}`"
-    elif isinstance(type_, selector.ColorRGBSelector):
+    if isinstance(type_, selector.ColorRGBSelector):
         return "RGB color"
-    else:
-        raise ValueError(f"Unknown type: {type_}")
+    msg = f"Unknown type: {type_}"
+    raise ValueError(msg)
 
 
 def generate_config_markdown_table():
@@ -85,7 +84,8 @@ def _schema_to_dict(schema: vol.Schema) -> dict[str, tuple[Any, Any]]:
 
 
 def _generate_service_markdown_table(
-    schema: dict[str, tuple[Any, Any]], alternative_docs: dict[str, str] = None
+    schema: dict[str, tuple[Any, Any]],
+    alternative_docs: dict[str, str] | None = None,
 ):
     schema = _schema_to_dict(schema)
     rows = []
@@ -112,5 +112,6 @@ def generate_apply_markdown_table():
 
 def generate_set_manual_control_markdown_table():
     return _generate_service_markdown_table(
-        SET_MANUAL_CONTROL_SCHEMA, DOCS_MANUAL_CONTROL
+        SET_MANUAL_CONTROL_SCHEMA,
+        DOCS_MANUAL_CONTROL,
     )

custom_components/adaptive_lighting/adaptation_utils.py

@@ -1,7 +1,7 @@
 """Utility functions for adaptation commands."""
+import logging
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
-import logging
 from typing import Any, Literal
 
 from homeassistant.components.light import (
@@ -13,6 +13,8 @@
     ATTR_COLOR_TEMP_KELVIN,
     ATTR_HS_COLOR,
     ATTR_RGB_COLOR,
+    ATTR_RGBW_COLOR,
+    ATTR_RGBWW_COLOR,
     ATTR_TRANSITION,
     ATTR_XY_COLOR,
 )
@@ -27,8 +29,11 @@
     ATTR_HS_COLOR,
     ATTR_RGB_COLOR,
     ATTR_XY_COLOR,
+    ATTR_RGBW_COLOR,
+    ATTR_RGBWW_COLOR,
 }
 
+
 BRIGHTNESS_ATTRS = {
     ATTR_BRIGHTNESS,
     ATTR_BRIGHTNESS_PCT,
@@ -40,10 +45,10 @@
 
 
 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.
-    """
+    """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}
 
@@ -61,41 +66,39 @@ def _split_service_call_data(service_data: ServiceData) -> list[ServiceData]:
 
     # Distribute the transition duration across all service calls
     if service_datas and (transition := service_data.get(ATTR_TRANSITION)) is not None:
-        transition = service_data[ATTR_TRANSITION] / len(service_datas)
+        transition /= len(service_datas)
 
         for service_data in service_datas:
             service_data[ATTR_TRANSITION] = transition
 
     return service_datas
 
 
-def _filter_service_data(service_data: ServiceData, state: State | None) -> ServiceData:
+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."""
-
-    if not state:
-        return service_data
-
-    filtered_service_data = {
-        k: service_data[k]
-        for k in service_data.keys()
-        if k not in state.attributes or service_data[k] != state.attributes[k]
+    in the target entity's state.
+    """
+    return {
+        k: v
+        for k, v in service_data.items()
+        if k not in state.attributes or v != state.attributes[k]
     }
 
-    return filtered_service_data
-
 
 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."""
+    change relevant attributes of an adapting entity, e.g., brightness or color.
+    """
     common_attrs = {ATTR_ENTITY_ID, ATTR_TRANSITION}
-    relevant_attrs = set(service_data) - common_attrs
 
-    return bool(relevant_attrs)
+    return any(attr not in common_attrs for attr in service_data)
 
 
 async def _create_service_call_data_iterator(
@@ -112,14 +115,16 @@ async def _create_service_call_data_iterator(
     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:
-                service_data = _filter_service_data(service_data, current_entity_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):
@@ -136,6 +141,7 @@ class AdaptationData:
     context: Context
     sleep_time: float
     service_call_datas: AsyncGenerator[ServiceData, None]
+    force: bool
     max_length: int
     which: Literal["brightness", "color", "both"]
     initial_sleep: bool = False
@@ -144,12 +150,26 @@ 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)
 
-
-class NoColorOrBrightnessInServiceData(Exception):
+    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"which={self.which}, "
+            f"initial_sleep={self.initial_sleep}"
+            ")"
+        )
+
+
+class NoColorOrBrightnessInServiceDataError(Exception):
     """Exception raised when no color or brightness attributes are found in service data."""
 
 
-def is_color_brightness_or_both(
+def _identify_lighting_type(
     service_data: ServiceData,
 ) -> Literal["brightness", "color", "both"]:
     """Extract the 'which' attribute from the service data."""
@@ -162,7 +182,7 @@ def is_color_brightness_or_both(
     if has_color:
         return "color"
     msg = f"Invalid service_data, no brightness or color attributes found: {service_data=}"
-    raise NoColorOrBrightnessInServiceData(msg)
+    raise NoColorOrBrightnessInServiceDataError(msg)
 
 
 def prepare_adaptation_data(
@@ -174,30 +194,38 @@ def prepare_adaptation_data(
     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 = (
-        [service_data] if not split else _split_service_call_data(service_data)
-    )
+    service_datas = _split_service_call_data(service_data) if split else [service_data]
+
+    service_datas_length = len(service_datas)
 
-    sleep_time = (
-        transition / max(1, len(service_datas)) if transition is not None else 0
-    ) + split_delay
+    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
+        hass,
+        service_datas,
+        filter_by_state,
     )
 
+    lighting_type = _identify_lighting_type(service_data)
+
     return AdaptationData(
-        entity_id,
-        context,
+        entity_id=entity_id,
+        context=context,
         sleep_time=sleep_time,
         service_call_datas=service_data_iterator,
-        max_length=len(service_datas),
-        which=is_color_brightness_or_both(service_data),
+        force=force,
+        max_length=service_datas_length,
+        which=lighting_type,
     )