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

feat: service call adaptation (#628)

Browse source 
* feat: service call adaptation

* feat: toggle-on service call adaptation

* feat: prefer service call transition
This commit is contained in:
Mario Guggenberger 2023-07-19 09:10:21 +02:00 • committed by GitHub
commit ed80bd7829
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 610 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

3
custom_components/adaptive_lighting/__init__.py
View file

@ -89,8 +89,7 @@ async def async_unload_entry(hass, config_entry: ConfigEntry) -> bool:
if len(data) == 1 and ATTR_TURN_ON_OFF_LISTENER in data:
# no more config_entries
turn_on_off_listener = data.pop(ATTR_TURN_ON_OFF_LISTENER)
turn_on_off_listener.remove_listener()
turn_on_off_listener.remove_listener2()
turn_on_off_listener.disable()
if not data:
hass.data.pop(DOMAIN)

2
custom_components/adaptive_lighting/adaptation_utils.py
View file

@ -133,6 +133,7 @@ class AdaptationData:
context: Context
sleep_time: float
service_call_datas: AsyncGenerator[ServiceData, None]
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."""
@ -149,6 +150,7 @@ def prepare_adaptation_data(
split: bool,
filter_by_state: bool,
) -> AdaptationData:
"Prepares a data object carrying all data required to execute an adaptation."
service_datas = (
[service_data] if not split else _split_service_call_data(service_data)
)

64
custom_components/adaptive_lighting/hass_utils.py Normal file
View file

@ -0,0 +1,64 @@
"""Utility functions for HA core."""
from collections.abc import Awaitable
from typing import Callable
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.util.read_only_dict import ReadOnlyDict
from .adaptation_utils import ServiceData
def setup_service_call_interceptor(
hass: HomeAssistant,
domain: str,
service: str,
intercept_func: Callable[[ServiceCall, ServiceData], Awaitable[None] | None],
) -> Callable[[], None]:
"""Inject a function into a registered service call to preprocess service data.
The injected interceptor function receives the service call and a writeable data dictionary
(the data of the service call is read-only) before the service call is executed."""
try:
# HACK: Access protected attribute of HA service registry.
# This is necessary to replace a registered service handler with our
# proxy handler to intercept calls.
registered_services = (
hass.services._services # pylint: disable=protected-access
)
except AttributeError as error:
raise RuntimeError(
"Intercept failed because registered services are no longer accessible "
"(internal API may have changed)"
) from error
if domain not in registered_services or service not in registered_services[domain]:
raise RuntimeError(
f"Intercept failed because service {domain}.{service} is not registered"
)
existing_service = registered_services[domain][service]
async def service_func_proxy(call: ServiceCall) -> None:
# Convert read-only data to writeable dictionary for modification by interceptor
data = dict(call.data)
# Call interceptor
await intercept_func(call, data)
# Convert data back to read-only
call.data = ReadOnlyDict(data)
# Call original service handler with processed data
await existing_service.job.target(call)
hass.services.async_register(
domain, service, service_func_proxy, existing_service.schema
)
def remove():
# Remove the interceptor by reinstalling the original service handler
hass.services.async_register(
domain, service, existing_service.job.target, existing_service.schema
)
return remove

315
custom_components/adaptive_lighting/switch.py
View file

@ -41,6 +41,7 @@ from homeassistant.components.light import (
SUPPORT_COLOR_TEMP,
SUPPORT_TRANSITION,
is_on,
preprocess_turn_on_alternatives,
)
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
@ -54,9 +55,11 @@ from homeassistant.const import (
ATTR_SERVICE_DATA,
ATTR_SUPPORTED_FEATURES,
CONF_NAME,
CONF_PARAMS,
EVENT_CALL_SERVICE,
EVENT_HOMEASSISTANT_STARTED,
EVENT_STATE_CHANGED,
SERVICE_TOGGLE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
@ -96,6 +99,7 @@ from .adaptation_utils import (
BRIGHTNESS_ATTRS,
COLOR_ATTRS,
AdaptationData,
ServiceData,
prepare_adaptation_data,
)
from .const import (
@ -156,6 +160,7 @@ from .const import (
apply_service_schema,
replace_none_str,
)
from .hass_utils import setup_service_call_interceptor
_SUPPORT_OPTS = {
COLOR_MODE_BRIGHTNESS: SUPPORT_BRIGHTNESS,
@ -182,6 +187,11 @@ _LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=10)
# A (non-user-configurable, thus internal) flag to control the proactive adaptation mode.
# This exists to disable the proactive adaptation in the unit tests and enable it
# only for specific unit tests and when running as integration."""
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION = "proactive_adaptation"
# Consider it a significant change when attribute changes more than
BRIGHTNESS_CHANGE = 25 # ≈10% of total range
COLOR_TEMP_CHANGE = 100 # ≈3% of total range (2000-6500)
@ -262,11 +272,17 @@ def create_context(
return Context(id=context_id, parent_id=parent_id)
def is_our_context_id(context_id: str | None) -> bool:
if context_id is None:
return False
return f":{_DOMAIN_SHORT}:" in context_id
def is_our_context(context: Context | None) -> bool:
"""Check whether this integration created 'context'."""
if context is None:
return False
return f":{_DOMAIN_SHORT}:" in context.id
return is_our_context_id(context.id)
def _get_switches_with_lights(
@ -284,7 +300,7 @@ def _get_switches_with_lights(
all_check_lights = _expand_light_groups(hass, lights)
switch._expand_light_groups()
# Check if any of the lights are in the switch's lights
if set(switch._lights) & set(all_check_lights):
if set(switch.lights) & set(all_check_lights):
switches.append(switch)
return switches
@ -385,12 +401,12 @@ async def handle_change_switch_settings(
data,
)
all_lights = switch._lights # pylint: disable=protected-access
all_lights = switch.lights # pylint: disable=protected-access
switch.turn_on_off_listener.reset(*all_lights, reset_manual_control=False)
if switch.is_on:
await switch._update_attrs_and_maybe_adapt_lights( # pylint: disable=protected-access
all_lights,
transition=switch._initial_transition,
transition=switch.initial_transition,
force=True,
context=switch.create_context("service", parent=service_call.context),
)
@ -424,8 +440,8 @@ async def async_setup_entry(
assert config_entry.entry_id in data
if ATTR_TURN_ON_OFF_LISTENER not in data:
data[ATTR_TURN_ON_OFF_LISTENER] = TurnOnOffListener(hass)
turn_on_off_listener = data[ATTR_TURN_ON_OFF_LISTENER]
data[ATTR_TURN_ON_OFF_LISTENER] = TurnOnOffListener(hass, config_entry)
turn_on_off_listener: TurnOnOffListener = data[ATTR_TURN_ON_OFF_LISTENER]
sleep_mode_switch = SimpleSwitch(
"Sleep Mode", False, hass, config_entry, ICON_SLEEP
)
@ -443,6 +459,7 @@ async def async_setup_entry(
adapt_color_switch,
adapt_brightness_switch,
)
turn_on_off_listener.adaptive_switch = switch
# save our switch instance, allows us to make switch's entity_id optional in service calls.
hass.data[DOMAIN][config_entry.entry_id]["instance"] = switch
@ -469,7 +486,7 @@ async def async_setup_entry(
lights = data[CONF_LIGHTS]
for switch in switches:
if not lights:
all_lights = switch._lights # pylint: disable=protected-access
all_lights = switch.lights
else:
all_lights = _expand_light_groups(switch.hass, lights)
switch.turn_on_off_listener.lights.update(all_lights)
@ -498,7 +515,7 @@ async def async_setup_entry(
lights = data[CONF_LIGHTS]
for switch in switches:
if not lights:
all_lights = switch._lights # pylint: disable=protected-access
all_lights = switch.lights
else:
all_lights = _expand_light_groups(switch.hass, lights)
if service_call.data[CONF_MANUAL_CONTROL]:
@ -510,7 +527,7 @@ async def async_setup_entry(
# pylint: disable=protected-access
await switch._update_attrs_and_maybe_adapt_lights(
all_lights,
transition=switch._initial_transition,
transition=switch.initial_transition,
force=True,
context=switch.create_context(
"service", parent=service_call.context
@ -523,7 +540,7 @@ async def async_setup_entry(
service=SERVICE_APPLY,
service_func=handle_apply,
schema=apply_service_schema(
switch._initial_transition
switch.initial_transition
), # pylint: disable=protected-access
)
@ -803,7 +820,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._name = data[CONF_NAME]
self._interval = data[CONF_INTERVAL]
self._lights = data[CONF_LIGHTS]
self.lights: list[str] = data[CONF_LIGHTS]
# backup data for use in change_switch_settings "configuration" CONF_USE_DEFAULTS
self._config_backup = deepcopy(data)
@ -835,7 +852,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
" config_entry.data: '%s',"
" config_entry.options: '%s', converted to '%s'.",
self._name,
self._lights,
self.lights,
config_entry.data,
config_entry.options,
data,
@ -868,7 +885,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
attrdata[k] = v.total_seconds()
self._config.update(attrdata)
self._initial_transition = data[CONF_INITIAL_TRANSITION]
self.initial_transition = data[CONF_INITIAL_TRANSITION]
self._sleep_transition = data[CONF_SLEEP_TRANSITION]
self._only_once = data[CONF_ONLY_ONCE]
self._prefer_rgb_color = data[CONF_PREFER_RGB_COLOR]
@ -921,7 +938,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
_LOGGER.debug(
"%s: Set switch settings for lights '%s'. now using data: '%s'",
self._name,
self._lights,
self.lights,
data,
)
@ -961,12 +978,12 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._remove_listeners()
def _expand_light_groups(self) -> None:
all_lights = _expand_light_groups(self.hass, self._lights)
all_lights = _expand_light_groups(self.hass, self.lights)
self.turn_on_off_listener.lights.update(all_lights)
self.turn_on_off_listener.set_auto_reset_manual_control_times(
all_lights, self._auto_reset_manual_control_time
)
self._lights = list(all_lights)
self.lights = list(all_lights)
async def _setup_listeners(self, _=None) -> None:
_LOGGER.debug("%s: Called '_setup_listeners'", self._name)
@ -987,10 +1004,10 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self.remove_listeners.extend([remove_interval, remove_sleep])
if self._lights:
if self.lights:
self._expand_light_groups()
remove_state = async_track_state_change_event(
self.hass, self._lights, self._light_event
self.hass, self.lights, self._light_event
)
self.remove_listeners.append(remove_state)
@ -1014,14 +1031,14 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
return extra_state_attributes
extra_state_attributes["manual_control"] = [
light
for light in self._lights
for light in self.lights
if self.turn_on_off_listener.manual_control.get(light)
]
extra_state_attributes.update(self._settings)
timers = self.turn_on_off_listener.auto_reset_manual_control_timers
extra_state_attributes["autoreset_time_remaining"] = {
light: time
for light in self._lights
for light in self.lights
if (timer := timers.get(light)) and (time := timer.remaining_time()) > 0
}
return extra_state_attributes
@ -1054,11 +1071,11 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if self.is_on:
return
self._state = True
self.turn_on_off_listener.reset(*self._lights)
self.turn_on_off_listener.reset(*self.lights)
await self._setup_listeners()
if adapt_lights:
await self._update_attrs_and_maybe_adapt_lights(
transition=self._initial_transition,
transition=self.initial_transition,
force=True,
context=self.create_context("turn_on"),
)
@ -1069,7 +1086,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
return
self._state = False
self._remove_listeners()
self.turn_on_off_listener.reset(*self._lights)
self.turn_on_off_listener.reset(*self.lights)
async def _async_update_at_interval(self, now=None) -> None:
await self._update_attrs_and_maybe_adapt_lights(
@ -1078,7 +1095,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
context=self.create_context("interval"),
)
async def _adapt_light( # noqa: C901
async def prepare_adaptation_data(
self,
light: str,
transition: int | None = None,
@ -1086,11 +1103,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
adapt_color: bool | None = None,
prefer_rgb_color: bool | None = None,
context: Context | None = None,
) -> None:
lock = self._locks.get(light)
if lock is not None and lock.locked():
_LOGGER.debug("%s: '%s' is locked", self._name, light)
return
) -> AdaptationData:
if transition is None:
transition = self._transition
if adapt_brightness is None:
@ -1141,7 +1154,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self.turn_on_off_listener.last_service_data[light] = service_data
data = prepare_adaptation_data(
return prepare_adaptation_data(
self.hass,
light,
context,
@ -1152,7 +1165,35 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
filter_by_state=self._skip_redundant_commands,
)
await self._execute_cancellable_adaptation_calls(data)
async def _adapt_light( # noqa: C901
self,
light: str,
transition: int | None = None,
adapt_brightness: bool | None = None,
adapt_color: bool | None = None,
prefer_rgb_color: bool | None = None,
context: Context | None = None,
) -> None:
lock = self._locks.get(light)
if lock is not None and lock.locked():
_LOGGER.debug("%s: '%s' is locked", self._name, light)
return
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)
return
data = await self.prepare_adaptation_data(
light,
transition,
adapt_brightness,
adapt_color,
prefer_rgb_color,
context,
)
await self.execute_cancellable_adaptation_calls(data)
async def _execute_adaptation_calls(self, data: AdaptationData):
"""Executes a sequence of adaptation service calls for the given service datas."""
@ -1163,7 +1204,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
index += 1
# Sleep between multiple service calls.
if not is_first_call:
if not is_first_call or data.initial_sleep:
await asyncio.sleep(data.sleep_time)
# Instead of directly iterating the generator in the while-loop, we get
@ -1189,7 +1230,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
context=data.context,
)
async def _execute_cancellable_adaptation_calls(
async def execute_cancellable_adaptation_calls(
self,
data: AdaptationData,
):
@ -1231,7 +1272,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self.async_write_ha_state()
if lights is None:
lights = self._lights
lights = self.lights
filtered_lights = []
if not force:
@ -1323,7 +1364,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
"%s: _sleep_mode_switch_state_event, event: '%s'", self._name, event
)
# Reset the manually controlled status when the "sleep mode" changes
self.turn_on_off_listener.reset(*self._lights)
self.turn_on_off_listener.reset(*self.lights)
await self._update_attrs_and_maybe_adapt_lights(
transition=self._sleep_transition,
force=True,
@ -1346,7 +1387,15 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
entity_id,
event.context.id,
)
self.turn_on_off_listener.reset(entity_id, reset_manual_control=False)
if (
event.context.parent_id
and not self.turn_on_off_listener.is_proactively_adapting(
event.context.id
)
):
self.turn_on_off_listener.reset(entity_id, reset_manual_control=False)
# Tracks 'off' → 'on' state changes
self._off_to_on_event[entity_id] = event
lock = self._locks.get(entity_id)
@ -1381,7 +1430,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
await self._update_attrs_and_maybe_adapt_lights(
lights=[entity_id],
transition=self._initial_transition,
transition=self.initial_transition,
force=True,
context=self.create_context("light_event", parent=event.context),
)
@ -1662,9 +1711,10 @@ class SunLightSettings:
class TurnOnOffListener:
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
def __init__(self, hass: HomeAssistant):
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry):
"""Initialize the TurnOnOffListener that is shared among all switches."""
self.hass = hass
data = validate(config_entry)
self.lights = set()
# Tracks 'light.turn_off' service calls
@ -1689,11 +1739,169 @@ class TurnOnOffListener:
# Track light transitions
self.transition_timers: dict[str, _AsyncSingleShotTimer] = {}
self.remove_listener = self.hass.bus.async_listen(
EVENT_CALL_SERVICE, self.turn_on_off_event_listener
self.listener_removers = []
self.listener_removers.append(
self.hass.bus.async_listen(
EVENT_CALL_SERVICE, self.turn_on_off_event_listener
)
)
self.remove_listener2 = self.hass.bus.async_listen(
EVENT_STATE_CHANGED, self.state_changed_event_listener
self.listener_removers.append(
self.hass.bus.async_listen(
EVENT_STATE_CHANGED, self.state_changed_event_listener
)
)
self.adaptive_switch: AdaptiveSwitch | None
self._proactively_adapting_contexts: dict[str, str] = {}
is_proactive_adaptation_enabled = (
data.get(INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION, True) is not False
)
if is_proactive_adaptation_enabled:
try:
self.listener_removers.append(
setup_service_call_interceptor(
hass,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
self._service_interceptor_turn_on_handler,
)
)
self.listener_removers.append(
setup_service_call_interceptor(
hass,
LIGHT_DOMAIN,
SERVICE_TOGGLE,
self._service_interceptor_turn_on_handler,
)
)
_LOGGER.debug("Proactive adaptation enabled")
except RuntimeError:
_LOGGER.warning(
"Failed to set up service call interceptors, "
"falling back to event-reactive mode",
exc_info=True,
)
def disable(self):
"""Disable the listener by removing all subscribed handlers."""
for remove in self.listener_removers:
remove()
def set_proactively_adapting(self, context_id: str, entity_id: str) -> None:
"""Declare the adaptation with the given context ID as proactively adapting,
and associate it to an entity ID."""
self._proactively_adapting_contexts[context_id] = entity_id
def is_proactively_adapting(self, context_id: str) -> bool:
"""Determine whether an adaptation with the given context ID is proactive."""
is_proactively_adapting_context = (
context_id in self._proactively_adapting_contexts
)
_LOGGER.debug(
"is_proactively_adapting_context %s %s",
context_id,
is_proactively_adapting_context,
)
return is_proactively_adapting_context
def clear_proactively_adapting(self, entity_id: str) -> None:
"""Clear all context IDs associated with the given entity ID.
Call this method to clear past context IDs and avoid a memory leak."""
keys = [
k for k, v in self._proactively_adapting_contexts.items() if v == entity_id
]
for key in keys:
self._proactively_adapting_contexts.pop(key)
async def _service_interceptor_turn_on_handler(
self, call: ServiceCall, data: ServiceData
):
# Don't adapt our own service calls
if is_our_context(call.context):
return
entity_ids = self._get_entity_list(data)
# For simplicity, only service calls affecting a single entity are currently handled.
#
# To add support for adapting multiple entities, the following properties
# need to hold for _all_ entities:
# - managed by this AL instance
# - not manually controlled
# - supporting the same relevant feature set
# - off state
if len(entity_ids) != 1:
return
entity_id = entity_ids[0]
if entity_id not in self.adaptive_switch.lights:
return
if self.manual_control.get(entity_id, False):
return
# Prevent adaptation of TURN_ON calls when light is already on,
# and of TOGGLE calls when toggling off.
if self.hass.states.is_state(entity_id, STATE_ON):
return
_LOGGER.debug(
"Intercepted TURN_ON call with data %s (%s)", data, call.context.id
)
self.reset(entity_id, reset_manual_control=False)
self.clear_proactively_adapting(entity_id)
adapt_brightness = self.adaptive_switch.adapt_brightness_switch.is_on or False
adapt_color = self.adaptive_switch.adapt_color_switch.is_on or False
transition = (
data[CONF_PARAMS].get(ATTR_TRANSITION, None)
or self.adaptive_switch.initial_transition
)
adaptation_data = await self.adaptive_switch.prepare_adaptation_data(
entity_id,
transition,
adapt_brightness,
adapt_color,
)
# Take first adaptation item to apply it to this service call
first_service_data = await adaptation_data.next_service_call_data()
if not first_service_data:
return
# Update/adapt service call data
first_service_data.pop(ATTR_ENTITY_ID, None)
# This is called as a preprocessing step by the schema validation of the original
# service call and needs to be repeated here to also process the added adaptation data.
# (A more generic alternative would be re-executing the validation, but that is more
# complicated and unstable because it requires transformation of the data object back
# into its original service call structure which cannot be reliably done due to the
# lack of a bijective mapping.)
preprocess_turn_on_alternatives(self.hass, first_service_data)
data[CONF_PARAMS].update(first_service_data)
# Schedule additional service calls for the remaining adaptation data.
# We cannot know here whether there is another call to follow (since the
# state can change until the next call), so we just schedule it and let
# it sort out by itself.
self.set_proactively_adapting(call.context.id, entity_id)
self.set_proactively_adapting(adaptation_data.context.id, entity_id)
adaptation_data.initial_sleep = True
asyncio.create_task( # Don't await to avoid blocking the service call
self.adaptive_switch.execute_cancellable_adaptation_calls(adaptation_data)
)
def _handle_timer(
@ -1772,7 +1980,7 @@ class TurnOnOffListener:
continue
await switch._update_attrs_and_maybe_adapt_lights(
[light],
transition=switch._initial_transition,
transition=switch.initial_transition,
force=True,
context=switch.create_context("autoreset"),
)
@ -1803,19 +2011,13 @@ class TurnOnOffListener:
self.last_service_data.pop(light, None)
self.cancel_ongoing_adaptation_calls(light)
async def turn_on_off_event_listener(self, event: Event) -> None:
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
domain = event.data.get(ATTR_DOMAIN)
if domain != LIGHT_DOMAIN:
return
def _get_entity_list(self, service_data: ServiceData) -> list[str]:
entity_ids = []
service = event.data[ATTR_SERVICE]
service_data = event.data[ATTR_SERVICE_DATA]
if ATTR_ENTITY_ID in service_data:
entity_ids = cv.ensure_list_csv(service_data[ATTR_ENTITY_ID])
elif ATTR_AREA_ID in service_data:
area_ids = cv.ensure_list_csv(service_data[ATTR_AREA_ID])
entity_ids = []
for area_id in area_ids:
area_entity_ids = area_entities(self.hass, area_id)
for entity_id in area_entity_ids:
@ -1828,8 +2030,19 @@ class TurnOnOffListener:
_LOGGER.debug(
"No entity_ids or area_ids found in service_data: %s", service_data
)
return entity_ids
async def turn_on_off_event_listener(self, event: Event) -> None:
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
domain = event.data.get(ATTR_DOMAIN)
if domain != LIGHT_DOMAIN:
return
service = event.data[ATTR_SERVICE]
service_data = event.data[ATTR_SERVICE_DATA]
entity_ids = self._get_entity_list(service_data)
if not any(eid in self.lights for eid in entity_ids):
return

81
tests/test_hass_utils.py Normal file
View file

@ -0,0 +1,81 @@
"""Tests for Adaptive Lighting HASS utils."""
from unittest.mock import AsyncMock
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.const import SERVICE_TURN_ON
from homeassistant.core import ServiceCall
from homeassistant.util.read_only_dict import ReadOnlyDict
from custom_components.adaptive_lighting.adaptation_utils import ServiceData
from custom_components.adaptive_lighting.hass_utils import (
setup_service_call_interceptor,
)
async def test_setup_service_call_interceptor(hass):
"""Test setup and removal of service call interceptor."""
service_func_mock = AsyncMock()
hass.services.async_register(LIGHT_DOMAIN, SERVICE_TURN_ON, service_func_mock)
async def service_call():
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{},
blocking=True,
)
# Test if service is called
await service_call()
assert service_func_mock.call_count == 1
# Test if interceptor is called after setup
intercept_func_mock = AsyncMock()
remove_interceptor = setup_service_call_interceptor(
hass,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
intercept_func_mock,
)
await service_call()
assert service_func_mock.call_count == 2
assert intercept_func_mock.call_count == 1
# Test if interceptor is no longer called after removal
remove_interceptor()
await service_call()
assert service_func_mock.call_count == 3
assert intercept_func_mock.call_count == 1
async def test_service_call_interceptor_data_manipulation(hass):
"""Test service call data manipulation by service call interceptor."""
service_func_mock = AsyncMock()
hass.services.async_register(LIGHT_DOMAIN, SERVICE_TURN_ON, service_func_mock)
async def intercept_func(call: ServiceCall, data: ServiceData):
data["test1"] = "changed"
data["test2"] = "added"
setup_service_call_interceptor(
hass,
LIGHT_DOMAIN,
SERVICE_TURN_ON,
intercept_func,
)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{"test1": "initial"},
blocking=True,
)
(service_call,) = service_func_mock.call_args[0]
assert service_call.data == {"test1": "changed", "test2": "added"}
assert isinstance(service_call.data, ReadOnlyDict)

212
tests/test_switch.py
View file

@ -6,7 +6,8 @@ import datetime
import itertools
import logging
from random import randint
from unittest.mock import MagicMock, patch
from typing import Any
from unittest.mock import MagicMock, Mock, patch
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -31,12 +32,14 @@ from homeassistant.const import (
ATTR_SUPPORTED_FEATURES,
CONF_LIGHTS,
CONF_NAME,
EVENT_CALL_SERVICE,
EVENT_STATE_CHANGED,
SERVICE_TOGGLE,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import Context, HomeAssistant, State
from homeassistant.core import Context, Event, HomeAssistant, State
from homeassistant.helpers import entity_registry
from homeassistant.helpers.entity_platform import async_get_platforms
from homeassistant.setup import async_setup_component
@ -86,12 +89,15 @@ from custom_components.adaptive_lighting.const import (
)
from custom_components.adaptive_lighting.switch import (
_SUPPORT_OPTS,
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION,
VALID_COLOR_MODES,
AdaptiveSwitch,
_attributes_have_changed,
_supported_features,
color_difference_redmean,
create_context,
is_our_context,
is_our_context_id,
)
_LOGGER = logging.getLogger(__name__)
@ -118,6 +124,7 @@ LAT_LONG_TZS = [
]
ENTITY_LIGHT = "light.bed_light"
ENTITY_LIGHT3 = "light.kitchen_lights"
_SWITCH_FMT = f"{SWITCH_DOMAIN}.{DOMAIN}"
ENTITY_SWITCH = f"{_SWITCH_FMT}_{DEFAULT_NAME}"
ENTITY_SLEEP_MODE_SWITCH = f"{_SWITCH_FMT}_sleep_mode_{DEFAULT_NAME}"
@ -143,9 +150,16 @@ def reset_time_zone():
dt_util.DEFAULT_TIME_ZONE = ORIG_TIMEZONE
async def setup_switch(hass, extra_data):
async def setup_switch(hass, extra_data) -> tuple[MockConfigEntry, AdaptiveSwitch]:
"""Create the switch entry."""
entry = MockConfigEntry(domain=DOMAIN, data={CONF_NAME: DEFAULT_NAME, **extra_data})
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_NAME: DEFAULT_NAME,
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION: False,
**extra_data,
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
@ -190,6 +204,7 @@ async def setup_lights(hass: HomeAssistant):
"set_level": None,
"set_temperature": None,
"set_color": None,
"supports_transition_template": True,
},
},
},
@ -212,7 +227,7 @@ async def setup_lights(hass: HomeAssistant):
return lights
async def setup_lights_and_switch(hass, extra_conf=None):
async def setup_lights_and_switch(hass, extra_conf=None, all_lights: bool = False):
"""Create switch and demo lights."""
# Setup demo lights and turn on
lights_instances = await setup_lights(hass)
@ -228,6 +243,10 @@ async def setup_lights_and_switch(hass, extra_conf=None):
ENTITY_LIGHT,
"light.ceiling_lights",
]
if all_lights:
lights.append(ENTITY_LIGHT3)
assert all(hass.states.get(light) is not None for light in lights)
_, switch = await setup_switch(
hass,
@ -421,7 +440,7 @@ async def test_adaptive_lighting_time_zones_and_sun_settings(
async def test_light_settings(hass):
"""Test that light settings are correctly applied."""
switch, _ = await setup_lights_and_switch(hass)
lights = switch._lights
lights = switch.lights
# Turn on "sleep mode"
await hass.services.async_call(
@ -525,7 +544,7 @@ async def test_turn_on_off_listener_not_tracking_untracked_lights(hass):
"""Test that lights that are not in a Adaptive Lighting switch aren't tracked."""
switch, _ = await setup_lights_and_switch(hass)
light = "light.kitchen_lights"
assert light not in switch._lights
assert light not in switch.lights
for state in [True, False]:
await hass.services.async_call(
LIGHT_DOMAIN,
@ -757,11 +776,11 @@ async def test_manual_control(hass):
await switch.adapt_brightness_switch.async_turn_on()
# Check that when no lights are specified, all are reset
await change_manual_control(True, {CONF_LIGHTS: switch._lights})
assert all([manual_control[eid] for eid in switch._lights])
await change_manual_control(True, {CONF_LIGHTS: switch.lights})
assert all([manual_control[eid] for eid in switch.lights])
# do not pass "lights" so reset all
await change_manual_control(False, {})
assert all([not manual_control[eid] for eid in switch._lights])
assert all([not manual_control[eid] for eid in switch.lights])
@pytest.mark.dependency(depends=[*GLOBAL_TEST_DEPENDENCIES, "test_manual_control"])
@ -820,7 +839,7 @@ async def test_apply_service(hass):
"""Test adaptive_lighting.apply service."""
switch, (_, _, light) = await setup_lights_and_switch(hass)
entity_id = light.entity_id
assert entity_id not in switch._lights
assert entity_id not in switch.lights
def increased_brightness():
return (light._attr_brightness + 100) % 255
@ -1074,7 +1093,7 @@ async def test_state_change_handlers(hass):
current_service_data = switch.turn_on_off_listener.last_service_data
assert current_service_data != last_service_data
for light in switch._lights:
for light in switch.lights:
# current_service_data should have changed after the last update.
assert current_service_data.get(light)
assert last_service_data.get(light)
@ -1359,7 +1378,7 @@ async def test_change_switch_settings_service(hass):
"""Test adaptive_lighting.change_switch_settings service."""
switch, (_, _, light) = await setup_lights_and_switch(hass)
entity_id = light.entity_id
assert entity_id not in switch._lights
assert entity_id not in switch.lights
async def change_switch_settings(**kwargs):
await hass.services.async_call(
@ -1427,7 +1446,7 @@ async def test_cancellable_service_calls_task(hass):
0,
_create_service_call_data_iterator(hass, [service_data]),
)
await switch._execute_cancellable_adaptation_calls(adaptation_data)
await switch.execute_cancellable_adaptation_calls(adaptation_data)
task = switch.turn_on_off_listener.adaptation_tasks.get(light.entity_id)
assert task is not None
@ -1451,3 +1470,168 @@ async def test_service_calls_task_cancellation(hass):
pass
assert task.cancelled()
async def _turn_on_and_track_event_contexts(
hass: HomeAssistant, context_id: str, entity_id
):
context = Context(id=context_id)
event_context_ids = []
async def turn_on_off_event_listener(event: Event) -> None:
event_context_ids.append(event.context.id)
hass.bus.async_listen(EVENT_CALL_SERVICE, turn_on_off_event_listener)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
context=context,
)
await hass.async_block_till_done()
return event_context_ids
def _mock_sun_light_settings(switch: AdaptiveSwitch, settings: dict[str, Any]):
sun_light_settings_mock = Mock()
sun_light_settings_mock.get_settings = Mock(return_value=settings)
switch._sun_light_settings = sun_light_settings_mock
async def test_proactive_adaptation(hass):
"""Validate that a proactive adaptation updates the original service call."""
switch, _ = await setup_lights_and_switch(
hass, {INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION: True}, True
)
_mock_sun_light_settings(
switch,
{
ATTR_BRIGHTNESS_PCT: 67,
ATTR_COLOR_TEMP_KELVIN: 3448,
},
)
event_context_ids = await _turn_on_and_track_event_contexts(
hass, "test_context", ENTITY_LIGHT3
)
# Expect a single service call
assert len(event_context_ids) == 1
assert event_context_ids == ["test_context"]
# Expect adapted light state
state = hass.states.get(ENTITY_LIGHT3)
# Sun light settings use %, state only contains absolute
assert state.attributes[ATTR_BRIGHTNESS] == 171 # == 67%
assert state.attributes[ATTR_COLOR_TEMP_KELVIN] == 3448
async def test_proactive_adaptation_with_separate_commands(hass):
"""Validate that a split proactive adaptation yields one additional service call."""
switch, _ = await setup_lights_and_switch(
hass,
{
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION: True,
CONF_SEPARATE_TURN_ON_COMMANDS: True,
},
True,
)
_mock_sun_light_settings(
switch,
{
ATTR_BRIGHTNESS_PCT: 67,
ATTR_COLOR_TEMP_KELVIN: 3448,
},
)
event_context_ids = await _turn_on_and_track_event_contexts(
hass, "test_context", ENTITY_LIGHT3
)
# Expect two service calls
assert len(event_context_ids) == 2
assert event_context_ids[0] == "test_context"
assert is_our_context_id(event_context_ids[1])
# Expect adapted light state
state = hass.states.get(ENTITY_LIGHT3)
assert state.attributes[ATTR_BRIGHTNESS] == 171
assert state.attributes[ATTR_COLOR_TEMP_KELVIN] == 3448
async def test_proactive_adaptation_toggle(hass):
"""Validate that a proactive adaptation updates service calls which toggle a light on,
but not those which toggle off.
This test is based on the fact that contexts of proactive adaptations are recorded.
"""
switch, _ = await setup_lights_and_switch(
hass, {INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION: True}, True
)
# Toggle ON
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TOGGLE,
{ATTR_ENTITY_ID: ENTITY_LIGHT3},
blocking=True,
context=Context(id="test1"),
)
assert switch.turn_on_off_listener.is_proactively_adapting("test1")
# Toggle OFF
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TOGGLE,
{ATTR_ENTITY_ID: ENTITY_LIGHT3},
blocking=True,
context=Context(id="test2"),
)
assert not switch.turn_on_off_listener.is_proactively_adapting("test2")
async def test_proactive_adaptation_transition_override(hass):
"""Validate that transitions in service calls are preferred over the default transition."""
switch, (_, _, light3) = await setup_lights_and_switch(
hass,
{
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION: True,
CONF_INITIAL_TRANSITION: 123,
},
True,
)
with patch.object(
light3, "async_turn_on", wraps=light3.async_turn_on
) as patched_async_turn_on:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: ENTITY_LIGHT3},
blocking=True,
)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: ENTITY_LIGHT3, ATTR_TRANSITION: 456},
blocking=True,
)
# Assert that default is used when no transition is specified in service call
kwargs = patched_async_turn_on.call_args_list[0].kwargs
assert set({ATTR_TRANSITION: 123}.items()).issubset(kwargs.items())
# Assert that specified service call transition takes precedence over default
kwargs = patched_async_turn_on.call_args_list[1].kwargs
assert set({ATTR_TRANSITION: 456}.items()).issubset(kwargs.items())
# Cleanup
switch.turn_on_off_listener.cancel_ongoing_adaptation_calls(ENTITY_LIGHT3)