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

Friendly names (#1258)

Browse source
This commit is contained in:
Dobby 2025-11-27 21:01:05 +01:00 • committed by GitHub
commit 19d30430bb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 140 additions and 90 deletions
Download patch file Download diff file Expand all files Collapse all files

18
custom_components/adaptive_lighting/__init__.py
View file

@ -7,7 +7,7 @@ import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.core import Event, HomeAssistant
from .const import (
_DOMAIN_SCHEMA,
@ -22,7 +22,7 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = ["switch"]
def _all_unique_names(value):
def _all_unique_names(value: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Validate that all entities have a unique profile name."""
hosts = [device[CONF_NAME] for device in value]
schema = vol.Schema(vol.Unique())
@ -36,12 +36,16 @@ CONFIG_SCHEMA = vol.Schema(
)
async def reload_configuration_yaml(event: dict, hass: HomeAssistant): # noqa: ARG001
async def reload_configuration_yaml(event: Event) -> None:
"""Reload configuration.yaml."""
await hass.services.async_call("homeassistant", "check_config", {})
hass: HomeAssistant | None = event.data.get("hass")
if hass is not None:
await hass.services.async_call("homeassistant", "check_config", {})
else:
_LOGGER.error("HomeAssistant instance not found in event data.")
async def async_setup(hass: HomeAssistant, config: dict[str, Any]):
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Import integration from config."""
if DOMAIN in config:
for entry in config[DOMAIN]:
@ -55,7 +59,7 @@ async def async_setup(hass: HomeAssistant, config: dict[str, Any]):
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up the component."""
data = hass.data.setdefault(DOMAIN, {})
@ -70,7 +74,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
return True
async def async_update_options(hass: HomeAssistant, config_entry: ConfigEntry):
async def async_update_options(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Update options."""
await hass.config_entries.async_reload(config_entry.entry_id)

21
custom_components/adaptive_lighting/_docs_helpers.py
View file

@ -15,7 +15,7 @@ from .const import (
)
def _format_voluptuous_instance(instance):
def _format_voluptuous_instance(instance: vol.All) -> str:
coerce_type = None
min_val = None
max_val = None
@ -56,8 +56,8 @@ def _type_to_str(type_: Any) -> str: # noqa: PLR0911
raise ValueError(msg)
def generate_config_markdown_table():
rows = []
def generate_config_markdown_table() -> str:
rows: list[dict[str, str]] = []
for k, default, type_ in VALIDATION_TUPLES:
description = DOCS[k]
row = {
@ -73,7 +73,7 @@ def generate_config_markdown_table():
def _schema_to_dict(schema: vol.Schema) -> dict[str, tuple[Any, Any]]:
result = {}
result: dict[str, tuple[Any, Any]] = {}
for key, value in schema.schema.items():
if isinstance(key, vol.Optional):
default_value = key.default
@ -82,11 +82,12 @@ def _schema_to_dict(schema: vol.Schema) -> dict[str, tuple[Any, Any]]:
def _generate_service_markdown_table(
schema: vol.Schema,
schema: dict[str, tuple[Any, Any]] | vol.Schema,
alternative_docs: dict[str, str] | None = None,
):
rows = []
for k, (default, type_) in _schema_to_dict(schema).items():
) -> str:
schema_dict = _schema_to_dict(schema) if isinstance(schema, vol.Schema) else schema
rows: list[dict[str, str]] = []
for k, (default, type_) in schema_dict.items():
if alternative_docs is not None and k in alternative_docs:
description = alternative_docs[k]
else:
@ -103,11 +104,11 @@ def _generate_service_markdown_table(
return df.to_markdown(index=False)
def generate_apply_markdown_table():
def generate_apply_markdown_table() -> str:
return _generate_service_markdown_table(apply_service_schema(), DOCS_APPLY)
def generate_set_manual_control_markdown_table():
def generate_set_manual_control_markdown_table() -> str:
return _generate_service_markdown_table(
SET_MANUAL_CONTROL_SCHEMA,
DOCS_MANUAL_CONTROL,

3
custom_components/adaptive_lighting/adaptation_utils.py
View file

@ -84,10 +84,11 @@ def _remove_redundant_attributes(
Removes all attributes from service call data whose values are already present
in the target entity's state.
"""
attributes: dict[str, Any] = dict(state.attributes)
return {
k: v
for k, v in service_data.items()
if k not in state.attributes or v != state.attributes[k]
if k not in attributes or v != attributes[k]
}

6
custom_components/adaptive_lighting/color_and_brightness.py
View file

@ -375,8 +375,8 @@ class SunLightSettings:
def get_settings(
self,
is_sleep,
transition,
is_sleep: bool,
transition: float | None,
) -> dict[str, float | int | tuple[float, float] | tuple[float, float, float]]:
"""Get all light settings.
@ -507,7 +507,7 @@ def lerp_color_hsv(
return cast("tuple[int, int, int]", rgb)
def lerp(x, x1, x2, y1, y2):
def lerp(x: float, x1: float, x2: float, y1: float, y2: float) -> float:
"""Linearly interpolate between two values."""
return y1 + (x - x1) * (y2 - y1) / (x2 - x1)

35
custom_components/adaptive_lighting/config_flow.py
View file

@ -16,6 +16,7 @@ from .const import ( # pylint: disable=unused-import
NONE_STR,
VALIDATION_TUPLES,
)
from .helpers import get_friendly_name
from .switch import _supported_features, validate
_LOGGER = logging.getLogger(__name__)
@ -26,9 +27,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_user(self, user_input=None):
async def async_step_user(self, user_input: dict[str, Any] | None = None):
"""Handle the initial step."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
await self.async_set_unique_id(user_input[CONF_NAME])
@ -41,8 +42,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_import(self, user_input: dict[str, Any]):
async def async_step_import(self, user_input: dict[str, Any] | None = None):
"""Handle configuration by YAML file."""
if user_input is None:
return self.async_abort(reason="no_data")
await self.async_set_unique_id(user_input[CONF_NAME])
# Keep a list of switches that are configured via YAML
data = self.hass.data.setdefault(DOMAIN, {})
@ -57,7 +61,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> "OptionsFlowHandler":
"""Get the options flow for this handler."""
if (MAJOR_VERSION, MINOR_VERSION) >= (2024, 12):
# https://github.com/home-assistant/core/pull/129651
@ -65,7 +71,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return OptionsFlowHandler(config_entry)
def validate_options(user_input, errors):
def validate_options(user_input: dict[str, Any], errors: dict[str, str]) -> None:
"""Validate the options in the OptionsFlow.
This is an extra validation step because the validators
@ -85,7 +91,7 @@ def validate_options(user_input, errors):
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for Adaptive Lighting."""
def __init__(self, *args, **kwargs) -> None:
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize options flow."""
if (MAJOR_VERSION, MINOR_VERSION) >= (2024, 12):
super().__init__(*args, **kwargs)
@ -93,7 +99,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
else:
self.config_entry = args[0]
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: dict[str, Any] | None = None):
"""Handle options flow."""
conf = self.config_entry
data = validate(conf)
@ -105,11 +111,12 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
if not errors:
return self.async_create_entry(title="", data=user_input)
all_lights = [
light
all_lights_with_names = {
light: get_friendly_name(self.hass, light)
for light in self.hass.states.async_entity_ids("light")
if _supported_features(self.hass, light)
]
}
all_lights = list(all_lights_with_names.keys())
for configured_light in data[CONF_LIGHTS]:
if configured_light not in all_lights:
errors = {CONF_LIGHTS: "entity_missing"}
@ -119,7 +126,13 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
configured_light,
)
all_lights.append(configured_light)
to_replace = {CONF_LIGHTS: cv.multi_select(sorted(all_lights))}
all_lights_with_names[configured_light] = configured_light
light_options = {
entity_id: f"{name} ({entity_id})"
for entity_id, name in all_lights_with_names.items()
}
to_replace = {CONF_LIGHTS: cv.multi_select(light_options)}
options_schema = {}
for name, default, validation in VALIDATION_TUPLES:

33
custom_components/adaptive_lighting/const.py
View file

@ -1,5 +1,8 @@
"""Constants for the Adaptive Lighting integration."""
from datetime import timedelta
from typing import Any
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant.components.light import VALID_TRANSITION
@ -291,13 +294,13 @@ DOCS_APPLY = {
}
def int_between(min_int, max_int):
def int_between(min_int: int, max_int: int) -> vol.All:
"""Return an integer between 'min_int' and 'max_int'."""
return vol.All(vol.Coerce(int), vol.Range(min=min_int, max=max_int))
VALIDATION_TUPLES = [
(CONF_LIGHTS, DEFAULT_LIGHTS, cv.entity_ids),
VALIDATION_TUPLES: list[tuple[str, Any, Any]] = [
(CONF_LIGHTS, DEFAULT_LIGHTS, cv.entity_ids), # type: ignore[arg-type]
(CONF_INTERVAL, DEFAULT_INTERVAL, cv.positive_int),
(CONF_TRANSITION, DEFAULT_TRANSITION, VALID_TRANSITION),
(CONF_INITIAL_TRANSITION, DEFAULT_INITIAL_TRANSITION, VALID_TRANSITION),
@ -310,7 +313,7 @@ VALIDATION_TUPLES = [
(
CONF_SLEEP_RGB_OR_COLOR_TEMP,
DEFAULT_SLEEP_RGB_OR_COLOR_TEMP,
selector.SelectSelector(
selector.SelectSelector( # type: ignore[arg-type]
selector.SelectSelectorConfig(
options=["color_temp", "rgb_color"],
multiple=False,
@ -322,7 +325,7 @@ VALIDATION_TUPLES = [
(
CONF_SLEEP_RGB_COLOR,
DEFAULT_SLEEP_RGB_COLOR,
selector.ColorRGBSelector(selector.ColorRGBSelectorConfig()),
selector.ColorRGBSelector(selector.ColorRGBSelectorConfig()), # type: ignore[arg-type]
),
(CONF_SLEEP_TRANSITION, DEFAULT_SLEEP_TRANSITION, VALID_TRANSITION),
(CONF_ADAPT_UNTIL_SLEEP, DEFAULT_ADAPT_UNTIL_SLEEP, bool),
@ -337,7 +340,7 @@ VALIDATION_TUPLES = [
(
CONF_BRIGHTNESS_MODE,
DEFAULT_BRIGHTNESS_MODE,
selector.SelectSelector(
selector.SelectSelector( # type: ignore[arg-type]
selector.SelectSelectorConfig(
options=["default", "linear", "tanh"],
multiple=False,
@ -370,7 +373,7 @@ VALIDATION_TUPLES = [
]
def timedelta_as_int(value):
def timedelta_as_int(value: timedelta) -> float:
"""Convert a `datetime.timedelta` object to an integer.
This integer can be serialized to json but a timedelta cannot.
@ -380,7 +383,7 @@ def timedelta_as_int(value):
# conf_option: (validator, coerce) tuples
# these validators cannot be serialized but can be serialized when coerced by coerce.
EXTRA_VALIDATION = {
EXTRA_VALIDATION: dict[str, tuple[Any, Any]] = {
CONF_INTERVAL: (cv.time_period, timedelta_as_int),
CONF_SUNRISE_OFFSET: (cv.time_period, timedelta_as_int),
CONF_SUNRISE_TIME: (cv.time, str),
@ -395,7 +398,7 @@ EXTRA_VALIDATION = {
}
def maybe_coerce(key, validation):
def maybe_coerce(key: str, validation: Any) -> vol.All | Any:
"""Coerce the validation into a json serializable type."""
validation, coerce = EXTRA_VALIDATION.get(key, (validation, None))
if coerce is not None:
@ -403,7 +406,7 @@ def maybe_coerce(key, validation):
return validation
def replace_none_str(value, replace_with=None):
def replace_none_str(value: Any, replace_with: Any | None = None) -> Any:
"""Replace "None" -> replace_with."""
return value if value != NONE_STR else replace_with
@ -421,12 +424,12 @@ _DOMAIN_SCHEMA = vol.Schema(
)
def apply_service_schema(initial_transition: int = 1):
def apply_service_schema(initial_transition: int = 1) -> vol.Schema:
"""Return the schema for the apply service."""
return vol.Schema(
{
vol.Optional(CONF_ENTITY_ID): cv.entity_ids,
vol.Optional(CONF_LIGHTS, default=[]): cv.entity_ids,
vol.Optional(CONF_ENTITY_ID): cv.entity_ids, # type: ignore[arg-type]
vol.Optional(CONF_LIGHTS, default=[]): cv.entity_ids, # type: ignore[arg-type]
vol.Optional(
CONF_TRANSITION,
default=initial_transition,
@ -441,8 +444,8 @@ def apply_service_schema(initial_transition: int = 1):
SET_MANUAL_CONTROL_SCHEMA = vol.Schema(
{
vol.Optional(CONF_ENTITY_ID): cv.entity_ids,
vol.Optional(CONF_LIGHTS, default=[]): cv.entity_ids,
vol.Optional(CONF_ENTITY_ID): cv.entity_ids, # type: ignore[arg-type]
vol.Optional(CONF_LIGHTS, default=[]): cv.entity_ids, # type: ignore[arg-type]
vol.Optional(CONF_MANUAL_CONTROL, default=True): cv.boolean,
},
)

16
custom_components/adaptive_lighting/hass_utils.py
View file

@ -47,7 +47,7 @@ def setup_service_call_interceptor(
# 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
hass.services._services # pylint: disable=protected-access # type: ignore[attr-defined]
)
except AttributeError as error:
msg = (
@ -68,7 +68,9 @@ def setup_service_call_interceptor(
data = dict(call.data)
# Call interceptor
await intercept_func(call, data)
result = intercept_func(call, data)
if result is not None:
await result
# Convert data back to read-only
call.data = ReadOnlyDict(data)
@ -79,7 +81,13 @@ def setup_service_call_interceptor(
call.data,
)
# Call original service handler with processed data
await existing_service.job.target(call)
import asyncio
target = existing_service.job.target
if asyncio.iscoroutinefunction(target):
await target(call)
else:
target(call)
hass.services.async_register(
domain,
@ -88,7 +96,7 @@ def setup_service_call_interceptor(
existing_service.schema,
)
def remove():
def remove() -> None:
# Remove the interceptor by reinstalling the original service handler
hass.services.async_register(
domain,

13
custom_components/adaptive_lighting/helpers.py
View file

@ -4,6 +4,10 @@ from __future__ import annotations
import base64
import math
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from homeassistant.core import HomeAssistant
def clamp(value: float, minimum: float, maximum: float) -> float:
@ -83,3 +87,12 @@ def color_difference_redmean(
green_term = 4 * delta_g**2
blue_term = (2 + (255 - r_hat) / 256) * delta_b**2
return math.sqrt(red_term + green_term + blue_term)
def get_friendly_name(hass: HomeAssistant, entity_id: str) -> str:
"""Retrieve the friendly name of an entity."""
state = hass.states.get(entity_id)
if state and hasattr(state, "attributes"):
attributes: dict[str, Any] = dict(getattr(state, "attributes", {}))
return attributes.get("friendly_name", entity_id)
return entity_id

85
custom_components/adaptive_lighting/switch.py
View file

@ -232,7 +232,7 @@ def _switches_with_lights(
"""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 = []
switches: list[AdaptiveSwitch] = []
all_check_lights = (
_expand_light_groups(hass, lights) if expand_light_groups else set(lights)
)
@ -369,7 +369,7 @@ def _fire_manual_control_event(
switch: AdaptiveSwitch,
light: str,
context: Context,
):
) -> None:
"""Fire an event that 'light' is marked as manual_control."""
hass = switch.hass
_LOGGER.debug(
@ -389,7 +389,7 @@ async def async_setup_entry( # noqa: PLR0915
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
):
) -> None:
"""Set up the AdaptiveLighting switch."""
assert hass is not None
data = hass.data[DOMAIN]
@ -456,7 +456,7 @@ async def async_setup_entry( # noqa: PLR0915
)
@callback
async def handle_apply(service_call: ServiceCall):
async def handle_apply(service_call: ServiceCall) -> None:
"""Handle the entity service apply."""
data = service_call.data
_LOGGER.debug(
@ -488,7 +488,7 @@ async def async_setup_entry( # noqa: PLR0915
)
@callback
async def handle_set_manual_control(service_call: ServiceCall):
async def handle_set_manual_control(service_call: ServiceCall) -> None:
"""Set or unset lights as 'manually controlled'."""
data = service_call.data
_LOGGER.debug(
@ -583,7 +583,7 @@ def validate(
return data
def _is_state_event(event: Event, from_or_to_state: Iterable[str]):
def _is_state_event(event: Event, 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
@ -598,7 +598,7 @@ def _expand_light_groups(
hass: HomeAssistant,
lights: list[str],
) -> list[str]:
all_lights = set()
all_lights: set[str] = set()
manager = hass.data[DOMAIN][ATTR_ADAPTIVE_LIGHTING_MANAGER]
for light in lights:
state = hass.states.get(light)
@ -625,15 +625,15 @@ 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 = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
supported_features = int(state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)) # type: ignore[arg-type]
assert isinstance(supported_features, int)
supported = set()
supported: set[str] = set()
if supported_features & LightEntityFeature.TRANSITION:
supported.add("transition")
supported_color_modes = state.attributes.get(ATTR_SUPPORTED_COLOR_MODES, set())
supported_color_modes = state.attributes.get(ATTR_SUPPORTED_COLOR_MODES, set()) # type: ignore[arg-type]
color_modes = {
ColorMode.RGB,
ColorMode.RGBW,
@ -838,7 +838,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
def __init__(
self,
hass,
hass: HomeAssistant,
config_entry: ConfigEntry,
manager: AdaptiveLightingManager,
sleep_mode_switch: SimpleSwitch,
@ -892,7 +892,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self,
data: dict[str, Any],
defaults: dict[str, Any] | None = None,
):
) -> None:
# Only pass settings users can change during runtime
data = validate(
config_entry=None,
@ -984,12 +984,12 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
)
@property
def name(self):
def name(self) -> str:
"""Return the name of the device if any."""
return f"Adaptive Lighting: {self._name}"
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return the unique ID of entity."""
return self._name
@ -1026,11 +1026,11 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._state = False
assert not self.remove_listeners
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Remove the listeners upon removing the component."""
self._remove_listeners()
def _expand_light_groups(self, hass=None) -> None:
def _expand_light_groups(self, hass: HomeAssistant | None = None) -> None:
hass = hass or self.hass
all_lights = _expand_light_groups(hass, self.lights)
self.manager.lights.update(all_lights)
@ -1187,7 +1187,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
force=True,
)
async def async_turn_off(self, **kwargs) -> None: # noqa: ARG002
async def async_turn_off(self, **kwargs: Any) -> None: # noqa: ARG002
"""Turn off adaptive lighting."""
if not self.is_on:
return
@ -1195,7 +1195,10 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._remove_listeners()
self.manager.reset(*self.lights)
async def _async_update_at_interval_action(self, now=None) -> None: # noqa: ARG002
async def _async_update_at_interval_action(
self,
now: Any = None, # noqa: ARG002
) -> None:
"""Update the attributes and maybe adapt the lights."""
await self._update_attrs_and_maybe_adapt_lights(
context=self.create_context("interval"),
@ -1328,7 +1331,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
await self.execute_cancellable_adaptation_calls(data)
async def _execute_adaptation_calls(self, data: AdaptationData):
async def _execute_adaptation_calls(self, data: AdaptationData) -> None:
"""Executes a sequence of adaptation service calls for the given service datas."""
for index in range(data.max_length):
is_first_call = index == 0
@ -1379,7 +1382,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
async def execute_cancellable_adaptation_calls(
self,
data: AdaptationData,
):
) -> None:
"""Executes a cancellable sequence of adaptation service calls for the given service datas.
Wraps the sequence of service calls in a task that can be cancelled from elsewhere, e.g.,
@ -1635,12 +1638,12 @@ class SimpleSwitch(SwitchEntity, RestoreEntity):
self._initial_state = initial_state
@property
def name(self):
def name(self) -> str:
"""Return the name of the device if any."""
return self._name
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return the unique ID of entity."""
return self._unique_id
@ -1676,12 +1679,12 @@ class SimpleSwitch(SwitchEntity, RestoreEntity):
else:
await self.async_turn_off()
async def async_turn_on(self, **kwargs) -> None: # noqa: ARG002
async def async_turn_on(self, **kwargs: Any) -> None: # noqa: ARG002
"""Turn on adaptive lighting sleep mode."""
_LOGGER.debug("%s: Turning on", self._name)
self._state = True
async def async_turn_off(self, **kwargs) -> None: # noqa: ARG002
async def async_turn_off(self, **kwargs: Any) -> None: # noqa: ARG002
"""Turn off adaptive lighting sleep mode."""
_LOGGER.debug("%s: Turning off", self._name)
self._state = False
@ -1769,7 +1772,7 @@ class AdaptiveLightingManager:
exc_info=True,
)
def disable(self):
def disable(self) -> None:
"""Disable the listener by removing all subscribed handlers."""
for remove in self.listener_removers:
remove()
@ -1991,7 +1994,7 @@ class AdaptiveLightingManager:
skipped,
)
def modify_service_data(service_data, entity_ids):
def modify_service_data(service_data, entity_ids) -> 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)
@ -2169,7 +2172,7 @@ class AdaptiveLightingManager:
light,
)
async def reset():
async def reset() -> None:
# Called when the timer expires, doesn't need to do anything
_LOGGER.debug(
"Transition finished for light %s",
@ -2178,7 +2181,11 @@ class AdaptiveLightingManager:
self._handle_timer(light, self.transition_timers, last_transition, reset)
def set_auto_reset_manual_control_times(self, lights: list[str], time: float):
def set_auto_reset_manual_control_times(
self,
lights: list[str],
time: float,
) -> None:
"""Set the time after which the lights are automatically reset."""
if time == 0:
return
@ -2201,7 +2208,7 @@ class AdaptiveLightingManager:
self.manual_control[light] = True
delay = self.auto_reset_manual_control_times.get(light)
async def reset():
async def reset() -> None:
_LOGGER.debug(
"Auto resetting 'manual_control' status of '%s' because"
" it was not manually controlled for %s seconds.",
@ -2227,7 +2234,7 @@ class AdaptiveLightingManager:
self,
light_id: str,
which: Literal["color", "brightness", "both"] = "both",
):
) -> None:
"""Cancel ongoing adaptation service calls for a specific light entity."""
brightness_task = self.adaptation_tasks_brightness.get(light_id)
color_task = self.adaptation_tasks_color.get(light_id)
@ -2256,7 +2263,7 @@ class AdaptiveLightingManager:
# color_task might be the same as brightness_task
color_task.cancel()
def reset(self, *lights, reset_manual_control: bool = True) -> None:
def reset(self, *lights: str, reset_manual_control: bool = True) -> None:
"""Reset the 'manual_control' status of the lights."""
for light in lights:
if reset_manual_control:
@ -2306,11 +2313,11 @@ class AdaptiveLightingManager:
if not any(eid in self.lights for eid in entity_ids):
return
def off(eid: str, event: Event):
def off(eid: str, event: Event) -> None:
self.turn_off_event[eid] = event
self.reset(eid)
def on(eid: str, event: Event):
def on(eid: str, event: Event) -> None:
task = self.sleep_tasks.get(eid)
if task is not None:
task.cancel()
@ -2735,14 +2742,14 @@ class AdaptiveLightingManager:
class _AsyncSingleShotTimer:
def __init__(self, delay, callback) -> None:
def __init__(self, delay: float, callback: Callable[[], None | Any]) -> None:
"""Initialize the timer."""
self.delay = delay
self.callback = callback
self.task = None
self.start_time: datetime.datetime | None = None
async def _run(self):
async def _run(self) -> None:
"""Run the timer. Don't call this directly, use start() instead."""
await asyncio.sleep(self.delay)
if self.callback:
@ -2751,11 +2758,11 @@ class _AsyncSingleShotTimer:
else:
self.callback()
def is_running(self):
def is_running(self) -> bool:
"""Return whether the timer is running."""
return self.task is not None and not self.task.done()
def start(self):
def start(self) -> None:
"""Start the timer."""
if self.task is not None and not self.task.done():
self.task.cancel()
@ -2765,13 +2772,13 @@ class _AsyncSingleShotTimer:
self.start_time = dt_util.utcnow()
self.task = asyncio.create_task(self._run())
def cancel(self):
def cancel(self) -> None:
"""Cancel the timer."""
if self.task:
self.task.cancel()
self.callback = None
def remaining_time(self):
def remaining_time(self) -> float:
"""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()