Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Water Heater support for Airzone Cloud (#115097) 

* airzone_cloud: add Water Heater support

Signed-off-by: Álvaro Fernández Rojas <noltari@gmail.com>

* run CI

---------

Signed-off-by: Álvaro Fernández Rojas <noltari@gmail.com>
pull/114546/head
Álvaro Fernández Rojas 2024-04-07 15:07:40 +02:00 committed by GitHub
parent e4fbe539aa
commit d4d06c0bc3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 397 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/airzone_cloud/__init__.py
View File

@ -17,6 +17,7 @@ PLATFORMS: list[Platform] = [
Platform.BINARY_SENSOR,
Platform.CLIMATE,
Platform.SENSOR,
Platform.WATER_HEATER,
]

42
homeassistant/components/airzone_cloud/entity.py
View File

@ -11,6 +11,7 @@ from aioairzone_cloud.const import (
AZD_AVAILABLE,
AZD_FIRMWARE,
AZD_GROUPS,
AZD_HOT_WATERS,
AZD_INSTALLATIONS,
AZD_NAME,
AZD_SYSTEM_ID,
@ -136,6 +137,47 @@ class AirzoneGroupEntity(AirzoneEntity):
self.coordinator.async_set_updated_data(self.coordinator.airzone.data())
class AirzoneHotWaterEntity(AirzoneEntity):
"""Define an Airzone Cloud Hot Water entity."""
def __init__(
self,
coordinator: AirzoneUpdateCoordinator,
dhw_id: str,
dhw_data: dict[str, Any],
) -> None:
"""Initialize."""
super().__init__(coordinator)
self.dhw_id = dhw_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, dhw_id)},
manufacturer=MANUFACTURER,
name=dhw_data[AZD_NAME],
via_device=(DOMAIN, dhw_data[AZD_WEBSERVER]),
)
def get_airzone_value(self, key: str) -> Any:
"""Return DHW value by key."""
value = None
if dhw := self.coordinator.data[AZD_HOT_WATERS].get(self.dhw_id):
value = dhw.get(key)
return value
async def _async_update_params(self, params: dict[str, Any]) -> None:
"""Send DHW parameters to Cloud API."""
_LOGGER.debug("dhw=%s: update_params=%s", self.entity_id, params)
try:
await self.coordinator.airzone.api_set_dhw_id_params(self.dhw_id, params)
except AirzoneCloudError as error:
raise HomeAssistantError(
f"Failed to set {self.entity_id} params: {error}"
) from error
self.coordinator.async_set_updated_data(self.coordinator.airzone.data())
class AirzoneInstallationEntity(AirzoneEntity):
"""Define an Airzone Cloud Installation entity."""

164
homeassistant/components/airzone_cloud/water_heater.py Normal file
View File

