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

Style with ruff and logging (#643)

Browse source 
* Formatting with ruff

* more logging

* Require on_only

* Different implementation

* More ruff

* Remove new logging statements

* Remove 'pylint: disable=protected'

* style

* fixes

* ruff

* Switch to ruff

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix .github

* Fix

* Add ignores

* fix arg

* fix B905

* Fix D

* Fix more

* Fix more

* order

* use path

* moer path

* Remove unused ignores

* fix tes deps

* fix typo

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Bas Nijholt 2023-07-23 14:24:22 -07:00 • committed by GitHub
commit bb84684bce
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 326 additions and 246 deletions
Download patch file Download diff file Expand all files Collapse all files

9
.github/update-services.py vendored
View file

@ -1,5 +1,6 @@
from pathlib import Path
"""Creates a services.yaml file with the latest docs."""
import sys
from pathlib import Path
import yaml
@ -7,8 +8,8 @@ sys.path.append(str(Path(__file__).parent.parent))
from custom_components.adaptive_lighting import const # noqa: E402
services_filename = "custom_components/adaptive_lighting/services.yaml"
with open(services_filename) as f:
services_filename = Path("custom_components") / "adaptive_lighting" / "services.yaml"
with open(services_filename) as f: # noqa: PTH123
services = yaml.safe_load(f)
for service_name, dct in services.items():
@ -20,6 +21,6 @@ for service_name, dct in services.items():
comment = "# This file is auto-generated by .github/update-services.py."
with open(services_filename, "w") as f:
with services_filename.open("w") as f:
f.write(comment + "\n")
yaml.dump(services, f, sort_keys=False, width=1000, allow_unicode=True)

16
.github/update-strings.py vendored
View file

@ -1,31 +1,33 @@
"""Update strings.json and en.json from const.py."""
import json
from pathlib import Path
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).parent.parent))
from custom_components.adaptive_lighting import const # noqa: E402
strings_fname = "custom_components/adaptive_lighting/strings.json"
en_fname = "custom_components/adaptive_lighting/translations/en.json"
with open(strings_fname) as f:
folder = Path("custom_components") / "adaptive_lighting"
strings_fname = folder / "strings.json"
en_fname = folder / "translations" / "en.json"
with strings_fname.open() as f:
strings = json.load(f)
data = {k: f"{k}: {const.DOCS[k]}" for k, _, _ in const.VALIDATION_TUPLES}
strings["options"]["step"]["init"]["data"] = data
with open(strings_fname, "w") as f:
with strings_fname.open("w") as f:
json.dump(strings, f, indent=2, ensure_ascii=False)
f.write("\n")
# Sync changes from strings.json to en.json
with open(en_fname) as f:
with en_fname.open() as f:
en = json.load(f)
en["config"]["step"]["user"] = strings["config"]["step"]["user"]
en["options"]["step"]["init"]["data"] = data
with open(en_fname, "w") as f:
with en_fname.open("w") as f:
json.dump(en, f, indent=2, ensure_ascii=False)
f.write("\n")

16
.pre-commit-config.yaml
View file

@ -7,20 +7,12 @@ repos:
- id: end-of-file-fixer
- id: mixed-line-ending
args: ["--fix=lf"]
- repo: https://github.com/pycqa/flake8
rev: 6.0.0
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.0.279
hooks:
- id: flake8
- id: ruff
args: ["--fix"]
- repo: https://github.com/psf/black
rev: 23.3.0
hooks:
- id: black
- repo: https://github.com/asottile/pyupgrade
rev: v3.7.0
hooks:
- id: pyupgrade
args: ["--py39-plus"]
- repo: https://github.com/PyCQA/isort
rev: 5.12.0
hooks:
- id: isort

55
.ruff.toml
View file

