remote/adaptive-lighting
1
0
Fork 0
mirror of https://github.com/basnijholt/adaptive-lighting.git synced 2026-09-11 14:24:03 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix multiple switches controlling one light (reactive path) (#644)

Browse source 
* Fix multiple switches controlling one light

* WIP

* Keep length in AdaptationData

* Do not proceed if there is nothing to do

* Simplify

* Set tasks correctly

* log more

* WIP

* realize that intercept and double switch not possible

* Remove which from find_switch_for_lights

* Rephrase

* Check which one to cancel

* length -> max_length

* Implement test_two_switches_for_single_light

* Clean up

* Simplify is_color_brightness_or_both

* Improve test

* no wait

* logging

* just block

* make test such that it fails on main
This commit is contained in:
Bas Nijholt 2023-07-21 14:46:18 -07:00 • committed by GitHub
commit 49857632cc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 183 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

45
custom_components/adaptive_lighting/adaptation_utils.py
View file

@ -1,7 +1,8 @@
"""Utility functions for adaptation commands."""
from collections.abc import AsyncGenerator
from dataclasses import dataclass
from typing import Any
import logging
from typing import Any, Literal
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -18,6 +19,8 @@ from homeassistant.components.light import (
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,
@ -98,7 +101,7 @@ def _has_relevant_service_data_attributes(service_data: ServiceData) -> bool:
async def _create_service_call_data_iterator(
hass: HomeAssistant,
service_datas: list[ServiceData],
filter_by_state: bool = False,
filter_by_state: bool,
) -> AsyncGenerator[ServiceData, None]:
"""Enumerates and filters a list of service datas on the fly.
@ -133,6 +136,8 @@ class AdaptationData:
context: Context
sleep_time: float
service_call_datas: AsyncGenerator[ServiceData, None]
max_length: int
which: Literal["brightness", "color", "both"]
initial_sleep: bool = False
async def next_service_call_data(self) -> ServiceData | None:
@ -140,6 +145,26 @@ class AdaptationData:
return await anext(self.service_call_datas, None)
class NoColorOrBrightnessInServiceData(Exception):
"""Exception raised when no color or brightness attributes are found in service data."""
def is_color_brightness_or_both(
service_data: ServiceData,
) -> Literal["brightness", "color", "both"]:
"""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)
if has_brightness and has_color:
return "both"
if has_brightness:
return "brightness"
if has_color:
return "color"
msg = f"Invalid service_data, no brightness or color attributes found: {service_data=}"
raise NoColorOrBrightnessInServiceData(msg)
def prepare_adaptation_data(
hass: HomeAssistant,
entity_id: str,
@ -150,7 +175,12 @@ def prepare_adaptation_data(
split: bool,
filter_by_state: bool,
) -> AdaptationData:
"Prepares a data object carrying all data required to execute an adaptation."
"""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)
)
@ -163,4 +193,11 @@ def prepare_adaptation_data(
hass, service_datas, filter_by_state
)
return AdaptationData(entity_id, context, sleep_time, service_data_iterator)
return AdaptationData(
entity_id,
context,
sleep_time=sleep_time,
service_call_datas=service_data_iterator,
max_length=len(service_datas),
which=is_color_brightness_or_both(service_data),
)

81
custom_components/adaptive_lighting/switch.py
View file

@ -1071,7 +1071,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
adapt_color: bool | None = None,
prefer_rgb_color: bool | None = None,
context: Context | None = None,
) -> AdaptationData:
) -> AdaptationData | None:
if transition is None:
transition = self._transition
if adapt_brightness is None:
@ -1081,6 +1081,15 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if prefer_rgb_color is None:
prefer_rgb_color = self._prefer_rgb_color
if not adapt_color and not adapt_brightness:
_LOGGER.debug(
"%s: Skipping adaptation of %s because both adapt_brightness and"
" adapt_color are False",
self._name,
light,
)
return None
# The switch might be off and not have _settings set.
self._settings = self._sun_light_settings.get_settings(
self.sleep_mode_switch.is_on, transition
@ -1150,7 +1159,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if self.turn_on_off_listener.is_proactively_adapting(context.parent_id):
# Skip if adaptation was already executed by the service call interceptor
_LOGGER.debug("Skipping reactive adaptation of %s", context.parent_id)
_LOGGER.debug(
"%s: Skipping reactive adaptation of %s", self._name, context.parent_id
)
return
data = await self.prepare_adaptation_data(
@ -1161,6 +1172,8 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
prefer_rgb_color,
context,
)
if data is None:
return None # nothing to adapt
await self.execute_cancellable_adaptation_calls(data)
@ -1209,15 +1222,32 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
to cancel an ongoing adaptation when a light is turned off.
"""
# Prevent overlap of multiple adaptation sequences
self.turn_on_off_listener.cancel_ongoing_adaptation_calls(data.entity_id)
listener = self.turn_on_off_listener
listener.cancel_ongoing_adaptation_calls(data.entity_id, which=data.which)
_LOGGER.debug(
"%s: execute_cancellable_adaptation_calls with data: %s"
"adaptation_tasks_brightness: %s"
"adaptation_tasks_color: %s",
self._name,
data,
listener.adaptation_tasks_brightness,
listener.adaptation_tasks_color,
)
# Execute adaptation calls within a task
try:
task = asyncio.ensure_future(self._execute_adaptation_calls(data))
self.turn_on_off_listener.adaptation_tasks[data.entity_id] = task
if data.which in ("both", "brightness"):
listener.adaptation_tasks_brightness[data.entity_id] = task
if data.which in ("both", "color"):
listener.adaptation_tasks_color[data.entity_id] = task
await task
except asyncio.CancelledError:
_LOGGER.debug("Ongoing adaptation of %s cancelled", data.entity_id)
_LOGGER.debug(
"%s: Ongoing adaptation of %s cancelled, with AdaptationData: %s",
self._name,
data.entity_id,
data,
)
async def _update_attrs_and_maybe_adapt_lights(
self,
@ -1699,7 +1729,8 @@ class TurnOnOffListener:
# Track last 'service_data' to 'light.turn_on' resulting from this integration
self.last_service_data: dict[str, dict[str, Any]] = {}
# Track ongoing split adaptations to be able to cancel them
self.adaptation_tasks: dict[str, asyncio.Task] = {}
self.adaptation_tasks_brightness: dict[str, asyncio.Task] = {}
self.adaptation_tasks_color: dict[str, asyncio.Task] = {}
# Track auto reset of manual_control
self.auto_reset_manual_control_timers: dict[str, _AsyncSingleShotTimer] = {}
@ -1815,6 +1846,11 @@ class TurnOnOffListener:
adaptive_switch = find_switch_for_lights(self.hass, [entity_id])
except NoSwitchFoundError:
# This might be a light that is not managed by this AL instance.
_LOGGER.debug(
"No (or multiple) adaptive switch(es) found for entity %s,"
" skipping adaptation by intercepting service call",
entity_id,
)
return
if not adaptive_switch.is_on:
@ -1851,6 +1887,8 @@ class TurnOnOffListener:
adapt_brightness,
adapt_color,
)
# if adaptation_data is None:
# return
# Take first adaptation item to apply it to this service call
first_service_data = await adaptation_data.next_service_call_data()
@ -1970,10 +2008,31 @@ class TurnOnOffListener:
self._handle_timer(light, self.auto_reset_manual_control_timers, delay, reset)
def cancel_ongoing_adaptation_calls(self, light_id: str):
"""Cancels an ongoing sequence of adaptation service calls for a specific light entity."""
if (previous_task := self.adaptation_tasks.get(light_id)) is not None:
previous_task.cancel()
def cancel_ongoing_adaptation_calls(
self, light_id: str, which: Literal["color", "brightness", "both"] = "both"
):
"""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:
_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
):
_LOGGER.debug(
"Cancelled ongoing color adaptation calls (%s) for '%s'",
color_task,
light_id,
)
# color_task might be the same as brightness_task
color_task.cancel()
def reset(self, *lights, reset_manual_control=True) -> None:
"""Reset the 'manual_control' status of the lights."""

76
tests/test_switch.py
View file

@ -1351,7 +1351,9 @@ async def test_cancellable_service_calls_task(hass):
_, switch = await setup_switch(hass, {CONF_SEPARATE_TURN_ON_COMMANDS: True})
context = switch.create_context("test")
assert switch.turn_on_off_listener.adaptation_tasks.get(light.entity_id) is None
assert (
switch.turn_on_off_listener.adaptation_tasks_color.get(light.entity_id) is None
)
service_data = {
ATTR_BRIGHTNESS: 10,
@ -1362,11 +1364,15 @@ async def test_cancellable_service_calls_task(hass):
light.entity_id,
context,
0,
_create_service_call_data_iterator(hass, [service_data]),
_create_service_call_data_iterator(hass, [service_data], False),
max_length=1,
which="both",
)
await switch.execute_cancellable_adaptation_calls(adaptation_data)
task = switch.turn_on_off_listener.adaptation_tasks.get(light.entity_id)
task = switch.turn_on_off_listener.adaptation_tasks_brightness.get(light.entity_id)
task2 = switch.turn_on_off_listener.adaptation_tasks_color.get(light.entity_id)
assert task is task2
assert task is not None
assert task.done()
@ -1378,7 +1384,7 @@ async def test_service_calls_task_cancellation(hass):
entity_id = "test_id"
task = asyncio.ensure_future(asyncio.sleep(1))
switch.turn_on_off_listener.adaptation_tasks[entity_id] = task
switch.turn_on_off_listener.adaptation_tasks_brightness[entity_id] = task
switch.turn_on_off_listener.cancel_ongoing_adaptation_calls(entity_id)
@ -1553,3 +1559,65 @@ async def test_proactive_adaptation_transition_override(hass):
# Cleanup
switch.turn_on_off_listener.cancel_ongoing_adaptation_calls(ENTITY_LIGHT3)
async def test_two_switches_for_single_light(hass):
"""Test the case where someone has two switches for a single light.
One switch for brightness and another for color.
"""
extra_conf = {INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION: True}
switch1, (light1, *_) = await setup_lights_and_switch(
hass, extra_conf | {CONF_NAME: "switch1"}, all_lights=True
)
switch2, (light2, *_) = await setup_lights_and_switch(
hass, extra_conf | {CONF_NAME: "switch2"}, all_lights=True
)
assert light1 is light2
# One switch controls brightness the other color
await switch1.adapt_color_switch.async_turn_off()
await switch2.adapt_brightness_switch.async_turn_off()
assert switch1.adapt_brightness_switch.is_on
assert switch2.adapt_color_switch.is_on
async def turn_light(state, **kwargs):
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON if state else SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: ENTITY_LIGHT, **kwargs},
blocking=True,
)
await hass.async_block_till_done()
_LOGGER.debug("Turn light %s, to %s", state, kwargs)
def increased_brightness():
return (light1._attr_brightness + 100) % 255
def increased_color_temp():
return max(
(light1._attr_color_temp + 100) % light1.max_color_temp_kelvin,
light1.min_color_temp_kelvin,
)
assert light1.is_on
await turn_light(True, brightness=increased_brightness())
await turn_light(True, color_temp=increased_color_temp())
attrs = hass.states.get(light1.entity_id).attributes
before_brightness = attrs[ATTR_BRIGHTNESS]
before_color_temp = attrs[ATTR_COLOR_TEMP_KELVIN]
# Turn off "light1"
await turn_light(False)
# Turn on "light1"
await turn_light(True)
# Assert that the brightness and color temp have changed
attrs = hass.states.get(light1.entity_id).attributes
after_brightness = attrs[ATTR_BRIGHTNESS]
after_color_temp = attrs[ATTR_COLOR_TEMP_KELVIN]
assert before_brightness != after_brightness
assert before_color_temp != after_color_temp