mirror of
https://github.com/basnijholt/adaptive-lighting.git
synced 2026-09-24 19:04:19 +02:00
Add a lot of types
Add the function to view friendly names instead of entity IDs.
This commit is contained in:
Dobby
parent
0a2d2e9fb5
commit
133057b953
9 changed files with 135 additions and 104 deletions
|
|
@ -1,13 +1,13 @@
|
|||
"""Adaptive Lighting integration in Home-Assistant."""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import Any, Optional
|
||||
import voluptuous as vol
|
||||
|
||||
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 HomeAssistant, Event
|
||||
|
||||
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())
|
||||
|
|
@ -35,13 +35,15 @@ CONFIG_SCHEMA = vol.Schema(
|
|||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
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: Optional[HomeAssistant] = event.data["hass"] if "hass" in event.data else None
|
||||
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 +57,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,12 +72,12 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
|
|||
return True
|
||||
|
||||
|
||||
async def async_update_options(hass, 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)
|
||||
|
||||
|
||||
async def async_unload_entry(hass, config_entry: ConfigEntry) -> bool:
|
||||
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_forward_entry_unload(
|
||||
config_entry,
|
||||
|
|
|
|||
|
|
@ -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,10 +56,10 @@ def _type_to_str(type_: Any) -> str: # noqa: PLR0911
|
|||
raise ValueError(msg)
|
||||
|
||||
|
||||
def generate_config_markdown_table():
|
||||
def generate_config_markdown_table() -> str:
|
||||
import pandas as pd
|
||||
|
||||
rows = []
|
||||
rows: list[dict[str, str]] = []
|
||||
for k, default, type_ in VALIDATION_TUPLES:
|
||||
description = DOCS[k]
|
||||
row = {
|
||||
|
|
@ -75,7 +75,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
|
||||
|
|
@ -84,12 +84,15 @@ def _schema_to_dict(schema: vol.Schema) -> dict[str, tuple[Any, Any]]:
|
|||
|
||||
|
||||
def _generate_service_markdown_table(
|
||||
schema: dict[str, tuple[Any, Any]],
|
||||
schema: dict[str, tuple[Any, Any]] | vol.Schema,
|
||||
alternative_docs: dict[str, str] | None = None,
|
||||
):
|
||||
schema = _schema_to_dict(schema)
|
||||
rows = []
|
||||
for k, (default, type_) in schema.items():
|
||||
) -> str:
|
||||
if isinstance(schema, vol.Schema):
|
||||
schema_dict = _schema_to_dict(schema)
|
||||
else:
|
||||
schema_dict = 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:
|
||||
|
|
@ -106,11 +109,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,
|
||||
|
|
|
|||
|
|
@ -54,7 +54,7 @@ def _split_service_call_data(service_data: ServiceData) -> list[ServiceData]:
|
|||
common_data = {k: service_data[k] for k in common_attrs if k in service_data}
|
||||
|
||||
attributes_split_sequence = [BRIGHTNESS_ATTRS, COLOR_ATTRS]
|
||||
service_datas = []
|
||||
service_datas: list[dict[str, Any]] = []
|
||||
|
||||
for attributes in attributes_split_sequence:
|
||||
split_data = {
|
||||
|
|
@ -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]
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -375,8 +375,8 @@ class SunLightSettings:
|
|||
|
||||
def get_settings(
|
||||
self,
|
||||
is_sleep,
|
||||
transition,
|
||||
is_sleep: bool,
|
||||
transition: float | int | 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)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
"""Config flow for Adaptive Lighting integration."""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
import voluptuous as vol
|
||||
|
|
@ -15,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__)
|
||||
|
|
@ -25,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])
|
||||
|
|
@ -40,8 +42,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
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, {})
|
||||
|
|
@ -56,7 +61,7 @@ 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
|
||||
|
|
@ -64,7 +69,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
|
||||
|
|
@ -84,7 +89,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)
|
||||
|
|
@ -92,7 +97,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)
|
||||
|
|
@ -104,11 +109,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"}
|
||||
|
|
@ -118,7 +124,10 @@ 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:
|
||||
|
|
|
|||
|
|
@ -1,5 +1,8 @@
|
|||
"""Constants for the Adaptive Lighting integration."""
|
||||
|
||||
from typing import Any, List, Tuple, Optional
|
||||
from datetime import timedelta
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
import voluptuous as vol
|
||||
from homeassistant.components.light import VALID_TRANSITION
|
||||
|
|
@ -290,14 +293,12 @@ DOCS_APPLY = {
|
|||
CONF_LIGHTS: "A light (or list of lights) to apply the settings to. 💡",
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
(CONF_INTERVAL, DEFAULT_INTERVAL, cv.positive_int),
|
||||
(CONF_TRANSITION, DEFAULT_TRANSITION, VALID_TRANSITION),
|
||||
(CONF_INITIAL_TRANSITION, DEFAULT_INITIAL_TRANSITION, VALID_TRANSITION),
|
||||
|
|
@ -310,7 +311,7 @@ VALIDATION_TUPLES = [
|
|||
(
|
||||
CONF_SLEEP_RGB_OR_COLOR_TEMP,
|
||||
DEFAULT_SLEEP_RGB_OR_COLOR_TEMP,
|
||||
selector.SelectSelector(
|
||||
selector.SelectSelector( # type: ignore
|
||||
selector.SelectSelectorConfig(
|
||||
options=["color_temp", "rgb_color"],
|
||||
multiple=False,
|
||||
|
|
@ -322,7 +323,7 @@ VALIDATION_TUPLES = [
|
|||
(
|
||||
CONF_SLEEP_RGB_COLOR,
|
||||
DEFAULT_SLEEP_RGB_COLOR,
|
||||
selector.ColorRGBSelector(selector.ColorRGBSelectorConfig()),
|
||||
selector.ColorRGBSelector(selector.ColorRGBSelectorConfig()), # type: ignore
|
||||
),
|
||||
(CONF_SLEEP_TRANSITION, DEFAULT_SLEEP_TRANSITION, VALID_TRANSITION),
|
||||
(CONF_ADAPT_UNTIL_SLEEP, DEFAULT_ADAPT_UNTIL_SLEEP, bool),
|
||||
|
|
@ -337,7 +338,7 @@ VALIDATION_TUPLES = [
|
|||
(
|
||||
CONF_BRIGHTNESS_MODE,
|
||||
DEFAULT_BRIGHTNESS_MODE,
|
||||
selector.SelectSelector(
|
||||
selector.SelectSelector( # type: ignore
|
||||
selector.SelectSelectorConfig(
|
||||
options=["default", "linear", "tanh"],
|
||||
multiple=False,
|
||||
|
|
@ -369,8 +370,7 @@ VALIDATION_TUPLES = [
|
|||
(CONF_INCLUDE_CONFIG_IN_ATTRIBUTES, DEFAULT_INCLUDE_CONFIG_IN_ATTRIBUTES, bool),
|
||||
]
|
||||
|
||||
|
||||
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 +380,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,15 +395,14 @@ 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:
|
||||
return vol.All(validation, vol.Coerce(coerce))
|
||||
return validation
|
||||
|
||||
|
||||
def replace_none_str(value, replace_with=None):
|
||||
def replace_none_str(value: Any, replace_with: Optional[Any] = None) -> Any:
|
||||
"""Replace "None" -> replace_with."""
|
||||
return value if value != NONE_STR else replace_with
|
||||
|
||||
|
|
@ -421,12 +420,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
|
||||
vol.Optional(CONF_LIGHTS, default=[]): cv.entity_ids, # type: ignore
|
||||
vol.Optional(
|
||||
CONF_TRANSITION,
|
||||
default=initial_transition,
|
||||
|
|
@ -441,8 +440,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
|
||||
vol.Optional(CONF_LIGHTS, default=[]): cv.entity_ids, # type: ignore
|
||||
vol.Optional(CONF_MANUAL_CONTROL, default=True): cv.boolean,
|
||||
},
|
||||
)
|
||||
|
|
|
|||
|
|
@ -27,7 +27,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
|
||||
)
|
||||
except AttributeError as error:
|
||||
msg = (
|
||||
|
|
@ -48,7 +48,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)
|
||||
|
|
@ -59,7 +61,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,
|
||||
|
|
@ -68,7 +76,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,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
"""Helper functions for the Adaptive Lighting custom components."""
|
||||
|
||||
from __future__ import annotations
|
||||
from typing import Any, Dict
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
import base64
|
||||
import math
|
||||
|
|
@ -83,3 +85,11 @@ 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
|
||||
|
|
@ -229,11 +229,12 @@ def _switches_with_lights(
|
|||
hass: HomeAssistant,
|
||||
lights: list[str],
|
||||
expand_light_groups: bool = True,
|
||||
) -> list[AdaptiveSwitch]:
|
||||
) -> list["AdaptiveSwitch"]:
|
||||
"""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 = []
|
||||
from typing import List
|
||||
switches: List["AdaptiveSwitch"] = []
|
||||
all_check_lights = (
|
||||
_expand_light_groups(hass, lights) if expand_light_groups else set(lights)
|
||||
)
|
||||
|
|
@ -370,7 +371,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(
|
||||
|
|
@ -390,7 +391,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]
|
||||
|
|
@ -457,7 +458,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(
|
||||
|
|
@ -489,7 +490,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(
|
||||
|
|
@ -584,7 +585,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
|
||||
|
|
@ -599,7 +600,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)
|
||||
|
|
@ -622,19 +623,18 @@ def _is_light_group(state: State) -> bool:
|
|||
False,
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
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
|
||||
color_modes = {
|
||||
ColorMode.RGB,
|
||||
ColorMode.RGBW,
|
||||
|
|
@ -659,7 +659,6 @@ def _supported_features(hass: HomeAssistant, light: str) -> set[str]:
|
|||
|
||||
return supported
|
||||
|
||||
|
||||
# All comparisons should be done with RGB since
|
||||
# converting anything to color temp is inaccurate.
|
||||
def _convert_attributes(attributes: dict[str, Any]) -> dict[str, Any]:
|
||||
|
|
@ -775,7 +774,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
hass,
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
manager: AdaptiveLightingManager,
|
||||
sleep_mode_switch: SimpleSwitch,
|
||||
|
|
@ -829,7 +828,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,
|
||||
|
|
@ -921,12 +920,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
|
||||
|
||||
|
|
@ -963,11 +962,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)
|
||||
|
|
@ -1124,7 +1123,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
|
||||
|
|
@ -1132,7 +1131,7 @@ 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) -> None: # noqa: ARG002
|
||||
"""Update the attributes and maybe adapt the lights."""
|
||||
await self._update_attrs_and_maybe_adapt_lights(
|
||||
context=self.create_context("interval"),
|
||||
|
|
@ -1265,7 +1264,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
|
||||
|
|
@ -1316,7 +1315,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.,
|
||||
|
|
@ -1572,12 +1571,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
|
||||
|
||||
|
|
@ -1613,12 +1612,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
|
||||
|
|
@ -1706,7 +1705,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()
|
||||
|
|
@ -1928,7 +1927,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)
|
||||
|
|
@ -2106,7 +2105,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",
|
||||
|
|
@ -2115,7 +2114,7 @@ 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
|
||||
|
|
@ -2138,7 +2137,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.",
|
||||
|
|
@ -2164,7 +2163,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)
|
||||
|
|
@ -2193,7 +2192,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:
|
||||
|
|
@ -2243,11 +2242,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()
|
||||
|
|
@ -2672,14 +2671,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."""
|
||||
self.start_time = dt_util.utcnow()
|
||||
await asyncio.sleep(self.delay)
|
||||
|
|
@ -2689,23 +2688,23 @@ 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()
|
||||
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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue