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

Merge branch 'main' into manual_control_lights_turned_on_with_custom_service_data

Browse source
This commit is contained in:
Benjamin Auquite 2023-04-10 13:00:25 -05:00
commit 93e40e4f7f
14 changed files with 674 additions and 243 deletions
Download patch file Download diff file Expand all files Collapse all files

2
custom_components/adaptive_lighting/const.py
View file

@ -290,6 +290,8 @@ VALIDATION_TUPLES = [
),
]
CONST_COLOR = "color"
def timedelta_as_int(value):
"""Convert a `datetime.timedelta` object to an integer.

4
custom_components/adaptive_lighting/manifest.json
View file

@ -7,6 +7,6 @@
"documentation": "https://github.com/basnijholt/adaptive-lighting#readme",
"iot_class": "calculated",
"issue_tracker": "https://github.com/basnijholt/adaptive-lighting/issues",
"requirements": [],
"version": "1.10.0"
"requirements": ["ulid-transform"],
"version": "1.11.0"
}

387
custom_components/adaptive_lighting/switch.py
View file

@ -23,7 +23,11 @@ from homeassistant.components.light import (
ATTR_COLOR_NAME,
ATTR_COLOR_TEMP_KELVIN,
ATTR_HS_COLOR,
ATTR_MAX_COLOR_TEMP_KELVIN,
ATTR_MIN_COLOR_TEMP_KELVIN,
ATTR_RGB_COLOR,
ATTR_RGBW_COLOR,
ATTR_RGBWW_COLOR,
ATTR_SUPPORTED_COLOR_MODES,
ATTR_TRANSITION,
ATTR_XY_COLOR,
@ -32,6 +36,7 @@ from homeassistant.components.light import (
COLOR_MODE_HS,
COLOR_MODE_RGB,
COLOR_MODE_RGBW,
COLOR_MODE_RGBWW,
COLOR_MODE_XY,
)
from homeassistant.components.light import (
@ -85,8 +90,10 @@ from homeassistant.util.color import (
color_RGB_to_xy,
color_temperature_to_rgb,
color_xy_to_hs,
color_xy_to_RGB,
)
import homeassistant.util.dt as dt_util
import ulid_transform
import voluptuous as vol
from .const import (
@ -128,6 +135,7 @@ from .const import (
CONF_TRANSITION,
CONF_TURN_ON_LIGHTS,
CONF_USE_DEFAULTS,
CONST_COLOR,
DOMAIN,
EXTRA_VALIDATION,
ICON_BRIGHTNESS,
@ -154,6 +162,16 @@ _SUPPORT_OPTS = {
"transition": SUPPORT_TRANSITION,
}
VALID_COLOR_MODES = {
COLOR_MODE_BRIGHTNESS: ATTR_BRIGHTNESS,
COLOR_MODE_COLOR_TEMP: ATTR_COLOR_TEMP_KELVIN,
COLOR_MODE_HS: ATTR_HS_COLOR,
COLOR_MODE_RGB: ATTR_RGB_COLOR,
COLOR_MODE_RGBW: ATTR_RGBW_COLOR,
COLOR_MODE_RGBWW: ATTR_RGBWW_COLOR,
COLOR_MODE_XY: ATTR_XY_COLOR,
}
_ORDER = (SUN_EVENT_SUNRISE, SUN_EVENT_NOON, SUN_EVENT_SUNSET, SUN_EVENT_MIDNIGHT)
_ALLOWED_ORDERS = {_ORDER[i:] + _ORDER[:i] for i in range(len(_ORDER))}
@ -182,21 +200,58 @@ BRIGHTNESS_ATTRS = {
}
# Keep a short domain version for the context instances (which can only be 36 chars)
_DOMAIN_SHORT = "adapt_lgt"
_DOMAIN_SHORT = "al"
def _int_to_bytes(i: int, signed: bool = False) -> bytes:
bits = i.bit_length()
if signed:
# Make room for the sign bit.
bits += 1
return i.to_bytes((bits + 7) // 8, "little", signed=signed)
def _int_to_base36(num: int) -> str:
"""
Convert an integer to its base-36 representation using numbers and uppercase letters.
Base-36 encoding uses digits 0-9 and uppercase letters A-Z, providing a case-insensitive
alphanumeric representation. The function takes an integer `num` as input and returns
its base-36 representation as a string.
Parameters
----------
num
The integer to convert to base-36.
Returns
-------
str
The base-36 representation of the input integer.
Examples
--------
>>> num = 123456
>>> base36_num = int_to_base36(num)
>>> print(base36_num)
'2N9'
"""
ALPHANUMERIC_CHARS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
if num == 0:
return ALPHANUMERIC_CHARS[0]
base36_str = ""
base = len(ALPHANUMERIC_CHARS)
while num:
num, remainder = divmod(num, base)
base36_str = ALPHANUMERIC_CHARS[remainder] + base36_str
return base36_str
def _short_hash(string: str, length: int = 4) -> str:
"""Create a hash of 'string' with length 'length'."""
str_hash_bytes = _int_to_bytes(hash(string), signed=True)
return base64.b85encode(str_hash_bytes)[:length]
return base64.b32encode(string.encode()).decode("utf-8").zfill(length)[:length]
def _remove_vowels(input_str: str, length: int = 4) -> str:
vowels = "aeiouAEIOU"
output_str = "".join([char for char in input_str if char not in vowels])
return output_str.zfill(length)[:length]
def create_context(
@ -204,12 +259,16 @@ def create_context(
) -> Context:
"""Create a context that can identify this integration."""
# Use a hash for the name because otherwise the context might become
# too long (max len == 36) to fit in the database.
name_hash = _short_hash(name)
# too long (max len == 26) to fit in the database.
# Pack index with base85 to maximize the number of contexts we can create
# before we exceed the 36-character limit and are forced to wrap.
index_packed = base64.b85encode(_int_to_bytes(index, signed=False))
context_id = f"{_DOMAIN_SHORT}:{name_hash}:{which}:{index_packed}"[:36]
# before we exceed the 26-character limit and are forced to wrap.
time_stamp = ulid_transform.ulid_now()[:10] # time part of a ULID
name_hash = _short_hash(name)
which_short = _remove_vowels(which)
context_id_start = f"{time_stamp}:{_DOMAIN_SHORT}:{name_hash}:{which_short}:"
chars_left = 26 - len(context_id_start)
index_packed = _int_to_base36(index).zfill(chars_left)[-chars_left:]
context_id = context_id_start + index_packed
parent_id = parent.id if parent else None
return Context(id=context_id, parent_id=parent_id)
@ -218,7 +277,7 @@ def is_our_context(context: Context | None) -> bool:
"""Check whether this integration created 'context'."""
if context is None:
return False
return context.id.startswith(_DOMAIN_SHORT)
return f":{_DOMAIN_SHORT}:" in context.id
def _split_service_data(service_data, adapt_brightness, adapt_color):
@ -382,6 +441,7 @@ def _fire_manual_control_event(
switch.entity_id,
light,
)
switch.turn_on_off_listener.mark_as_manual_control(light)
fire(
f"{DOMAIN}.manual_control",
{ATTR_ENTITY_ID: light, SWITCH_DOMAIN: switch.entity_id},
@ -477,7 +537,6 @@ async def async_setup_entry(
all_lights = _expand_light_groups(switch.hass, lights)
if service_call.data[CONF_MANUAL_CONTROL]:
for light in all_lights:
switch.turn_on_off_listener.mark_as_manual_control(light)
_fire_manual_control_event(switch, light, service_call.context)
else:
switch.turn_on_off_listener.reset(*all_lights)
@ -581,33 +640,51 @@ def _expand_light_groups(hass: HomeAssistant, lights: list[str]) -> list[str]:
return list(all_lights)
def _supported_to_attributes(supported):
supported_attributes = {}
supports_colors = False
for mode, attr in VALID_COLOR_MODES.items():
if mode not in supported:
continue
supported_attributes[attr] = True
if (
not supports_colors
and mode != COLOR_MODE_BRIGHTNESS
and mode != COLOR_MODE_COLOR_TEMP
):
supports_colors = True
return supported_attributes, supports_colors
def _supported_features(hass: HomeAssistant, light: str):
state = hass.states.get(light)
supported_features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
supported = {
key for key, value in _SUPPORT_OPTS.items() if supported_features & value
legacy_supported_features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
legacy_supported = {
key for key, value in _SUPPORT_OPTS.items() if legacy_supported_features & value
}
supported_color_modes = state.attributes.get(ATTR_SUPPORTED_COLOR_MODES, set())
if COLOR_MODE_RGB in supported_color_modes:
supported.add("color")
supported, supports_colors = _supported_to_attributes(
legacy_supported.union(supported_color_modes)
)
min_kelvin = state.attributes.get(ATTR_MIN_COLOR_TEMP_KELVIN)
max_kelvin = state.attributes.get(ATTR_MAX_COLOR_TEMP_KELVIN)
supported.update(
{
ATTR_MIN_COLOR_TEMP_KELVIN: min_kelvin,
ATTR_MAX_COLOR_TEMP_KELVIN: max_kelvin,
}
)
if supports_colors:
# Adding brightness here, see
# comment https://github.com/basnijholt/adaptive-lighting/issues/112#issuecomment-836944011
supported.add("brightness")
if COLOR_MODE_RGBW in supported_color_modes:
supported.add("color")
supported.add("brightness") # see above url
if COLOR_MODE_XY in supported_color_modes:
supported.add("color")
supported.add("brightness") # see above url
if COLOR_MODE_HS in supported_color_modes:
supported.add("color")
supported.add("brightness") # see above url
if COLOR_MODE_COLOR_TEMP in supported_color_modes:
supported.add("color_temp")
supported.add("brightness") # see above url
if COLOR_MODE_BRIGHTNESS in supported_color_modes:
supported.add("brightness")
return supported
supported[ATTR_BRIGHTNESS] = True
if CONST_COLOR not in legacy_supported:
# supports_colors = False
_LOGGER.debug(
"'supported_color_modes' supports color but the legacy 'supported_features'"
" bitfield says we do not. Despite this we'll assume light '%s' supports colors",
)
return supported, supports_colors
def color_difference_redmean(
@ -629,6 +706,41 @@ def color_difference_redmean(
return math.sqrt(red_term + green_term + blue_term)
# All comparisons should be done with RGB since
# converting anything to color temp is inaccurate.
def _convert_attributes(attributes: dict[str, Any]) -> dict[str, Any]:
if ATTR_RGB_COLOR in attributes:
return attributes
rgb = None
if ATTR_COLOR_TEMP_KELVIN in attributes:
rgb = color_temperature_to_rgb(attributes[ATTR_COLOR_TEMP_KELVIN])
elif ATTR_XY_COLOR in attributes:
rgb = color_xy_to_RGB(*attributes[ATTR_XY_COLOR])
if rgb is not None:
attributes[ATTR_RGB_COLOR] = rgb
_LOGGER.debug(f"Converted {attributes} to rgb {rgb}")
else:
_LOGGER.debug("No suitable conversion found")
return attributes
def _add_missing_attributes(
old_attributes: dict[str, Any],
new_attributes: dict[str, Any],
) -> dict[str, Any]:
if not any(
attr in old_attributes and attr in new_attributes
for attr in [ATTR_COLOR_TEMP_KELVIN, ATTR_RGB_COLOR]
):
old_attributes = _convert_attributes(old_attributes)
new_attributes = _convert_attributes(new_attributes)
return old_attributes, new_attributes
def _attributes_have_changed(
light: str,
old_attributes: dict[str, Any],
@ -637,6 +749,11 @@ def _attributes_have_changed(
adapt_color: bool,
context: Context,
) -> bool:
if adapt_color:
old_attributes, new_attributes = _add_missing_attributes(
old_attributes, new_attributes
)
if (
adapt_brightness
and ATTR_BRIGHTNESS in old_attributes
@ -691,21 +808,6 @@ def _attributes_have_changed(
context.id,
)
return True
switched_color_temp = (
ATTR_RGB_COLOR in old_attributes and ATTR_RGB_COLOR not in new_attributes
)
switched_to_rgb_color = (
ATTR_COLOR_TEMP_KELVIN in old_attributes
and ATTR_COLOR_TEMP_KELVIN not in new_attributes
)
if switched_color_temp or switched_to_rgb_color:
# Light switched from RGB mode to color_temp or visa versa
_LOGGER.debug(
"'%s' switched from RGB mode to color_temp or visa versa",
light,
)
return True
return False
@ -950,6 +1052,12 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
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
if (timer := timers.get(light)) and (time := timer.remaining_time()) > 0
}
return extra_state_attributes
def create_context(
@ -1018,9 +1126,6 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if lock is not None and lock.locked():
_LOGGER.debug("%s: '%s' is locked", self._name, light)
return
service_data = {ATTR_ENTITY_ID: light}
features = _supported_features(self.hass, light)
if transition is None:
transition = self._transition
if adapt_brightness is None:
@ -1030,16 +1135,19 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if prefer_rgb_color is None:
prefer_rgb_color = self._prefer_rgb_color
# Check transition == 0 to fix #378
if "transition" in features and transition > 0:
service_data[ATTR_TRANSITION] = transition
# 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
)
if "brightness" in features and adapt_brightness:
# Build service data.
service_data = {ATTR_ENTITY_ID: light}
features, supports_colors = _supported_features(self.hass, light)
# Check transition == 0 to fix #378
if ATTR_TRANSITION in features and transition > 0:
service_data[ATTR_TRANSITION] = transition
if ATTR_BRIGHTNESS in features and adapt_brightness:
brightness = round(255 * self._settings["brightness_pct"] / 100)
service_data[ATTR_BRIGHTNESS] = brightness
@ -1048,39 +1156,26 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
and self._sun_light_settings.sleep_rgb_or_color_temp == "rgb_color"
)
if (
"color_temp" in features
ATTR_COLOR_TEMP_KELVIN in features
and adapt_color
and not (prefer_rgb_color and "color" in features)
and not (sleep_rgb and "color" in features)
and not (prefer_rgb_color and supports_colors)
and not (sleep_rgb and supports_colors)
):
_LOGGER.debug("%s: Setting color_temp of light %s", self._name, light)
attributes = self.hass.states.get(light).attributes
min_kelvin = attributes["min_color_temp_kelvin"]
max_kelvin = attributes["max_color_temp_kelvin"]
min_kelvin = features[ATTR_MIN_COLOR_TEMP_KELVIN]
max_kelvin = features[ATTR_MAX_COLOR_TEMP_KELVIN]
color_temp_kelvin = self._settings["color_temp_kelvin"]
color_temp_kelvin = max(min(color_temp_kelvin, max_kelvin), min_kelvin)
service_data[ATTR_COLOR_TEMP_KELVIN] = color_temp_kelvin
elif "color" in features and adapt_color:
elif supports_colors and adapt_color:
_LOGGER.debug("%s: Setting rgb_color of light %s", self._name, light)
service_data[ATTR_RGB_COLOR] = self._settings["rgb_color"]
context = context or self.create_context("adapt_lights")
if (
self._take_over_control
and self._detect_non_ha_changes
and not force
and await self.turn_on_off_listener.significant_change(
self,
light,
adapt_brightness,
adapt_color,
context,
)
):
return
# See #80. Doesn't check if transitions differ but it does the job.
last_service_data = self.turn_on_off_listener.last_service_data
if last_service_data.get(light) == service_data:
if not force and last_service_data.get(light) == service_data:
_LOGGER.debug(
"%s: Cancelling adapt to light %s, there's no new values to set (context.id='%s')",
self._name,
@ -1145,21 +1240,32 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if lights is None:
lights = self._lights
if not force and self._only_once:
return
filtered_lights = []
for light in lights:
# Don't adapt lights that haven't finished prior transitions.
if force or not self.turn_on_off_listener.transition_timers.get(light):
filtered_lights.append(light)
if not force:
if self._only_once:
return
for light in lights:
# Don't adapt lights that haven't finished prior transitions.
timer = self.turn_on_off_listener.transition_timers.get(light)
if timer is not None and timer.is_running():
_LOGGER.debug(
"%s: Light '%s' is still transitioning",
self._name,
light,
)
else:
filtered_lights.append(light)
else:
filtered_lights = lights
if not filtered_lights:
return
await self._adapt_lights(filtered_lights, transition, force, context)
await self._update_manual_control_and_maybe_adapt(
filtered_lights, transition, force, context
)
async def _adapt_lights(
async def _update_manual_control_and_maybe_adapt(
self,
lights: list[str],
transition: int | None,
@ -1168,34 +1274,53 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
) -> None:
assert context is not None
_LOGGER.debug(
"%s: '_adapt_lights(%s, %s, force=%s, context.id=%s)' called",
"%s: '_update_manual_control_and_maybe_adapt(%s, %s, force=%s, context.id=%s)' called",
self.name,
lights,
transition,
force,
context.id,
)
adapt_brightness = self.adapt_brightness_switch.is_on
adapt_color = self.adapt_color_switch.is_on
for light in lights:
if not is_on(self.hass, light):
continue
if (
self._take_over_control
and self.turn_on_off_listener.is_manually_controlled(
manually_controlled = self.turn_on_off_listener.is_manually_controlled(
self,
light,
force,
adapt_brightness,
adapt_color,
)
significant_change = (
self._detect_non_ha_changes
and not force
and await self.turn_on_off_listener.significant_change(
self,
light,
force,
self.adapt_brightness_switch.is_on,
self.adapt_color_switch.is_on,
adapt_brightness,
adapt_color,
context,
)
):
_LOGGER.debug(
"%s: '%s' is being manually controlled, stop adapting, context.id=%s.",
self._name,
light,
context.id,
)
continue
await self._adapt_light(light, transition, force=force, context=context)
)
if self._take_over_control and (manually_controlled or significant_change):
if manually_controlled:
_LOGGER.debug(
"%s: '%s' is being manually controlled, stop adapting, context.id=%s.",
self._name,
light,
context.id,
)
else:
_fire_manual_control_event(self, light, context)
else:
await self._adapt_light(light, transition, force=force, context=context)
async def _sleep_mode_switch_state_event(self, event: Event) -> None:
if not match_switch_state_event(event, (STATE_ON, STATE_OFF)):
@ -1598,33 +1723,28 @@ class TurnOnOffListener:
def start_transition_timer(self, light: str) -> None:
"""Mark a light as manually controlled."""
_LOGGER.debug("Start transition timer for %s", light)
last_service_data = self.last_service_data
if (
not last_service_data
or light not in last_service_data
or ATTR_TRANSITION not in last_service_data[light]
):
last_service_data = self.last_service_data.get(light)
if not last_service_data:
_LOGGER.debug("This should not ever happen. Please report to the devs.")
return
delay = last_service_data[light][ATTR_TRANSITION]
last_transition = last_service_data.get(ATTR_TRANSITION)
if not last_transition:
_LOGGER.debug(
"No transition in last adapt for light %s, continuing...", light
)
return
_LOGGER.debug(
"Start transition timer of %s seconds for light %s", last_transition, light
)
async def reset():
ValueError("TEST")
_LOGGER.debug(
"Transition finished for light %s",
light,
)
switches = _get_switches_with_lights(self.hass, [light])
for switch in switches:
if not switch.is_on:
continue
await switch._update_attrs_and_maybe_adapt_lights(
[light],
force=False,
context=switch.create_context("transit"),
)
self._handle_timer(light, self.transition_timers, delay, reset)
self._handle_timer(light, self.transition_timers, last_transition, reset)
def set_auto_reset_manual_control_times(self, lights: list[str], time: float):
"""Set the time after which the lights are automatically reset."""
@ -1778,7 +1898,7 @@ class TurnOnOffListener:
async def state_changed_event_listener(self, event: Event) -> None:
"""Track 'state_changed' events."""
entity_id = event.data.get(ATTR_ENTITY_ID, "")
if entity_id not in self.lights or entity_id.split(".")[0] != LIGHT_DOMAIN:
if entity_id not in self.lights:
return
new_state = event.data.get("new_state")
@ -1823,6 +1943,10 @@ class TurnOnOffListener:
entity_id,
)
self.last_state_change[entity_id] = [new_state]
_LOGGER.debug(
"Last transition: %s",
self.last_service_data[entity_id].get(ATTR_TRANSITION),
)
self.start_transition_timer(entity_id)
elif old_state is not None:
self.last_state_change[entity_id].append(new_state)
@ -1853,7 +1977,7 @@ class TurnOnOffListener:
):
# Light was already on and 'light.turn_on' was not called by
# the adaptive_lighting integration.
manual_control = self.mark_as_manual_control(light)
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"
@ -1921,8 +2045,6 @@ class TurnOnOffListener:
light,
context.id,
)
self.mark_as_manual_control(light)
_fire_manual_control_event(switch, light, context, is_async=False)
return True
_LOGGER.debug(
"%s: Light '%s' correctly matches our last adapt's service data, continuing..."
@ -2065,3 +2187,10 @@ class _AsyncSingleShotTimer:
if self.task:
self.task.cancel()
self.callback = None
def remaining_time(self):
"""Return the remaining time before the timer expires."""
if self.start_time is not None:
elapsed_time = (dt_util.utcnow() - self.start_time).total_seconds()
return max(0, self.delay - elapsed_time)
return 0