From 5886eee04cecb30a0951ced38e53ce8edaaa50eb Mon Sep 17 00:00:00 2001 From: Mario Guggenberger Date: Fri, 12 Dec 2025 22:37:42 +0100 Subject: [PATCH] Code cleanup (#1348) --- .../adaptive_lighting/__init__.py | 2 +- .../adaptive_lighting/color_and_brightness.py | 76 ++++--- .../adaptive_lighting/config_flow.py | 21 +- custom_components/adaptive_lighting/switch.py | 192 +++++++++--------- tests/test_color_and_brightness.py | 11 +- 5 files changed, 155 insertions(+), 147 deletions(-) diff --git a/custom_components/adaptive_lighting/__init__.py b/custom_components/adaptive_lighting/__init__.py index 44151c84..8e61e93b 100644 --- a/custom_components/adaptive_lighting/__init__.py +++ b/custom_components/adaptive_lighting/__init__.py @@ -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, diff --git a/custom_components/adaptive_lighting/color_and_brightness.py b/custom_components/adaptive_lighting/color_and_brightness.py index caee67c2..f3ae3efe 100644 --- a/custom_components/adaptive_lighting/color_and_brightness.py +++ b/custom_components/adaptive_lighting/color_and_brightness.py @@ -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) diff --git a/custom_components/adaptive_lighting/config_flow.py b/custom_components/adaptive_lighting/config_flow.py index 5e21e67b..5b0dd13c 100644 --- a/custom_components/adaptive_lighting/config_flow.py +++ b/custom_components/adaptive_lighting/config_flow.py @@ -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", diff --git a/custom_components/adaptive_lighting/switch.py b/custom_components/adaptive_lighting/switch.py index 2d667d97..9217e8d9 100644 --- a/custom_components/adaptive_lighting/switch.py +++ b/custom_components/adaptive_lighting/switch.py @@ -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 diff --git a/tests/test_color_and_brightness.py b/tests/test_color_and_brightness.py index 6c76175f..cc3e8c4d 100644 --- a/tests/test_color_and_brightness.py +++ b/tests/test_color_and_brightness.py @@ -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()