@ -0,0 +1,164 @@
"""Support for the Airzone Cloud water heater."""
from __future__ import annotations
from typing import Any, Final
from aioairzone_cloud.common import HotWaterOperation, TemperatureUnit
from aioairzone_cloud.const import (
API_OPTS,
API_POWER,
API_POWERFUL_MODE,
API_SETPOINT,
API_UNITS,
API_VALUE,
AZD_HOT_WATERS,
AZD_OPERATION,
AZD_OPERATIONS,
AZD_TEMP,
AZD_TEMP_SET,
AZD_TEMP_SET_MAX,
AZD_TEMP_SET_MIN,
)
from homeassistant.components.water_heater import (
STATE_ECO,
STATE_PERFORMANCE,
WaterHeaterEntity,
WaterHeaterEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, STATE_OFF, UnitOfTemperature
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .coordinator import AirzoneUpdateCoordinator
from .entity import AirzoneHotWaterEntity
OPERATION_LIB_TO_HASS: Final[dict[HotWaterOperation, str]] = {
HotWaterOperation.Off: STATE_OFF,
HotWaterOperation.On: STATE_ECO,
HotWaterOperation.Powerful: STATE_PERFORMANCE,
}
OPERATION_MODE_TO_DHW_PARAMS: Final[dict[str, dict[str, Any]]] = {
STATE_OFF: {
API_POWER: {
API_VALUE: False,
},
},
STATE_ECO: {
API_POWER: {
API_VALUE: True,
},
API_POWERFUL_MODE: {
API_VALUE: False,
},
},
STATE_PERFORMANCE: {
API_POWER: {
API_VALUE: True,
},
API_POWERFUL_MODE: {
API_VALUE: True,
},
},
}
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Add Airzone Cloud Water Heater from a config_entry."""
coordinator: AirzoneUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
AirzoneWaterHeater(
coordinator,
dhw_id,
dhw_data,
)
for dhw_id, dhw_data in coordinator.data.get(AZD_HOT_WATERS, {}).items()
)
class AirzoneWaterHeater(AirzoneHotWaterEntity, WaterHeaterEntity):
"""Define an Airzone Cloud Water Heater."""
_attr_name = None
_attr_supported_features = (
WaterHeaterEntityFeature.TARGET_TEMPERATURE
| WaterHeaterEntityFeature.ON_OFF
| WaterHeaterEntityFeature.OPERATION_MODE
)
_attr_temperature_unit = UnitOfTemperature.CELSIUS
def __init__(
self,
coordinator: AirzoneUpdateCoordinator,
dhw_id: str,
dhw_data: dict,
) -> None:
"""Initialize Airzone Cloud Water Heater."""
super().__init__(coordinator, dhw_id, dhw_data)
self._attr_unique_id = dhw_id
self._attr_operation_list = [
OPERATION_LIB_TO_HASS[operation]
for operation in self.get_airzone_value(AZD_OPERATIONS)
]
self._async_update_attrs()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the water heater off."""
params = {
API_POWER: {
API_VALUE: False,
},
}
await self._async_update_params(params)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the water heater off."""
params = {
API_POWER: {
API_VALUE: True,
},
}
await self._async_update_params(params)
async def async_set_operation_mode(self, operation_mode: str) -> None:
"""Set new target operation mode."""
params = OPERATION_MODE_TO_DHW_PARAMS.get(operation_mode, {})
await self._async_update_params(params)
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
params: dict[str, Any] = {}
if ATTR_TEMPERATURE in kwargs:
params[API_SETPOINT] = {
API_VALUE: kwargs[ATTR_TEMPERATURE],
API_OPTS: {
API_UNITS: TemperatureUnit.CELSIUS.value,
},
}
await self._async_update_params(params)
@callback
def _handle_coordinator_update(self) -> None:
"""Update attributes when the coordinator updates."""
self._async_update_attrs()
super()._handle_coordinator_update()
@callback
def _async_update_attrs(self) -> None:
"""Update water heater attributes."""
self._attr_current_temperature = self.get_airzone_value(AZD_TEMP)
self._attr_current_operation = OPERATION_LIB_TO_HASS[
self.get_airzone_value(AZD_OPERATION)
]
self._attr_max_temp = self.get_airzone_value(AZD_TEMP_SET_MAX)
self._attr_min_temp = self.get_airzone_value(AZD_TEMP_SET_MIN)
self._attr_target_temperature = self.get_airzone_value(AZD_TEMP_SET)

2
tests/components/airzone_cloud/snapshots/test_diagnostics.ambr
View File

@ -292,8 +292,10 @@
'operations': list([
0,
1,
2,
]),
'power': False,
'power-mode': False,
'problems': False,
'temperature': 45.5,
'temperature-setpoint': 48,

186
tests/components/airzone_cloud/test_water_heater.py Normal file
View File

@ -0,0 +1,186 @@
"""The water heater tests for the Airzone Cloud platform."""
from unittest.mock import patch
from aioairzone_cloud.exceptions import AirzoneCloudError
import pytest
from homeassistant.components.water_heater import (
ATTR_CURRENT_TEMPERATURE,
ATTR_MAX_TEMP,
ATTR_MIN_TEMP,
ATTR_OPERATION_MODE,
DOMAIN as WATER_HEATER_DOMAIN,
SERVICE_SET_OPERATION_MODE,
SERVICE_SET_TEMPERATURE,
STATE_ECO,
STATE_PERFORMANCE,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_TEMPERATURE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .util import async_init_integration
async def test_airzone_create_water_heater(hass: HomeAssistant) -> None:
"""Test creation of water heater."""
await async_init_integration(hass)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.state == STATE_OFF
assert state.attributes[ATTR_CURRENT_TEMPERATURE] == 45.5
assert state.attributes[ATTR_MAX_TEMP] == 60
assert state.attributes[ATTR_MIN_TEMP] == 40
assert state.attributes[ATTR_TEMPERATURE] == 48
async def test_airzone_water_heater_turn_on_off(hass: HomeAssistant) -> None:
"""Test turning on/off."""
await async_init_integration(hass)
with patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
return_value=None,
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.state == STATE_ECO
with patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
return_value=None,
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.state == STATE_OFF
async def test_airzone_water_heater_set_operation(hass: HomeAssistant) -> None:
"""Test setting the Operation mode."""
await async_init_integration(hass)
with patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
return_value=None,
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_OPERATION_MODE,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
ATTR_OPERATION_MODE: STATE_ECO,
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.state == STATE_ECO
with patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
return_value=None,
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_OPERATION_MODE,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
ATTR_OPERATION_MODE: STATE_PERFORMANCE,
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.state == STATE_PERFORMANCE
with patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
return_value=None,
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_OPERATION_MODE,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
ATTR_OPERATION_MODE: STATE_OFF,
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.state == STATE_OFF
async def test_airzone_water_heater_set_temp(hass: HomeAssistant) -> None:
"""Test setting the target temperature."""
await async_init_integration(hass)
with patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
return_value=None,
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
ATTR_TEMPERATURE: 50,
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.attributes[ATTR_TEMPERATURE] == 50
async def test_airzone_water_heater_set_temp_error(hass: HomeAssistant) -> None:
"""Test error when setting the target temperature."""
await async_init_integration(hass)
with (
patch(
"homeassistant.components.airzone_cloud.AirzoneCloudApi.api_patch_device",
side_effect=AirzoneCloudError,
),
pytest.raises(HomeAssistantError),
):
await hass.services.async_call(
WATER_HEATER_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: "water_heater.airzone_cloud_dhw",
ATTR_TEMPERATURE: 80,
},
blocking=True,
)
state = hass.states.get("water_heater.airzone_cloud_dhw")
assert state.attributes[ATTR_TEMPERATURE] == 48

2
tests/components/airzone_cloud/util.py
View File

@ -41,6 +41,7 @@ from aioairzone_cloud.const import (
API_NAME,
API_OLD_ID,
API_POWER,
API_POWERFUL_MODE,
API_RANGE_MAX_AIR,
API_RANGE_MIN_AIR,
API_RANGE_SP_MAX_ACS,
@ -315,6 +316,7 @@ def mock_get_device_status(device: Device) -> dict[str, Any]:
API_ACTIVE: False,
API_ERRORS: [],
API_POWER: False,
API_POWERFUL_MODE: False,
API_SETPOINT: {API_CELSIUS: 48, API_FAH: 118},
API_RANGE_SP_MAX_ACS: {API_CELSIUS: 60, API_FAH: 140},
API_RANGE_SP_MIN_ACS: {API_CELSIUS: 40, API_FAH: 104},