Individual manual control of brightness and color (#1356)

* refactor: introduce light control parameter enum

* refactor: replace manual control flag with parameter enum

* test: update deprecated color temp attribute

* build: set execution bits on task scripts

* feat: individual manual control of brightness and color

* test: add tests for individual manual control evaluation

* fix: sequential manual changes not always detected

If multiple attributes of a light were changed within an interval, only the last change was detected because the check in the interval only used the latest event. For example, if there was a brightness change and a following color change, only the color attribute was detected as manually controlled. To fix this, the manual control attribute flags are now set directly from the event handler so that all events are processed.

* fix: invalid service description

* docs: fix missing space in config description

* refactor: pluralize multivalued bitmask enum name
This commit is contained in:
Mario Guggenberger 2025-12-23 07:55:16 +01:00 committed by GitHub
commit f84ee445b7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 846 additions and 293 deletions

View file

@ -8,7 +8,7 @@ import logging
import zoneinfo
from copy import deepcopy
from datetime import timedelta
from typing import TYPE_CHECKING, Any, Literal
from typing import TYPE_CHECKING, Any
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
@ -17,8 +17,6 @@ import voluptuous as vol
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP_KELVIN,
ATTR_EFFECT,
ATTR_FLASH,
ATTR_RGB_COLOR,
ATTR_SUPPORTED_COLOR_MODES,
ATTR_TRANSITION,
@ -76,10 +74,12 @@ from homeassistant.util.color import (
)
from .adaptation_utils import (
BRIGHTNESS_ATTRS,
COLOR_ATTRS,
AdaptationData,
LightControlAttributes,
ServiceData,
get_light_control_attributes,
has_effect_attribute,
manual_control_event_attribute_to_flags,
prepare_adaptation_data,
)
from .color_and_brightness import SunLightSettings
@ -127,6 +127,7 @@ from .const import (
CONF_SUNSET_OFFSET,
CONF_SUNSET_TIME,
CONF_TAKE_OVER_CONTROL,
CONF_TAKE_OVER_CONTROL_MODE,
CONF_TRANSITION,
CONF_TURN_ON_LIGHTS,
CONF_USE_DEFAULTS,
@ -143,6 +144,7 @@ from .const import (
SLEEP_MODE_SWITCH,
TURNING_OFF_DELAY,
VALIDATION_TUPLES,
TakeOverControlMode,
apply_service_schema,
replace_none_str,
)
@ -359,27 +361,6 @@ async def handle_change_switch_settings(
)
@callback
def _fire_manual_control_event(
switch: AdaptiveSwitch,
light: str,
context: Context,
) -> None:
"""Fire an event that 'light' is marked as manual_control."""
hass = switch.hass
_LOGGER.debug(
"'adaptive_lighting.manual_control' event fired for %s for light %s",
switch.entity_id,
light,
)
switch.manager.mark_as_manual_control(light)
hass.bus.async_fire(
f"{DOMAIN}.manual_control",
{ATTR_ENTITY_ID: light, SWITCH_DOMAIN: switch.entity_id},
context=context,
)
async def async_setup_entry( # noqa: PLR0915
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -497,9 +478,21 @@ async def async_setup_entry( # noqa: PLR0915
all_lights = switch.lights
else:
all_lights = _expand_light_groups(hass, lights)
if service_call.data[CONF_MANUAL_CONTROL]:
manual_attributes = manual_control_event_attribute_to_flags(
service_call.data[CONF_MANUAL_CONTROL],
)
if manual_attributes:
for light in all_lights:
_fire_manual_control_event(switch, light, service_call.context)
switch.manager.set_manual_control_attributes(
light,
manual_attributes,
)
switch.fire_manual_control_event(
light,
service_call.context,
)
else:
switch.manager.reset(*all_lights)
if switch.is_on:
@ -753,36 +746,32 @@ def _attributes_have_changed(
light: str,
old_attributes: dict[str, Any],
new_attributes: dict[str, Any],
adapt_brightness: bool,
adapt_color: bool,
context: Context,
) -> bool:
) -> LightControlAttributes:
# 2023-11-19: HA core no longer removes light domain attributes when off
# so we must protect for `None` here
# see https://github.com/home-assistant/core/pull/101946
changed_attributes = LightControlAttributes.NONE
# Check for color mode changes BEFORE attribute conversion
# This detects external changes like Hue scenes switching from color_temp to RGB
# See: https://github.com/basnijholt/adaptive-lighting/issues/1275
if adapt_color and _has_color_mode_changed(
if _has_color_mode_changed(
light,
old_attributes,
new_attributes,
context,
):
return True
changed_attributes |= LightControlAttributes.COLOR
if adapt_color:
if LightControlAttributes.COLOR not in changed_attributes:
old_attributes, new_attributes = _add_missing_attributes(
old_attributes,
new_attributes,
)
if (
adapt_brightness
and old_attributes.get(ATTR_BRIGHTNESS)
and new_attributes.get(ATTR_BRIGHTNESS)
):
if old_attributes.get(ATTR_BRIGHTNESS) and new_attributes.get(ATTR_BRIGHTNESS):
last_brightness = old_attributes[ATTR_BRIGHTNESS]
current_brightness = new_attributes[ATTR_BRIGHTNESS]
if abs(current_brightness - last_brightness) > BRIGHTNESS_CHANGE:
@ -794,10 +783,10 @@ def _attributes_have_changed(
current_brightness,
context.id,
)
return True
changed_attributes |= LightControlAttributes.BRIGHTNESS
if (
adapt_color
LightControlAttributes.COLOR not in changed_attributes
and old_attributes.get(ATTR_COLOR_TEMP_KELVIN)
and new_attributes.get(ATTR_COLOR_TEMP_KELVIN)
):
@ -812,10 +801,10 @@ def _attributes_have_changed(
current_color_temp,
context.id,
)
return True
changed_attributes |= LightControlAttributes.COLOR
if (
adapt_color
LightControlAttributes.COLOR not in changed_attributes
and old_attributes.get(ATTR_RGB_COLOR)
and new_attributes.get(ATTR_RGB_COLOR)
):
@ -831,9 +820,9 @@ def _attributes_have_changed(
current_rgb_color,
context.id,
)
return True
changed_attributes |= LightControlAttributes.COLOR
return False
return changed_attributes
class AdaptiveSwitch(SwitchEntity, RestoreEntity):
@ -937,6 +926,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._name,
)
self._take_over_control = True
self._take_over_control_mode = TakeOverControlMode(
data[CONF_TAKE_OVER_CONTROL_MODE],
)
self._detect_non_ha_changes = data[CONF_DETECT_NON_HA_CHANGES]
self._adapt_only_on_bare_turn_on = data[CONF_ADAPT_ONLY_ON_BARE_TURN_ON]
self._auto_reset_manual_control_time = data[CONF_AUTORESET_CONTROL]
@ -1203,12 +1195,19 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
context: Context | None = None,
) -> AdaptationData | None:
"""Prepare `AdaptationData` for adapting a light."""
adaptation_attributes = self.manager.get_adaption_control_attributes(
self,
light,
)
if transition is None:
transition = self._transition
if adapt_brightness is None:
adapt_brightness = self.adapt_brightness_switch.is_on
adapt_brightness = (
LightControlAttributes.BRIGHTNESS in adaptation_attributes
)
if adapt_color is None:
adapt_color = self.adapt_color_switch.is_on
adapt_color = LightControlAttributes.COLOR in adaptation_attributes
if prefer_rgb_color is None:
prefer_rgb_color = self._prefer_rgb_color
@ -1375,7 +1374,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
to cancel an ongoing adaptation when a light is turned off.
"""
# Prevent overlap of multiple adaptation sequences
self.manager.cancel_ongoing_adaptation_calls(data.entity_id, which=data.which)
self.manager.cancel_ongoing_adaptation_calls(data.entity_id)
_LOGGER.debug(
"%s: execute_cancellable_adaptation_calls with data: %s",
self._name,
@ -1384,9 +1383,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
# Execute adaptation calls within a task
try:
task = asyncio.ensure_future(self._execute_adaptation_calls(data))
if data.which in ("both", "brightness"):
if LightControlAttributes.BRIGHTNESS in data.attributes:
self.manager.adaptation_tasks_brightness[data.entity_id] = task
if data.which in ("both", "color"):
if LightControlAttributes.COLOR in data.attributes:
self.manager.adaptation_tasks_color[data.entity_id] = task
await task
except asyncio.CancelledError:
@ -1397,7 +1396,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
data,
)
async def _update_attrs_and_maybe_adapt_lights( # noqa: PLR0912
async def _update_attrs_and_maybe_adapt_lights(
self,
*,
context: Context,
@ -1468,49 +1467,26 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if not filtered_lights:
return
adapt_brightness = self.adapt_brightness_switch.is_on
adapt_color = self.adapt_color_switch.is_on
assert isinstance(adapt_brightness, bool)
assert isinstance(adapt_color, bool)
tasks: list[asyncio.Task[None]] = []
for light in filtered_lights:
manually_controlled = (
self._take_over_control
and self.manager.is_manually_controlled(
self,
light,
force,
adapt_brightness,
adapt_color,
)
await self.manager.update_manually_controlled_from_untracked_change(
self,
light,
force,
context,
)
if manually_controlled:
# Performance optimization: Skip adaptation task if all attributes are
# manually controlled and the task wouldn't actually do anything.
if self.manager.get_adaption_control_attributes(self, light).has_none():
_LOGGER.debug(
"%s: '%s' is being manually controlled, stop adapting, context.id=%s.",
"%s: '%s' is being manually controlled, skip adaptation, context.id=%s.",
self._name,
light,
context.id,
)
continue
significant_change = (
self._take_over_control
and self._detect_non_ha_changes
and not force
# Note: This call updates the state of the light
# so it might suddenly be off.
and await self.manager.significant_change(
self,
light,
adapt_brightness,
adapt_color,
context,
)
)
if significant_change:
_fire_manual_control_event(self, light, context)
continue
_LOGGER.debug(
"%s: Calling _adapt_light from _update_attrs_and_maybe_adapt_lights:"
" '%s' with transition %s and context.id=%s",
@ -1553,7 +1529,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
entity_id,
event.context.id,
)
self.manager.mark_as_manual_control(entity_id)
self.manager.set_manual_control_attributes(entity_id)
return
if (
@ -1607,6 +1583,28 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
force=True,
)
def fire_manual_control_event(
self,
light: str,
context: Context,
) -> None:
"""Fire an event that 'light' is marked as manual_control."""
_LOGGER.debug(
"'adaptive_lighting.manual_control' event fired for %s for light %s",
self.entity_id,
light,
)
manual_attributes = self.manager.get_manual_control_attributes(light)
self.hass.bus.async_fire(
f"{DOMAIN}.manual_control",
{
ATTR_ENTITY_ID: light,
SWITCH_DOMAIN: self.entity_id,
CONF_MANUAL_CONTROL: manual_attributes,
},
context=context,
)
class SimpleSwitch(SwitchEntity, RestoreEntity):
"""Representation of a Adaptive Lighting switch."""
@ -1711,7 +1709,7 @@ class AdaptiveLightingManager:
# Locks that prevent light adjusting when waiting for a light to 'turn_off'
self.turn_off_locks: dict[str, asyncio.Lock] = {}
# Tracks which lights are manually controlled
self.manual_control: dict[str, bool] = {}
self.manual_control: dict[str, LightControlAttributes] = {}
# Track 'state_changed' events of self.lights resulting from this integration
self.our_last_state_on_change: dict[str, list[State]] = {}
# Track last 'service_data' to 'light.turn_on' resulting from this integration
@ -1954,10 +1952,7 @@ class AdaptiveLightingManager:
# were skipped by us
return
if (
ATTR_EFFECT in service_data[CONF_PARAMS]
or ATTR_FLASH in service_data[CONF_PARAMS]
):
if has_effect_attribute(service_data[CONF_PARAMS]):
return
_LOGGER.debug(
@ -2204,10 +2199,26 @@ class AdaptiveLightingManager:
)
self.auto_reset_manual_control_times[light] = time
def mark_as_manual_control(self, light: str) -> None:
"""Mark a light as manually controlled."""
_LOGGER.debug("Marking '%s' as manually controlled.", light)
self.manual_control[light] = True
def get_manual_control_attributes(
self,
light: str,
) -> LightControlAttributes:
"""Get the attributes for a light that are manually controlled."""
return self.manual_control.get(light, LightControlAttributes.NONE)
def set_manual_control_attributes(
self,
light: str,
attributes: LightControlAttributes = LightControlAttributes.ALL,
) -> None:
"""Mark attributes of a light as manually controlled."""
_LOGGER.debug(
"Light %s: Setting manual control attributes to %s (from %s).",
light,
attributes,
self.manual_control[light],
)
self.manual_control[light] = attributes
delay = self.auto_reset_manual_control_times.get(light)
async def reset() -> None:
@ -2228,35 +2239,87 @@ class AdaptiveLightingManager:
transition=switch.initial_transition,
force=True,
)
assert not self.manual_control[light]
assert self.manual_control[light] == LightControlAttributes.NONE
self._handle_timer(light, self.auto_reset_manual_control_timers, delay, reset)
def add_manual_control_attributes(
self,
light: str,
attributes: LightControlAttributes,
) -> None:
"""Add attributes to the manual control status of a light."""
current = self.get_manual_control_attributes(light)
_LOGGER.debug(
"Light %s: Adding manual control attributes %s (current: %s).",
light,
attributes,
current,
)
new = current | attributes
self.set_manual_control_attributes(light, new)
def get_adaption_control_attributes(
self,
switch: AdaptiveSwitch,
light: str,
) -> LightControlAttributes:
"""Get the attributes that should be adapted for a light.
Determines the attributes that should actually be adapted from the attributes
marked as manually controlled, the state of adaptation switches, and the adaptation
configuration.
Example 1: When no attributes are marked as manually controlled and all adaptation
switches are on, all attributes are returned.
Example 2: When no attributes are marked as manually controlled and the brightness
adaptation switch is off, only the color attribute is returned.
Example 3: When only brightness is marked as manually controlled, but the configuration
specifies to pause all adaptations on manual change, no attributes are returned so that
color is also not adapted.
"""
denied_adaptation_attributes = self.get_manual_control_attributes(light)
if (
denied_adaptation_attributes.has_any()
and switch._take_over_control_mode == TakeOverControlMode.PAUSE_ALL
):
# Extend to pausing all only if there is at least one manually controlled attribute
denied_adaptation_attributes = LightControlAttributes.ALL
enabled_adaptation_attributes = (
LightControlAttributes.BRIGHTNESS
if switch.adapt_brightness_switch.is_on
else LightControlAttributes.NONE
) | (
LightControlAttributes.COLOR
if switch.adapt_color_switch.is_on
else LightControlAttributes.NONE
)
return (
LightControlAttributes.ALL
& ~denied_adaptation_attributes
& enabled_adaptation_attributes
)
def cancel_ongoing_adaptation_calls(
self,
light_id: str,
which: Literal["color", "brightness", "both"] = "both",
) -> None:
"""Cancel ongoing adaptation service calls for a specific light entity."""
brightness_task = self.adaptation_tasks_brightness.get(light_id)
color_task = self.adaptation_tasks_color.get(light_id)
if (
which in ("both", "brightness")
and brightness_task is not None
and not brightness_task.done()
):
if brightness_task is not None and not brightness_task.done():
_LOGGER.debug(
"Cancelled ongoing brightness adaptation calls (%s) for '%s'",
brightness_task,
light_id,
)
brightness_task.cancel()
if (
which in ("both", "color")
and color_task is not None
and color_task is not brightness_task
and not color_task.done()
):
if color_task is not None and not color_task.done():
_LOGGER.debug(
"Cancelled ongoing color adaptation calls (%s) for '%s'",
color_task,
@ -2269,7 +2332,11 @@ class AdaptiveLightingManager:
"""Reset the 'manual_control' status of the lights."""
for light in lights:
if reset_manual_control:
self.manual_control[light] = False
_LOGGER.debug(
"Light %s: Clearing manual control attributes.",
light,
)
self.manual_control[light] = LightControlAttributes.NONE
if timer := self.auto_reset_manual_control_timers.pop(light, None):
timer.cancel()
self.our_last_state_on_change.pop(light, None)
@ -2319,11 +2386,29 @@ class AdaptiveLightingManager:
self.turn_off_event[eid] = event
self.reset(eid)
def on(eid: str, event: Event) -> None:
async def on(eid: str, event: Event) -> None:
task = self.sleep_tasks.get(eid)
if task is not None:
task.cancel()
self.turn_on_event[eid] = event
try:
switch = _switch_with_lights(
self.hass,
[eid],
expand_light_groups=False,
)
await self.update_manually_controlled_from_event(
switch,
eid,
force=False,
)
except NoSwitchFoundError:
_LOGGER.debug(
"No switch found for entity_id='%s' in 'on' event listener",
eid,
)
timer = self.auto_reset_manual_control_timers.get(eid)
if (
timer is not None
@ -2351,7 +2436,7 @@ class AdaptiveLightingManager:
event.context.id,
)
for eid in entity_ids:
on(eid, event)
await on(eid, event)
elif service == SERVICE_TOGGLE:
_LOGGER.debug(
@ -2366,7 +2451,7 @@ class AdaptiveLightingManager:
if state.state == STATE_ON: # is turning off
off(eid, event)
elif state.state == STATE_OFF: # is turning on
on(eid, event)
await on(eid, event)
async def state_changed_event_listener(
self,
@ -2489,68 +2574,95 @@ class AdaptiveLightingManager:
event,
)
def is_manually_controlled(
async def update_manually_controlled_from_event(
self,
switch: AdaptiveSwitch,
light: str,
force: bool,
adapt_brightness: bool,
adapt_color: bool,
) -> bool:
"""Check if the light has been 'on' and is now manually controlled."""
manual_control = self.manual_control.setdefault(light, False)
if manual_control:
# Manually controlled until light is turned on and off
return True
) -> None:
"""Check if the light has been manually controlled by the latest turn on event."""
if not switch._take_over_control:
return
turn_on_event = self.turn_on_event.get(light)
if (
turn_on_event is not None
and not self.is_proactively_adapting(turn_on_event.context.id)
and not is_our_context(turn_on_event.context)
and not force
turn_on_event is None
or self.is_proactively_adapting(turn_on_event.context.id)
or is_our_context(turn_on_event.context)
or force
):
keys = turn_on_event.data[ATTR_SERVICE_DATA].keys()
if (
(adapt_color and COLOR_ATTRS.intersection(keys))
or (adapt_brightness and BRIGHTNESS_ATTRS.intersection(keys))
or (ATTR_FLASH in keys)
or (ATTR_EFFECT in keys)
):
# Light was already on and 'light.turn_on' was not called by
# the adaptive_lighting integration.
manual_control = True
_fire_manual_control_event(switch, light, turn_on_event.context)
_LOGGER.debug(
"'%s' was already on and 'light.turn_on' was not called by the"
" adaptive_lighting integration (context.id='%s'), the Adaptive"
" Lighting will stop adapting the light until the switch or the"
" light turns off and then on again.",
light,
turn_on_event.context.id,
)
return manual_control
return
turn_on_attributes = get_light_control_attributes(
turn_on_event.data[ATTR_SERVICE_DATA],
)
if not turn_on_attributes:
return
# Light was already on and 'light.turn_on' was not called by
# the adaptive_lighting integration.
self.add_manual_control_attributes(light, turn_on_attributes)
switch.fire_manual_control_event(light, turn_on_event.context)
_LOGGER.debug(
"'%s' was already on and 'light.turn_on' was not called by the"
" adaptive_lighting integration (context.id='%s'), the Adaptive"
" Lighting will stop adapting %s of the light until the switch or the"
" light turns off and then on again.",
light,
turn_on_event.context.id,
turn_on_attributes,
)
async def update_manually_controlled_from_untracked_change(
self,
switch: AdaptiveSwitch,
light: str,
force: bool,
context: Context,
) -> None:
"""Check if the light has been manually controlled from an untracked change.
An untracked change is a change that has been made outsideof HA and is
therefore not visible through events.
"""
if not switch._take_over_control or not switch._detect_non_ha_changes or force:
return
# Note: This call updates the state of the light
# so it might suddenly be off.
significantly_changed_attributes = await self.significant_change(
switch,
light,
context,
)
if not significantly_changed_attributes:
return
self.add_manual_control_attributes(
light,
significantly_changed_attributes,
)
switch.fire_manual_control_event(light, context)
async def significant_change(
self,
switch: AdaptiveSwitch,
light: str,
adapt_brightness: bool,
adapt_color: bool,
context: Context, # just for logging
) -> bool:
) -> LightControlAttributes:
"""Has the light made a significant change since last update.
This method will detect changes that were made to the light without
calling 'light.turn_on', so outside of Home Assistant. If a change is
detected, we mark the light as 'manually controlled' until the light
or switch is turned 'off' and 'on' again.
calling 'light.turn_on', so outside of Home Assistant.
"""
assert switch._detect_non_ha_changes
last_service_data = self.last_service_data.get(light)
if last_service_data is None:
return False
return LightControlAttributes.NONE
# Update state and check for a manual change not done in HA.
# Ensure HASS is correctly updating your light's state with
# light.turn_on calls if any problems arise. This
@ -2559,33 +2671,32 @@ class AdaptiveLightingManager:
refreshed_state = self.hass.states.get(light)
assert refreshed_state is not None
changed = _attributes_have_changed(
changed_attributes = _attributes_have_changed(
old_attributes=last_service_data,
new_attributes=refreshed_state.attributes,
light=light,
adapt_brightness=adapt_brightness,
adapt_color=adapt_color,
context=context,
)
if changed:
if changed_attributes:
_LOGGER.debug(
"%s: State attributes of '%s' changed (%s) wrt 'last_service_data' (%s) (context.id=%s)",
"%s: State attributes %s of '%s' changed (%s) wrt 'last_service_data' (%s) (context.id=%s)",
switch._name,
changed_attributes,
light,
refreshed_state.attributes,
last_service_data,
context.id,
)
else:
_LOGGER.debug(
"%s: State attributes of '%s' did not change (%s) wrt 'last_service_data' (%s) (context.id=%s)",
switch._name,
light,
refreshed_state.attributes,
last_service_data,
context.id,
)
return True
_LOGGER.debug(
"%s: State attributes of '%s' did not change (%s) wrt 'last_service_data' (%s) (context.id=%s)",
switch._name,
light,
refreshed_state.attributes,
last_service_data,
context.id,
)
return False
return changed_attributes
def _off_to_on_state_event_is_from_turn_on(
self,
@ -2745,12 +2856,12 @@ class AdaptiveLightingManager:
entity_id,
service_data,
)
if any(
attr in service_data
for attr in COLOR_ATTRS | BRIGHTNESS_ATTRS | {ATTR_EFFECT}
):
self.mark_as_manual_control(entity_id)
manual_control_attributes = get_light_control_attributes(service_data)
if manual_control_attributes:
self.set_manual_control_attributes(entity_id, manual_control_attributes)
return True
return False