mirror of
https://github.com/basnijholt/adaptive-lighting.git
synced 2026-10-29 06:37:13 +01:00
Merge 8ba7aabf05 into b2ca188df3
This commit is contained in:
commit
cdbf2fc021
15 changed files with 1478 additions and 78 deletions
6
.github/update-services.py
vendored
6
.github/update-services.py
vendored
|
|
@ -14,7 +14,11 @@ with open(services_filename) as f: # noqa: PTH123
|
|||
services = yaml.safe_load(f)
|
||||
|
||||
for service_name, dct in services.items():
|
||||
_docs = {"set_manual_control": const.DOCS_MANUAL_CONTROL, "apply": const.DOCS_APPLY}
|
||||
_docs = {
|
||||
"set_manual_control": const.DOCS_MANUAL_CONTROL,
|
||||
"apply": const.DOCS_APPLY,
|
||||
"get_commands": const.DOCS_GET_COMMANDS,
|
||||
}
|
||||
alternative_docs = _docs.get(service_name, const.DOCS)
|
||||
for field_name, field in dct["fields"].items():
|
||||
description = alternative_docs.get(field_name, const.DOCS[field_name])
|
||||
|
|
|
|||
57
README.md
57
README.md
|
|
@ -78,6 +78,7 @@ The attributes are absent when the Adaptive Lighting switch is off. Use a fallba
|
|||
- [:memo: Options](#memo-options)
|
||||
- [:hammer_and_wrench: Services](#hammer_and_wrench-services)
|
||||
- [`adaptive_lighting.apply`](#adaptive_lightingapply)
|
||||
- [`adaptive_lighting.get_commands`](#adaptive_lightingget_commands)
|
||||
- [`adaptive_lighting.set_manual_control`](#adaptive_lightingset_manual_control)
|
||||
- [`adaptive_lighting.change_switch_settings`](#adaptive_lightingchange_switch_settings)
|
||||
- [:robot: Automation examples](#robot-automation-examples)
|
||||
|
|
@ -229,6 +230,62 @@ Provide a switch in `entity_id`, a list of `lights`, or both.
|
|||
| `turn_on_lights` | Whether to turn on lights that are currently off. 🔆 | ❌ | bool |
|
||||
|
||||
<!-- OUTPUT:END -->
|
||||
|
||||
#### `adaptive_lighting.get_commands`
|
||||
|
||||
`adaptive_lighting.get_commands` returns the `light.turn_on` calls Adaptive Lighting would make to adapt lights, without making them: for now, or for a given `time` of day (today).
|
||||
Provide a switch in `entity_id`, a list of `lights`, or both.
|
||||
|
||||
<!-- CODE:START -->
|
||||
<!-- from adaptive_lighting._docs_helpers import generate_get_commands_markdown_table -->
|
||||
<!-- print(generate_get_commands_markdown_table()) -->
|
||||
<!-- CODE:END -->
|
||||
|
||||
<!-- OUTPUT:START -->
|
||||
<!-- ⚠️ This content is auto-generated by `markdown-code-runner`. -->
|
||||
| Service data attribute | Description | Required | Type |
|
||||
|:-------------------------|:----------------------------------------------------------------------------------------------------------------------|:-----------|:---------------------|
|
||||
| `entity_id` | The `entity_id` of the switch whose settings to use. 📝 | ❌ | list of `entity_id`s |
|
||||
| `lights` | Lights or light groups in the switch's `lights` to get the commands for. 💡 | ❌ | list of `entity_id`s |
|
||||
| `time` | Time of day (HH:MM:SS) to get the commands for, instead of now. 🕰️ | ❌ | `str` (HH:MM[:SS]) |
|
||||
| `transition` | Duration of transition when lights change, in seconds (default: the switch's `initial_transition`). 🕑 | ❌ | `float` 0-6553 |
|
||||
| `adapt_brightness` | Whether to adapt the brightness of the light (default: the switch's Adapt Brightness switch). 🌞 | ❌ | bool |
|
||||
| `adapt_color` | Whether to adapt the color on supporting lights (default: the switch's Adapt Color switch). 🌈 | ❌ | bool |
|
||||
| `prefer_rgb_color` | Whether to prefer RGB color adjustment over light color temperature when possible (default: the switch's setting). 🌈 | ❌ | bool |
|
||||
|
||||
<!-- OUTPUT:END -->
|
||||
|
||||
<!-- SECTION:get-commands:START -->
|
||||
Call it with a `response_variable`. For each light, the response lists the calls in order, each with its `service_data` and the `delay` in seconds before it (only nonzero between split calls, see `separate_turn_on_commands`). Only the lights and light groups in a switch's `lights` can be passed; without `entity_id`, each uses the switch it's in (of several, the one that's on). A light group is replaced by its members if the switch expands light groups. Options you leave out follow the switch, with its `initial_transition`, and in sleep mode the calls are for the sleep settings, whatever the time. With `skip_redundant_commands`, what a light already has is left out, as when Adaptive Lighting adapts it; that compares with the light's state when you call the service, so make the calls right away. Manual control doesn't change the calls. In YAML files, quote the time (`"22:00:00"`), since `22:00:00` is read as a number.
|
||||
|
||||
Making the calls is a manual change for Adaptive Lighting: with `take_over_control`, it stops adapting those lights until manual control is reset. A light that's off is adapted to the current settings as it turns on (unless `take_over_control` and `adapt_only_on_bare_turn_on` are enabled), so this example only changes lights that are on:
|
||||
|
||||
```yaml
|
||||
# Set the lights to the settings for 22:00, e.g., from a button
|
||||
sequence:
|
||||
- action: adaptive_lighting.get_commands
|
||||
data:
|
||||
entity_id: switch.adaptive_lighting_living_room
|
||||
time: "22:00:00"
|
||||
response_variable: commands
|
||||
- repeat:
|
||||
for_each: "{{ commands.items() | list }}"
|
||||
sequence:
|
||||
- variables:
|
||||
light: "{{ repeat.item[0] }}"
|
||||
calls: "{{ repeat.item[1] }}"
|
||||
# Only lights that are on, see above
|
||||
- if: "{{ is_state(light, 'on') }}"
|
||||
then:
|
||||
- repeat:
|
||||
for_each: "{{ calls }}"
|
||||
sequence:
|
||||
- delay: "{{ repeat.item.delay }}"
|
||||
- action: light.turn_on
|
||||
data: "{{ repeat.item.service_data }}"
|
||||
```
|
||||
<!-- SECTION:get-commands:END -->
|
||||
|
||||
#### `adaptive_lighting.set_manual_control`
|
||||
|
||||
`adaptive_lighting.set_manual_control` can mark (or unmark) whether a light is "manually controlled", meaning that when a light has `manual_control`, the light is not adapted.
|
||||
|
|
|
|||
|
|
@ -8,7 +8,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, Platform
|
||||
from homeassistant.core import Event, HomeAssistant
|
||||
from homeassistant.core import Event, HomeAssistant, SupportsResponse
|
||||
from homeassistant.helpers import service
|
||||
|
||||
from .const import (
|
||||
|
|
@ -18,15 +18,18 @@ from .const import (
|
|||
DOMAIN,
|
||||
SERVICE_APPLY,
|
||||
SERVICE_CHANGE_SWITCH_SETTINGS,
|
||||
SERVICE_GET_COMMANDS,
|
||||
SERVICE_SET_MANUAL_CONTROL,
|
||||
SET_MANUAL_CONTROL_SCHEMA,
|
||||
UNDO_UPDATE_LISTENER,
|
||||
apply_service_schema,
|
||||
change_switch_settings_schema,
|
||||
get_commands_service_schema,
|
||||
)
|
||||
from .switch import (
|
||||
handle_apply_service,
|
||||
handle_change_switch_settings,
|
||||
handle_get_commands_service,
|
||||
handle_set_manual_control_service,
|
||||
)
|
||||
|
||||
|
|
@ -67,6 +70,14 @@ async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
|
|||
schema=apply_service_schema(),
|
||||
)
|
||||
|
||||
hass.services.async_register(
|
||||
domain=DOMAIN,
|
||||
service=SERVICE_GET_COMMANDS,
|
||||
service_func=partial(handle_get_commands_service, hass),
|
||||
schema=get_commands_service_schema(),
|
||||
supports_response=SupportsResponse.ONLY,
|
||||
)
|
||||
|
||||
hass.services.async_register(
|
||||
domain=DOMAIN,
|
||||
service=SERVICE_SET_MANUAL_CONTROL,
|
||||
|
|
|
|||
|
|
@ -8,10 +8,12 @@ from homeassistant.helpers import selector
|
|||
from .const import (
|
||||
DOCS,
|
||||
DOCS_APPLY,
|
||||
DOCS_GET_COMMANDS,
|
||||
DOCS_MANUAL_CONTROL,
|
||||
SET_MANUAL_CONTROL_SCHEMA,
|
||||
VALIDATION_TUPLES,
|
||||
apply_service_schema,
|
||||
get_commands_service_schema,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -44,6 +46,8 @@ def _type_to_str(type_: Any) -> str: # noqa: PLR0911
|
|||
return f"`{type_.__name__}`"
|
||||
if type_ == cv.boolean:
|
||||
return "bool"
|
||||
if type_ == cv.time:
|
||||
return "`str` (HH:MM[:SS])"
|
||||
if isinstance(type_, vol.All):
|
||||
return _format_voluptuous_instance(type_)
|
||||
if isinstance(type_, vol.Any):
|
||||
|
|
@ -109,6 +113,13 @@ def generate_apply_markdown_table() -> str:
|
|||
return _generate_service_markdown_table(apply_service_schema(), DOCS_APPLY)
|
||||
|
||||
|
||||
def generate_get_commands_markdown_table() -> str:
|
||||
return _generate_service_markdown_table(
|
||||
get_commands_service_schema(),
|
||||
DOCS_GET_COMMANDS,
|
||||
)
|
||||
|
||||
|
||||
def generate_set_manual_control_markdown_table() -> str:
|
||||
return _generate_service_markdown_table(
|
||||
SET_MANUAL_CONTROL_SCHEMA,
|
||||
|
|
|
|||
|
|
@ -176,6 +176,19 @@ def _has_relevant_service_data_attributes(service_data: ServiceData) -> bool:
|
|||
return any(attr not in common_attrs for attr in service_data)
|
||||
|
||||
|
||||
def remove_redundant_service_data(
|
||||
service_data: ServiceData,
|
||||
state: State | None,
|
||||
) -> ServiceData | None:
|
||||
"""Remove what the light's `state` already has from the data of a service call.
|
||||
|
||||
Returns `None` if nothing is left that justifies the service call.
|
||||
"""
|
||||
if state is not None:
|
||||
service_data = _remove_redundant_attributes(service_data, state=state)
|
||||
return service_data if _has_relevant_service_data_attributes(service_data) else None
|
||||
|
||||
|
||||
async def _create_service_call_data_iterator(
|
||||
hass: HomeAssistant,
|
||||
service_datas: list[ServiceData],
|
||||
|
|
@ -192,18 +205,12 @@ async def _create_service_call_data_iterator(
|
|||
"""
|
||||
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( # noqa: PLW2901
|
||||
service_data,
|
||||
state=current_entity_state,
|
||||
)
|
||||
|
||||
# Emit service data if it still contains relevant attributes (else try next)
|
||||
if _has_relevant_service_data_attributes(service_data):
|
||||
yield service_data
|
||||
if filtered := remove_redundant_service_data(
|
||||
service_data,
|
||||
hass.states.get(entity_id),
|
||||
):
|
||||
yield filtered
|
||||
else:
|
||||
yield service_data
|
||||
|
||||
|
|
@ -265,6 +272,26 @@ def _identify_light_control_attributes(
|
|||
return parameters
|
||||
|
||||
|
||||
def plan_service_calls(
|
||||
service_data: ServiceData,
|
||||
transition: float | None,
|
||||
split_delay: float,
|
||||
*,
|
||||
split: bool,
|
||||
) -> tuple[list[ServiceData], float]:
|
||||
"""Plan the `light.turn_on` calls for adapting a light with `service_data`.
|
||||
|
||||
Returns the service data of the calls in order, split into brightness and color
|
||||
if `split`, and the time to wait before each call after the first.
|
||||
"""
|
||||
service_datas = _split_service_call_data(service_data) if split else [service_data]
|
||||
if transition is not None:
|
||||
sleep_time = transition / max(1, len(service_datas)) + split_delay
|
||||
else:
|
||||
sleep_time = split_delay
|
||||
return service_datas, sleep_time
|
||||
|
||||
|
||||
def prepare_adaptation_data(
|
||||
hass: HomeAssistant,
|
||||
entity_id: str,
|
||||
|
|
@ -283,15 +310,12 @@ def prepare_adaptation_data(
|
|||
entity_id,
|
||||
service_data,
|
||||
)
|
||||
service_datas = _split_service_call_data(service_data) if split else [service_data]
|
||||
|
||||
service_datas_length = len(service_datas)
|
||||
|
||||
if transition is not None:
|
||||
transition_duration_per_data = transition / max(1, service_datas_length)
|
||||
sleep_time = transition_duration_per_data + split_delay
|
||||
else:
|
||||
sleep_time = split_delay
|
||||
service_datas, sleep_time = plan_service_calls(
|
||||
service_data,
|
||||
transition,
|
||||
split_delay,
|
||||
split=split,
|
||||
)
|
||||
|
||||
# Keep the original split timing, but omit attributes carried by an
|
||||
# intercepted turn-on shared with other lights. Do this before state
|
||||
|
|
|
|||
|
|
@ -464,12 +464,17 @@ class SunLightSettings:
|
|||
self,
|
||||
is_sleep: bool,
|
||||
transition: float | None,
|
||||
dt: datetime.datetime | None = None,
|
||||
) -> dict[str, float | int | tuple[float, float] | tuple[float, float, float]]:
|
||||
"""Get all light settings.
|
||||
|
||||
For when a `transition` started now (or at `dt`) ends.
|
||||
|
||||
Calculating all values takes <0.5ms.
|
||||
"""
|
||||
dt = utcnow() + timedelta(seconds=transition or 0)
|
||||
if dt is None:
|
||||
dt = utcnow()
|
||||
dt += timedelta(seconds=transition or 0)
|
||||
return self.brightness_and_color(dt, is_sleep)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -316,6 +316,11 @@ DOCS[CONF_MANUAL_CONTROL] = "Whether to manually control the lights. 🔒"
|
|||
SERVICE_APPLY = "apply"
|
||||
CONF_TURN_ON_LIGHTS = "turn_on_lights"
|
||||
DOCS[CONF_TURN_ON_LIGHTS] = "Whether to turn on lights that are currently off. 🔆"
|
||||
SERVICE_GET_COMMANDS = "get_commands"
|
||||
CONF_COMMANDS_TIME = "time"
|
||||
DOCS[CONF_COMMANDS_TIME] = (
|
||||
"Time of day (HH:MM:SS) to get the commands for, instead of now. 🕰️"
|
||||
)
|
||||
SERVICE_CHANGE_SWITCH_SETTINGS = "change_switch_settings"
|
||||
CONF_USE_DEFAULTS = "use_defaults"
|
||||
DOCS[CONF_USE_DEFAULTS] = (
|
||||
|
|
@ -341,6 +346,20 @@ DOCS_APPLY = {
|
|||
CONF_LIGHTS: "A light (or list of lights) to apply the settings to. 💡",
|
||||
}
|
||||
|
||||
DOCS_GET_COMMANDS = {
|
||||
CONF_ENTITY_ID: "The `entity_id` of the switch whose settings to use. 📝",
|
||||
CONF_LIGHTS: "Lights or light groups in the switch's `lights` to get the"
|
||||
" commands for. 💡",
|
||||
CONF_TRANSITION: "Duration of transition when lights change, in seconds"
|
||||
" (default: the switch's `initial_transition`). 🕑",
|
||||
ATTR_ADAPT_BRIGHTNESS: "Whether to adapt the brightness of the light"
|
||||
" (default: the switch's Adapt Brightness switch). 🌞",
|
||||
ATTR_ADAPT_COLOR: "Whether to adapt the color on supporting lights"
|
||||
" (default: the switch's Adapt Color switch). 🌈",
|
||||
CONF_PREFER_RGB_COLOR: "Whether to prefer RGB color adjustment over light color"
|
||||
" temperature when possible (default: the switch's setting). 🌈",
|
||||
}
|
||||
|
||||
# Basic options shown at top level in options flow (not in collapsed section)
|
||||
BASIC_OPTIONS: set[str] = {
|
||||
CONF_LIGHTS,
|
||||
|
|
@ -525,6 +544,21 @@ def apply_service_schema() -> vol.Schema:
|
|||
)
|
||||
|
||||
|
||||
def get_commands_service_schema() -> vol.Schema:
|
||||
"""Return the schema for the get_commands service."""
|
||||
return vol.Schema(
|
||||
{
|
||||
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_COMMANDS_TIME): cv.time,
|
||||
vol.Optional(CONF_TRANSITION): VALID_TRANSITION,
|
||||
vol.Optional(ATTR_ADAPT_BRIGHTNESS): cv.boolean,
|
||||
vol.Optional(ATTR_ADAPT_COLOR): cv.boolean,
|
||||
vol.Optional(CONF_PREFER_RGB_COLOR): cv.boolean,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def change_switch_settings_schema() -> dict[vol.Marker, Any]:
|
||||
"""Return the schema for the change_switch_settings service."""
|
||||
args: dict[vol.Marker, Any] = {
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ def _transform_readme_links(content: str) -> str:
|
|||
"#memo-options": "configuration.md#all-options",
|
||||
"#hammer_and_wrench-services": "services.md",
|
||||
"#adaptive_lightingapply": "services.md#adaptive_lightingapply",
|
||||
"#adaptive_lightingget_commands": "services.md#adaptive_lightingget_commands",
|
||||
"#adaptive_lightingset_manual_control": "services.md#adaptive_lightingset_manual_control",
|
||||
"#adaptive_lightingchange_switch_settings": "services.md#adaptive_lightingchange_switch_settings",
|
||||
"#robot-automation-examples": "automation-examples.md",
|
||||
|
|
|
|||
|
|
@ -40,6 +40,47 @@ apply:
|
|||
example: false
|
||||
selector:
|
||||
boolean: null
|
||||
get_commands:
|
||||
description: Returns the `light.turn_on` calls Adaptive Lighting would make to adapt lights, now or for a given time of day, without making them.
|
||||
fields:
|
||||
entity_id:
|
||||
description: The `entity_id` of the switch whose settings to use. 📝
|
||||
selector:
|
||||
entity:
|
||||
integration: adaptive_lighting
|
||||
domain: switch
|
||||
multiple: false
|
||||
lights:
|
||||
description: Lights or light groups in the switch's `lights` to get the commands for. 💡
|
||||
selector:
|
||||
entity:
|
||||
domain: light
|
||||
multiple: true
|
||||
time:
|
||||
description: Time of day (HH:MM:SS) to get the commands for, instead of now. 🕰️
|
||||
example: '22:00:00'
|
||||
selector:
|
||||
time: null
|
||||
transition:
|
||||
description: 'Duration of transition when lights change, in seconds (default: the switch''s `initial_transition`). 🕑'
|
||||
example: 1
|
||||
selector:
|
||||
text: null
|
||||
adapt_brightness:
|
||||
description: 'Whether to adapt the brightness of the light (default: the switch''s Adapt Brightness switch). 🌞'
|
||||
example: true
|
||||
selector:
|
||||
boolean: null
|
||||
adapt_color:
|
||||
description: 'Whether to adapt the color on supporting lights (default: the switch''s Adapt Color switch). 🌈'
|
||||
example: true
|
||||
selector:
|
||||
boolean: null
|
||||
prefer_rgb_color:
|
||||
description: 'Whether to prefer RGB color adjustment over light color temperature when possible (default: the switch''s setting). 🌈'
|
||||
example: false
|
||||
selector:
|
||||
boolean: null
|
||||
set_manual_control:
|
||||
description: Mark whether a light is 'manually controlled'.
|
||||
fields:
|
||||
|
|
|
|||
|
|
@ -146,6 +146,40 @@
|
|||
}
|
||||
}
|
||||
},
|
||||
"get_commands": {
|
||||
"name": "get_commands",
|
||||
"description": "Returns the `light.turn_on` calls Adaptive Lighting would make to adapt lights, now or for a given time of day, without making them.",
|
||||
"fields": {
|
||||
"entity_id": {
|
||||
"description": "The `entity_id` of the switch whose settings to use. 📝",
|
||||
"name": "entity_id"
|
||||
},
|
||||
"lights": {
|
||||
"description": "Lights or light groups in the switch's `lights` to get the commands for. 💡",
|
||||
"name": "lights"
|
||||
},
|
||||
"time": {
|
||||
"description": "Time of day (HH:MM:SS) to get the commands for, instead of now. 🕰️",
|
||||
"name": "time"
|
||||
},
|
||||
"transition": {
|
||||
"description": "Duration of transition when lights change, in seconds (default: the switch's `initial_transition`). 🕑",
|
||||
"name": "transition"
|
||||
},
|
||||
"adapt_brightness": {
|
||||
"description": "Whether to adapt the brightness of the light (default: the switch's Adapt Brightness switch). 🌞",
|
||||
"name": "adapt_brightness"
|
||||
},
|
||||
"adapt_color": {
|
||||
"description": "Whether to adapt the color on supporting lights (default: the switch's Adapt Color switch). 🌈",
|
||||
"name": "adapt_color"
|
||||
},
|
||||
"prefer_rgb_color": {
|
||||
"description": "Whether to prefer RGB color adjustment over light color temperature when possible (default: the switch's setting). 🌈",
|
||||
"name": "prefer_rgb_color"
|
||||
}
|
||||
}
|
||||
},
|
||||
"set_manual_control": {
|
||||
"name": "set_manual_control",
|
||||
"description": "Mark whether a light is 'manually controlled'.",
|
||||
|
|
|
|||
|
|
@ -57,6 +57,7 @@ from homeassistant.core import (
|
|||
Event,
|
||||
HomeAssistant,
|
||||
ServiceCall,
|
||||
ServiceResponse,
|
||||
State,
|
||||
callback,
|
||||
)
|
||||
|
|
@ -84,7 +85,9 @@ from .adaptation_utils import (
|
|||
get_light_control_attributes,
|
||||
has_effect_attribute,
|
||||
manual_control_event_attribute_to_flags,
|
||||
plan_service_calls,
|
||||
prepare_adaptation_data,
|
||||
remove_redundant_service_data,
|
||||
)
|
||||
from .color_and_brightness import SunLightSettings
|
||||
from .const import (
|
||||
|
|
@ -100,6 +103,7 @@ from .const import (
|
|||
CONF_BRIGHTNESS_MODE,
|
||||
CONF_BRIGHTNESS_MODE_TIME_DARK,
|
||||
CONF_BRIGHTNESS_MODE_TIME_LIGHT,
|
||||
CONF_COMMANDS_TIME,
|
||||
CONF_DETECT_NON_HA_CHANGES,
|
||||
CONF_EXPAND_LIGHT_GROUPS,
|
||||
CONF_INCLUDE_CONFIG_IN_ATTRIBUTES,
|
||||
|
|
@ -238,21 +242,81 @@ def is_our_context(context: Context | None, which: str | None = None) -> bool:
|
|||
return is_our_context_id(context.id, which)
|
||||
|
||||
|
||||
def _loaded_switches(hass: HomeAssistant) -> AdaptiveSwitches:
|
||||
"""Get the switches of all loaded config entries."""
|
||||
data = hass.data.get(DOMAIN, {})
|
||||
loaded_switches: AdaptiveSwitches = []
|
||||
for config in hass.config_entries.async_entries(DOMAIN):
|
||||
entry = data.get(config.entry_id)
|
||||
if not isinstance(entry, dict) or SWITCH_DOMAIN not in entry:
|
||||
continue
|
||||
loaded_switches.append(entry[SWITCH_DOMAIN])
|
||||
return loaded_switches
|
||||
|
||||
|
||||
def _not_configured_msg(
|
||||
hass: HomeAssistant,
|
||||
lights: list[str],
|
||||
switches: AdaptiveSwitches,
|
||||
*,
|
||||
any_switch: bool,
|
||||
) -> str:
|
||||
"""Explain that `lights` aren't among the configured lights of `switches`.
|
||||
|
||||
With `any_switch`, `switches` are all switches, else the switch that was passed.
|
||||
"""
|
||||
groups = {
|
||||
light: configured
|
||||
for switch in switches
|
||||
for configured in switch._configured_lights
|
||||
for light in lights
|
||||
if light != configured and light in _expand_light_groups(hass, [configured])
|
||||
}
|
||||
where = (
|
||||
"the lights of any switch"
|
||||
if any_switch
|
||||
else f"the lights of {', '.join(repr(switch.entity_id) for switch in switches)}"
|
||||
)
|
||||
verb = "isn't" if len(lights) == 1 else "aren't"
|
||||
msg = f"adaptive-lighting: {', '.join(map(repr, lights))} {verb} among {where}."
|
||||
if groups:
|
||||
msg += " Pass the light group instead: " + ", ".join(
|
||||
f"{group!r} for {light!r}" for light, group in groups.items()
|
||||
)
|
||||
msg += "."
|
||||
return msg
|
||||
|
||||
|
||||
def _switch_configured_with(hass: HomeAssistant, light: str) -> AdaptiveSwitch:
|
||||
"""Get the switch with `light` (a light or group) among its configured lights.
|
||||
|
||||
Of several, the one that's on.
|
||||
"""
|
||||
owners = [
|
||||
switch
|
||||
for switch in _loaded_switches(hass)
|
||||
if light in switch._configured_lights
|
||||
]
|
||||
if len(owners) > 1 and len(on := [s for s in owners if s.is_on]) == 1:
|
||||
owners = on
|
||||
if len(owners) == 1:
|
||||
return owners[0]
|
||||
msg = (
|
||||
f"adaptive-lighting: '{light}' is among the lights of more than one switch,"
|
||||
" and none, or more than one, of them is on. Pass one of them in 'entity_id'."
|
||||
if owners
|
||||
else _not_configured_msg(hass, [light], _loaded_switches(hass), any_switch=True)
|
||||
)
|
||||
raise ServiceValidationError(msg)
|
||||
|
||||
|
||||
def _switches_with_lights(
|
||||
hass: HomeAssistant,
|
||||
lights: list[str],
|
||||
expand_light_groups: bool = True,
|
||||
) -> AdaptiveSwitches:
|
||||
"""Get all switches that control at least one of the lights passed."""
|
||||
config_entries = hass.config_entries.async_entries(DOMAIN)
|
||||
data = hass.data.get(DOMAIN, {})
|
||||
loaded_switches: AdaptiveSwitches = []
|
||||
for config in config_entries:
|
||||
entry = data.get(config.entry_id)
|
||||
if not isinstance(entry, dict) or SWITCH_DOMAIN not in entry:
|
||||
continue
|
||||
loaded_switches.append(entry[SWITCH_DOMAIN])
|
||||
|
||||
loaded_switches = _loaded_switches(hass)
|
||||
if not loaded_switches:
|
||||
return []
|
||||
|
||||
|
|
@ -413,6 +477,62 @@ async def handle_change_switch_settings(
|
|||
)
|
||||
|
||||
|
||||
async def handle_get_commands_service(
|
||||
hass: HomeAssistant,
|
||||
service_call: ServiceCall,
|
||||
) -> ServiceResponse:
|
||||
"""Handle the get_commands service, which only returns data.
|
||||
|
||||
Returns, for each light, the `light.turn_on` calls Adaptive Lighting would make to
|
||||
adapt it now (or for the time of day given), without making them.
|
||||
"""
|
||||
data = service_call.data
|
||||
_LOGGER.debug(
|
||||
"Called 'adaptive_lighting.get_commands' service with '%s'",
|
||||
data,
|
||||
)
|
||||
lights: list[str] = data[CONF_LIGHTS]
|
||||
if not lights and not data.get(ATTR_ENTITY_ID):
|
||||
msg = "adaptive-lighting: Pass a switch in 'entity_id', 'lights', or both."
|
||||
raise ServiceValidationError(msg)
|
||||
# Only the lights (and groups) configured in the switches
|
||||
targets: list[tuple[AdaptiveSwitch, list[str] | None]]
|
||||
if data.get(ATTR_ENTITY_ID):
|
||||
switches = _switches_from_service_call(hass, service_call)
|
||||
for switch in switches if lights else ():
|
||||
if others := [x for x in lights if x not in switch._configured_lights]:
|
||||
raise ServiceValidationError(
|
||||
_not_configured_msg(hass, others, [switch], any_switch=False),
|
||||
)
|
||||
targets = [(switch, lights or None) for switch in switches]
|
||||
else:
|
||||
lights_by_switch: dict[str, tuple[AdaptiveSwitch, list[str]]] = {}
|
||||
for light in lights:
|
||||
switch = _switch_configured_with(hass, light)
|
||||
_, owned = lights_by_switch.setdefault(switch.entity_id, (switch, []))
|
||||
owned.append(light)
|
||||
targets = list(lights_by_switch.values())
|
||||
commands: dict[str, Any] = {}
|
||||
for switch, switch_lights in targets:
|
||||
switch_commands = switch.get_adaptation_commands(
|
||||
switch_lights,
|
||||
time=data.get(CONF_COMMANDS_TIME),
|
||||
transition=data.get(CONF_TRANSITION),
|
||||
adapt_brightness=data.get(ATTR_ADAPT_BRIGHTNESS),
|
||||
adapt_color=data.get(ATTR_ADAPT_COLOR),
|
||||
prefer_rgb_color=data.get(CONF_PREFER_RGB_COLOR),
|
||||
)
|
||||
if shared := set(commands) & set(switch_commands):
|
||||
msg = (
|
||||
f"adaptive-lighting: Lights {sorted(shared)} are in more than one of"
|
||||
" the switches, so their commands would be ambiguous. Pass one of"
|
||||
" the switches in 'entity_id'."
|
||||
)
|
||||
raise ServiceValidationError(msg)
|
||||
commands.update(switch_commands)
|
||||
return commands
|
||||
|
||||
|
||||
async def handle_apply_service(hass: HomeAssistant, service_call: ServiceCall) -> None:
|
||||
"""Handle the entity service apply."""
|
||||
data = service_call.data
|
||||
|
|
@ -1305,6 +1425,153 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
|
|||
force=False,
|
||||
)
|
||||
|
||||
def _build_service_data(
|
||||
self,
|
||||
light: str,
|
||||
transition: float,
|
||||
*,
|
||||
adapt_brightness: bool,
|
||||
adapt_color: bool,
|
||||
prefer_rgb_color: bool,
|
||||
settings: dict[str, Any],
|
||||
) -> tuple[dict[str, Any], float] | None:
|
||||
"""Build the `light.turn_on` service data for adapting a light to `settings`.
|
||||
|
||||
Returns the service data and the transition it uses (0 if the light doesn't
|
||||
support transitions), or `None` if there's nothing to adapt.
|
||||
"""
|
||||
service_data: dict[str, Any] = {ATTR_ENTITY_ID: light}
|
||||
features = _supported_features(self.hass, light)
|
||||
|
||||
# Check transition == 0 to fix #378
|
||||
use_transition = "transition" in features and transition > 0
|
||||
if use_transition:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
|
||||
if "brightness" in features and adapt_brightness:
|
||||
brightness = round(255 * settings["brightness_pct"] / 100)
|
||||
service_data[ATTR_BRIGHTNESS] = brightness
|
||||
|
||||
sleep_rgb = (
|
||||
self.sleep_mode_switch.is_on
|
||||
and self._sun_light_settings.sleep_rgb_or_color_temp == "rgb_color"
|
||||
)
|
||||
if (
|
||||
"color_temp" in features
|
||||
and adapt_color
|
||||
and not (prefer_rgb_color and "color" in features)
|
||||
and not (sleep_rgb and "color" in features)
|
||||
and not (settings["force_rgb_color"] and "color" in features)
|
||||
):
|
||||
_LOGGER.debug("%s: Setting color_temp of light %s", self._name, light)
|
||||
state = self.hass.states.get(light)
|
||||
assert isinstance(state, State)
|
||||
attributes = state.attributes
|
||||
min_kelvin = attributes["min_color_temp_kelvin"]
|
||||
max_kelvin = attributes["max_color_temp_kelvin"]
|
||||
color_temp_kelvin = settings["color_temp_kelvin"]
|
||||
color_temp_kelvin = clamp(color_temp_kelvin, min_kelvin, max_kelvin)
|
||||
service_data[ATTR_COLOR_TEMP_KELVIN] = color_temp_kelvin
|
||||
elif "color" in features and adapt_color:
|
||||
_LOGGER.debug("%s: Setting rgb_color of light %s", self._name, light)
|
||||
service_data[ATTR_RGB_COLOR] = settings["rgb_color"]
|
||||
|
||||
required_attrs = [ATTR_RGB_COLOR, ATTR_COLOR_TEMP_KELVIN, ATTR_BRIGHTNESS]
|
||||
if not any(attr in service_data for attr in required_attrs):
|
||||
_LOGGER.debug(
|
||||
"%s: Skipping adaptation of %s because no relevant attributes"
|
||||
" are set in service_data: %s",
|
||||
self._name,
|
||||
light,
|
||||
service_data,
|
||||
)
|
||||
return None
|
||||
return service_data, transition if use_transition else 0
|
||||
|
||||
def get_adaptation_commands(
|
||||
self,
|
||||
lights: list[str] | None,
|
||||
*,
|
||||
time: datetime.time | None = None,
|
||||
transition: float | None = None,
|
||||
adapt_brightness: bool | None = None,
|
||||
adapt_color: bool | None = None,
|
||||
prefer_rgb_color: bool | None = None,
|
||||
) -> dict[str, list[dict[str, Any]]]:
|
||||
"""Get the `light.turn_on` calls this switch would make to adapt lights.
|
||||
|
||||
For each light (all of the switch's if `lights` is `None`), the calls in
|
||||
order, each with its service data and the time to wait before it, adapting
|
||||
now or at the `time` of day today (lights without a state are left out). With
|
||||
`skip_redundant_commands`, what a light already has (now) is left out.
|
||||
Omitted options default to the switch's settings (unlike for `apply`), and
|
||||
`transition` to its `initial_transition` (like for `apply`). Manual control
|
||||
doesn't change the calls, and nothing is sent.
|
||||
"""
|
||||
if transition is None:
|
||||
transition = self.initial_transition
|
||||
if adapt_brightness is None:
|
||||
adapt_brightness = bool(self.adapt_brightness_switch.is_on)
|
||||
if adapt_color is None:
|
||||
adapt_color = bool(self.adapt_color_switch.is_on)
|
||||
if prefer_rgb_color is None:
|
||||
prefer_rgb_color = self._prefer_rgb_color
|
||||
at_time = None
|
||||
if time is not None:
|
||||
tz = self._sun_light_settings.timezone
|
||||
# In UTC, like the time the regular adaptation calculates for
|
||||
at_time = dt_util.as_utc(
|
||||
datetime.datetime.combine(dt_util.now(tz).date(), time, tzinfo=tz),
|
||||
)
|
||||
settings = self._sun_light_settings.get_settings(
|
||||
bool(self.sleep_mode_switch.is_on),
|
||||
transition,
|
||||
at_time,
|
||||
)
|
||||
commands: dict[str, list[dict[str, Any]]] = {}
|
||||
for light in self._resolve_lights(lights):
|
||||
if self.hass.states.get(light) is None:
|
||||
continue
|
||||
commands[light] = []
|
||||
if not (adapt_brightness or adapt_color):
|
||||
continue
|
||||
built = self._build_service_data(
|
||||
light,
|
||||
transition,
|
||||
adapt_brightness=adapt_brightness,
|
||||
adapt_color=adapt_color,
|
||||
prefer_rgb_color=prefer_rgb_color,
|
||||
settings=settings,
|
||||
)
|
||||
if built is None:
|
||||
continue
|
||||
service_data, light_transition = built
|
||||
service_datas, sleep_time = plan_service_calls(
|
||||
service_data,
|
||||
light_transition,
|
||||
self._send_split_delay / 1000.0,
|
||||
split=self._separate_turn_on_commands,
|
||||
)
|
||||
if self._skip_redundant_commands:
|
||||
# Leave out what the light already has, as when adapting it
|
||||
state = self.hass.states.get(light)
|
||||
service_datas = [
|
||||
filtered
|
||||
for data in service_datas
|
||||
if (filtered := remove_redundant_service_data(data, state))
|
||||
]
|
||||
commands[light] = [
|
||||
{
|
||||
"service_data": {
|
||||
key: list(value) if isinstance(value, (list, tuple)) else value
|
||||
for key, value in data.items()
|
||||
},
|
||||
"delay": round(sleep_time, 3) if index else 0,
|
||||
}
|
||||
for index, data in enumerate(service_datas)
|
||||
]
|
||||
return commands
|
||||
|
||||
async def prepare_adaptation_data(
|
||||
self,
|
||||
light: str,
|
||||
|
|
@ -1348,53 +1615,17 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
|
|||
transition,
|
||||
)
|
||||
|
||||
# Build service data.
|
||||
service_data: dict[str, Any] = {ATTR_ENTITY_ID: light}
|
||||
features = _supported_features(self.hass, light)
|
||||
|
||||
# Check transition == 0 to fix #378
|
||||
use_transition = "transition" in features and transition > 0
|
||||
if use_transition:
|
||||
service_data[ATTR_TRANSITION] = transition
|
||||
|
||||
if "brightness" in features and adapt_brightness:
|
||||
brightness = round(255 * self._settings["brightness_pct"] / 100)
|
||||
service_data[ATTR_BRIGHTNESS] = brightness
|
||||
|
||||
sleep_rgb = (
|
||||
self.sleep_mode_switch.is_on
|
||||
and self._sun_light_settings.sleep_rgb_or_color_temp == "rgb_color"
|
||||
built = self._build_service_data(
|
||||
light,
|
||||
transition,
|
||||
adapt_brightness=adapt_brightness,
|
||||
adapt_color=adapt_color,
|
||||
prefer_rgb_color=prefer_rgb_color,
|
||||
settings=self._settings,
|
||||
)
|
||||
if (
|
||||
"color_temp" in features
|
||||
and adapt_color
|
||||
and not (prefer_rgb_color and "color" in features)
|
||||
and not (sleep_rgb and "color" in features)
|
||||
and not (self._settings["force_rgb_color"] and "color" in features)
|
||||
):
|
||||
_LOGGER.debug("%s: Setting color_temp of light %s", self._name, light)
|
||||
state = self.hass.states.get(light)
|
||||
assert isinstance(state, State)
|
||||
attributes = state.attributes
|
||||
min_kelvin = attributes["min_color_temp_kelvin"]
|
||||
max_kelvin = attributes["max_color_temp_kelvin"]
|
||||
color_temp_kelvin = self._settings["color_temp_kelvin"]
|
||||
color_temp_kelvin = clamp(color_temp_kelvin, min_kelvin, max_kelvin)
|
||||
service_data[ATTR_COLOR_TEMP_KELVIN] = color_temp_kelvin
|
||||
elif "color" in features and adapt_color:
|
||||
_LOGGER.debug("%s: Setting rgb_color of light %s", self._name, light)
|
||||
service_data[ATTR_RGB_COLOR] = self._settings["rgb_color"]
|
||||
|
||||
required_attrs = [ATTR_RGB_COLOR, ATTR_COLOR_TEMP_KELVIN, ATTR_BRIGHTNESS]
|
||||
if not any(attr in service_data for attr in required_attrs):
|
||||
_LOGGER.debug(
|
||||
"%s: Skipping adaptation of %s because no relevant attributes"
|
||||
" are set in service_data: %s",
|
||||
self._name,
|
||||
light,
|
||||
service_data,
|
||||
)
|
||||
if built is None:
|
||||
return None
|
||||
service_data, adapted_transition = built
|
||||
|
||||
context = context or self.create_context("adapt_lights")
|
||||
|
||||
|
|
@ -1402,7 +1633,7 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
|
|||
self.hass,
|
||||
light,
|
||||
context,
|
||||
transition if use_transition else 0,
|
||||
adapted_transition,
|
||||
self._send_split_delay / 1000.0,
|
||||
service_data,
|
||||
split=self._separate_turn_on_commands,
|
||||
|
|
|
|||
|
|
@ -147,6 +147,40 @@
|
|||
}
|
||||
}
|
||||
},
|
||||
"get_commands": {
|
||||
"name": "get_commands",
|
||||
"description": "Returns the `light.turn_on` calls Adaptive Lighting would make to adapt lights, now or for a given time of day, without making them.",
|
||||
"fields": {
|
||||
"entity_id": {
|
||||
"description": "The `entity_id` of the switch whose settings to use. 📝",
|
||||
"name": "entity_id"
|
||||
},
|
||||
"lights": {
|
||||
"description": "Lights or light groups in the switch's `lights` to get the commands for. 💡",
|
||||
"name": "lights"
|
||||
},
|
||||
"time": {
|
||||
"description": "Time of day (HH:MM:SS) to get the commands for, instead of now. 🕰️",
|
||||
"name": "time"
|
||||
},
|
||||
"transition": {
|
||||
"description": "Duration of transition when lights change, in seconds (default: the switch's `initial_transition`). 🕑",
|
||||
"name": "transition"
|
||||
},
|
||||
"adapt_brightness": {
|
||||
"description": "Whether to adapt the brightness of the light (default: the switch's Adapt Brightness switch). 🌞",
|
||||
"name": "adapt_brightness"
|
||||
},
|
||||
"adapt_color": {
|
||||
"description": "Whether to adapt the color on supporting lights (default: the switch's Adapt Color switch). 🌈",
|
||||
"name": "adapt_color"
|
||||
},
|
||||
"prefer_rgb_color": {
|
||||
"description": "Whether to prefer RGB color adjustment over light color temperature when possible (default: the switch's setting). 🌈",
|
||||
"name": "prefer_rgb_color"
|
||||
}
|
||||
}
|
||||
},
|
||||
"set_manual_control": {
|
||||
"name": "set_manual_control",
|
||||
"description": "Mark whether a light is 'manually controlled'.",
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ icon: lucide/zap
|
|||
|
||||
# Services
|
||||
|
||||
Adaptive Lighting provides three services for programmatic control, allowing you to integrate with automations and scripts.
|
||||
Adaptive Lighting provides four services for programmatic control, allowing you to integrate with automations and scripts.
|
||||
|
||||
## adaptive_lighting.apply
|
||||
|
||||
|
|
@ -56,6 +56,70 @@ data:
|
|||
|
||||
---
|
||||
|
||||
## adaptive_lighting.get_commands
|
||||
|
||||
Returns the `light.turn_on` calls Adaptive Lighting would make to adapt lights, without making them: for now, or for a given `time` of day (today).
|
||||
Provide a switch in `entity_id`, a list of `lights`, or both.
|
||||
|
||||
### Parameters
|
||||
|
||||
<!-- CODE:START -->
|
||||
<!-- from adaptive_lighting._docs_helpers import generate_get_commands_markdown_table -->
|
||||
<!-- print(generate_get_commands_markdown_table()) -->
|
||||
<!-- CODE:END -->
|
||||
<!-- OUTPUT:START -->
|
||||
<!-- ⚠️ This content is auto-generated by `markdown-code-runner`. -->
|
||||
| Service data attribute | Description | Required | Type |
|
||||
|:-------------------------|:----------------------------------------------------------------------------------------------------------------------|:-----------|:---------------------|
|
||||
| `entity_id` | The `entity_id` of the switch whose settings to use. 📝 | ❌ | list of `entity_id`s |
|
||||
| `lights` | Lights or light groups in the switch's `lights` to get the commands for. 💡 | ❌ | list of `entity_id`s |
|
||||
| `time` | Time of day (HH:MM:SS) to get the commands for, instead of now. 🕰️ | ❌ | `str` (HH:MM[:SS]) |
|
||||
| `transition` | Duration of transition when lights change, in seconds (default: the switch's `initial_transition`). 🕑 | ❌ | `float` 0-6553 |
|
||||
| `adapt_brightness` | Whether to adapt the brightness of the light (default: the switch's Adapt Brightness switch). 🌞 | ❌ | bool |
|
||||
| `adapt_color` | Whether to adapt the color on supporting lights (default: the switch's Adapt Color switch). 🌈 | ❌ | bool |
|
||||
| `prefer_rgb_color` | Whether to prefer RGB color adjustment over light color temperature when possible (default: the switch's setting). 🌈 | ❌ | bool |
|
||||
|
||||
<!-- OUTPUT:END -->
|
||||
|
||||
<!-- CODE:START -->
|
||||
<!-- from adaptive_lighting.docs_gen import _transform_readme_links -->
|
||||
<!-- print(_transform_readme_links(include_section("../README.md", "get-commands"))) -->
|
||||
<!-- CODE:END -->
|
||||
<!-- OUTPUT:START -->
|
||||
<!-- ⚠️ This content is auto-generated by `markdown-code-runner`. -->
|
||||
Call it with a `response_variable`. For each light, the response lists the calls in order, each with its `service_data` and the `delay` in seconds before it (only nonzero between split calls, see `separate_turn_on_commands`). Only the lights and light groups in a switch's `lights` can be passed; without `entity_id`, each uses the switch it's in (of several, the one that's on). A light group is replaced by its members if the switch expands light groups. Options you leave out follow the switch, with its `initial_transition`, and in sleep mode the calls are for the sleep settings, whatever the time. With `skip_redundant_commands`, what a light already has is left out, as when Adaptive Lighting adapts it; that compares with the light's state when you call the service, so make the calls right away. Manual control doesn't change the calls. In YAML files, quote the time (`"22:00:00"`), since `22:00:00` is read as a number.
|
||||
|
||||
Making the calls is a manual change for Adaptive Lighting: with `take_over_control`, it stops adapting those lights until manual control is reset. A light that's off is adapted to the current settings as it turns on (unless `take_over_control` and `adapt_only_on_bare_turn_on` are enabled), so this example only changes lights that are on:
|
||||
|
||||
```yaml
|
||||
# Set the lights to the settings for 22:00, e.g., from a button
|
||||
sequence:
|
||||
- action: adaptive_lighting.get_commands
|
||||
data:
|
||||
entity_id: switch.adaptive_lighting_living_room
|
||||
time: "22:00:00"
|
||||
response_variable: commands
|
||||
- repeat:
|
||||
for_each: "{{ commands.items() | list }}"
|
||||
sequence:
|
||||
- variables:
|
||||
light: "{{ repeat.item[0] }}"
|
||||
calls: "{{ repeat.item[1] }}"
|
||||
# Only lights that are on, see above
|
||||
- if: "{{ is_state(light, 'on') }}"
|
||||
then:
|
||||
- repeat:
|
||||
for_each: "{{ calls }}"
|
||||
sequence:
|
||||
- delay: "{{ repeat.item.delay }}"
|
||||
- action: light.turn_on
|
||||
data: "{{ repeat.item.service_data }}"
|
||||
```
|
||||
|
||||
<!-- OUTPUT:END -->
|
||||
|
||||
---
|
||||
|
||||
## adaptive_lighting.set_manual_control
|
||||
|
||||
Marks or unmarks a light as "manually controlled". When a light is marked as manually controlled, Adaptive Lighting will not adjust it until the manual control flag is cleared.
|
||||
|
|
|
|||
846
tests/test_get_commands.py
Normal file
846
tests/test_get_commands.py
Normal file
|
|
@ -0,0 +1,846 @@
|
|||
"""Tests for the `adaptive_lighting.get_commands` service."""
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import homeassistant.util.dt as dt_util
|
||||
import pytest
|
||||
import yaml
|
||||
from homeassistant.components.adaptive_lighting.const import (
|
||||
ATTR_ADAPT_BRIGHTNESS,
|
||||
ATTR_ADAPT_COLOR,
|
||||
CONF_ADAPT_ONLY_ON_BARE_TURN_ON,
|
||||
CONF_ADAPT_UNTIL_SLEEP,
|
||||
CONF_COMMANDS_TIME,
|
||||
CONF_EXPAND_LIGHT_GROUPS,
|
||||
CONF_INITIAL_TRANSITION,
|
||||
CONF_LIGHTS,
|
||||
CONF_MAX_BRIGHTNESS,
|
||||
CONF_MIN_BRIGHTNESS,
|
||||
CONF_MIN_COLOR_TEMP,
|
||||
CONF_PREFER_RGB_COLOR,
|
||||
CONF_SEND_SPLIT_DELAY,
|
||||
CONF_SEPARATE_TURN_ON_COMMANDS,
|
||||
CONF_SKIP_REDUNDANT_COMMANDS,
|
||||
CONF_SLEEP_RGB_OR_COLOR_TEMP,
|
||||
CONF_TRANSITION,
|
||||
DEFAULT_SLEEP_BRIGHTNESS,
|
||||
DEFAULT_SLEEP_RGB_COLOR,
|
||||
DOMAIN,
|
||||
SERVICE_APPLY,
|
||||
SERVICE_GET_COMMANDS,
|
||||
)
|
||||
from homeassistant.components.light import (
|
||||
ATTR_BRIGHTNESS,
|
||||
ATTR_COLOR_TEMP_KELVIN,
|
||||
ATTR_RGB_COLOR,
|
||||
ATTR_TRANSITION,
|
||||
)
|
||||
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_NAME,
|
||||
SERVICE_TURN_OFF,
|
||||
SERVICE_TURN_ON,
|
||||
)
|
||||
from homeassistant.exceptions import ServiceValidationError
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .test_switch import (
|
||||
CONF_INTERCEPT,
|
||||
ENTITY_LIGHT_1,
|
||||
ENTITY_LIGHT_2,
|
||||
ENTITY_LIGHT_3,
|
||||
SUNRISE,
|
||||
_track_adaptive_light_calls,
|
||||
async_process_ha_core_config,
|
||||
cleanup, # noqa: F401 # fixture
|
||||
reset_time_zone, # noqa: F401 # fixture
|
||||
setup_lights,
|
||||
setup_lights_and_switch,
|
||||
setup_switch,
|
||||
)
|
||||
|
||||
UTCNOW = "homeassistant.components.adaptive_lighting.color_and_brightness.utcnow"
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("cleanup")
|
||||
|
||||
|
||||
def _noon() -> datetime.datetime:
|
||||
"""Noon (in the test's time zone) on the day of the fixed sun events."""
|
||||
return SUNRISE.replace(hour=12, tzinfo=dt_util.DEFAULT_TIME_ZONE).astimezone(
|
||||
dt_util.UTC,
|
||||
)
|
||||
|
||||
|
||||
def _today_at(switch, time: datetime.time) -> datetime.datetime:
|
||||
"""The time of day today, in UTC, in the switch's time zone."""
|
||||
tz = switch._sun_light_settings.timezone
|
||||
return dt_util.as_utc(
|
||||
datetime.datetime.combine(dt_util.now(tz).date(), time, tzinfo=tz),
|
||||
)
|
||||
|
||||
|
||||
async def _get_commands(hass, switch, **data):
|
||||
return await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
{ATTR_ENTITY_ID: switch.entity_id, **data},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_get_commands_match_what_apply_sends(hass):
|
||||
"""With the switch's defaults, the commands are exactly what `apply` sends."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
# In the evening, when brightness and color change by the minute
|
||||
with patch(UTCNOW, return_value=_today_at(switch, datetime.time(23, 0))):
|
||||
commands = await _get_commands(hass, switch)
|
||||
calls = _track_adaptive_light_calls(hass)
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_APPLY,
|
||||
{ATTR_ENTITY_ID: switch.entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert set(commands) == set(switch.lights)
|
||||
sent = {call[ATTR_ENTITY_ID]: call for call in calls}
|
||||
for light in switch.lights:
|
||||
assert [c["service_data"] for c in commands[light]] == [sent[light]]
|
||||
assert commands[light][0]["delay"] == 0
|
||||
|
||||
|
||||
async def test_get_commands_for_a_time(hass):
|
||||
"""The commands for a time are those the regular adaptation makes at it."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
time = datetime.time(23, 30)
|
||||
with patch(UTCNOW, return_value=_today_at(switch, time)):
|
||||
at_that_time = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
with patch(UTCNOW, return_value=_noon()):
|
||||
now = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
for_time = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_COMMANDS_TIME: "23:30:00"},
|
||||
)
|
||||
assert for_time == at_that_time
|
||||
assert for_time != now
|
||||
assert list(for_time) == [ENTITY_LIGHT_1]
|
||||
|
||||
|
||||
async def test_get_commands_is_read_only(hass):
|
||||
"""Nothing is sent, and manual control and the switch's settings are kept."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
unmarked = await _get_commands(hass, switch, **{CONF_COMMANDS_TIME: "23:30:00"})
|
||||
switch.manager.set_manual_control_attributes(ENTITY_LIGHT_1)
|
||||
manual_control = dict(switch.manager.manual_control)
|
||||
settings = dict(switch._settings)
|
||||
calls = _track_adaptive_light_calls(hass, ours_only=False)
|
||||
commands = await _get_commands(hass, switch, **{CONF_COMMANDS_TIME: "23:30:00"})
|
||||
await hass.async_block_till_done()
|
||||
assert calls == []
|
||||
assert switch.manager.manual_control == manual_control
|
||||
assert switch._settings == settings
|
||||
# Manual control doesn't change the commands
|
||||
assert commands == unmarked
|
||||
|
||||
|
||||
async def test_get_commands_split(hass):
|
||||
"""With separate commands, they're returned in order with the waits between."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_SEPARATE_TURN_ON_COMMANDS: True, CONF_SEND_SPLIT_DELAY: 100},
|
||||
all_lights=True,
|
||||
)
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_3], # supports transitions
|
||||
**{CONF_TRANSITION: 2},
|
||||
)
|
||||
calls = commands[ENTITY_LIGHT_3]
|
||||
assert [set(c["service_data"]) for c in calls] == [
|
||||
{ATTR_ENTITY_ID, ATTR_BRIGHTNESS, ATTR_TRANSITION},
|
||||
{ATTR_ENTITY_ID, ATTR_COLOR_TEMP_KELVIN, ATTR_TRANSITION},
|
||||
]
|
||||
assert [c["service_data"][ATTR_TRANSITION] for c in calls] == [1, 1]
|
||||
assert [c["delay"] for c in calls] == [0, pytest.approx(1.1)]
|
||||
|
||||
|
||||
async def test_get_commands_follow_adapt_switches(hass):
|
||||
"""What's adapted follows the switch's adapt switches, unless specified."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
await hass.services.async_call(
|
||||
SWITCH_DOMAIN,
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: switch.adapt_color_switch.entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert ATTR_BRIGHTNESS in call["service_data"]
|
||||
assert ATTR_COLOR_TEMP_KELVIN not in call["service_data"]
|
||||
assert ATTR_RGB_COLOR not in call["service_data"]
|
||||
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{ATTR_ADAPT_COLOR: True, ATTR_ADAPT_BRIGHTNESS: False},
|
||||
)
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert ATTR_BRIGHTNESS not in call["service_data"]
|
||||
assert ATTR_COLOR_TEMP_KELVIN in call["service_data"]
|
||||
|
||||
# Nothing to adapt
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{ATTR_ADAPT_BRIGHTNESS: False},
|
||||
)
|
||||
assert commands == {ENTITY_LIGHT_1: []}
|
||||
|
||||
|
||||
async def test_get_commands_color_choice(hass):
|
||||
"""RGB follows the switch's `prefer_rgb_color`; dimmers get no color."""
|
||||
switch, _ = await setup_lights_and_switch(hass, {CONF_PREFER_RGB_COLOR: True})
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert isinstance(call["service_data"][ATTR_RGB_COLOR], list)
|
||||
assert ATTR_COLOR_TEMP_KELVIN not in call["service_data"]
|
||||
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_PREFER_RGB_COLOR: False},
|
||||
)
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert ATTR_COLOR_TEMP_KELVIN in call["service_data"]
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.adaptive_lighting.switch._supported_features",
|
||||
return_value={"brightness"},
|
||||
):
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert set(call["service_data"]) == {ATTR_ENTITY_ID, ATTR_BRIGHTNESS}
|
||||
|
||||
|
||||
async def test_get_commands_sleep_mode(hass):
|
||||
"""In sleep mode, the commands are for the sleep settings, whatever the time."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
await hass.services.async_call(
|
||||
SWITCH_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: switch.sleep_mode_switch.entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_COMMANDS_TIME: "12:00:00"},
|
||||
)
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert call["service_data"][ATTR_BRIGHTNESS] == round(
|
||||
255 * DEFAULT_SLEEP_BRIGHTNESS / 100,
|
||||
)
|
||||
|
||||
|
||||
async def test_get_commands_all_lights_of_switches(hass):
|
||||
"""Without lights, all lights of the switch, whether on or off."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: ENTITY_LIGHT_2},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(hass, switch)
|
||||
assert set(commands) == {ENTITY_LIGHT_1, ENTITY_LIGHT_2}
|
||||
assert all(commands.values())
|
||||
|
||||
|
||||
async def test_get_commands_needs_response(hass):
|
||||
"""It only returns data, so calling it without asking for it is an error."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
with pytest.raises(ServiceValidationError):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
{ATTR_ENTITY_ID: switch.entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
|
||||
async def test_get_commands_far_from_utc(
|
||||
hass,
|
||||
reset_time_zone, # noqa: F811 # pylint: disable=redefined-outer-name
|
||||
):
|
||||
"""A time is calculated like the regular adaptation, also in UTC+14."""
|
||||
await async_process_ha_core_config(
|
||||
hass,
|
||||
{
|
||||
"latitude": 1.87,
|
||||
"longitude": -157.4,
|
||||
"time_zone": "Pacific/Kiritimati",
|
||||
"country": "KI",
|
||||
},
|
||||
)
|
||||
await setup_lights(hass)
|
||||
_, switch = await setup_switch(
|
||||
hass,
|
||||
{CONF_LIGHTS: [ENTITY_LIGHT_1], CONF_INITIAL_TRANSITION: 0},
|
||||
)
|
||||
for time in (datetime.time(0, 10), datetime.time(12)):
|
||||
with patch(UTCNOW, return_value=_today_at(switch, time)):
|
||||
at_that_time = await _get_commands(hass, switch)
|
||||
for_time = await _get_commands(hass, switch, **{CONF_COMMANDS_TIME: time})
|
||||
assert for_time == at_that_time
|
||||
|
||||
|
||||
async def test_get_commands_time_like_now_with_transition(hass):
|
||||
"""With a transition, a time is (like now) for when the transition ends."""
|
||||
switch, _ = await setup_lights_and_switch(hass, all_lights=True)
|
||||
time = datetime.time(21, 58)
|
||||
data = {"lights": [ENTITY_LIGHT_3], CONF_TRANSITION: 300} # supports transitions
|
||||
with patch(UTCNOW, return_value=_today_at(switch, time)):
|
||||
at_that_time = await _get_commands(hass, switch, **data)
|
||||
for_time = await _get_commands(hass, switch, **data, **{CONF_COMMANDS_TIME: time})
|
||||
assert for_time == at_that_time
|
||||
(call,) = for_time[ENTITY_LIGHT_3]
|
||||
assert call["service_data"][ATTR_TRANSITION] == 300
|
||||
|
||||
|
||||
async def test_get_commands_for_a_daytime(hass):
|
||||
"""Also during the day, where the color temperature changes by the minute."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
time = datetime.time(6, 30)
|
||||
with patch(UTCNOW, return_value=_today_at(switch, time)):
|
||||
at_that_time = await _get_commands(hass, switch)
|
||||
for_time = await _get_commands(hass, switch, **{CONF_COMMANDS_TIME: time})
|
||||
assert for_time == at_that_time
|
||||
with patch(UTCNOW, return_value=_today_at(switch, datetime.time(7, 30))):
|
||||
an_hour_later = await _get_commands(hass, switch)
|
||||
assert for_time != an_hour_later
|
||||
|
||||
|
||||
async def test_get_commands_split_without_transition_support(hass):
|
||||
"""A light without transitions gets none, and only the split delay between."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_SEPARATE_TURN_ON_COMMANDS: True, CONF_SEND_SPLIT_DELAY: 100},
|
||||
)
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_TRANSITION: 2},
|
||||
)
|
||||
calls = commands[ENTITY_LIGHT_1]
|
||||
assert all(ATTR_TRANSITION not in c["service_data"] for c in calls)
|
||||
assert [c["delay"] for c in calls] == [0, pytest.approx(0.1)]
|
||||
|
||||
|
||||
async def test_get_commands_sleep_rgb(hass):
|
||||
"""In sleep mode with RGB, the sleep color is sent as RGB."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_SLEEP_RGB_OR_COLOR_TEMP: "rgb_color"},
|
||||
)
|
||||
await hass.services.async_call(
|
||||
SWITCH_DOMAIN,
|
||||
SERVICE_TURN_ON,
|
||||
{ATTR_ENTITY_ID: switch.sleep_mode_switch.entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert call["service_data"][ATTR_RGB_COLOR] == list(DEFAULT_SLEEP_RGB_COLOR)
|
||||
|
||||
|
||||
async def test_get_commands_rgb_until_sleep(hass):
|
||||
"""After sunset with `transition_until_sleep` and RGB, the color is RGB."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_SLEEP_RGB_OR_COLOR_TEMP: "rgb_color", CONF_ADAPT_UNTIL_SLEEP: True},
|
||||
)
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_COMMANDS_TIME: "23:30:00"},
|
||||
)
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
assert ATTR_RGB_COLOR in call["service_data"]
|
||||
assert ATTR_COLOR_TEMP_KELVIN not in call["service_data"]
|
||||
|
||||
|
||||
async def test_get_commands_clamps_color_temp(hass):
|
||||
"""The color temperature is clamped to what the light supports."""
|
||||
switch, _ = await setup_lights_and_switch(hass, {CONF_MIN_COLOR_TEMP: 1000})
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_COMMANDS_TIME: "23:30:00"},
|
||||
)
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
minimum = hass.states.get(ENTITY_LIGHT_1).attributes["min_color_temp_kelvin"]
|
||||
assert minimum > 1000
|
||||
assert call["service_data"][ATTR_COLOR_TEMP_KELVIN] == minimum
|
||||
|
||||
|
||||
async def test_get_commands_several_switches(hass):
|
||||
"""Several switches' lights are combined, but a shared light is an error."""
|
||||
await setup_lights(hass)
|
||||
_, switch_1 = await setup_switch(
|
||||
hass,
|
||||
{CONF_NAME: "one", CONF_LIGHTS: [ENTITY_LIGHT_1], CONF_INITIAL_TRANSITION: 0},
|
||||
)
|
||||
_, switch_2 = await setup_switch(
|
||||
hass,
|
||||
{CONF_NAME: "two", CONF_LIGHTS: [ENTITY_LIGHT_2], CONF_INITIAL_TRANSITION: 0},
|
||||
)
|
||||
commands = await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
{ATTR_ENTITY_ID: [switch_1.entity_id, switch_2.entity_id]},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
assert set(commands) == {ENTITY_LIGHT_1, ENTITY_LIGHT_2}
|
||||
|
||||
_, switch_3 = await setup_switch(
|
||||
hass,
|
||||
{CONF_NAME: "three", CONF_LIGHTS: [ENTITY_LIGHT_1]},
|
||||
)
|
||||
with pytest.raises(ServiceValidationError, match="more than one"):
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
{ATTR_ENTITY_ID: [switch_1.entity_id, switch_3.entity_id]},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("expand", [True, False])
|
||||
async def test_get_commands_light_groups(hass, expand):
|
||||
"""Light groups are expanded to their members, as the switch does."""
|
||||
await setup_lights(hass, with_group=True)
|
||||
_, switch = await setup_switch(
|
||||
hass,
|
||||
{CONF_LIGHTS: ["light.light_group"], CONF_EXPAND_LIGHT_GROUPS: expand},
|
||||
)
|
||||
commands = await _get_commands(hass, switch)
|
||||
expected = {"light.light_4", "light.light_5"} if expand else {"light.light_group"}
|
||||
assert set(commands) == expected
|
||||
|
||||
|
||||
async def test_get_commands_unknown_lights(hass):
|
||||
"""Unknown lights, e.g. typos, are an error rather than left out."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
with pytest.raises(ServiceValidationError, match=r"light\.kitchn"):
|
||||
await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1, "light.kitchn"])
|
||||
|
||||
|
||||
def _readme_example() -> dict:
|
||||
"""The README's example script for `adaptive_lighting.get_commands`."""
|
||||
readme = Path(__file__).resolve().parents[1].joinpath("README.md").read_text()
|
||||
section = readme[readme.index("#### `adaptive_lighting.get_commands`") :]
|
||||
match = re.search(r"```yaml\n(.*?)```", section, re.DOTALL)
|
||||
assert match is not None
|
||||
return yaml.safe_load(match.group(1))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("intercept", "split"),
|
||||
[(False, False), (True, False), (False, True)],
|
||||
)
|
||||
async def test_get_commands_readme_example(hass, intercept, split):
|
||||
"""The README's example script makes the calls, for lights that are on."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{
|
||||
CONF_INTERCEPT: intercept,
|
||||
CONF_SEPARATE_TURN_ON_COMMANDS: split,
|
||||
CONF_SEND_SPLIT_DELAY: 100,
|
||||
},
|
||||
)
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: ENTITY_LIGHT_2},
|
||||
blocking=True,
|
||||
)
|
||||
config = _readme_example()
|
||||
# At noon, for a time with other settings than now
|
||||
config["sequence"][0]["data"][ATTR_ENTITY_ID] = switch.entity_id
|
||||
config["sequence"][0]["data"]["time"] = "23:30:00"
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"script",
|
||||
{"script": {"example": config}},
|
||||
)
|
||||
with patch(UTCNOW, return_value=_noon()):
|
||||
await switch._update_attrs_and_maybe_adapt_lights(
|
||||
context=switch.create_context("test"),
|
||||
transition=0,
|
||||
force=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
before = dict(hass.states.get(ENTITY_LIGHT_1).attributes)
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
**{CONF_COMMANDS_TIME: "23:30:00"},
|
||||
)
|
||||
await hass.services.async_call("script", "example", blocking=True)
|
||||
await hass.async_block_till_done()
|
||||
calls = commands[ENTITY_LIGHT_1]
|
||||
assert [c["delay"] > 0 for c in calls] == ([False, True] if split else [False])
|
||||
expected = {}
|
||||
for call in calls:
|
||||
expected.update(call["service_data"])
|
||||
state = hass.states.get(ENTITY_LIGHT_1)
|
||||
for key in (ATTR_BRIGHTNESS, ATTR_COLOR_TEMP_KELVIN):
|
||||
assert state.attributes[key] == expected[key]
|
||||
assert before[key] != expected[key]
|
||||
assert hass.states.get(ENTITY_LIGHT_2).state == "off"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bare_only", [False, True])
|
||||
async def test_get_commands_turning_light_on(hass, bare_only):
|
||||
"""Turning a light on with the calls only sticks with `adapt_only_on_bare_turn_on`.
|
||||
|
||||
As documented: otherwise Adaptive Lighting adapts the light for now as it
|
||||
turns on, replacing the values.
|
||||
"""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_INTERCEPT: True, CONF_ADAPT_ONLY_ON_BARE_TURN_ON: bare_only},
|
||||
)
|
||||
with patch(UTCNOW, return_value=_noon()):
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_OFF,
|
||||
{ATTR_ENTITY_ID: ENTITY_LIGHT_1},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(
|
||||
hass,
|
||||
switch,
|
||||
lights=[ENTITY_LIGHT_1],
|
||||
**{CONF_COMMANDS_TIME: "23:30:00"},
|
||||
)
|
||||
(call,) = commands[ENTITY_LIGHT_1]
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_ON,
|
||||
call["service_data"],
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
brightness = hass.states.get(ENTITY_LIGHT_1).attributes[ATTR_BRIGHTNESS]
|
||||
assert (brightness == call["service_data"][ATTR_BRIGHTNESS]) == bare_only
|
||||
|
||||
|
||||
async def test_get_commands_default_transition(hass):
|
||||
"""The transition defaults to the switch's `initial_transition`."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_INITIAL_TRANSITION: 2, CONF_TRANSITION: 30},
|
||||
all_lights=True,
|
||||
)
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_3])
|
||||
(call,) = commands[ENTITY_LIGHT_3] # supports transitions
|
||||
assert call["service_data"][ATTR_TRANSITION] == 2
|
||||
|
||||
|
||||
async def test_get_commands_switch_off(hass):
|
||||
"""The commands are the same when the switch is off."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
data = {CONF_COMMANDS_TIME: "23:30:00"}
|
||||
on = await _get_commands(hass, switch, **data)
|
||||
await switch.async_turn_off()
|
||||
await hass.async_block_till_done()
|
||||
assert await _get_commands(hass, switch, **data) == on
|
||||
|
||||
|
||||
async def test_get_commands_lights_by_their_switches(hass):
|
||||
"""Without a switch, each light's commands are for the switch it's in."""
|
||||
await setup_lights(hass)
|
||||
_, dim = await setup_switch(
|
||||
hass,
|
||||
{
|
||||
CONF_NAME: "dim",
|
||||
CONF_LIGHTS: [ENTITY_LIGHT_1],
|
||||
CONF_INITIAL_TRANSITION: 0,
|
||||
CONF_MAX_BRIGHTNESS: 20,
|
||||
},
|
||||
)
|
||||
_, bright = await setup_switch(
|
||||
hass,
|
||||
{
|
||||
CONF_NAME: "bright",
|
||||
CONF_LIGHTS: [ENTITY_LIGHT_2],
|
||||
CONF_INITIAL_TRANSITION: 0,
|
||||
CONF_MIN_BRIGHTNESS: 80,
|
||||
},
|
||||
)
|
||||
# Which switches are on doesn't matter
|
||||
await bright.async_turn_off()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
async def get_commands(**data):
|
||||
return await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
data,
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
|
||||
time = {CONF_COMMANDS_TIME: "12:00:00"}
|
||||
commands = await get_commands(lights=[ENTITY_LIGHT_1, ENTITY_LIGHT_2], **time)
|
||||
assert commands == {
|
||||
**await _get_commands(hass, dim, lights=[ENTITY_LIGHT_1], **time),
|
||||
**await _get_commands(hass, bright, lights=[ENTITY_LIGHT_2], **time),
|
||||
}
|
||||
brightness = {
|
||||
light: calls[0]["service_data"][ATTR_BRIGHTNESS]
|
||||
for light, calls in commands.items()
|
||||
}
|
||||
assert brightness[ENTITY_LIGHT_1] < brightness[ENTITY_LIGHT_2]
|
||||
|
||||
with pytest.raises(ServiceValidationError, match="isn't among the lights of any"):
|
||||
await get_commands(lights=[ENTITY_LIGHT_3])
|
||||
# A light in several switches: the one that's on, like for `apply`
|
||||
_, also_dim = await setup_switch(
|
||||
hass,
|
||||
{CONF_NAME: "also dim", CONF_LIGHTS: [ENTITY_LIGHT_1], CONF_MAX_BRIGHTNESS: 50},
|
||||
)
|
||||
with pytest.raises(ServiceValidationError, match="more than one switch"):
|
||||
await get_commands(lights=[ENTITY_LIGHT_1])
|
||||
await dim.async_turn_off()
|
||||
await hass.async_block_till_done()
|
||||
assert await get_commands(lights=[ENTITY_LIGHT_1], **time) == (
|
||||
await _get_commands(hass, also_dim, lights=[ENTITY_LIGHT_1], **time)
|
||||
)
|
||||
with pytest.raises(ServiceValidationError, match="Pass a switch"):
|
||||
await get_commands()
|
||||
|
||||
|
||||
async def test_get_commands_without_switch_is_read_only(hass):
|
||||
"""Finding the lights' switches doesn't update the switches' lights."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
with patch.object(
|
||||
type(switch),
|
||||
"_expand_light_groups",
|
||||
side_effect=AssertionError("updated the switch's lights"),
|
||||
):
|
||||
commands = await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
{CONF_LIGHTS: [ENTITY_LIGHT_1]},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
assert list(commands) == [ENTITY_LIGHT_1]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("expand", [True, False])
|
||||
async def test_get_commands_group_without_switch(hass, expand):
|
||||
"""A group in `lights` is resolved to the switch with the group (or members)."""
|
||||
await setup_lights(hass, with_group=True)
|
||||
_, switch = await setup_switch(
|
||||
hass,
|
||||
{CONF_LIGHTS: ["light.light_group"], CONF_EXPAND_LIGHT_GROUPS: expand},
|
||||
)
|
||||
commands = await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
{CONF_LIGHTS: ["light.light_group"]},
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
assert commands == await _get_commands(hass, switch)
|
||||
|
||||
|
||||
async def test_get_commands_only_the_switchs_lights(hass):
|
||||
"""Only lights in the switch's `lights` are allowed, e.g. not typos."""
|
||||
switch, _ = await setup_lights_and_switch(hass)
|
||||
assert ENTITY_LIGHT_3 not in switch.lights
|
||||
with pytest.raises(ServiceValidationError, match="isn't among the lights"):
|
||||
await _get_commands(hass, switch, lights=[ENTITY_LIGHT_3])
|
||||
with pytest.raises(
|
||||
ServiceValidationError,
|
||||
match=rf"'{ENTITY_LIGHT_3}', 'light\.typo' aren't among the lights of"
|
||||
rf" '{switch.entity_id}'\.$",
|
||||
):
|
||||
await _get_commands(hass, switch, lights=[ENTITY_LIGHT_3, "light.typo"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("expand", [True, False])
|
||||
async def test_get_commands_only_configured_entities(hass, expand):
|
||||
"""Only the entities in the switches' `lights` are allowed, not their members."""
|
||||
await setup_lights(hass, with_group=True)
|
||||
_, switch = await setup_switch(
|
||||
hass,
|
||||
{CONF_LIGHTS: ["light.light_group"], CONF_EXPAND_LIGHT_GROUPS: expand},
|
||||
)
|
||||
|
||||
async def get_commands(**data):
|
||||
return await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_GET_COMMANDS,
|
||||
data,
|
||||
blocking=True,
|
||||
return_response=True,
|
||||
)
|
||||
|
||||
for data in (
|
||||
{CONF_LIGHTS: ["light.light_4"]},
|
||||
{CONF_LIGHTS: ["light.light_4"], ATTR_ENTITY_ID: switch.entity_id},
|
||||
):
|
||||
# The error names the light group to pass instead
|
||||
with pytest.raises(
|
||||
ServiceValidationError,
|
||||
match=r"isn't among the lights of .* 'light\.light_group' for",
|
||||
):
|
||||
await get_commands(**data)
|
||||
assert await get_commands(lights=["light.light_group"]) == (
|
||||
await _get_commands(hass, switch)
|
||||
)
|
||||
|
||||
|
||||
async def test_get_commands_leaves_out_lights_without_state(hass):
|
||||
"""A switch's light without a state (e.g., not loaded) is left out."""
|
||||
await setup_lights(hass)
|
||||
_, switch = await setup_switch(
|
||||
hass,
|
||||
{CONF_LIGHTS: [ENTITY_LIGHT_1, "light.not_loaded"]},
|
||||
)
|
||||
commands = await _get_commands(hass, switch)
|
||||
assert list(commands) == [ENTITY_LIGHT_1]
|
||||
|
||||
|
||||
async def test_get_commands_response_is_json(hass):
|
||||
"""The response can be serialized, e.g., for automations and the frontend."""
|
||||
switch, _ = await setup_lights_and_switch(hass, {CONF_PREFER_RGB_COLOR: True})
|
||||
commands = await _get_commands(hass, switch)
|
||||
assert json.loads(json.dumps(commands)) == commands
|
||||
|
||||
|
||||
@pytest.mark.parametrize("skip_redundant", [True, False])
|
||||
async def test_get_commands_skip_redundant(hass, skip_redundant):
|
||||
"""With `skip_redundant_commands`, what a light already has is left out."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_SKIP_REDUNDANT_COMMANDS: skip_redundant},
|
||||
)
|
||||
with patch(UTCNOW, return_value=_noon()):
|
||||
full = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
(call,) = full[ENTITY_LIGHT_1]
|
||||
target = call["service_data"]
|
||||
# The light has the brightness already, but another color temperature
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_ON,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_LIGHT_1,
|
||||
ATTR_BRIGHTNESS: target[ATTR_BRIGHTNESS],
|
||||
ATTR_COLOR_TEMP_KELVIN: target[ATTR_COLOR_TEMP_KELVIN] - 1000,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
partial = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
# And then everything
|
||||
await hass.services.async_call("light", SERVICE_TURN_ON, target, blocking=True)
|
||||
nothing = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
if not skip_redundant:
|
||||
assert partial == nothing == full
|
||||
return
|
||||
(call,) = partial[ENTITY_LIGHT_1]
|
||||
assert set(call["service_data"]) == {ATTR_ENTITY_ID, ATTR_COLOR_TEMP_KELVIN}
|
||||
assert nothing == {ENTITY_LIGHT_1: []}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("already_has", ["brightness", "color"])
|
||||
async def test_get_commands_skip_redundant_split(hass, already_has):
|
||||
"""A split call with nothing left is dropped, and the rest isn't delayed."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{
|
||||
CONF_SKIP_REDUNDANT_COMMANDS: True,
|
||||
CONF_SEPARATE_TURN_ON_COMMANDS: True,
|
||||
CONF_SEND_SPLIT_DELAY: 100,
|
||||
},
|
||||
)
|
||||
with patch(UTCNOW, return_value=_noon()):
|
||||
full = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
brightness_call, color_call = full[ENTITY_LIGHT_1]
|
||||
brightness = brightness_call["service_data"][ATTR_BRIGHTNESS]
|
||||
color_temp = color_call["service_data"][ATTR_COLOR_TEMP_KELVIN]
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_ON,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_LIGHT_1,
|
||||
ATTR_BRIGHTNESS: (
|
||||
brightness if already_has == "brightness" else brightness - 50
|
||||
),
|
||||
ATTR_COLOR_TEMP_KELVIN: (
|
||||
color_temp if already_has == "color" else color_temp - 1000
|
||||
),
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
remaining = color_call if already_has == "brightness" else brightness_call
|
||||
assert commands == {
|
||||
ENTITY_LIGHT_1: [{"service_data": remaining["service_data"], "delay": 0}],
|
||||
}
|
||||
|
||||
|
||||
async def test_get_commands_skip_redundant_match_what_apply_sends(hass):
|
||||
"""With `skip_redundant_commands`, the calls are still what `apply` sends."""
|
||||
switch, _ = await setup_lights_and_switch(
|
||||
hass,
|
||||
{CONF_SKIP_REDUNDANT_COMMANDS: True},
|
||||
)
|
||||
with patch(UTCNOW, return_value=_noon()):
|
||||
(call,) = (await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1]))[
|
||||
ENTITY_LIGHT_1
|
||||
]
|
||||
await hass.services.async_call(
|
||||
"light",
|
||||
SERVICE_TURN_ON,
|
||||
{
|
||||
ATTR_ENTITY_ID: ENTITY_LIGHT_1,
|
||||
ATTR_BRIGHTNESS: call["service_data"][ATTR_BRIGHTNESS],
|
||||
ATTR_COLOR_TEMP_KELVIN: (
|
||||
call["service_data"][ATTR_COLOR_TEMP_KELVIN] - 1000
|
||||
),
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
commands = await _get_commands(hass, switch, lights=[ENTITY_LIGHT_1])
|
||||
calls = _track_adaptive_light_calls(hass)
|
||||
await hass.services.async_call(
|
||||
DOMAIN,
|
||||
SERVICE_APPLY,
|
||||
{ATTR_ENTITY_ID: switch.entity_id, CONF_LIGHTS: [ENTITY_LIGHT_1]},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert [c["service_data"] for c in commands[ENTITY_LIGHT_1]] == calls
|
||||
|
|
@ -8,6 +8,7 @@ from homeassistant.components.adaptive_lighting.const import (
|
|||
DEFAULT_NAME,
|
||||
SERVICE_APPLY,
|
||||
SERVICE_CHANGE_SWITCH_SETTINGS,
|
||||
SERVICE_GET_COMMANDS,
|
||||
SERVICE_SET_MANUAL_CONTROL,
|
||||
UNDO_UPDATE_LISTENER,
|
||||
)
|
||||
|
|
@ -70,10 +71,12 @@ async def test_services_survive_entry_unload_and_reload(hass):
|
|||
service_names = (
|
||||
SERVICE_APPLY,
|
||||
SERVICE_CHANGE_SWITCH_SETTINGS,
|
||||
SERVICE_GET_COMMANDS,
|
||||
SERVICE_SET_MANUAL_CONTROL,
|
||||
)
|
||||
services = hass.services.async_services()[adaptive_lighting.DOMAIN]
|
||||
assert SERVICE_APPLY in services
|
||||
assert SERVICE_GET_COMMANDS in services
|
||||
assert SERVICE_SET_MANUAL_CONTROL in services
|
||||
if hasattr(service, "async_register_platform_entity_service"):
|
||||
assert SERVICE_CHANGE_SWITCH_SETTINGS in services
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue