Add Binary Sensors for Airzone Systems (#69736)

pull/71601/head
Álvaro Fernández Rojas 2022-05-09 16:07:11 +02:00 committed by GitHub
parent 539ce7ff0e
commit 88c2c5c36c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 176 additions and 65 deletions

View File

@ -4,17 +4,7 @@ from __future__ import annotations
import logging
from typing import Any
from aioairzone.const import (
AZD_ID,
AZD_MAC,
AZD_NAME,
AZD_SYSTEM,
AZD_THERMOSTAT_FW,
AZD_THERMOSTAT_MODEL,
AZD_WEBSERVER,
AZD_ZONES,
DEFAULT_SYSTEM_ID,
)
from aioairzone.const import AZD_MAC, AZD_WEBSERVER, DEFAULT_SYSTEM_ID
from aioairzone.localapi import AirzoneLocalApi, ConnectionOptions
from homeassistant.config_entries import ConfigEntry
@ -25,10 +15,8 @@ from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER
from .const import DOMAIN
from .coordinator import AirzoneUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.CLIMATE, Platform.SENSOR]
@ -36,52 +24,6 @@ PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.CLIMATE, Platform.
_LOGGER = logging.getLogger(__name__)
class AirzoneEntity(CoordinatorEntity[AirzoneUpdateCoordinator]):
"""Define an Airzone entity."""
def get_airzone_value(self, key) -> Any:
"""Return Airzone entity value by key."""
raise NotImplementedError()
class AirzoneZoneEntity(AirzoneEntity):
"""Define an Airzone Zone entity."""
def __init__(
self,
coordinator: AirzoneUpdateCoordinator,
entry: ConfigEntry,
system_zone_id: str,
zone_data: dict[str, Any],
) -> None:
"""Initialize."""
super().__init__(coordinator)
self.system_id = zone_data[AZD_SYSTEM]
self.system_zone_id = system_zone_id
self.zone_id = zone_data[AZD_ID]
self._attr_device_info: DeviceInfo = {
"identifiers": {(DOMAIN, f"{entry.entry_id}_{system_zone_id}")},
"manufacturer": MANUFACTURER,
"model": self.get_airzone_value(AZD_THERMOSTAT_MODEL),
"name": f"Airzone [{system_zone_id}] {zone_data[AZD_NAME]}",
"sw_version": self.get_airzone_value(AZD_THERMOSTAT_FW),
}
self._attr_unique_id = (
entry.entry_id if entry.unique_id is None else entry.unique_id
)
def get_airzone_value(self, key) -> Any:
"""Return zone value by key."""
value = None
if self.system_zone_id in self.coordinator.data[AZD_ZONES]:
zone = self.coordinator.data[AZD_ZONES][self.system_zone_id]
if key in zone:
value = zone[key]
return value
async def _async_migrate_unique_ids(
hass: HomeAssistant,
entry: ConfigEntry,

View File

@ -12,6 +12,7 @@ from aioairzone.const import (
AZD_FLOOR_DEMAND,
AZD_NAME,
AZD_PROBLEMS,
AZD_SYSTEMS,
AZD_ZONES,
)
@ -25,9 +26,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import AirzoneEntity, AirzoneZoneEntity
from .const import DOMAIN
from .coordinator import AirzoneUpdateCoordinator
from .entity import AirzoneEntity, AirzoneSystemEntity, AirzoneZoneEntity
@dataclass
@ -37,6 +38,18 @@ class AirzoneBinarySensorEntityDescription(BinarySensorEntityDescription):
attributes: dict[str, str] | None = None
SYSTEM_BINARY_SENSOR_TYPES: Final[tuple[AirzoneBinarySensorEntityDescription, ...]] = (
AirzoneBinarySensorEntityDescription(
attributes={
"errors": AZD_ERRORS,
},
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
key=AZD_PROBLEMS,
name="Problem",
),
)
ZONE_BINARY_SENSOR_TYPES: Final[tuple[AirzoneBinarySensorEntityDescription, ...]] = (
AirzoneBinarySensorEntityDescription(
device_class=BinarySensorDeviceClass.RUNNING,
@ -72,6 +85,20 @@ async def async_setup_entry(
coordinator = hass.data[DOMAIN][entry.entry_id]
binary_sensors: list[AirzoneBinarySensor] = []
for system_id, system_data in coordinator.data[AZD_SYSTEMS].items():
for description in SYSTEM_BINARY_SENSOR_TYPES:
if description.key in system_data:
binary_sensors.append(
AirzoneSystemBinarySensor(
coordinator,
description,
entry,
system_id,
system_data,
)
)
for system_zone_id, zone_data in coordinator.data[AZD_ZONES].items():
for description in ZONE_BINARY_SENSOR_TYPES:
if description.key in zone_data:
@ -109,6 +136,24 @@ class AirzoneBinarySensor(AirzoneEntity, BinarySensorEntity):
return self.get_airzone_value(self.entity_description.key)
class AirzoneSystemBinarySensor(AirzoneSystemEntity, AirzoneBinarySensor):
"""Define an Airzone System binary sensor."""
def __init__(
self,
coordinator: AirzoneUpdateCoordinator,
description: AirzoneBinarySensorEntityDescription,
entry: ConfigEntry,
system_id: str,
system_data: dict[str, Any],
) -> None:
"""Initialize."""
super().__init__(coordinator, entry, system_data)
self._attr_name = f"System {system_id} {description.name}"
self._attr_unique_id = f"{self._attr_unique_id}_{system_id}_{description.key}"
self.entity_description = description
class AirzoneZoneBinarySensor(AirzoneZoneEntity, AirzoneBinarySensor):
"""Define an Airzone Zone binary sensor."""

View File

@ -39,9 +39,9 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import AirzoneZoneEntity
from .const import API_TEMPERATURE_STEP, DOMAIN, TEMP_UNIT_LIB_TO_HASS
from .coordinator import AirzoneUpdateCoordinator
from .entity import AirzoneZoneEntity
_LOGGER = logging.getLogger(__name__)

View File

@ -0,0 +1,105 @@
"""Entity classes for the Airzone integration."""
from __future__ import annotations
from typing import Any
from aioairzone.const import (
AZD_FIRMWARE,
AZD_FULL_NAME,
AZD_ID,
AZD_MODEL,
AZD_NAME,
AZD_SYSTEM,
AZD_SYSTEMS,
AZD_THERMOSTAT_FW,
AZD_THERMOSTAT_MODEL,
AZD_ZONES,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER
from .coordinator import AirzoneUpdateCoordinator
class AirzoneEntity(CoordinatorEntity[AirzoneUpdateCoordinator]):
"""Define an Airzone entity."""
def get_airzone_value(self, key) -> Any:
"""Return Airzone entity value by key."""
raise NotImplementedError()
class AirzoneSystemEntity(AirzoneEntity):
"""Define an Airzone System entity."""
def __init__(
self,
coordinator: AirzoneUpdateCoordinator,
entry: ConfigEntry,
system_data: dict[str, Any],
) -> None:
"""Initialize."""
super().__init__(coordinator)
self.system_id = system_data[AZD_ID]
self._attr_device_info: DeviceInfo = {
"identifiers": {(DOMAIN, f"{entry.entry_id}_{self.system_id}")},
"manufacturer": MANUFACTURER,
"model": self.get_airzone_value(AZD_MODEL),
"name": self.get_airzone_value(AZD_FULL_NAME),
"sw_version": self.get_airzone_value(AZD_FIRMWARE),
"via_device": (DOMAIN, f"{entry.entry_id}_ws"),
}
self._attr_unique_id = (
entry.entry_id if entry.unique_id is None else entry.unique_id
)
def get_airzone_value(self, key) -> Any:
"""Return system value by key."""
value = None
if system := self.coordinator.data[AZD_SYSTEMS].get(self.system_id):
if key in system:
value = system[key]
return value
class AirzoneZoneEntity(AirzoneEntity):
"""Define an Airzone Zone entity."""
def __init__(
self,
coordinator: AirzoneUpdateCoordinator,
entry: ConfigEntry,
system_zone_id: str,
zone_data: dict[str, Any],
) -> None:
"""Initialize."""
super().__init__(coordinator)
self.system_id = zone_data[AZD_SYSTEM]
self.system_zone_id = system_zone_id
self.zone_id = zone_data[AZD_ID]
self._attr_device_info: DeviceInfo = {
"identifiers": {(DOMAIN, f"{entry.entry_id}_{system_zone_id}")},
"manufacturer": MANUFACTURER,
"model": self.get_airzone_value(AZD_THERMOSTAT_MODEL),
"name": f"Airzone [{system_zone_id}] {zone_data[AZD_NAME]}",
"sw_version": self.get_airzone_value(AZD_THERMOSTAT_FW),
"via_device": (DOMAIN, f"{entry.entry_id}_{self.system_id}"),
}
self._attr_unique_id = (
entry.entry_id if entry.unique_id is None else entry.unique_id
)
def get_airzone_value(self, key) -> Any:
"""Return zone value by key."""
value = None
if zone := self.coordinator.data[AZD_ZONES].get(self.system_zone_id):
if key in zone:
value = zone[key]
return value

View File

@ -16,9 +16,9 @@ from homeassistant.const import PERCENTAGE, TEMP_CELSIUS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import AirzoneEntity, AirzoneZoneEntity
from .const import DOMAIN, TEMP_UNIT_LIB_TO_HASS
from .coordinator import AirzoneUpdateCoordinator
from .entity import AirzoneEntity, AirzoneZoneEntity
ZONE_SENSOR_TYPES: Final[tuple[SensorEntityDescription, ...]] = (
SensorEntityDescription(

View File

@ -13,6 +13,11 @@ async def test_airzone_create_binary_sensors(hass: HomeAssistant) -> None:
await async_init_integration(hass)
# Systems
state = hass.states.get("binary_sensor.system_1_problem")
assert state.state == STATE_OFF
# Zones
state = hass.states.get("binary_sensor.despacho_air_demand")
assert state.state == STATE_OFF

View File

@ -19,9 +19,12 @@ from aioairzone.const import (
API_MODES,
API_NAME,
API_ON,
API_POWER,
API_ROOM_TEMP,
API_SET_POINT,
API_SYSTEM_FIRMWARE,
API_SYSTEM_ID,
API_SYSTEM_TYPE,
API_SYSTEMS,
API_THERMOS_FIRMWARE,
API_THERMOS_RADIO,
@ -31,7 +34,7 @@ from aioairzone.const import (
API_WIFI_RSSI,
API_ZONE_ID,
)
from aioairzone.exceptions import InvalidMethod, SystemOutOfRange
from aioairzone.exceptions import InvalidMethod
from homeassistant.components.airzone import DOMAIN
from homeassistant.const import CONF_HOST, CONF_ID, CONF_PORT
@ -178,6 +181,17 @@ HVAC_MOCK = {
]
}
HVAC_SYSTEMS_MOCK = {
API_SYSTEMS: [
{
API_SYSTEM_ID: 1,
API_POWER: 0,
API_SYSTEM_FIRMWARE: "3.31",
API_SYSTEM_TYPE: 1,
}
]
}
HVAC_WEBSERVER_MOCK = {
API_MAC: "11:22:33:44:55:66",
API_WIFI_CHANNEL: 6,
@ -202,7 +216,7 @@ async def async_init_integration(
return_value=HVAC_MOCK,
), patch(
"homeassistant.components.airzone.AirzoneLocalApi.get_hvac_systems",
side_effect=SystemOutOfRange,
return_value=HVAC_SYSTEMS_MOCK,
), patch(
"homeassistant.components.airzone.AirzoneLocalApi.get_webserver",
side_effect=InvalidMethod,