389 lines
15 KiB
Python
389 lines
15 KiB
Python
"""Climate entities proxied through the standalone GREE Controller."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from homeassistant.components.climate import ClimateEntity
|
|
from homeassistant.components.climate.const import (
|
|
SWING_OFF,
|
|
SWING_ON,
|
|
ClimateEntityFeature,
|
|
HVACMode,
|
|
)
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import ConfigEntryError
|
|
from homeassistant.helpers import entity_registry as er
|
|
from homeassistant.helpers.entity import DeviceInfo
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
from homeassistant.util import slugify
|
|
|
|
from . import GreeControllerRuntimeData
|
|
from .const import DOMAIN
|
|
from .coordinator import GreeControllerCoordinator
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
MODE_TO_HA = {
|
|
"auto": HVACMode.AUTO,
|
|
"cool": HVACMode.COOL,
|
|
"dry": HVACMode.DRY,
|
|
"fan": HVACMode.FAN_ONLY,
|
|
"heat": HVACMode.HEAT,
|
|
}
|
|
HA_TO_MODE = {value: key for key, value in MODE_TO_HA.items()}
|
|
FAN_TO_NAME = {0: "auto", 1: "low", 2: "medium_low", 3: "medium", 4: "medium_high", 5: "high"}
|
|
NAME_TO_FAN = {value: key for key, value in FAN_TO_NAME.items()}
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
"""Create climate entities for all devices exposed by the controller."""
|
|
runtime: GreeControllerRuntimeData = entry.runtime_data
|
|
registry = er.async_get(hass)
|
|
entities: list[ClimateEntity] = []
|
|
|
|
for device_id, device in runtime.coordinator.data.items():
|
|
desired_entity_id = runtime.entity_map.get(device_id)
|
|
unique_id = f"{device_id}-climate"
|
|
if desired_entity_id:
|
|
if not desired_entity_id.startswith("climate."):
|
|
raise ConfigEntryError(f"Mapped entity ID must use the climate domain: {desired_entity_id}")
|
|
existing = registry.async_get(desired_entity_id)
|
|
if existing and not (existing.platform == DOMAIN and existing.unique_id == unique_id):
|
|
raise ConfigEntryError(
|
|
f"Entity ID {desired_entity_id} is still reserved by integration {existing.platform}. "
|
|
"Disable/remove the previous GREE integration and remove its entity registry entry before takeover."
|
|
)
|
|
if hass.states.get(desired_entity_id) is not None and existing is None:
|
|
raise ConfigEntryError(
|
|
f"Entity ID {desired_entity_id} is still active in Home Assistant. "
|
|
"Unload the previous integration before takeover."
|
|
)
|
|
entities.append(
|
|
GreeControllerClimate(runtime.coordinator, device_id, desired_entity_id)
|
|
)
|
|
|
|
for zone in runtime.coordinator.plan.get("zones", []):
|
|
zone_id = str(zone.get("zone_id") or "").strip()
|
|
if zone_id:
|
|
_ensure_zone_thermostat_entity_id(registry, zone)
|
|
entities.append(GreeControllerZoneClimate(runtime.coordinator, zone_id))
|
|
|
|
async_add_entities(entities)
|
|
|
|
|
|
def _ensure_zone_thermostat_entity_id(
|
|
registry: er.EntityRegistry, zone: dict[str, Any]
|
|
) -> None:
|
|
"""Keep zone climate entity IDs stable and descriptive.
|
|
|
|
Zone climate entities use ``climate.<zone_name>_thermostat``. Existing
|
|
registry entries are migrated to the same scheme, which also fixes stale
|
|
IDs left behind after a zone was renamed (for example ``climate.jan_2``
|
|
for a zone currently named Igor).
|
|
"""
|
|
zone_id = str(zone.get("zone_id") or "").strip()
|
|
if not zone_id:
|
|
return
|
|
|
|
unique_id = f"{zone_id}-zone-climate"
|
|
current_entity_id = registry.async_get_entity_id("climate", DOMAIN, unique_id)
|
|
if current_entity_id is None:
|
|
return
|
|
|
|
zone_name = str(zone.get("zone_name") or zone_id).strip() or zone_id
|
|
object_id = slugify(zone_name) or slugify(zone_id) or "zone"
|
|
desired_entity_id = f"climate.{object_id}_thermostat"
|
|
if current_entity_id == desired_entity_id:
|
|
return
|
|
|
|
occupied = registry.async_get(desired_entity_id)
|
|
if occupied is not None and occupied.entity_id != current_entity_id:
|
|
_LOGGER.warning(
|
|
"Cannot rename zone thermostat %s to %s because that entity ID is already in use",
|
|
current_entity_id,
|
|
desired_entity_id,
|
|
)
|
|
return
|
|
|
|
registry.async_update_entity(current_entity_id, new_entity_id=desired_entity_id)
|
|
_LOGGER.info(
|
|
"Renamed zone thermostat entity %s to %s",
|
|
current_entity_id,
|
|
desired_entity_id,
|
|
)
|
|
|
|
|
|
class GreeControllerClimate(CoordinatorEntity[GreeControllerCoordinator], ClimateEntity):
|
|
"""Home Assistant climate entity controlled through the Rust service."""
|
|
|
|
_attr_has_entity_name = True
|
|
_attr_temperature_unit = UnitOfTemperature.CELSIUS
|
|
_attr_min_temp = 8.0
|
|
_attr_max_temp = 30.0
|
|
_attr_target_temperature_step = 1.0
|
|
_attr_hvac_modes = [HVACMode.OFF, HVACMode.AUTO, HVACMode.COOL, HVACMode.HEAT, HVACMode.DRY, HVACMode.FAN_ONLY]
|
|
_attr_fan_modes = list(NAME_TO_FAN)
|
|
_attr_swing_modes = [SWING_OFF, SWING_ON]
|
|
_attr_swing_horizontal_modes = [SWING_OFF, SWING_ON]
|
|
_attr_supported_features = (
|
|
ClimateEntityFeature.TARGET_TEMPERATURE
|
|
| ClimateEntityFeature.FAN_MODE
|
|
| ClimateEntityFeature.SWING_MODE
|
|
| ClimateEntityFeature.SWING_HORIZONTAL_MODE
|
|
| ClimateEntityFeature.TURN_ON
|
|
| ClimateEntityFeature.TURN_OFF
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: GreeControllerCoordinator,
|
|
device_id: str,
|
|
requested_entity_id: str | None,
|
|
) -> None:
|
|
super().__init__(coordinator)
|
|
self._device_id = device_id
|
|
self._attr_unique_id = f"{device_id}-climate"
|
|
self._attr_name = "Direct control"
|
|
if requested_entity_id:
|
|
# This is intentionally limited to same-domain takeover migrations.
|
|
# The setup guard above prevents accidental collisions.
|
|
self.entity_id = requested_entity_id
|
|
|
|
@property
|
|
def _device(self) -> dict[str, Any]:
|
|
return self.coordinator.data.get(self._device_id, {})
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
return super().available and bool(self._device.get("online", False))
|
|
|
|
@property
|
|
def device_info(self) -> DeviceInfo:
|
|
device = self._device
|
|
return DeviceInfo(
|
|
identifiers={(DOMAIN, self._device_id)},
|
|
name=str(device.get("name") or self._device_id),
|
|
manufacturer="GREE",
|
|
model=str(device.get("model") or "GREE HVAC"),
|
|
sw_version=str(device.get("firmware") or "") or None,
|
|
)
|
|
|
|
@property
|
|
def current_temperature(self) -> float | None:
|
|
value = self._device.get("current_temperature")
|
|
return float(value) if value is not None else None
|
|
|
|
@property
|
|
def target_temperature(self) -> float | None:
|
|
value = self._device.get("target_temperature")
|
|
return float(value) if value is not None else None
|
|
|
|
@property
|
|
def hvac_mode(self) -> HVACMode:
|
|
device = self._device
|
|
if not device.get("power", False):
|
|
return HVACMode.OFF
|
|
return MODE_TO_HA.get(str(device.get("mode", "auto")), HVACMode.AUTO)
|
|
|
|
@property
|
|
def fan_mode(self) -> str:
|
|
return FAN_TO_NAME.get(int(self._device.get("fan_speed", 0)), "auto")
|
|
|
|
@property
|
|
def swing_mode(self) -> str:
|
|
return SWING_ON if self._device.get("swing_vertical", False) else SWING_OFF
|
|
|
|
@property
|
|
def swing_horizontal_mode(self) -> str:
|
|
return SWING_ON if self._device.get("swing_horizontal", False) else SWING_OFF
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any]:
|
|
device = self._device
|
|
return {
|
|
"controller_device_id": self._device_id,
|
|
"controller_online": bool(device.get("online", False)),
|
|
"quiet": bool(device.get("quiet", False)),
|
|
"turbo": bool(device.get("turbo", False)),
|
|
"light": bool(device.get("light", False)),
|
|
"xfan": bool(device.get("xfan", False)),
|
|
"air": bool(device.get("air", False)),
|
|
"health": bool(device.get("health", False)),
|
|
"sleep": bool(device.get("sleep", False)),
|
|
"last_seen": device.get("last_seen"),
|
|
}
|
|
|
|
async def _command(self, payload: dict[str, Any]) -> None:
|
|
await self.coordinator.async_device_command(self._device_id, payload)
|
|
|
|
async def async_turn_on(self) -> None:
|
|
await self._command({"power": True})
|
|
|
|
async def async_turn_off(self) -> None:
|
|
await self._command({"power": False})
|
|
|
|
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
|
|
if hvac_mode == HVACMode.OFF:
|
|
await self._command({"power": False})
|
|
return
|
|
mode = HA_TO_MODE.get(hvac_mode)
|
|
if mode is None:
|
|
return
|
|
await self._command({"power": True, "mode": mode})
|
|
|
|
async def async_set_temperature(self, **kwargs: Any) -> None:
|
|
temperature = kwargs.get(ATTR_TEMPERATURE)
|
|
if temperature is not None:
|
|
await self._command({"target_temperature": float(temperature)})
|
|
|
|
async def async_set_fan_mode(self, fan_mode: str) -> None:
|
|
if fan_mode in NAME_TO_FAN:
|
|
await self._command({"fan_speed": NAME_TO_FAN[fan_mode]})
|
|
|
|
async def async_set_swing_mode(self, swing_mode: str) -> None:
|
|
await self._command({"swing_vertical": swing_mode == SWING_ON})
|
|
|
|
async def async_set_swing_horizontal_mode(self, swing_horizontal_mode: str) -> None:
|
|
await self._command({"swing_horizontal": swing_horizontal_mode == SWING_ON})
|
|
|
|
class GreeControllerZoneClimate(CoordinatorEntity[GreeControllerCoordinator], ClimateEntity):
|
|
"""Full climate entity for a controller thermostat zone."""
|
|
|
|
_attr_has_entity_name = True
|
|
_attr_name = "Thermostat"
|
|
_attr_temperature_unit = UnitOfTemperature.CELSIUS
|
|
_attr_min_temp = 8.0
|
|
_attr_max_temp = 30.0
|
|
_attr_target_temperature_step = 0.5
|
|
_attr_hvac_modes = [HVACMode.OFF, HVACMode.AUTO, HVACMode.COOL, HVACMode.HEAT]
|
|
_attr_preset_modes = ["auto", "comfort", "sleep", "away"]
|
|
_attr_supported_features = (
|
|
ClimateEntityFeature.TARGET_TEMPERATURE
|
|
| ClimateEntityFeature.PRESET_MODE
|
|
| ClimateEntityFeature.TURN_ON
|
|
| ClimateEntityFeature.TURN_OFF
|
|
)
|
|
|
|
def __init__(self, coordinator: GreeControllerCoordinator, zone_id: str) -> None:
|
|
super().__init__(coordinator)
|
|
self._zone_id = zone_id
|
|
self._attr_unique_id = f"{zone_id}-zone-climate"
|
|
|
|
@property
|
|
def _zone(self) -> dict[str, Any]:
|
|
for zone in self.coordinator.plan.get("zones", []):
|
|
if str(zone.get("zone_id")) == self._zone_id:
|
|
return zone
|
|
return {}
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
return super().available and bool(self._zone)
|
|
|
|
@property
|
|
def device_info(self) -> DeviceInfo:
|
|
zone = self._zone
|
|
return DeviceInfo(
|
|
identifiers={(DOMAIN, f"zone:{self._zone_id}")},
|
|
name=str(zone.get("zone_name") or self._zone_id),
|
|
manufacturer="GREE Controller",
|
|
model="Zone thermostat",
|
|
via_device=(DOMAIN, "controller"),
|
|
)
|
|
|
|
@property
|
|
def current_temperature(self) -> float | None:
|
|
value = self._zone.get("current_temperature")
|
|
return float(value) if value is not None else None
|
|
|
|
@property
|
|
def target_temperature(self) -> float | None:
|
|
value = self._zone.get("target_temperature")
|
|
if value is None:
|
|
value = self._zone.get("setpoint")
|
|
return float(value) if value is not None else None
|
|
|
|
@property
|
|
def hvac_mode(self) -> HVACMode:
|
|
zone = self._zone
|
|
if not zone.get("enabled", False) or zone.get("local_thermostat_power") is False:
|
|
return HVACMode.OFF
|
|
if zone.get("inherit_house_mode", True):
|
|
return HVACMode.AUTO
|
|
return HVACMode.HEAT if str(zone.get("configured_mode", zone.get("mode", "cool"))) == "heat" else HVACMode.COOL
|
|
|
|
@property
|
|
def preset_mode(self) -> str:
|
|
return str(self._zone.get("preset_override") or "auto")
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict[str, Any]:
|
|
zone = self._zone
|
|
return {
|
|
"controller_zone_id": self._zone_id,
|
|
"effective_enabled": bool(zone.get("effective_enabled", zone.get("enabled", False))),
|
|
"local_thermostat_power": zone.get("local_thermostat_power"),
|
|
"local_thermostat_resume_at": zone.get("local_thermostat_resume_at"),
|
|
"effective_mode": zone.get("mode"),
|
|
"follows_house_mode": bool(zone.get("inherit_house_mode", True)),
|
|
"auto_mode_meaning": "Follow the GREE Controller whole-house heating/cooling mode",
|
|
"active_preset": zone.get("preset"),
|
|
"preset_override": zone.get("preset_override"),
|
|
"demand": bool(zone.get("demand", False)),
|
|
"device_setpoint": zone.get("device_setpoint"),
|
|
"temperature_source": zone.get("control_source"),
|
|
"control_owner": zone.get("control_owner"),
|
|
"control_source": zone.get("control_command_source"),
|
|
"control_since": zone.get("control_since"),
|
|
"resume_at": zone.get("resume_at"),
|
|
"control_reason": zone.get("control_reason"),
|
|
"blocked_reason": zone.get("blocked_reason"),
|
|
"lockout_until": zone.get("lockout_until"),
|
|
"desired_power": zone.get("desired_power"),
|
|
"desired_mode": zone.get("desired_mode"),
|
|
"actual_power": zone.get("actual_power"),
|
|
"actual_mode": zone.get("actual_mode"),
|
|
"actual_setpoint": zone.get("actual_setpoint"),
|
|
"current_schedule": zone.get("current_schedule_name"),
|
|
}
|
|
|
|
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
|
|
if hvac_mode == HVACMode.OFF:
|
|
payload = {"power": False}
|
|
elif hvac_mode == HVACMode.AUTO:
|
|
payload = {"power": True, "mode": "house"}
|
|
elif hvac_mode == HVACMode.COOL:
|
|
payload = {"power": True, "mode": "cool"}
|
|
elif hvac_mode == HVACMode.HEAT:
|
|
payload = {"power": True, "mode": "heat"}
|
|
else:
|
|
return
|
|
await self.coordinator.async_zone_control(self._zone_id, payload)
|
|
|
|
async def async_set_temperature(self, **kwargs: Any) -> None:
|
|
temperature = kwargs.get(ATTR_TEMPERATURE)
|
|
if temperature is None:
|
|
return
|
|
await self.coordinator.async_zone_control(self._zone_id, {"setpoint": float(temperature)})
|
|
|
|
async def async_set_preset_mode(self, preset_mode: str) -> None:
|
|
if preset_mode not in self._attr_preset_modes:
|
|
return
|
|
await self.coordinator.async_zone_control(self._zone_id, {"preset": preset_mode})
|
|
|
|
async def async_turn_on(self) -> None:
|
|
await self.coordinator.async_zone_control(self._zone_id, {"power": True})
|
|
|
|
async def async_turn_off(self) -> None:
|
|
await self.coordinator.async_zone_control(self._zone_id, {"power": False})
|
|
|