Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 17:49:37 +01:00
Code Activity

Add water heater support to ESPHome (#159201)

Browse Source 
Co-authored-by: Ludovic BOUÉ <lboue@users.noreply.github.com>
Co-authored-by: Ludovic BOUÉ <ludovic.boue@gmail.com>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Douwe
2026-01-27 19:30:58 +01:00
committed by GitHub
parent 11f713209d
commit eaa1798443
4 changed files with 301 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/esphome/entry_data.py
+2
View File
@@ -44,6 +44,7 @@ from aioesphomeapi import (
UpdateInfo,
UserService,
ValveInfo,
WaterHeaterInfo,
build_unique_id,
)
from aioesphomeapi.model import ButtonInfo
@@ -96,6 +97,7 @@ INFO_TYPE_TO_PLATFORM: dict[type[EntityInfo], Platform] = {
TimeInfo: Platform.TIME,
UpdateInfo: Platform.UPDATE,
ValveInfo: Platform.VALVE,
WaterHeaterInfo: Platform.WATER_HEATER,
}
homeassistant/components/esphome/water_heater.py
+110
View File
@@ -0,0 +1,110 @@
"""Support for ESPHome water heaters."""
from __future__ import annotations
from functools import partial
from typing import Any
from aioesphomeapi import EntityInfo, WaterHeaterInfo, WaterHeaterMode, WaterHeaterState
from homeassistant.components.water_heater import (
WaterHeaterEntity,
WaterHeaterEntityFeature,
)
from homeassistant.const import ATTR_TEMPERATURE, PRECISION_TENTHS, UnitOfTemperature
from homeassistant.core import callback
from .entity import (
EsphomeEntity,
convert_api_error_ha_error,
esphome_float_state_property,
esphome_state_property,
platform_async_setup_entry,
)
from .enum_mapper import EsphomeEnumMapper
PARALLEL_UPDATES = 0
_WATER_HEATER_MODES: EsphomeEnumMapper[WaterHeaterMode, str] = EsphomeEnumMapper(
{
WaterHeaterMode.OFF: "off",
WaterHeaterMode.ECO: "eco",
WaterHeaterMode.ELECTRIC: "electric",
WaterHeaterMode.PERFORMANCE: "performance",
WaterHeaterMode.HIGH_DEMAND: "high_demand",
WaterHeaterMode.HEAT_PUMP: "heat_pump",
WaterHeaterMode.GAS: "gas",
}
)
class EsphomeWaterHeater(
EsphomeEntity[WaterHeaterInfo, WaterHeaterState], WaterHeaterEntity
):
"""A water heater implementation for ESPHome."""
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_precision = PRECISION_TENTHS
@callback
def _on_static_info_update(self, static_info: EntityInfo) -> None:
"""Set attrs from static info."""
super()._on_static_info_update(static_info)
static_info = self._static_info
self._attr_min_temp = static_info.min_temperature
self._attr_max_temp = static_info.max_temperature
features = WaterHeaterEntityFeature.TARGET_TEMPERATURE
if static_info.supported_modes:
features |= WaterHeaterEntityFeature.OPERATION_MODE
self._attr_operation_list = [
_WATER_HEATER_MODES.from_esphome(mode)
for mode in static_info.supported_modes
]
else:
self._attr_operation_list = None
self._attr_supported_features = features
@property
@esphome_float_state_property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self._state.current_temperature
@property
@esphome_float_state_property
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self._state.target_temperature
@property
@esphome_state_property
def current_operation(self) -> str | None:
"""Return current operation mode."""
return _WATER_HEATER_MODES.from_esphome(self._state.mode)
@convert_api_error_ha_error
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
self._client.water_heater_command(
key=self._key,
target_temperature=kwargs[ATTR_TEMPERATURE],
device_id=self._static_info.device_id,
)
@convert_api_error_ha_error
async def async_set_operation_mode(self, operation_mode: str) -> None:
"""Set new operation mode."""
self._client.water_heater_command(
key=self._key,
mode=_WATER_HEATER_MODES.from_hass(operation_mode),
device_id=self._static_info.device_id,
)
async_setup_entry = partial(
platform_async_setup_entry,
info_type=WaterHeaterInfo,
entity_type=EsphomeWaterHeater,
state_type=WaterHeaterState,
)
tests/components/esphome/conftest.py
+4 -3
View File
@@ -560,9 +560,10 @@ async def _mock_generic_device_entry(
async def mock_try_connect(self):
"""Set an event when ReconnectLogic._try_connect has been awaited."""
result = await super()._try_connect()
try_connect_done.set()
return result
try:
return await super()._try_connect()
finally:
try_connect_done.set()
def stop_callback(self) -> None:
"""Stop the reconnect logic."""
tests/components/esphome/test_water_heater.py
+185
View File
@@ -0,0 +1,185 @@
"""Test ESPHome water heaters."""
from unittest.mock import call
from aioesphomeapi import APIClient, WaterHeaterInfo, WaterHeaterMode, WaterHeaterState
from homeassistant.components.water_heater import (
ATTR_OPERATION_LIST,
DOMAIN as WATER_HEATER_DOMAIN,
SERVICE_SET_OPERATION_MODE,
SERVICE_SET_TEMPERATURE,
)
from homeassistant.const import ATTR_ENTITY_ID, ATTR_TEMPERATURE
from homeassistant.core import HomeAssistant
from .conftest import MockGenericDeviceEntryType
async def test_water_heater_entity(
hass: HomeAssistant,
mock_client: APIClient,
mock_generic_device_entry: MockGenericDeviceEntryType,
) -> None:
"""Test a generic water heater entity."""
entity_info = [
WaterHeaterInfo(
object_id="my_boiler",
key=1,
name="My Boiler",
min_temperature=10.0,
max_temperature=85.0,
supported_modes=[
WaterHeaterMode.ECO,
WaterHeaterMode.GAS,
],
)
]
states = [
WaterHeaterState(
key=1,
mode=WaterHeaterMode.ECO,
current_temperature=45.0,
target_temperature=50.0,
)
]
await mock_generic_device_entry(
mock_client=mock_client,
entity_info=entity_info,
states=states,
)
state = hass.states.get("water_heater.test_my_boiler")
assert state is not None
assert state.state == "eco"
assert state.attributes["current_temperature"] == 45.0
assert state.attributes["temperature"] == 50.0
assert state.attributes["min_temp"] == 10.0
assert state.attributes["max_temp"] == 85.0
assert state.attributes["operation_list"] == ["eco", "gas"]
async def test_water_heater_entity_no_modes(
hass: HomeAssistant,
mock_client: APIClient,
mock_generic_device_entry: MockGenericDeviceEntryType,
) -> None:
"""Test a water heater entity without operation modes."""
entity_info = [
WaterHeaterInfo(
object_id="my_boiler",
key=1,
name="My Boiler",
min_temperature=10.0,
max_temperature=85.0,
)
]
states = [
WaterHeaterState(
key=1,
current_temperature=45.0,
target_temperature=50.0,
)
]
await mock_generic_device_entry(
mock_client=mock_client,
entity_info=entity_info,
states=states,
)
state = hass.states.get("water_heater.test_my_boiler")
assert state is not None
assert state.attributes["min_temp"] == 10.0
assert state.attributes["max_temp"] == 85.0
assert state.attributes.get(ATTR_OPERATION_LIST) is None
async def test_water_heater_set_temperature(
hass: HomeAssistant,
mock_client: APIClient,
mock_generic_device_entry: MockGenericDeviceEntryType,
) -> None:
"""Test setting the target temperature."""
entity_info = [
WaterHeaterInfo(
object_id="my_boiler",
key=1,
name="My Boiler",
min_temperature=10.0,
max_temperature=85.0,
)
]
states = [
WaterHeaterState(
key=1,
mode=WaterHeaterMode.ECO,
target_temperature=45.0,
)
]
await mock_generic_device_entry(
mock_client=mock_client,
entity_info=entity_info,
states=states,
)
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: "water_heater.test_my_boiler",
ATTR_TEMPERATURE: 55,
},
blocking=True,
)
mock_client.water_heater_command.assert_has_calls(
[call(key=1, target_temperature=55.0, device_id=0)]
)
async def test_water_heater_set_operation_mode(
hass: HomeAssistant,
mock_client: APIClient,
mock_generic_device_entry: MockGenericDeviceEntryType,
) -> None:
"""Test setting the operation mode."""
entity_info = [
WaterHeaterInfo(
object_id="my_boiler",
key=1,
name="My Boiler",
supported_modes=[
WaterHeaterMode.ECO,
WaterHeaterMode.GAS,
],
)
]
states = [
WaterHeaterState(
key=1,
mode=WaterHeaterMode.ECO,
)
]
await mock_generic_device_entry(
mock_client=mock_client,
entity_info=entity_info,
states=states,
)
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_OPERATION_MODE,
{
ATTR_ENTITY_ID: "water_heater.test_my_boiler",
"operation_mode": "gas",
},
blocking=True,
)
mock_client.water_heater_command.assert_has_calls(
[call(key=1, mode=WaterHeaterMode.GAS, device_id=0)]
)