Add water_heater conditions (#166335)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

homeassistant/components/automation/__init__.py

@@ -141,6 +141,7 @@ _EXPERIMENTAL_CONDITION_PLATFORMS = {
     "switch",
     "text",
     "vacuum",
+    "water_heater",
     "window",
 }
 

homeassistant/components/water_heater/condition.py

@@ -0,0 +1,102 @@
+"""Provides conditions for water heaters."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import voluptuous as vol
+
+from homeassistant.const import (
+    ATTR_TEMPERATURE,
+    CONF_OPTIONS,
+    CONF_TARGET,
+    STATE_OFF,
+    UnitOfTemperature,
+)
+from homeassistant.core import HomeAssistant, State
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.automation import DomainSpec, NumericalDomainSpec
+from homeassistant.helpers.condition import (
+    ATTR_BEHAVIOR,
+    BEHAVIOR_ALL,
+    BEHAVIOR_ANY,
+    Condition,
+    ConditionConfig,
+    EntityConditionBase,
+    EntityNumericalConditionWithUnitBase,
+    make_entity_state_condition,
+)
+from homeassistant.util.unit_conversion import TemperatureConverter
+
+from .const import DOMAIN
+
+ATTR_OPERATION_MODE = "operation_mode"
+
+
+_OPERATION_MODE_CONDITION_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
+        vol.Required(CONF_OPTIONS): {
+            vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
+                [BEHAVIOR_ANY, BEHAVIOR_ALL]
+            ),
+            vol.Required(ATTR_OPERATION_MODE): vol.All(
+                cv.ensure_list, vol.Length(min=1), [str]
+            ),
+        },
+    }
+)
+
+
+class WaterHeaterOnCondition(EntityConditionBase):
+    """Condition for water heater being on."""
+
+    _domain_specs = {DOMAIN: DomainSpec()}
+
+    def is_valid_state(self, entity_state: State) -> bool:
+        """Check if the water heater is in a non-off state."""
+        return entity_state.state != STATE_OFF
+
+
+class WaterHeaterOperationModeCondition(EntityConditionBase):
+    """Condition for water heater operation mode."""
+
+    _domain_specs = {DOMAIN: DomainSpec()}
+    _schema = _OPERATION_MODE_CONDITION_SCHEMA
+
+    def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
+        """Initialize the operation mode condition."""
+        super().__init__(hass, config)
+        if TYPE_CHECKING:
+            assert config.options is not None
+        self._operation_modes: set[str] = set(config.options[ATTR_OPERATION_MODE])
+
+    def is_valid_state(self, entity_state: State) -> bool:
+        """Check if the state matches any of the expected operation modes."""
+        return entity_state.state in self._operation_modes
+
+
+class WaterHeaterTargetTemperatureCondition(EntityNumericalConditionWithUnitBase):
+    """Condition for water heater target temperature."""
+
+    _base_unit = UnitOfTemperature.CELSIUS
+    _domain_specs = {DOMAIN: NumericalDomainSpec(value_source=ATTR_TEMPERATURE)}
+    _unit_converter = TemperatureConverter
+
+    def _get_entity_unit(self, entity_state: State) -> str | None:
+        """Get the temperature unit of a water heater entity from its state."""
+        # Water heater entities convert temperatures to the system unit via show_temp
+        return self._hass.config.units.temperature_unit
+
+
+CONDITIONS: dict[str, type[Condition]] = {
+    "is_off": make_entity_state_condition(DOMAIN, STATE_OFF),
+    "is_on": WaterHeaterOnCondition,
+    "is_operation_mode": WaterHeaterOperationModeCondition,
+    "is_target_temperature": WaterHeaterTargetTemperatureCondition,
+}
+
+
+async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
+    """Return the water heater conditions."""
+    return CONDITIONS

homeassistant/components/water_heater/conditions.yaml

@@ -0,0 +1,72 @@
+.condition_common: &condition_common
+  target: &condition_water_heater_target
+    entity:
+      domain: water_heater
+  fields:
+    behavior: &condition_behavior
+      required: true
+      default: any
+      selector:
+        select:
+          translation_key: condition_behavior
+          options:
+            - all
+            - any
+
+.number_or_entity_temperature: &number_or_entity_temperature
+  required: false
+  selector:
+    choose:
+      choices:
+        number:
+          selector:
+            number:
+              mode: box
+        entity:
+          selector:
+            entity:
+              filter:
+                - domain: input_number
+                  unit_of_measurement:
+                    - "°C"
+                    - "°F"
+                - domain: sensor
+                  device_class: temperature
+                - domain: number
+                  device_class: temperature
+      translation_key: number_or_entity
+
+.condition_unit_temperature: &condition_unit_temperature
+  required: false
+  selector:
+    select:
+      options:
+        - "°C"
+        - "°F"
+
+is_off: *condition_common
+is_on: *condition_common
+
+is_operation_mode:
+  target: *condition_water_heater_target
+  fields:
+    behavior: *condition_behavior
+    operation_mode:
+      context:
+        filter_target: target
+      required: true
+      selector:
+        state:
+          attribute: operation_mode
+          hide_states:
+            - unavailable
+            - unknown
+          multiple: true
+
+is_target_temperature:
+  target: *condition_water_heater_target
+  fields:
+    behavior: *condition_behavior
+    above: *number_or_entity_temperature
+    below: *number_or_entity_temperature
+    unit: *condition_unit_temperature

homeassistant/components/water_heater/icons.json

@@ -1,4 +1,18 @@
 {
+  "conditions": {
+    "is_off": {
+      "condition": "mdi:water-boiler-off"
+    },
+    "is_on": {
+      "condition": "mdi:water-boiler"
+    },
+    "is_operation_mode": {
+      "condition": "mdi:water-boiler"
+    },
+    "is_target_temperature": {
+      "condition": "mdi:thermometer"
+    }
+  },
   "entity_component": {
     "_": {
       "default": "mdi:water-boiler",

homeassistant/components/water_heater/strings.json

@@ -1,8 +1,68 @@
 {
   "common": {
+    "condition_behavior_description": "How the state should match on the targeted water heaters.",
+    "condition_behavior_name": "Behavior",
     "trigger_behavior_description": "The behavior of the targeted water heaters to trigger on.",
     "trigger_behavior_name": "Behavior"
   },
+  "conditions": {
+    "is_off": {
+      "description": "Tests if one or more water heaters are off.",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::water_heater::common::condition_behavior_description%]",
+          "name": "[%key:component::water_heater::common::condition_behavior_name%]"
+        }
+      },
+      "name": "Water heater is off"
+    },
+    "is_on": {
+      "description": "Tests if one or more water heaters are on.",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::water_heater::common::condition_behavior_description%]",
+          "name": "[%key:component::water_heater::common::condition_behavior_name%]"
+        }
+      },
+      "name": "Water heater is on"
+    },
+    "is_operation_mode": {
+      "description": "Tests if one or more water heaters are set to a specific operation mode.",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::water_heater::common::condition_behavior_description%]",
+          "name": "[%key:component::water_heater::common::condition_behavior_name%]"
+        },
+        "operation_mode": {
+          "description": "The operation mode to check for.",
+          "name": "Operation mode"
+        }
+      },
+      "name": "Water heater operation mode"
+    },
+    "is_target_temperature": {
+      "description": "Tests the temperature setpoint of one or more water heaters.",
+      "fields": {
+        "above": {
+          "description": "Require the target temperature to be above this value.",
+          "name": "Above"
+        },
+        "behavior": {
+          "description": "[%key:component::water_heater::common::condition_behavior_description%]",
+          "name": "[%key:component::water_heater::common::condition_behavior_name%]"
+        },
+        "below": {
+          "description": "Require the target temperature to be below this value.",
+          "name": "Below"
+        },
+        "unit": {
+          "description": "All values will be converted to this unit when evaluating the condition.",
+          "name": "Unit of measurement"
+        }
+      },
+      "name": "Water heater target temperature"
+    }
+  },
   "device_automation": {
     "action_type": {
       "turn_off": "[%key:common::device_automation::action_type::turn_off%]",
@@ -59,6 +119,12 @@
     }
   },
   "selector": {
+    "condition_behavior": {
+      "options": {
+        "all": "All",
+        "any": "Any"
+      }
+    },
     "number_or_entity": {
       "choices": {
         "entity": "Entity",

tests/components/water_heater/test_condition.py

@@ -0,0 +1,323 @@
+"""Test water heater conditions."""
+
+from typing import Any
+
+import pytest
+
+from homeassistant.components.water_heater import (
+    STATE_ECO,
+    STATE_ELECTRIC,
+    STATE_GAS,
+    STATE_HEAT_PUMP,
+    STATE_HIGH_DEMAND,
+    STATE_PERFORMANCE,
+)
+from homeassistant.const import (
+    ATTR_TEMPERATURE,
+    ATTR_UNIT_OF_MEASUREMENT,
+    CONF_ABOVE,
+    CONF_BELOW,
+    STATE_OFF,
+    STATE_ON,
+    UnitOfTemperature,
+)
+from homeassistant.core import HomeAssistant
+
+from tests.components.common import (
+    ConditionStateDescription,
+    assert_condition_behavior_all,
+    assert_condition_behavior_any,
+    assert_condition_gated_by_labs_flag,
+    assert_numerical_condition_unit_conversion,
+    parametrize_condition_states_all,
+    parametrize_condition_states_any,
+    parametrize_numerical_attribute_condition_above_below_all,
+    parametrize_numerical_attribute_condition_above_below_any,
+    parametrize_target_entities,
+    target_entities,
+)
+
+_TEMPERATURE_CONDITION_OPTIONS = {"unit": UnitOfTemperature.CELSIUS}
+
+_ALL_STATES = [
+    STATE_ECO,
+    STATE_ELECTRIC,
+    STATE_GAS,
+    STATE_HEAT_PUMP,
+    STATE_HIGH_DEMAND,
+    STATE_OFF,
+    STATE_ON,
+    STATE_PERFORMANCE,
+]
+
+_ON_STATES = [s for s in _ALL_STATES if s != STATE_OFF]
+
+
+@pytest.fixture
+async def target_water_heaters(hass: HomeAssistant) -> dict[str, list[str]]:
+    """Create multiple water heater entities associated with different targets."""
+    return await target_entities(hass, "water_heater")
+
+
+@pytest.mark.parametrize(
+    "condition",
+    [
+        "water_heater.is_off",
+        "water_heater.is_on",
+        "water_heater.is_operation_mode",
+        "water_heater.is_target_temperature",
+    ],
+)
+async def test_water_heater_conditions_gated_by_labs_flag(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture, condition: str
+) -> None:
+    """Test the water heater conditions are gated by the labs flag."""
+    await assert_condition_gated_by_labs_flag(hass, caplog, condition)
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+@pytest.mark.parametrize(
+    ("condition_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("water_heater"),
+)
+@pytest.mark.parametrize(
+    ("condition", "condition_options", "states"),
+    [
+        *parametrize_condition_states_any(
+            condition="water_heater.is_off",
+            target_states=[STATE_OFF],
+            other_states=_ON_STATES,
+        ),
+        *parametrize_condition_states_any(
+            condition="water_heater.is_on",
+            target_states=_ON_STATES,
+            other_states=[STATE_OFF],
+        ),
+        *(
+            param
+            for mode in _ALL_STATES
+            for param in parametrize_condition_states_any(
+                condition="water_heater.is_operation_mode",
+                condition_options={"operation_mode": [mode]},
+                target_states=[mode],
+                other_states=[s for s in _ALL_STATES if s != mode],
+            )
+        ),
+        *parametrize_condition_states_any(
+            condition="water_heater.is_operation_mode",
+            condition_options={"operation_mode": ["eco", "electric"]},
+            target_states=["eco", "electric"],
+            other_states=[s for s in _ALL_STATES if s not in ("eco", "electric")],
+        ),
+    ],
+)
+async def test_water_heater_state_condition_behavior_any(
+    hass: HomeAssistant,
+    target_water_heaters: dict[str, list[str]],
+    condition_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    condition: str,
+    condition_options: dict[str, Any],
+    states: list[ConditionStateDescription],
+) -> None:
+    """Test the water heater state condition with the 'any' behavior."""
+    await assert_condition_behavior_any(
+        hass,
+        target_entities=target_water_heaters,
+        condition_target_config=condition_target_config,
+        entity_id=entity_id,
+        entities_in_target=entities_in_target,
+        condition=condition,
+        condition_options=condition_options,
+        states=states,
+    )
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+@pytest.mark.parametrize(
+    ("condition_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("water_heater"),
+)
+@pytest.mark.parametrize(
+    ("condition", "condition_options", "states"),
+    [
+        *parametrize_condition_states_all(
+            condition="water_heater.is_off",
+            target_states=[STATE_OFF],
+            other_states=_ON_STATES,
+        ),
+        *parametrize_condition_states_all(
+            condition="water_heater.is_on",
+            target_states=_ON_STATES,
+            other_states=[STATE_OFF],
+        ),
+        *(
+            param
+            for mode in _ALL_STATES
+            for param in parametrize_condition_states_all(
+                condition="water_heater.is_operation_mode",
+                condition_options={"operation_mode": [mode]},
+                target_states=[mode],
+                other_states=[s for s in _ALL_STATES if s != mode],
+            )
+        ),
+        *parametrize_condition_states_all(
+            condition="water_heater.is_operation_mode",
+            condition_options={"operation_mode": ["eco", "electric"]},
+            target_states=["eco", "electric"],
+            other_states=[s for s in _ALL_STATES if s not in ("eco", "electric")],
+        ),
+    ],
+)
+async def test_water_heater_state_condition_behavior_all(
+    hass: HomeAssistant,
+    target_water_heaters: dict[str, list[str]],
+    condition_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    condition: str,
+    condition_options: dict[str, Any],
+    states: list[ConditionStateDescription],
+) -> None:
+    """Test the water heater state condition with the 'all' behavior."""
+    await assert_condition_behavior_all(
+        hass,
+        target_entities=target_water_heaters,
+        condition_target_config=condition_target_config,
+        entity_id=entity_id,
+        entities_in_target=entities_in_target,
+        condition=condition,
+        condition_options=condition_options,
+        states=states,
+    )
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+@pytest.mark.parametrize(
+    ("condition_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("water_heater"),
+)
+@pytest.mark.parametrize(
+    ("condition", "condition_options", "states"),
+    [
+        *parametrize_numerical_attribute_condition_above_below_any(
+            "water_heater.is_target_temperature",
+            "eco",
+            ATTR_TEMPERATURE,
+            condition_options=_TEMPERATURE_CONDITION_OPTIONS,
+        ),
+    ],
+)
+async def test_water_heater_numerical_condition_behavior_any(
+    hass: HomeAssistant,
+    target_water_heaters: dict[str, list[str]],
+    condition_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    condition: str,
+    condition_options: dict[str, Any],
+    states: list[ConditionStateDescription],
+) -> None:
+    """Test the water heater numerical condition with the 'any' behavior."""
+    await assert_condition_behavior_any(
+        hass,
+        target_entities=target_water_heaters,
+        condition_target_config=condition_target_config,
+        entity_id=entity_id,
+        entities_in_target=entities_in_target,
+        condition=condition,
+        condition_options=condition_options,
+        states=states,
+    )
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+@pytest.mark.parametrize(
+    ("condition_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("water_heater"),
+)
+@pytest.mark.parametrize(
+    ("condition", "condition_options", "states"),
+    [
+        *parametrize_numerical_attribute_condition_above_below_all(
+            "water_heater.is_target_temperature",
+            "eco",
+            ATTR_TEMPERATURE,
+            condition_options=_TEMPERATURE_CONDITION_OPTIONS,
+        ),
+    ],
+)
+async def test_water_heater_numerical_condition_behavior_all(
+    hass: HomeAssistant,
+    target_water_heaters: dict[str, list[str]],
+    condition_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    condition: str,
+    condition_options: dict[str, Any],
+    states: list[ConditionStateDescription],
+) -> None:
+    """Test the water heater numerical condition with the 'all' behavior."""
+    await assert_condition_behavior_all(
+        hass,
+        target_entities=target_water_heaters,
+        condition_target_config=condition_target_config,
+        entity_id=entity_id,
+        entities_in_target=entities_in_target,
+        condition=condition,
+        condition_options=condition_options,
+        states=states,
+    )
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_water_heater_numerical_condition_unit_conversion(
+    hass: HomeAssistant,
+) -> None:
+    """Test that the water heater numerical condition converts units correctly."""
+    _unit_celsius = {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.CELSIUS}
+    _unit_fahrenheit = {ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT}
+    _unit_invalid = {ATTR_UNIT_OF_MEASUREMENT: "not_a_valid_unit"}
+
+    await assert_numerical_condition_unit_conversion(
+        hass,
+        condition="water_heater.is_target_temperature",
+        entity_id="water_heater.test",
+        pass_states=[{"state": "eco", "attributes": {ATTR_TEMPERATURE: 55}}],
+        fail_states=[
+            {
+                "state": "eco",
+                "attributes": {ATTR_TEMPERATURE: 40},
+            }
+        ],
+        numerical_condition_options=[
+            {CONF_ABOVE: 120, CONF_BELOW: 140, "unit": UnitOfTemperature.FAHRENHEIT},
+            {CONF_ABOVE: 49, CONF_BELOW: 60, "unit": UnitOfTemperature.CELSIUS},
+        ],
+        limit_entity_condition_options={
+            CONF_ABOVE: "sensor.above",
+            CONF_BELOW: "sensor.below",
+        },
+        limit_entities=("sensor.above", "sensor.below"),
+        limit_entity_states=[
+            (
+                {"state": "120", "attributes": _unit_fahrenheit},  # ≈48.9°C
+                {"state": "140", "attributes": _unit_fahrenheit},  # ≈60.0°C
+            ),
+            (
+                {"state": "49", "attributes": _unit_celsius},
+                {"state": "60", "attributes": _unit_celsius},
+            ),
+        ],
+        invalid_limit_entity_states=[
+            (
+                {"state": "120", "attributes": _unit_invalid},
+                {"state": "140", "attributes": _unit_invalid},
+            ),
+            (
+                {"state": "49", "attributes": _unit_invalid},
+                {"state": "60", "attributes": _unit_invalid},
+            ),
+        ],
+    )