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

Code cleanup (#1348)

Browse source
This commit is contained in:
Mario Guggenberger 2025-12-12 22:37:42 +01:00 • committed by GitHub
commit 5886eee04c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 155 additions and 147 deletions
Download patch file Download diff file Expand all files Collapse all files

2
custom_components/adaptive_lighting/__init__.py
View file

@ -10,7 +10,7 @@ from homeassistant.const import CONF_SOURCE
from homeassistant.core import Event, HomeAssistant
from .const import (
_DOMAIN_SCHEMA,
_DOMAIN_SCHEMA, # pyright: ignore[reportPrivateUsage]
ATTR_ADAPTIVE_LIGHTING_MANAGER,
CONF_NAME,
DOMAIN,

76
custom_components/adaptive_lighting/color_and_brightness.py
View file

@ -9,6 +9,7 @@ import logging
import math
from dataclasses import dataclass
from datetime import UTC, timedelta
from enum import Enum
from functools import cached_property, partial
from typing import TYPE_CHECKING, Any, Literal, cast
@ -21,15 +22,19 @@ from homeassistant.util.color import (
if TYPE_CHECKING:
import astral.location
# Same as homeassistant.const.SUN_EVENT_SUNRISE and homeassistant.const.SUN_EVENT_SUNSET
# We re-define them here to not depend on homeassistant in this file.
SUN_EVENT_SUNRISE = "sunrise"
SUN_EVENT_SUNSET = "sunset"
SUN_EVENT_NOON = "solar_noon"
SUN_EVENT_MIDNIGHT = "solar_midnight"
class SunEvent(str, Enum):
"""A set of sun events that happen during a day."""
_ORDER = (SUN_EVENT_SUNRISE, SUN_EVENT_NOON, SUN_EVENT_SUNSET, SUN_EVENT_MIDNIGHT)
# Same as homeassistant.const.SUN_EVENT_SUNRISE and homeassistant.const.SUN_EVENT_SUNSET
# We re-define them here to not depend on homeassistant in this file.
SUNRISE = "sunrise"
SUNSET = "sunset"
NOON = "solar_noon"
MIDNIGHT = "solar_midnight"
_ORDER = (SunEvent.SUNRISE, SunEvent.NOON, SunEvent.SUNSET, SunEvent.MIDNIGHT)
_ALLOWED_ORDERS = {_ORDER[i:] + _ORDER[:i] for i in range(len(_ORDER))}
utcnow: partial[datetime.datetime] = partial(datetime.datetime.now, UTC)
@ -126,21 +131,21 @@ class SunEvents:
noon = midnight + timedelta(hours=12) * (1 if midnight.hour < 12 else -1)
return noon, midnight
def sun_events(self, dt: datetime.datetime) -> list[tuple[str, float]]:
def sun_events(self, dt: datetime.datetime) -> list[tuple[SunEvent, float]]:
"""Get the four sun event's timestamps at 'dt'."""
sunrise = self.sunrise(dt)
sunset = self.sunset(dt)
solar_noon, solar_midnight = self.noon_and_midnight(dt, sunset, sunrise)
events = [
(SUN_EVENT_SUNRISE, sunrise.timestamp()),
(SUN_EVENT_SUNSET, sunset.timestamp()),
(SUN_EVENT_NOON, solar_noon.timestamp()),
(SUN_EVENT_MIDNIGHT, solar_midnight.timestamp()),
events: list[tuple[SunEvent, float]] = [
(SunEvent.SUNRISE, sunrise.timestamp()),
(SunEvent.SUNSET, sunset.timestamp()),
(SunEvent.NOON, solar_noon.timestamp()),
(SunEvent.MIDNIGHT, solar_midnight.timestamp()),
]
self._validate_sun_event_order(events)
return events
def _validate_sun_event_order(self, events: list[tuple[str, float]]) -> None:
def _validate_sun_event_order(self, events: list[tuple[SunEvent, float]]) -> None:
"""Check if the sun events are in the expected order."""
events = sorted(events, key=lambda x: x[1])
events_names, _ = zip(*events, strict=True)
@ -154,7 +159,10 @@ class SunEvents:
_LOGGER.error(msg)
raise ValueError(msg)
def prev_and_next_events(self, dt: datetime.datetime) -> list[tuple[str, float]]:
def prev_and_next_events(
self,
dt: datetime.datetime,
) -> list[tuple[SunEvent, float]]:
"""Get the previous and next sun event."""
events = [
event
@ -171,23 +179,26 @@ class SunEvents:
(_, prev_ts), (next_event, next_ts) = self.prev_and_next_events(dt)
h, x = (
(prev_ts, next_ts)
if next_event in (SUN_EVENT_SUNSET, SUN_EVENT_SUNRISE)
if next_event in (SunEvent.SUNSET, SunEvent.SUNRISE)
else (next_ts, prev_ts)
)
# k = -1 between sunset and sunrise (sun below horizon)
# k = 1 between sunrise and sunset (sun above horizon)
k = 1 if next_event in (SUN_EVENT_SUNSET, SUN_EVENT_NOON) else -1
k = 1 if next_event in (SunEvent.SUNSET, SunEvent.NOON) else -1
return k * (1 - ((target_ts - h) / (h - x)) ** 2)
def closest_event(self, dt: datetime.datetime) -> tuple[str, float]:
def closest_event(
self,
dt: datetime.datetime,
) -> tuple[Literal[SunEvent.SUNRISE, SunEvent.SUNSET], float]:
"""Get the closest sunset or sunrise event."""
(prev_event, prev_ts), (next_event, next_ts) = self.prev_and_next_events(dt)
if SUN_EVENT_SUNRISE in (prev_event, next_event):
ts_event = prev_ts if prev_event == SUN_EVENT_SUNRISE else next_ts
return SUN_EVENT_SUNRISE, ts_event
if SUN_EVENT_SUNSET in (prev_event, next_event):
ts_event = prev_ts if prev_event == SUN_EVENT_SUNSET else next_ts
return SUN_EVENT_SUNSET, ts_event
if SunEvent.SUNRISE in (prev_event, next_event):
ts_event = prev_ts if prev_event == SunEvent.SUNRISE else next_ts
return SunEvent.SUNRISE, ts_event
if SunEvent.SUNSET in (prev_event, next_event):
ts_event = prev_ts if prev_event == SunEvent.SUNSET else next_ts
return SunEvent.SUNSET, ts_event
msg = "No sunrise or sunset event found."
raise ValueError(msg)
@ -249,7 +260,7 @@ class SunLightSettings:
event, ts_event = self.sun.closest_event(dt)
dark = self.brightness_mode_time_dark.total_seconds()
light = self.brightness_mode_time_light.total_seconds()
if event == SUN_EVENT_SUNRISE:
if event == SunEvent.SUNRISE:
brightness = scaled_tanh(
dt.timestamp() - ts_event,
x1=-dark,
@ -259,7 +270,7 @@ class SunLightSettings:
y_min=self.min_brightness,
y_max=self.max_brightness,
)
elif event == SUN_EVENT_SUNSET:
elif event == SunEvent.SUNSET:
brightness = scaled_tanh(
dt.timestamp() - ts_event,
x1=-light, # shifted timestamp for the start of sunset
@ -269,6 +280,9 @@ class SunLightSettings:
y_min=self.min_brightness,
y_max=self.max_brightness,
)
else:
msg = "Unsupported sun event"
raise ValueError(msg)
return clamp(brightness, self.min_brightness, self.max_brightness)
def _brightness_pct_linear(self, dt: datetime.datetime) -> float:
@ -277,7 +291,7 @@ class SunLightSettings:
# at ts_event + dt_end, brightness == end_brightness
dark = self.brightness_mode_time_dark.total_seconds()
light = self.brightness_mode_time_light.total_seconds()
if event == SUN_EVENT_SUNRISE:
if event == SunEvent.SUNRISE:
brightness = lerp(
dt.timestamp() - ts_event,
x1=-dark,
@ -285,7 +299,7 @@ class SunLightSettings:
y1=self.min_brightness,
y2=self.max_brightness,
)
elif event == SUN_EVENT_SUNSET:
elif event == SunEvent.SUNSET:
brightness = lerp(
dt.timestamp() - ts_event,
x1=-light,
@ -293,6 +307,9 @@ class SunLightSettings:
y1=self.max_brightness,
y2=self.min_brightness,
)
else:
msg = "Unsupported sun event"
raise ValueError(msg)
return clamp(brightness, self.min_brightness, self.max_brightness)
def brightness_pct(self, dt: datetime.datetime, is_sleep: bool) -> float | None:
@ -356,7 +373,8 @@ class SunLightSettings:
force_rgb_color = True
else:
color_temp_kelvin = self.color_temp_kelvin(sun_position)
rgb_color = color_temperature_to_rgb(color_temp_kelvin)
r, g, b = color_temperature_to_rgb(color_temp_kelvin)
rgb_color = (round(r), round(g), round(b))
# backwards compatibility for versions < 1.3.1 - see #403
color_temp_mired: float = math.floor(1000000 / color_temp_kelvin)
xy_color: tuple[float, float] = color_RGB_to_xy(*rgb_color)

21
custom_components/adaptive_lighting/config_flow.py
View file

@ -5,7 +5,7 @@ from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_NAME, MAJOR_VERSION, MINOR_VERSION
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
from homeassistant.helpers.selector import EntitySelector, EntitySelectorConfig
@ -96,13 +96,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
config_entry: config_entries.ConfigEntry, # noqa: ARG004
) -> "OptionsFlowHandler":
"""Get the options flow for this handler."""
if (MAJOR_VERSION, MINOR_VERSION) >= (2024, 12):
# https://github.com/home-assistant/core/pull/129651
return OptionsFlowHandler()
return OptionsFlowHandler(config_entry)
return OptionsFlowHandler()
def validate_options(user_input: dict[str, Any], errors: dict[str, str]) -> None:
@ -125,21 +122,13 @@ def validate_options(user_input: dict[str, Any], errors: dict[str, str]) -> None
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for Adaptive Lighting."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize options flow."""
if (MAJOR_VERSION, MINOR_VERSION) >= (2024, 12):
super().__init__(*args, **kwargs)
# https://github.com/home-assistant/core/pull/129651
else:
self.config_entry = args[0]
async def async_step_init(self, user_input: dict[str, Any] | None = None):
"""Handle options flow."""
conf = self.config_entry
data = validate(conf)
if conf.source == config_entries.SOURCE_IMPORT:
return self.async_show_form(step_id="init", data_schema=None)
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
validate_options(user_input, errors)
if not errors:
@ -156,7 +145,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
configured_light,
)
to_replace = {
to_replace: dict[str, Any] = {
CONF_LIGHTS: EntitySelector(
EntitySelectorConfig(
domain="light",

192
custom_components/adaptive_lighting/switch.py
View file

@ -44,8 +44,6 @@ from homeassistant.const import (
EVENT_CALL_SERVICE,
EVENT_HOMEASSISTANT_STARTED,
EVENT_STATE_CHANGED,
MAJOR_VERSION,
MINOR_VERSION,
SERVICE_TOGGLE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
@ -62,14 +60,10 @@ from homeassistant.core import (
callback,
)
from homeassistant.helpers import entity_platform, entity_registry
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_component import async_update_entity
if [MAJOR_VERSION, MINOR_VERSION] < [2023, 9]:
from homeassistant.helpers.entity import DeviceInfo
else:
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.event import (
EventStateChangedData,
async_track_state_change_event,
async_track_time_interval,
)
@ -166,6 +160,7 @@ if TYPE_CHECKING:
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import NoEventData, VolDictType
_LOGGER = logging.getLogger(__name__)
@ -228,11 +223,11 @@ def _switches_with_lights(
hass: HomeAssistant,
lights: list[str],
expand_light_groups: bool = True,
) -> list[AdaptiveSwitch]:
) -> AdaptiveSwitches:
"""Get all switches that control at least one of the lights passed."""
config_entries = hass.config_entries.async_entries(DOMAIN)
data = hass.data[DOMAIN]
switches: list[AdaptiveSwitch] = []
switches: AdaptiveSwitches = []
all_check_lights = (
_expand_light_groups(hass, lights) if expand_light_groups else set(lights)
)
@ -285,7 +280,7 @@ def _switch_with_lights(
def _switches_from_service_call(
hass: HomeAssistant,
service_call: ServiceCall,
) -> list[AdaptiveSwitch]:
) -> AdaptiveSwitches:
data = service_call.data
lights = data[CONF_LIGHTS]
switch_entity_ids: list[str] | None = data.get("entity_id")
@ -307,7 +302,7 @@ def _switches_from_service_call(
f" Invalid service data received: {service_call.data}"
)
raise ValueError(msg)
switches = []
switches: AdaptiveSwitches = []
ent_reg = entity_registry.async_get(hass)
for entity_id in switch_entity_ids:
ent_entry = ent_reg.async_get(entity_id)
@ -536,7 +531,7 @@ async def async_setup_entry( # noqa: PLR0915
schema=SET_MANUAL_CONTROL_SCHEMA,
)
args = {vol.Optional(CONF_USE_DEFAULTS, default="current"): cv.string}
args: VolDictType = {vol.Optional(CONF_USE_DEFAULTS, default="current"): cv.string}
# Modifying these after init isn't possible
skip = (CONF_INTERVAL, CONF_NAME, CONF_LIGHTS)
for k, _, valid in VALIDATION_TUPLES:
@ -583,7 +578,10 @@ def validate(
return data
def _is_state_event(event: Event, from_or_to_state: Iterable[str]) -> bool:
def _is_state_event(
event: Event[EventStateChangedData],
from_or_to_state: Iterable[str],
) -> bool:
"""Match state event when either 'from_state' or 'to_state' matches."""
return (
(old_state := event.data.get("old_state")) is not None
@ -625,7 +623,9 @@ def _is_light_group(state: State) -> bool:
def _supported_features(hass: HomeAssistant, light: str) -> set[str]:
state = hass.states.get(light)
assert state is not None
supported_features = int(state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)) # type: ignore[arg-type]
supported_features = int(
state.attributes.get(ATTR_SUPPORTED_FEATURES, 0),
) # type: ignore[arg-type]
assert isinstance(supported_features, int)
supported: set[str] = set()
@ -633,7 +633,10 @@ def _supported_features(hass: HomeAssistant, light: str) -> set[str]:
if supported_features & LightEntityFeature.TRANSITION:
supported.add("transition")
supported_color_modes = state.attributes.get(ATTR_SUPPORTED_COLOR_MODES, set()) # type: ignore[arg-type]
supported_color_modes = state.attributes.get(
ATTR_SUPPORTED_COLOR_MODES,
set(),
) # type: ignore[arg-type]
color_modes = {
ColorMode.RGB,
ColorMode.RGBW,
@ -920,7 +923,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._only_once = data[CONF_ONLY_ONCE]
self._prefer_rgb_color = data[CONF_PREFER_RGB_COLOR]
self._separate_turn_on_commands = data[CONF_SEPARATE_TURN_ON_COMMANDS]
self._transition = data[CONF_TRANSITION]
self._transition: int = data[CONF_TRANSITION]
self._adapt_delay = data[CONF_ADAPT_DELAY]
self._send_split_delay = data[CONF_SEND_SPLIT_DELAY]
self._take_over_control = data[CONF_TAKE_OVER_CONTROL]
@ -1040,29 +1043,12 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
)
self.lights = list(all_lights)
async def _setup_listeners(self, _=None) -> None:
async def _setup_listeners(self, _: Event[NoEventData] | None = None) -> None:
_LOGGER.debug("%s: Called '_setup_listeners'", self._name)
if not self.is_on or not self.hass.is_running:
_LOGGER.debug("%s: Cancelled '_setup_listeners'", self._name)
return
while not all(
sw._state is not None
for sw in [
self.sleep_mode_switch,
self.adapt_brightness_switch,
self.adapt_color_switch,
]
):
# Waits until `async_added_to_hass` is done, which in SimpleSwitch
# is when `_state` is set to `True` or `False`.
# Fixes first issue in https://github.com/basnijholt/adaptive-lighting/issues/682
_LOGGER.debug(
"%s: Waiting for simple switches to be initialized",
self._name,
)
await asyncio.sleep(0.1)
assert not self.remove_listeners
self._update_time_interval_listener()
@ -1449,7 +1435,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if force:
filtered_lights = on_lights
else:
filtered_lights = []
filtered_lights: list[str] = []
for light in on_lights:
# Don't adapt lights that haven't finished prior transitions.
timer = self.manager.transition_timers.get(light)
@ -1486,7 +1472,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
adapt_color = self.adapt_color_switch.is_on
assert isinstance(adapt_brightness, bool)
assert isinstance(adapt_color, bool)
tasks = []
tasks: list[asyncio.Task[None]] = []
for light in filtered_lights:
manually_controlled = (
self._take_over_control
@ -1541,7 +1527,11 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if tasks:
await asyncio.gather(*tasks)
async def _respond_to_off_to_on_event(self, entity_id: str, event: Event) -> None:
async def _respond_to_off_to_on_event(
self,
entity_id: str,
event: Event[EventStateChangedData],
) -> None:
assert not self.manager.is_proactively_adapting(event.context.id)
from_turn_on = self.manager._off_to_on_state_event_is_from_turn_on(
entity_id,
@ -1597,7 +1587,10 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
force=True,
)
async def _sleep_mode_switch_state_event_action(self, event: Event) -> None:
async def _sleep_mode_switch_state_event_action(
self,
event: Event[EventStateChangedData],
) -> None:
if not _is_state_event(event, (STATE_ON, STATE_OFF)):
_LOGGER.debug("%s: Ignoring sleep event %s", self._name, event)
return
@ -1690,6 +1683,10 @@ class SimpleSwitch(SwitchEntity, RestoreEntity):
self._state = False
type AdaptiveSwitches = list[AdaptiveSwitch]
type AdaptiveSwitchMap = dict[AdaptiveSwitch, list[str]]
class AdaptiveLightingManager:
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
@ -1706,11 +1703,11 @@ class AdaptiveLightingManager:
# Tracks 'light.toggle' service calls
self.toggle_event: dict[str, Event] = {}
# Tracks 'on' → 'off' state changes
self.on_to_off_event: dict[str, Event] = {}
self.on_to_off_event: dict[str, Event[EventStateChangedData]] = {}
# Tracks 'off' → 'on' state changes
self.off_to_on_event: dict[str, Event] = {}
self.off_to_on_event: dict[str, Event[EventStateChangedData]] = {}
# Keep 'asyncio.sleep' tasks that can be cancelled by 'light.turn_on' events
self.sleep_tasks: dict[str, asyncio.Task] = {}
self.sleep_tasks: dict[str, asyncio.Task[None]] = {}
# 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
@ -1720,8 +1717,8 @@ class AdaptiveLightingManager:
# 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_brightness: dict[str, asyncio.Task] = {}
self.adaptation_tasks_color: dict[str, asyncio.Task] = {}
self.adaptation_tasks_brightness: dict[str, asyncio.Task[None]] = {}
self.adaptation_tasks_color: dict[str, asyncio.Task[None]] = {}
# Track auto reset of manual_control
self.auto_reset_manual_control_timers: dict[str, _AsyncSingleShotTimer] = {}
@ -1731,7 +1728,7 @@ class AdaptiveLightingManager:
self.transition_timers: dict[str, _AsyncSingleShotTimer] = {}
# Track _execute_cancellable_adaptation_calls tasks
self.adaptation_tasks = set()
self.adaptation_tasks: set[asyncio.Task[None]] = set()
# Setup listeners and its callbacks to remove them later
self.listener_removers = [
@ -1823,15 +1820,11 @@ class AdaptiveLightingManager:
def _separate_entity_ids(
self,
entity_ids: list[str],
data,
) -> tuple[list[str], list[str]]:
data: ServiceData,
) -> tuple[AdaptiveSwitchMap, list[str]]:
# Create a mapping from switch to entity IDs
# AdaptiveSwitch.name → entity_ids mapping
switch_to_eids: dict[str, list[str]] = {}
# AdaptiveSwitch.name → AdaptiveSwitch mapping
switch_name_mapping: dict[str, AdaptiveSwitch] = {}
# Note: In HA≥2023.5, AdaptiveSwitch is hashable, so we can
# use dict[AdaptiveSwitch, list[str]]
# AdaptiveSwitch → entity_ids mapping
switch_to_eids: AdaptiveSwitchMap = {}
skipped: list[str] = []
for entity_id in entity_ids:
try:
@ -1855,7 +1848,7 @@ class AdaptiveLightingManager:
not switch.is_on
or not switch._intercept
# Never adapt on light groups, because HA will make a separate light.turn_on
or _is_light_group(self.hass.states.get(entity_id))
or ((e := self.hass.states.get(entity_id)) and _is_light_group(e))
# Prevent adaptation of TURN_ON calls when light is already on,
# and of TOGGLE calls when toggling off.
or self.hass.states.is_state(entity_id, STATE_ON)
@ -1881,19 +1874,17 @@ class AdaptiveLightingManager:
)
skipped.append(entity_id)
else:
switch_to_eids.setdefault(switch.name, []).append(entity_id)
switch_name_mapping[switch.name] = switch
return switch_to_eids, switch_name_mapping, skipped
switch_to_eids.setdefault(switch, []).append(entity_id)
return switch_to_eids, skipped
def _correct_for_multi_light_intercept(
self,
entity_ids,
switch_to_eids,
switch_name_mapping,
skipped,
entity_ids: list[str],
switch_to_eids: AdaptiveSwitchMap,
skipped: list[str],
):
# Check for `multi_light_intercept: true/false`
mli = [sw._multi_light_intercept for sw in switch_name_mapping.values()]
mli = [sw._multi_light_intercept for sw in switch_to_eids]
more_than_one_switch = len(switch_to_eids) > 1
single_switch_with_multiple_lights = (
len(switch_to_eids) == 1 and len(next(iter(switch_to_eids.values()))) > 1
@ -1919,7 +1910,7 @@ class AdaptiveLightingManager:
)
skipped = entity_ids
switch_to_eids = {}
return switch_to_eids, switch_name_mapping, skipped
return switch_to_eids, skipped
async def _service_interceptor_turn_on_handler(
self,
@ -1984,19 +1975,17 @@ class AdaptiveLightingManager:
# we skip them and rely on the followup call that HA will make
# with the expanded entity IDs.
switch_to_eids, switch_name_mapping, skipped = self._separate_entity_ids(
switch_to_eids, skipped = self._separate_entity_ids(
entity_ids,
service_data,
)
(
switch_to_eids,
switch_name_mapping,
skipped,
) = self._correct_for_multi_light_intercept(
entity_ids,
switch_to_eids,
switch_name_mapping,
skipped,
)
_LOGGER.debug(
@ -2005,7 +1994,10 @@ class AdaptiveLightingManager:
skipped,
)
def modify_service_data(service_data, entity_ids) -> dict[str, Any]:
def modify_service_data(
service_data: ServiceData,
entity_ids: list[str],
) -> dict[str, Any]:
"""Modify the service data to contain the entity IDs."""
service_data.pop(ATTR_ENTITY_ID, None)
service_data.pop(ATTR_AREA_ID, None)
@ -2014,8 +2006,7 @@ class AdaptiveLightingManager:
# Intercept the call for first switch and call _adapt_light for the rest
has_intercepted = False # Can only intercept a turn_on call once
for adaptive_switch_name, _entity_ids in switch_to_eids.items():
switch = switch_name_mapping[adaptive_switch_name]
for switch, _entity_ids in switch_to_eids.items():
transition = service_data[CONF_PARAMS].get(
ATTR_TRANSITION,
switch.initial_transition,
@ -2289,8 +2280,8 @@ class AdaptiveLightingManager:
if ATTR_ENTITY_ID in service_data:
return cv.ensure_list_csv(service_data[ATTR_ENTITY_ID])
if ATTR_AREA_ID in service_data:
entity_ids = []
area_ids = cv.ensure_list_csv(service_data[ATTR_AREA_ID])
entity_ids: list[str] = []
area_ids: list[str] = cv.ensure_list_csv(service_data[ATTR_AREA_ID])
for area_id in area_ids:
area_entity_ids = area_entities(self.hass, area_id)
eids = [
@ -2369,14 +2360,18 @@ class AdaptiveLightingManager:
event.context.id,
)
for eid in entity_ids:
state = self.hass.states.get(eid).state
state = self.hass.states.get(eid)
assert state
self.toggle_event[eid] = event
if state == STATE_ON: # is turning off
if state.state == STATE_ON: # is turning off
off(eid, event)
elif state == STATE_OFF: # is turning on
elif state.state == STATE_OFF: # is turning on
on(eid, event)
async def state_changed_event_listener(self, event: Event) -> None:
async def state_changed_event_listener(
self,
event: Event[EventStateChangedData],
) -> None:
"""Track 'state_changed' events."""
entity_id = event.data.get(ATTR_ENTITY_ID, "")
if entity_id not in self.lights:
@ -2385,17 +2380,29 @@ class AdaptiveLightingManager:
old_state = event.data.get("old_state")
new_state = event.data.get("new_state")
new_on = new_state is not None and new_state.state == STATE_ON
new_off = new_state is not None and new_state.state == STATE_OFF
old_on = old_state is not None and old_state.state == STATE_ON
old_off = old_state is not None and old_state.state == STATE_OFF
new_on = (
new_state if new_state is not None and new_state.state == STATE_ON else None
)
new_off = (
new_state
if new_state is not None and new_state.state == STATE_OFF
else None
)
old_on = (
old_state if old_state is not None and old_state.state == STATE_ON else None
)
old_off = (
old_state
if old_state is not None and old_state.state == STATE_OFF
else None
)
if new_on:
_LOGGER.debug(
"Detected a '%s' 'state_changed' event: '%s' with context.id='%s'",
entity_id,
new_state.attributes,
new_state.context.id,
new_on.attributes,
new_on.context.id,
)
# It is possible to have multiple state change events with the same context.
# This can happen because a `turn_on.light(brightness_pct=100, transition=30)`
@ -2411,29 +2418,29 @@ class AdaptiveLightingManager:
last_state: list[State] | None = self.our_last_state_on_change.get(
entity_id,
)
if is_our_context(new_state.context):
if is_our_context(new_on.context):
if (
last_state is not None
and last_state[0].context.id == new_state.context.id
and last_state[0].context.id == new_on.context.id
):
_LOGGER.debug(
"AdaptiveLightingManager: State change event of '%s' is already"
" in 'self.our_last_state_on_change' (%s)"
" adding this state also",
entity_id,
new_state.context.id,
new_on.context.id,
)
self.our_last_state_on_change[entity_id].append(new_state)
self.our_last_state_on_change[entity_id].append(new_on)
else:
_LOGGER.debug(
"AdaptiveLightingManager: New adapt '%s' found for %s",
new_state,
new_on,
entity_id,
)
self.our_last_state_on_change[entity_id] = [new_state]
self.our_last_state_on_change[entity_id] = [new_on]
self.start_transition_timer(entity_id)
elif last_state is not None:
self.our_last_state_on_change[entity_id].append(new_state)
self.our_last_state_on_change[entity_id].append(new_on)
if old_on and new_off:
# Tracks 'on' → 'off' state changes
@ -2583,7 +2590,7 @@ class AdaptiveLightingManager:
def _off_to_on_state_event_is_from_turn_on(
self,
entity_id: str,
off_to_on_event: Event,
off_to_on_event: Event[EventStateChangedData],
) -> bool:
# Adaptive Lighting should never turn on lights itself
if is_our_context(off_to_on_event.context) and not is_our_context(
@ -2601,11 +2608,7 @@ class AdaptiveLightingManager:
)
turn_on_event: Event | None = self.turn_on_event.get(entity_id)
id_off_to_on = off_to_on_event.context.id
return (
turn_on_event is not None
and id_off_to_on is not None
and id_off_to_on == turn_on_event.context.id
)
return turn_on_event is not None and id_off_to_on == turn_on_event.context.id
async def just_turned_off( # noqa: PLR0911
self,
@ -2662,7 +2665,6 @@ class AdaptiveLightingManager:
if (
turn_off_event is not None
and id_on_to_off == turn_off_event.context.id
and id_on_to_off is not None
and transition is not None # 'turn_off' is called with transition=...
):
# State change 'on' → 'off' and 'light.turn_off(..., transition=...)' come

11
tests/test_color_and_brightness.py
View file

@ -5,8 +5,7 @@ import pytest
from astral import LocationInfo
from astral.location import Location
from homeassistant.components.adaptive_lighting.color_and_brightness import (
SUN_EVENT_NOON,
SUN_EVENT_SUNRISE,
SunEvent,
SunEvents,
)
@ -167,7 +166,7 @@ def test_sun_events(tzinfo_and_location):
date = dt.datetime(2022, 1, 1)
events = sun_events.sun_events(date)
assert len(events) == 4
assert (SUN_EVENT_SUNRISE, location.sunrise(date).timestamp()) in events
assert (SunEvent.SUNRISE, location.sunrise(date).timestamp()) in events
def test_prev_and_next_events(tzinfo_and_location):
@ -186,8 +185,8 @@ def test_prev_and_next_events(tzinfo_and_location):
datetime = dt.datetime(2022, 1, 1, 10, 0)
after_sunrise = sun_events.sunrise(datetime.date()) + dt.timedelta(hours=1)
prev_event, next_event = sun_events.prev_and_next_events(after_sunrise)
assert prev_event[0] == SUN_EVENT_SUNRISE
assert next_event[0] == SUN_EVENT_NOON
assert prev_event[0] == SunEvent.SUNRISE
assert next_event[0] == SunEvent.NOON
def test_closest_event(tzinfo_and_location):
@ -206,5 +205,5 @@ def test_closest_event(tzinfo_and_location):
datetime = dt.datetime(2022, 1, 1, 6, 0)
sunrise = sun_events.sunrise(datetime.date())
event_name, ts = sun_events.closest_event(sunrise)
assert event_name == SUN_EVENT_SUNRISE
assert event_name == SunEvent.SUNRISE
assert ts == location.sunrise(sunrise.date()).timestamp()