diff --git a/custom_components/adaptive_lighting/__init__.py b/custom_components/adaptive_lighting/__init__.py index 0c8bad80..44151c84 100644 --- a/custom_components/adaptive_lighting/__init__.py +++ b/custom_components/adaptive_lighting/__init__.py @@ -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) diff --git a/custom_components/adaptive_lighting/_docs_helpers.py b/custom_components/adaptive_lighting/_docs_helpers.py index 49899395..afbb9d66 100644 --- a/custom_components/adaptive_lighting/_docs_helpers.py +++ b/custom_components/adaptive_lighting/_docs_helpers.py @@ -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, diff --git a/custom_components/adaptive_lighting/adaptation_utils.py b/custom_components/adaptive_lighting/adaptation_utils.py index 2c339223..14c28ae8 100644 --- a/custom_components/adaptive_lighting/adaptation_utils.py +++ b/custom_components/adaptive_lighting/adaptation_utils.py @@ -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] } diff --git a/custom_components/adaptive_lighting/color_and_brightness.py b/custom_components/adaptive_lighting/color_and_brightness.py index 2fd93e67..34da23c6 100644 --- a/custom_components/adaptive_lighting/color_and_brightness.py +++ b/custom_components/adaptive_lighting/color_and_brightness.py @@ -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) diff --git a/custom_components/adaptive_lighting/config_flow.py b/custom_components/adaptive_lighting/config_flow.py index 00214fdb..d1799888 100644 --- a/custom_components/adaptive_lighting/config_flow.py +++ b/custom_components/adaptive_lighting/config_flow.py @@ -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: diff --git a/custom_components/adaptive_lighting/const.py b/custom_components/adaptive_lighting/const.py index 3fc9c5d4..f7c3620b 100644 --- a/custom_components/adaptive_lighting/const.py +++ b/custom_components/adaptive_lighting/const.py @@ -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, }, ) diff --git a/custom_components/adaptive_lighting/hass_utils.py b/custom_components/adaptive_lighting/hass_utils.py index 21ea67b3..550ae350 100644 --- a/custom_components/adaptive_lighting/hass_utils.py +++ b/custom_components/adaptive_lighting/hass_utils.py @@ -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, diff --git a/custom_components/adaptive_lighting/helpers.py b/custom_components/adaptive_lighting/helpers.py index fa3af6ef..df2abb95 100644 --- a/custom_components/adaptive_lighting/helpers.py +++ b/custom_components/adaptive_lighting/helpers.py @@ -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 diff --git a/custom_components/adaptive_lighting/switch.py b/custom_components/adaptive_lighting/switch.py index 59b98a0c..dbb72ea1 100644 --- a/custom_components/adaptive_lighting/switch.py +++ b/custom_components/adaptive_lighting/switch.py @@ -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()