@ -2,41 +2,30 @@
target-version = "py310"
select = [
"B007", # Loop control variable {name} not used within loop body
"B014", # Exception handler with duplicate exception
"C", # complexity
"D", # docstrings
"E", # pycodestyle
"F", # pyflakes/autoflake
"ICN001", # import concentions; {name} should be imported as {asname}
"PGH004", # Use specific rule codes when using noqa
"PLC0414", # Useless import alias. Import alias does not rename original package.
"SIM105", # Use contextlib.suppress({exception}) instead of try-except-pass
"SIM117", # Merge with-statements that use the same scope
"SIM118", # Use {key} in {dict} instead of {key} in {dict}.keys()
"SIM201", # Use {left} != {right} instead of not {left} == {right}
"SIM212", # Use {a} if {a} else {b} instead of {b} if not {a} else {a}
"SIM300", # Yoda conditions. Use 'age == 42' instead of '42 == age'.
"SIM401", # Use get from dict with default instead of an if block
"T20", # flake8-print
"TRY004", # Prefer TypeError exception for invalid type
"RUF006", # Store a reference to the return value of asyncio.create_task
"UP", # pyupgrade
"W", # pycodestyle
select = ["ALL"]
# All the ones without a comment were the ones that are currently violated
# by the codebase. The plan is to fix them all (when sensible) and then enable them.
ignore = [
"ANN",
"ANN101", # Missing type annotation for {name} in method
"ANN401", # Dynamically typed expressions (typing.Any) are disallowed in {name}
"D401", # First line of docstring should be in imperative mood
"E501", # line too long
"FBT001", # Boolean positional arg in function definition
"FBT002", # Boolean default value in function definition
"FIX004", # Line contains HACK, consider resolving the issue
"PD901", # df is a bad variable name. Be kinder to your future self.
"PERF203",# `try`-`except` within a loop incurs performance overhead
"PLR0913", # Too many arguments to function call (N > 5)
"PLR2004", # Magic value used in comparison, consider replacing X with a constant variable
"S101", # Use of assert detected
"SLF001", # Private member accessed
]
ignore = [
"D202", # No blank lines allowed after function docstring
"D203", # 1 blank line required before class docstring
"D213", # Multi-line docstring summary should start at the second line
"D404", # First word of the docstring should not be This
"D406", # Section name should end with a newline
"D407", # Section name underlining
"D411", # Missing blank line before section
"E501", # line too long
"E731", # do not assign a lambda expression, use a def
]
[per-file-ignores]
"tests/*.py" = ["ALL"]
".github/*py" = ["INP001"]
[flake8-pytest-style]
fixture-parentheses = false

18
custom_components/adaptive_lighting/__init__.py Executable file → Normal file
View file

@ -2,11 +2,11 @@
import logging
from typing import Any
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
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from .const import (
_DOMAIN_SCHEMA,
@ -35,20 +35,21 @@ CONFIG_SCHEMA = vol.Schema(
)
async def reload_configuration_yaml(event: dict, hass: HomeAssistant):
async def reload_configuration_yaml(event: dict, hass: HomeAssistant): # noqa: ARG001
"""Reload configuration.yaml."""
await hass.services.async_call("homeassistant", "check_config", {})
async def async_setup(hass: HomeAssistant, config: dict[str, Any]):
"""Import integration from config."""
if DOMAIN in config:
for entry in config[DOMAIN]:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_IMPORT}, data=entry
)
DOMAIN,
context={CONF_SOURCE: SOURCE_IMPORT},
data=entry,
),
)
return True
@ -65,7 +66,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
data[config_entry.entry_id] = {UNDO_UPDATE_LISTENER: undo_listener}
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, platform)
hass.config_entries.async_forward_entry_setup(config_entry, platform),
)
return True
@ -79,7 +80,8 @@ async def async_update_options(hass, config_entry: ConfigEntry):
async def async_unload_entry(hass, config_entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_forward_entry_unload(
config_entry, "switch"
config_entry,
"switch",
)
data = hass.data[DOMAIN]
data[config_entry.entry_id][UNDO_UPDATE_LISTENER]()

35
custom_components/adaptive_lighting/_docs_helpers.py
View file

@ -1,9 +1,9 @@
from typing import Any
from homeassistant.helpers import selector
import homeassistant.helpers.config_validation as cv
import pandas as pd
import voluptuous as vol
from homeassistant.helpers import selector
from .const import (
DOCS,
@ -23,38 +23,37 @@ def _format_voluptuous_instance(instance):
for validator in instance.validators:
if isinstance(validator, vol.Coerce):
coerce_type = validator.type.__name__
elif isinstance(validator, (vol.Clamp, vol.Range)):
elif isinstance(validator, vol.Clamp | vol.Range):
min_val = validator.min
max_val = validator.max
if min_val is not None and max_val is not None:
return f"`{coerce_type}` {min_val}-{max_val}"
elif min_val is not None:
if min_val is not None:
return f"`{coerce_type} > {min_val}`"
elif max_val is not None:
if max_val is not None:
return f"`{coerce_type} < {max_val}`"
else:
return f"`{coerce_type}`"
return f"`{coerce_type}`"
def _type_to_str(type_: Any) -> str:
def _type_to_str(type_: Any) -> str: # noqa: PLR0911
"""Convert a (voluptuous) type to a string."""
if type_ == cv.entity_ids:
return "list of `entity_id`s"
elif type_ in (bool, int, float, str):
if type_ in (bool, int, float, str):
return f"`{type_.__name__}`"
elif type_ == cv.boolean:
if type_ == cv.boolean:
return "bool"
elif isinstance(type_, vol.All):
if isinstance(type_, vol.All):
return _format_voluptuous_instance(type_)
elif isinstance(type_, vol.In):
if isinstance(type_, vol.In):
return f"one of `{type_.container}`"
elif isinstance(type_, selector.SelectSelector):
if isinstance(type_, selector.SelectSelector):
return f"one of `{type_.config['options']}`"
elif isinstance(type_, selector.ColorRGBSelector):
if isinstance(type_, selector.ColorRGBSelector):
return "RGB color"
else:
raise ValueError(f"Unknown type: {type_}")
msg = f"Unknown type: {type_}"
raise ValueError(msg)
def generate_config_markdown_table():
@ -85,7 +84,8 @@ def _schema_to_dict(schema: vol.Schema) -> dict[str, tuple[Any, Any]]:
def _generate_service_markdown_table(
schema: dict[str, tuple[Any, Any]], alternative_docs: dict[str, str] = None
schema: dict[str, tuple[Any, Any]],
alternative_docs: dict[str, str] | None = None,
):
schema = _schema_to_dict(schema)
rows = []
@ -112,5 +112,6 @@ def generate_apply_markdown_table():
def generate_set_manual_control_markdown_table():
return _generate_service_markdown_table(
SET_MANUAL_CONTROL_SCHEMA, DOCS_MANUAL_CONTROL
SET_MANUAL_CONTROL_SCHEMA,
DOCS_MANUAL_CONTROL,
)

37
custom_components/adaptive_lighting/adaptation_utils.py
View file

@ -1,7 +1,7 @@
"""Utility functions for adaptation commands."""
import logging
from collections.abc import AsyncGenerator
from dataclasses import dataclass
import logging
from typing import Any, Literal
from homeassistant.components.light import (
@ -40,10 +40,10 @@ ServiceData = dict[str, Any]
def _split_service_call_data(service_data: ServiceData) -> list[ServiceData]:
"""Splits the service data by the adapted attributes, i.e., into separate data
items for brightness and color.
"""
"""Splits the service data by the adapted attributes.
i.e., into separate data items for brightness and color.
"""
common_attrs = {ATTR_ENTITY_ID}
common_data = {k: service_data[k] for k in common_attrs if k in service_data}
@ -70,13 +70,14 @@ def _split_service_call_data(service_data: ServiceData) -> list[ServiceData]:
def _remove_redundant_attributes(
service_data: ServiceData, state: State
service_data: ServiceData,
state: State,
) -> ServiceData:
"""Filter service data by removing attributes that already equal the given state.
Removes all attributes from service call data whose values are already present
in the target entity's state."""
in the target entity's state.
"""
return {
k: v
for k, v in service_data.items()
@ -88,7 +89,8 @@ def _has_relevant_service_data_attributes(service_data: ServiceData) -> bool:
"""Determines whether the service data justifies an adaptation service call.
A service call is not justified for data which does not contain any entries that
change relevant attributes of an adapting entity, e.g., brightness or color."""
change relevant attributes of an adapting entity, e.g., brightness or color.
"""
common_attrs = {ATTR_ENTITY_ID, ATTR_TRANSITION}
return any(attr not in common_attrs for attr in service_data)
@ -108,15 +110,15 @@ async def _create_service_call_data_iterator(
at the time when the service data is read instead of up front. This gives greater
flexibility because entity states can change while the items are iterated.
"""
for service_data in service_datas:
if filter_by_state and (entity_id := service_data.get(ATTR_ENTITY_ID)):
current_entity_state = hass.states.get(entity_id)
# Filter data to remove attributes that equal the current state
if current_entity_state is not None:
service_data = _remove_redundant_attributes(
service_data, current_entity_state
service_data = _remove_redundant_attributes( # noqa: PLW2901
service_data,
state=current_entity_state,
)
# Emit service data if it still contains relevant attributes (else try next)
@ -143,7 +145,7 @@ class AdaptationData:
return await anext(self.service_call_datas, None)
class NoColorOrBrightnessInServiceData(Exception):
class NoColorOrBrightnessInServiceDataError(Exception):
"""Exception raised when no color or brightness attributes are found in service data."""
@ -160,7 +162,7 @@ def _identify_lighting_type(
if has_color:
return "color"
msg = f"Invalid service_data, no brightness or color attributes found: {service_data=}"
raise NoColorOrBrightnessInServiceData(msg)
raise NoColorOrBrightnessInServiceDataError(msg)
def prepare_adaptation_data(
@ -179,10 +181,7 @@ def prepare_adaptation_data(
entity_id,
service_data,
)
if split:
service_datas = _split_service_call_data(service_data)
else:
service_datas = [service_data]
service_datas = _split_service_call_data(service_data) if split else [service_data]
service_datas_length = len(service_datas)
@ -193,7 +192,9 @@ def prepare_adaptation_data(
sleep_time = split_delay
service_data_iterator = _create_service_call_data_iterator(
hass, service_datas, filter_by_state
hass,
service_datas,
filter_by_state,
)
lighting_type = _identify_lighting_type(service_data)

10
custom_components/adaptive_lighting/config_flow.py
View file

@ -1,11 +1,11 @@
"""Config flow for Adaptive Lighting integration."""
import logging
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_NAME
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from .const import ( # pylint: disable=unused-import
CONF_LIGHTS,
@ -75,7 +75,7 @@ def validate_options(user_input, errors):
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for Adaptive Lighting."""
def __init__(self, config_entry: config_entries.ConfigEntry):
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
@ -114,5 +114,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
options_schema[key] = value
return self.async_show_form(
step_id="init", data_schema=vol.Schema(options_schema), errors=errors
step_id="init",
data_schema=vol.Schema(options_schema),
errors=errors,
)

34
custom_components/adaptive_lighting/const.py
View file

@ -1,10 +1,10 @@
"""Constants for the Adaptive Lighting integration."""
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant.components.light import VALID_TRANSITION
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.helpers import selector
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
ICON_MAIN = "mdi:theme-light-dark"
ICON_BRIGHTNESS = "mdi:brightness-4"
@ -49,9 +49,9 @@ DOCS[CONF_INITIAL_TRANSITION] = (
)
CONF_SLEEP_TRANSITION, DEFAULT_SLEEP_TRANSITION = "sleep_transition", 1
DOCS[CONF_SLEEP_TRANSITION] = (
'Duration of transition when "sleep mode" is toggled ' "in seconds. 😴"
)
DOCS[
CONF_SLEEP_TRANSITION
] = 'Duration of transition when "sleep mode" is toggled in seconds. 😴'
CONF_INTERVAL, DEFAULT_INTERVAL = "interval", 90
DOCS[CONF_INTERVAL] = "Frequency to adapt the lights, in seconds. 🔄"
@ -99,22 +99,22 @@ DOCS[CONF_SLEEP_COLOR_TEMP] = (
)
CONF_SLEEP_RGB_COLOR, DEFAULT_SLEEP_RGB_COLOR = "sleep_rgb_color", [255, 56, 0]
DOCS[CONF_SLEEP_RGB_COLOR] = (
"RGB color in sleep mode (used when " '`sleep_rgb_or_color_temp` is "rgb_color"). 🌈'
)
DOCS[
CONF_SLEEP_RGB_COLOR
] = 'RGB color in sleep mode (used when `sleep_rgb_or_color_temp` is "rgb_color"). 🌈'
CONF_SLEEP_RGB_OR_COLOR_TEMP, DEFAULT_SLEEP_RGB_OR_COLOR_TEMP = (
"sleep_rgb_or_color_temp",
"color_temp",
)
DOCS[CONF_SLEEP_RGB_OR_COLOR_TEMP] = (
'Use either `"rgb_color"` or `"color_temp"` ' "in sleep mode. 🌙"
)
DOCS[
CONF_SLEEP_RGB_OR_COLOR_TEMP
] = 'Use either `"rgb_color"` or `"color_temp"` in sleep mode. 🌙'
CONF_SUNRISE_OFFSET, DEFAULT_SUNRISE_OFFSET = "sunrise_offset", 0
DOCS[CONF_SUNRISE_OFFSET] = (
"Adjust sunrise time with a positive or negative offset " "in seconds. ⏰"
)
DOCS[
CONF_SUNRISE_OFFSET
] = "Adjust sunrise time with a positive or negative offset in seconds. ⏰"
CONF_SUNRISE_TIME = "sunrise_time"
DOCS[CONF_SUNRISE_TIME] = "Set a fixed time (HH:MM:SS) for sunrise. 🌅"
@ -333,7 +333,7 @@ _DOMAIN_SCHEMA = vol.Schema(
{
vol.Optional(key, default=replace_none_str(default, vol.UNDEFINED)): validation
for key, default, validation in _yaml_validation_tuples
}
},
)
@ -351,7 +351,7 @@ def apply_service_schema(initial_transition: int = 1):
vol.Optional(ATTR_ADAPT_COLOR, default=True): cv.boolean,
vol.Optional(CONF_PREFER_RGB_COLOR, default=False): cv.boolean,
vol.Optional(CONF_TURN_ON_LIGHTS, default=False): cv.boolean,
}
},
)
@ -360,5 +360,5 @@ 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_MANUAL_CONTROL, default=True): cv.boolean,
}
},
)

