-
-
Notifications
You must be signed in to change notification settings - Fork 30.1k
/
device_condition.py
127 lines (100 loc) · 3.78 KB
/
device_condition.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""Provide the device automations for Climate."""
from __future__ import annotations
import voluptuous as vol
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_CONDITION,
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_ENTITY_ID,
CONF_TYPE,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
condition,
config_validation as cv,
entity_registry as er,
)
from homeassistant.helpers.config_validation import DEVICE_CONDITION_BASE_SCHEMA
from homeassistant.helpers.entity import get_capability, get_supported_features
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from . import DOMAIN, const
CONDITION_TYPES = {"is_hvac_mode", "is_preset_mode"}
HVAC_MODE_CONDITION = DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): "is_hvac_mode",
vol.Required(const.ATTR_HVAC_MODE): vol.In(const.HVAC_MODES),
}
)
PRESET_MODE_CONDITION = DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TYPE): "is_preset_mode",
vol.Required(const.ATTR_PRESET_MODE): str,
}
)
CONDITION_SCHEMA = vol.Any(HVAC_MODE_CONDITION, PRESET_MODE_CONDITION)
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
"""List device conditions for Climate devices."""
registry = er.async_get(hass)
conditions = []
# Get all the integrations entities for this device
for entry in er.async_entries_for_device(registry, device_id):
if entry.domain != DOMAIN:
continue
supported_features = get_supported_features(hass, entry.entity_id)
base_condition = {
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
}
conditions.append({**base_condition, CONF_TYPE: "is_hvac_mode"})
if supported_features & const.SUPPORT_PRESET_MODE:
conditions.append({**base_condition, CONF_TYPE: "is_preset_mode"})
return conditions
@callback
def async_condition_from_config(
hass: HomeAssistant, config: ConfigType
) -> condition.ConditionCheckerType:
"""Create a function to test a device condition."""
def test_is_state(hass: HomeAssistant, variables: TemplateVarsType) -> bool:
"""Test if an entity is a certain state."""
if (state := hass.states.get(config[ATTR_ENTITY_ID])) is None:
return False
if config[CONF_TYPE] == "is_hvac_mode":
return state.state == config[const.ATTR_HVAC_MODE]
return (
state.attributes.get(const.ATTR_PRESET_MODE)
== config[const.ATTR_PRESET_MODE]
)
return test_is_state
async def async_get_condition_capabilities(
hass: HomeAssistant, config: ConfigType
) -> dict[str, vol.Schema]:
"""List condition capabilities."""
condition_type = config[CONF_TYPE]
fields = {}
if condition_type == "is_hvac_mode":
try:
hvac_modes = (
get_capability(hass, config[ATTR_ENTITY_ID], const.ATTR_HVAC_MODES)
or []
)
except HomeAssistantError:
hvac_modes = []
fields[vol.Required(const.ATTR_HVAC_MODE)] = vol.In(hvac_modes)
elif condition_type == "is_preset_mode":
try:
preset_modes = (
get_capability(hass, config[ATTR_ENTITY_ID], const.ATTR_PRESET_MODES)
or []
)
except HomeAssistantError:
preset_modes = []
fields[vol.Required(const.ATTR_PRESET_MODE)] = vol.In(preset_modes)
return {"extra_fields": vol.Schema(fields)}