30
custom_components/adaptive_lighting/hass_utils.py
View file

@ -1,6 +1,5 @@
"""Utility functions for HA core."""
from collections.abc import Awaitable
from typing import Callable
from collections.abc import Awaitable, Callable
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.util.read_only_dict import ReadOnlyDict
@ -17,7 +16,8 @@ def setup_service_call_interceptor(
"""Inject a function into a registered service call to preprocess service data.
The injected interceptor function receives the service call and a writeable data dictionary
(the data of the service call is read-only) before the service call is executed."""
(the data of the service call is read-only) before the service call is executed.
"""
try:
# HACK: Access protected attribute of HA service registry.
# This is necessary to replace a registered service handler with our
@ -26,15 +26,15 @@ def setup_service_call_interceptor(
hass.services._services # pylint: disable=protected-access
)
except AttributeError as error:
raise RuntimeError(
"Intercept failed because registered services are no longer accessible "
"(internal API may have changed)"
) from error
msg = (
"Intercept failed because registered services are no longer"
" accessible (internal API may have changed)"
)
raise RuntimeError(msg) from error
if domain not in registered_services or service not in registered_services[domain]:
raise RuntimeError(
f"Intercept failed because service {domain}.{service} is not registered"
)
msg = f"Intercept failed because service {domain}.{service} is not registered"
raise RuntimeError(msg)
existing_service = registered_services[domain][service]
@ -52,13 +52,19 @@ def setup_service_call_interceptor(
await existing_service.job.target(call)
hass.services.async_register(
domain, service, service_func_proxy, existing_service.schema
domain,
service,
service_func_proxy,
existing_service.schema,
)
def remove():
# Remove the interceptor by reinstalling the original service handler
hass.services.async_register(
domain, service, existing_service.job.target, existing_service.schema
domain,
service,
existing_service.job.target,
existing_service.schema,
)
return remove

304
custom_components/adaptive_lighting/switch.py
View file

@ -4,17 +4,19 @@ from __future__ import annotations
import asyncio
import base64
import bisect
from collections.abc import Callable, Coroutine, Iterable
from copy import deepcopy
from dataclasses import dataclass
import datetime
from datetime import timedelta
import functools
import logging
import math
from typing import Any, Literal
from copy import deepcopy
from dataclasses import dataclass
from datetime import timedelta
from typing import TYPE_CHECKING, Any, Literal
import astral
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
import ulid_transform
import voluptuous as vol
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP_KELVIN,
@ -29,8 +31,6 @@ from homeassistant.components.light import (
COLOR_MODE_RGBW,
COLOR_MODE_RGBWW,
COLOR_MODE_XY,
)
from homeassistant.components.light import (
SUPPORT_BRIGHTNESS,
SUPPORT_COLOR,
SUPPORT_COLOR_TEMP,
@ -41,7 +41,6 @@ from homeassistant.components.light import (
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DOMAIN,
@ -71,7 +70,6 @@ from homeassistant.core import (
callback,
)
from homeassistant.helpers import entity_platform, entity_registry
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import (
async_track_state_change_event,
async_track_time_interval,
@ -86,9 +84,6 @@ from homeassistant.util.color import (
color_xy_to_hs,
color_xy_to_RGB,
)
import homeassistant.util.dt as dt_util
import ulid_transform
import voluptuous as vol
from .adaptation_utils import (
BRIGHTNESS_ATTRS,
@ -156,6 +151,13 @@ from .const import (
)
from .hass_utils import setup_service_call_interceptor
if TYPE_CHECKING:
from collections.abc import Callable, Coroutine, Iterable
import astral
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity_platform import AddEntitiesCallback
_SUPPORT_OPTS = {
"brightness": SUPPORT_BRIGHTNESS,
"color_temp": SUPPORT_COLOR_TEMP,
@ -209,17 +211,17 @@ def _int_to_base36(num: int) -> str:
>>> print(base36_num)
'2N9'
"""
ALPHANUMERIC_CHARS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
alphanumeric_chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
if num == 0:
return ALPHANUMERIC_CHARS[0]
return alphanumeric_chars[0]
base36_str = ""
base = len(ALPHANUMERIC_CHARS)
base = len(alphanumeric_chars)
while num:
num, remainder = divmod(num, base)
base36_str = ALPHANUMERIC_CHARS[remainder] + base36_str
base36_str = alphanumeric_chars[remainder] + base36_str
return base36_str
@ -236,7 +238,10 @@ def _remove_vowels(input_str: str, length: int = 4) -> str:
def create_context(
name: str, which: str, index: int, parent: Context | None = None
name: str,
which: str,
index: int,
parent: Context | None = None,
) -> Context:
"""Create a context that can identify this integration."""
# Use a hash for the name because otherwise the context might become
@ -255,6 +260,7 @@ def create_context(
def is_our_context_id(context_id: str | None) -> bool:
"""Check whether this integration created 'context_id'."""
if context_id is None:
return False
return f":{_DOMAIN_SHORT}:" in context_id
@ -268,7 +274,8 @@ def is_our_context(context: Context | None) -> bool:
def _switches_with_lights(
hass: HomeAssistant, lights: list[str]
hass: HomeAssistant,
lights: list[str],
) -> list[AdaptiveSwitch]:
"""Get all switches that control at least one of the lights passed."""
config_entries = hass.config_entries.async_entries(DOMAIN)
@ -299,48 +306,52 @@ def _switch_with_lights(
switches = _switches_with_lights(hass, lights)
if len(switches) == 1:
return switches[0]
elif len(switches) > 1:
if len(switches) > 1:
on_switches = [s for s in switches if s.is_on]
if len(on_switches) == 1:
# Of the multiple switches, only one is on
return on_switches[0]
raise NoSwitchFoundError(
msg = (
f"_switch_with_lights: Light(s) {lights} found in multiple switch configs"
f" ({[s.entity_id for s in switches]}). You must pass a switch under"
f" 'entity_id'."
)
else:
raise NoSwitchFoundError(
f"_switch_with_lights: Light(s) {lights} not found in any switch's"
f" configuration. You must either include the light(s) that is/are"
f" in the integration config, or pass a switch under 'entity_id'."
" 'entity_id'."
)
raise NoSwitchFoundError(msg)
msg = (
f"_switch_with_lights: Light(s) {lights} not found in any switch's"
" configuration. You must either include the light(s) that is/are"
" in the integration config, or pass a switch under 'entity_id'."
)
raise NoSwitchFoundError(msg)
# For documentation on this function, see integration_entities() from HomeAssistant Core:
# https://github.com/home-assistant/core/blob/dev/homeassistant/helpers/template.py#L1109
def _switches_from_service_call(
hass: HomeAssistant, service_call: ServiceCall
hass: HomeAssistant,
service_call: ServiceCall,
) -> list[AdaptiveSwitch]:
data = service_call.data
lights = data[CONF_LIGHTS]
switch_entity_ids: list[str] | None = data.get("entity_id")
if not lights and not switch_entity_ids:
raise ValueError(
msg = (
"adaptive-lighting: Neither a switch nor a light was provided in the service call."
" If you intend to adapt all lights on all switches, please inform the developers at"
" https://github.com/basnijholt/adaptive-lighting about your use case."
" Currently, you must pass either an adaptive-lighting switch or the lights to an"
" `adaptive_lighting` service call."
" If you intend to adapt all lights on all switches, please inform the"
" developers at https://github.com/basnijholt/adaptive-lighting about your"
" use case. Currently, you must pass either an adaptive-lighting switch or"
" the lights to an `adaptive_lighting` service call."
)
raise ValueError(msg)
if switch_entity_ids is not None:
if len(switch_entity_ids) > 1 and lights:
raise ValueError(
f"adaptive-lighting: Cannot pass multiple switches with lights argument."
msg = (
"adaptive-lighting: Cannot pass multiple switches with lights argument."
f" Invalid service data received: {service_call.data}"
)
raise ValueError(msg)
switches = []
ent_reg = entity_registry.async_get(hass)
for entity_id in switch_entity_ids:
@ -353,14 +364,16 @@ def _switches_from_service_call(
switch = _switch_with_lights(hass, lights)
return [switch]
raise ValueError(
f"adaptive-lighting: Incorrect data provided in service call."
msg = (
"adaptive-lighting: Incorrect data provided in service call."
f" Entities not found in the integration. Service data: {service_call.data}"
)
raise ValueError(msg)
async def handle_change_switch_settings(
switch: AdaptiveSwitch, service_call: ServiceCall
switch: AdaptiveSwitch,
service_call: ServiceCall,
) -> None:
"""Allows HASS to change config values via a service call."""
data = service_call.data
@ -396,7 +409,10 @@ async def handle_change_switch_settings(
@callback
def _fire_manual_control_event(
switch: AdaptiveSwitch, light: str, context: Context, is_async=True
switch: AdaptiveSwitch,
light: str,
context: Context,
is_async: bool = True,
):
"""Fire an event that 'light' is marked as manual_control."""
hass = switch.hass
@ -414,8 +430,10 @@ def _fire_manual_control_event(
)
async def async_setup_entry(
hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: bool
async def async_setup_entry( # noqa: PLR0915
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
):
"""Set up the AdaptiveLighting switch."""
data = hass.data[DOMAIN]
@ -423,17 +441,30 @@ async def async_setup_entry(
if ATTR_ADAPTIVE_LIGHTING_MANAGER not in data:
data[ATTR_ADAPTIVE_LIGHTING_MANAGER] = AdaptiveLightingManager(
hass, config_entry
hass,
config_entry,
)
manager: AdaptiveLightingManager = data[ATTR_ADAPTIVE_LIGHTING_MANAGER]
sleep_mode_switch = SimpleSwitch(
"Sleep Mode", False, hass, config_entry, ICON_SLEEP
which="Sleep Mode",
initial_state=False,
hass=hass,
config_entry=config_entry,
icon=ICON_SLEEP,
)
adapt_color_switch = SimpleSwitch(
"Adapt Color", True, hass, config_entry, ICON_COLOR_TEMP
which="Adapt Color",
initial_state=True,
hass=hass,
config_entry=config_entry,
icon=ICON_COLOR_TEMP,
)
adapt_brightness_switch = SimpleSwitch(
"Adapt Brightness", True, hass, config_entry, ICON_BRIGHTNESS
which="Adapt Brightness",
initial_state=True,
hass=hass,
config_entry=config_entry,
icon=ICON_BRIGHTNESS,
)
switch = AdaptiveSwitch(
hass,
@ -482,7 +513,8 @@ async def async_setup_entry(
data[ATTR_ADAPT_COLOR],
data[CONF_PREFER_RGB_COLOR],
context=switch.create_context(
"service", parent=service_call.context
"service",
parent=service_call.context,
),
)
@ -513,7 +545,8 @@ async def async_setup_entry(
transition=switch.initial_transition,
force=True,
context=switch.create_context(
"service", parent=service_call.context
"service",
parent=service_call.context,
),
)
@ -637,7 +670,8 @@ def _supported_features(hass: HomeAssistant, light: str) -> set[str]:
def color_difference_redmean(
rgb1: tuple[float, float, float], rgb2: tuple[float, float, float]
rgb1: tuple[float, float, float],
rgb2: tuple[float, float, float],
) -> float:
"""Distance between colors in RGB space (redmean metric).
@ -648,7 +682,9 @@ def color_difference_redmean(
- https://www.compuphase.com/cmetric.htm
"""
r_hat = (rgb1[0] + rgb2[0]) / 2
delta_r, delta_g, delta_b = ((col1 - col2) for col1, col2 in zip(rgb1, rgb2))
delta_r, delta_g, delta_b = (
(col1 - col2) for col1, col2 in zip(rgb1, rgb2, strict=True)
)
red_term = (2 + r_hat / 256) * delta_r**2
green_term = 4 * delta_g**2
blue_term = (2 + (255 - r_hat) / 256) * delta_b**2
@ -700,7 +736,8 @@ def _attributes_have_changed(
) -> bool:
if adapt_color:
old_attributes, new_attributes = _add_missing_attributes(
old_attributes, new_attributes
old_attributes,
new_attributes,
)
if (
@ -771,7 +808,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
sleep_mode_switch: SimpleSwitch,
adapt_color_switch: SimpleSwitch,
adapt_brightness_switch: SimpleSwitch,
):
) -> None:
"""Initialize the Adaptive Lighting switch."""
# Set attributes that can't be modified during runtime
self.hass = hass
@ -920,7 +957,8 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
await self._setup_listeners()
else:
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STARTED, self._setup_listeners
EVENT_HOMEASSISTANT_STARTED,
self._setup_listeners,
)
last_state = await self.async_get_last_state()
is_new_entry = last_state is None # newly added to HA
@ -938,7 +976,8 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
all_lights = _expand_light_groups(self.hass, self.lights)
self.manager.lights.update(all_lights)
self.manager.set_auto_reset_manual_control_times(
all_lights, self._auto_reset_manual_control_time
all_lights,
self._auto_reset_manual_control_time,
)
self.lights = list(all_lights)
@ -963,7 +1002,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if self.lights:
self._expand_light_groups()
remove_state = async_track_state_change_event(
self.hass, entity_ids=self.lights, action=self._light_event_action
self.hass,
entity_ids=self.lights,
action=self._light_event_action,
)
self.remove_listeners.append(remove_state)
@ -1029,7 +1070,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
return extra_state_attributes
def create_context(
self, which: str = "default", parent: Context | None = None
self,
which: str = "default",
parent: Context | None = None,
) -> Context:
"""Create a context that identifies this Adaptive Lighting instance."""
context = create_context(self._name, which, self._context_cnt, parent=parent)
@ -1037,11 +1080,14 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
return context
async def async_turn_on( # pylint: disable=arguments-differ
self, adapt_lights: bool = True
self,
adapt_lights: bool = True,
) -> None:
"""Turn on adaptive lighting."""
_LOGGER.debug(
"%s: Called 'async_turn_on', current state is '%s'", self._name, self._state
"%s: Called 'async_turn_on', current state is '%s'",
self._name,
self._state,
)
if self.is_on:
return
@ -1055,7 +1101,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
context=self.create_context("turn_on"),
)
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs) -> None: # noqa: ARG002
"""Turn off adaptive lighting."""
if not self.is_on:
return
@ -1063,7 +1109,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
self._remove_listeners()
self.manager.reset(*self.lights)
async def _async_update_at_interval_action(self, now=None) -> None:
async def _async_update_at_interval_action(self, now=None) -> None: # noqa: ARG002
await self._update_attrs_and_maybe_adapt_lights(
transition=self._transition,
force=False,
@ -1079,6 +1125,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
prefer_rgb_color: bool | None = None,
context: Context | None = None,
) -> AdaptationData | None:
"""Prepare `AdaptationData` for adapting a light."""
if transition is None:
transition = self._transition
if adapt_brightness is None:
@ -1099,7 +1146,8 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
# The switch might be off and not have _settings set.
self._settings = self._sun_light_settings.get_settings(
self.sleep_mode_switch.is_on, transition
self.sleep_mode_switch.is_on,
transition,
)
# Build service data.
@ -1151,7 +1199,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
filter_by_state=self._skip_redundant_commands,
)
async def _adapt_light( # noqa: C901
async def _adapt_light(
self,
light: str,
transition: int | None = None,
@ -1167,7 +1215,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
if self.manager.is_proactively_adapting(context.parent_id):
# Skip if adaptation was already executed by the service call interceptor
_LOGGER.debug(
"%s: Skipping reactive adaptation of %s", self._name, context.parent_id
"%s: Skipping reactive adaptation of %s",
self._name,
context.parent_id,
)
return
@ -1180,13 +1230,12 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
context,
)
if data is None:
return None # nothing to adapt
return # nothing to adapt
await self.execute_cancellable_adaptation_calls(data)
async def _execute_adaptation_calls(self, data: AdaptationData):
"""Executes a sequence of adaptation service calls for the given service datas."""
for index in range(data.max_length):
is_first_call = index == 0
@ -1250,7 +1299,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
data,
)
async def _update_attrs_and_maybe_adapt_lights(
async def _update_attrs_and_maybe_adapt_lights( # noqa: PLR0912
self,
lights: list[str] | None = None,
transition: int | None = None,
@ -1270,8 +1319,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
assert self.is_on
self._settings.update(
self._sun_light_settings.get_settings(
self.sleep_mode_switch.is_on, transition
)
self.sleep_mode_switch.is_on,
transition,
),
)
self.async_write_ha_state()
@ -1352,7 +1402,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
_LOGGER.debug("%s: Ignoring sleep event %s", self._name, event)
return
_LOGGER.debug(
"%s: _sleep_mode_switch_state_event_action, event: '%s'", self._name, event
"%s: _sleep_mode_switch_state_event_action, event: '%s'",
self._name,
event,
)
# Reset the manually controlled status when the "sleep mode" changes
self.manager.reset(*self.lights)
@ -1380,7 +1432,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
)
if event.context.parent_id and not self.manager.is_proactively_adapting(
event.context.id
event.context.id,
):
self.manager.reset(entity_id, reset_manual_control=False)
@ -1395,7 +1447,9 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
):
# Stop if a rapid 'off' → 'on' → 'off' happens.
_LOGGER.debug(
"%s: Cancelling adjusting lights for %s", self._name, entity_id
"%s: Cancelling adjusting lights for %s",
self._name,
entity_id,
)
return
@ -1441,7 +1495,7 @@ class SimpleSwitch(SwitchEntity, RestoreEntity):
hass: HomeAssistant,
config_entry: ConfigEntry,
icon: str,
):
) -> None:
"""Initialize the Adaptive Lighting switch."""
self.hass = hass
data = validate(config_entry)
@ -1484,19 +1538,21 @@ class SimpleSwitch(SwitchEntity, RestoreEntity):
else:
await self.async_turn_off()
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs) -> 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:
async def async_turn_off(self, **kwargs) -> None: # noqa: ARG002
"""Turn off adaptive lighting sleep mode."""
_LOGGER.debug("%s: Turning off", self._name)
self._state = False
def lerp_color(
rgb1: tuple[int, int, int], rgb2: tuple[int, int, int], t: float
rgb1: tuple[int, int, int],
rgb2: tuple[int, int, int],
t: float,
) -> tuple[int, int, int]:
"""Linearly interpolate between two RGB colors."""
return (
@ -1536,13 +1592,13 @@ class SunLightSettings:
def _replace_time(date: datetime.datetime, key: str) -> datetime.datetime:
time = getattr(self, f"{key}_time")
date_time = datetime.datetime.combine(date, time)
utc_time = date_time.replace(tzinfo=dt_util.DEFAULT_TIME_ZONE).astimezone(
dt_util.UTC
return date_time.replace(tzinfo=dt_util.DEFAULT_TIME_ZONE).astimezone(
dt_util.UTC,
)
return utc_time
def calculate_noon_and_midnight(
sunset: datetime.datetime, sunrise: datetime.datetime
sunset: datetime.datetime,
sunrise: datetime.datetime,
) -> tuple[datetime.datetime, datetime.datetime]:
middle = abs(sunset - sunrise) / 2
if sunset > sunrise:
@ -1597,7 +1653,7 @@ class SunLightSettings:
]
# Check whether order is correct
events = sorted(events, key=lambda x: x[1])
events_names, _ = zip(*events)
events_names, _ = zip(*events, strict=True)
if events_names not in _ALLOWED_ORDERS:
msg = (
f"{self.name}: The sun events {events_names} are not in the expected"
@ -1635,8 +1691,7 @@ class SunLightSettings:
else (next_ts, prev_ts)
)
k = 1 if next_event in (SUN_EVENT_SUNSET, SUN_EVENT_NOON) else -1
percentage = (0 - k) * ((target_ts - h) / (h - x)) ** 2 + k
return percentage
return (0 - k) * ((target_ts - h) / (h - x)) ** 2 + k
def calc_brightness_pct(self, percent: float, is_sleep: bool) -> float:
"""Calculate the brightness in %."""
@ -1660,9 +1715,12 @@ class SunLightSettings:
delta = abs(self.min_color_temp - self.sleep_color_temp)
ct = (delta * abs(1 + percent)) + self.sleep_color_temp
return 5 * round(ct / 5) # round to nearest 5
return None
def get_settings(
self, is_sleep, transition
self,
is_sleep,
transition,
) -> dict[str, float | int | tuple[float, float] | tuple[float, float, float]]:
"""Get all light settings.
@ -1715,7 +1773,7 @@ class SunLightSettings:
class AdaptiveLightingManager:
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry):
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Initialize the AdaptiveLightingManager that is shared among all switches."""
self.hass = hass
data = validate(config_entry)
@ -1759,7 +1817,8 @@ class AdaptiveLightingManager:
self._proactively_adapting_contexts: dict[str, str] = {}
is_proactive_adaptation_enabled = data.get(
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION, True
INTERNAL_CONF_PROACTIVE_SERVICE_CALL_ADAPTATION,
True,
)
if is_proactive_adaptation_enabled:
@ -1770,7 +1829,7 @@ class AdaptiveLightingManager:
LIGHT_DOMAIN,
SERVICE_TURN_ON,
self._service_interceptor_turn_on_handler,
)
),
)
self.listener_removers.append(
@ -1779,7 +1838,7 @@ class AdaptiveLightingManager:
LIGHT_DOMAIN,
SERVICE_TOGGLE,
self._service_interceptor_turn_on_handler,
)
),
)
_LOGGER.debug("Proactive adaptation enabled")
@ -1796,20 +1855,21 @@ class AdaptiveLightingManager:
remove()
def set_proactively_adapting(self, context_id: str, entity_id: str) -> None:
"""Declare the adaptation with the given context ID as proactively adapting,
and associate it to an entity ID."""
"""Declare the adaptation with context_id as proactively adapting,
and associate it to an entity_id.
""" # noqa: D205
self._proactively_adapting_contexts[context_id] = entity_id
def is_proactively_adapting(self, context_id: str) -> bool:
"""Determine whether an adaptation with the given context ID is proactive."""
"""Determine whether an adaptation with the given context_id is proactive."""
is_proactively_adapting_context = (
context_id in self._proactively_adapting_contexts
)
_LOGGER.debug(
"is_proactively_adapting_context %s %s",
context_id,
"is_proactively_adapting_context='%s', context_id='%s'",
is_proactively_adapting_context,
context_id,
)
return is_proactively_adapting_context
@ -1817,16 +1877,19 @@ class AdaptiveLightingManager:
def clear_proactively_adapting(self, entity_id: str) -> None:
"""Clear all context IDs associated with the given entity ID.
Call this method to clear past context IDs and avoid a memory leak."""
Call this method to clear past context IDs and avoid a memory leak.
"""
# First get the keys to avoid modifying the dict while iterating it
keys = [
k for k, v in self._proactively_adapting_contexts.items() if v == entity_id
]
for key in keys:
self._proactively_adapting_contexts.pop(key)
async def _service_interceptor_turn_on_handler(
self, call: ServiceCall, data: ServiceData
async def _service_interceptor_turn_on_handler( # noqa: PLR0911
self,
call: ServiceCall,
data: ServiceData,
):
# Don't adapt our own service calls
if is_our_context(call.context):
@ -1872,14 +1935,17 @@ class AdaptiveLightingManager:
return
_LOGGER.debug(
"Intercepted TURN_ON call with data %s (%s)", data, call.context.id
"Intercepted TURN_ON call with data %s (%s)",
data,
call.context.id,
)
self.reset(entity_id, reset_manual_control=False)
self.clear_proactively_adapting(entity_id)
transition = data[CONF_PARAMS].get(
ATTR_TRANSITION, adaptive_switch.initial_transition
ATTR_TRANSITION,
adaptive_switch.initial_transition,
)
adaptation_data = await adaptive_switch.prepare_adaptation_data(
@ -1913,8 +1979,8 @@ class AdaptiveLightingManager:
self.set_proactively_adapting(call.context.id, entity_id)
self.set_proactively_adapting(adaptation_data.context.id, entity_id)
adaptation_data.initial_sleep = True
asyncio.create_task( # Don't await to avoid blocking the service call
adaptive_switch.execute_cancellable_adaptation_calls(adaptation_data)
_ = asyncio.create_task( # Don't await to avoid blocking the service call
adaptive_switch.execute_cancellable_adaptation_calls(adaptation_data),
)
def _handle_timer(
@ -1946,11 +2012,14 @@ class AdaptiveLightingManager:
last_transition = last_service_data.get(ATTR_TRANSITION)
if not last_transition:
_LOGGER.debug(
"No transition in last adapt for light %s, continuing...", light
"No transition in last adapt for light %s, continuing...",
light,
)
return
_LOGGER.debug(
"Start transition timer of %s seconds for light %s", last_transition, light
"Start transition timer of %s seconds for light %s",
last_transition,
light,
)
async def reset():
@ -2008,7 +2077,9 @@ class AdaptiveLightingManager:
self._handle_timer(light, self.auto_reset_manual_control_timers, delay, reset)
def cancel_ongoing_adaptation_calls(
self, light_id: str, which: Literal["color", "brightness", "both"] = "both"
self,
light_id: str,
which: Literal["color", "brightness", "both"] = "both",
):
"""Cancel ongoing adaptation service calls for a specific light entity."""
brightness_task = self.adaptation_tasks_brightness.get(light_id)
@ -2054,15 +2125,21 @@ class AdaptiveLightingManager:
area_ids = cv.ensure_list_csv(service_data[ATTR_AREA_ID])
for area_id in area_ids:
area_entity_ids = area_entities(self.hass, area_id)
for entity_id in area_entity_ids:
if entity_id.startswith(LIGHT_DOMAIN):
entity_ids.append(entity_id)
eids = [
entity_id
for entity_id in area_entity_ids
if entity_id.startswith(LIGHT_DOMAIN)
]
entity_ids.extend(eids)
_LOGGER.debug(
"Found entity_ids '%s' for area_id '%s'", entity_ids, area_id
"Found entity_ids '%s' for area_id '%s'",
entity_ids,
area_id,
)
else:
_LOGGER.debug(
"No entity_ids or area_ids found in service_data: %s", service_data
"No entity_ids or area_ids found in service_data: %s",
service_data,
)
return entity_ids
@ -2219,7 +2296,7 @@ class AdaptiveLightingManager:
"""
last_service_data = self.last_service_data.get(light)
if last_service_data is None:
return
return None
compare_to = functools.partial(
_attributes_have_changed,
light=light,
@ -2268,8 +2345,11 @@ class AdaptiveLightingManager:
)
return False
async def maybe_cancel_adjusting(
self, entity_id: str, off_to_on_event: Event, on_to_off_event: Event | None
async def maybe_cancel_adjusting( # noqa: PLR0911, PLR0912
self,
entity_id: str,
off_to_on_event: Event,
on_to_off_event: Event | None,
) -> bool:
"""Cancel the adjusting of a light if it has just been turned off.
@ -2368,7 +2448,7 @@ class AdaptiveLightingManager:
class _AsyncSingleShotTimer:
def __init__(self, delay, callback):
def __init__(self, delay, callback) -> None:
"""Initialize the timer."""
self.delay = delay
self.callback = callback

10
test_dependencies.py
View file

@ -1,13 +1,17 @@
"""Extracts the dependencies of the components required for testing."""
from collections import defaultdict
from pathlib import Path
deps = defaultdict(list)
components, packages = [], []
with open("core/requirements_test_all.txt") as f:
requirements = Path("core") / "requirements_test_all.txt"
with requirements.open() as f:
lines = f.readlines()
for line in lines:
line = line.strip()
line = line.strip() # noqa: PLW2901
if line.startswith("# homeassistant."):
if components and packages:
@ -34,4 +38,4 @@ required = [
]
to_install = [package for r in required for package in deps[r]]
print(" ".join(to_install))
print(" ".join(to_install)) # noqa: T201