homeassistant-dkncloudna/custom_components/dkncloudna/climate.py

"""Climate entity for DKN Cloud NA."""

from __future__ import annotations

from typing import Any, ClassVar

from homeassistant.components.climate import (
    ClimateEntity,
    ClimateEntityFeature,
    HVACAction,
    HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, PRECISION_WHOLE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .const import (
    DEVICE_MODE_AUTO,
    DEVICE_MODE_COOL,
    DEVICE_MODE_DRY,
    DEVICE_MODE_FAN,
    DEVICE_MODE_HEAT,
    DOMAIN,
    SPEED_20,
    SPEED_40,
    SPEED_60,
    SPEED_80,
    SPEED_100,
    SPEED_AUTO,
)
from .coordinator import DknCoordinator
from .entity import DknEntity
from .model import (
    available_fan_speeds,
    current_temperature as model_current_temperature,
    fan_mode_labels,
    inferred_hvac_action,
    requested_mode,
    supports_swing,
    target_temperature as model_target_temperature,
    target_temperature_key,
    to_device_temperature,
    writable_target_temperature_key,
)

_MODE_TO_HVAC: dict[int, HVACMode] = {
    DEVICE_MODE_AUTO: HVACMode.AUTO,
    DEVICE_MODE_COOL: HVACMode.COOL,
    DEVICE_MODE_HEAT: HVACMode.HEAT,
    DEVICE_MODE_FAN: HVACMode.FAN_ONLY,
    DEVICE_MODE_DRY: HVACMode.DRY,
}
_HVAC_TO_MODE: dict[HVACMode, int] = {v: k for k, v in _MODE_TO_HVAC.items()}

_SPEED_TO_FAN: dict[int, str] = {
    SPEED_AUTO: "auto",
    SPEED_20: "20%",
    SPEED_40: "40%",
    SPEED_60: "60%",
    SPEED_80: "80%",
    SPEED_100: "100%",
}
_FAN_TO_SPEED: dict[str, int] = {v: k for k, v in _SPEED_TO_FAN.items()}

_NO_TARGET_TEMP_MODES = {HVACMode.FAN_ONLY, HVACMode.DRY, HVACMode.OFF}


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up climate entities from a config entry."""
    coordinator: DknCoordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
    async_add_entities(
        DknClimateEntity(coordinator, mac) for mac in (coordinator.data or {})
    )


class DknClimateEntity(DknEntity, ClimateEntity):
    """Climate entity representing one DKN Cloud NA AC unit."""

    _attr_name = None
    _attr_temperature_unit = UnitOfTemperature.CELSIUS
    _attr_precision = PRECISION_WHOLE
    _attr_hvac_modes: ClassVar[list[HVACMode]] = [
        HVACMode.OFF,
        HVACMode.AUTO,
        HVACMode.COOL,
        HVACMode.HEAT,
        HVACMode.DRY,
        HVACMode.FAN_ONLY,
    ]
    _attr_swing_modes: ClassVar[list[str]] = ["off", "swing"]
    _attr_min_temp = 16
    _attr_max_temp = 32
    _attr_target_temperature_step = 1

    def __init__(self, coordinator: DknCoordinator, mac: str) -> None:
        """Initialise the climate entity for a single indoor unit."""
        super().__init__(coordinator, mac)
        self._attr_unique_id = f"{DOMAIN}_{mac}"

    @property
    def supported_features(self) -> ClimateEntityFeature:
        """Return feature flags appropriate for the current HVAC mode."""
        features = ClimateEntityFeature.TURN_ON | ClimateEntityFeature.TURN_OFF
        if self.hvac_mode not in _NO_TARGET_TEMP_MODES:
            features |= ClimateEntityFeature.TARGET_TEMPERATURE
        if available_fan_speeds(self._device_data):
            features |= ClimateEntityFeature.FAN_MODE
        if supports_swing(self._device_data):
            features |= ClimateEntityFeature.SWING_MODE
        return features

    @property
    def fan_modes(self) -> list[str] | None:
        """Return the fan speeds this unit reports as available."""
        labels = fan_mode_labels(self._device_data)
        return labels or None

    @property
    def hvac_mode(self) -> HVACMode:
        """Return the current mode, preferring an unexpired optimistic write."""
        data = self._device_data
        power = self._optimistic_get("power", data.get("power", False))
        if not power:
            return HVACMode.OFF
        optimistic_mode = self._optimistic_get("hvac_mode", None)
        if optimistic_mode is not None:
            return optimistic_mode
        mode_int = requested_mode(data) or DEVICE_MODE_AUTO
        return _MODE_TO_HVAC.get(int(mode_int), HVACMode.AUTO)

    @property
    def hvac_action(self) -> HVACAction | None:
        """Return what the unit is inferred to be doing right now.

        The cloud API reports no explicit action, so this is derived from the
        mode and the gap between room and target temperature.
        """
        mode = self.hvac_mode
        if mode == HVACMode.OFF:
            return HVACAction.OFF

        action = inferred_hvac_action(
            self._device_data,
            mode=_HVAC_TO_MODE.get(mode),
            target=self.target_temperature,
            power=True,
        )
        if action == "heating":
            return HVACAction.HEATING
        if action == "cooling":
            return HVACAction.COOLING
        if action == "idle":
            return HVACAction.IDLE
        if action == "fan":
            return HVACAction.FAN
        if action == "drying":
            return HVACAction.DRYING
        return None

    @property
    def current_temperature(self) -> float | None:
        """Return the room temperature in Celsius."""
        return model_current_temperature(self._device_data)

    @property
    def target_temperature(self) -> float | None:
        """Return the setpoint for the effective mode, in Celsius."""
        mode = self.hvac_mode
        if mode in _NO_TARGET_TEMP_MODES:
            return None
        # Read the setpoint for the effective mode so the value shown matches
        # the key a write would target, even while a mode change is still
        # propagating through the cloud.
        fallback = model_target_temperature(self._device_data, _HVAC_TO_MODE.get(mode))
        if fallback is None:
            fallback = model_target_temperature(self._device_data)
        if fallback is None:
            return None
        return self._optimistic_get("target_temp", fallback)

    @property
    def target_temperature_high(self) -> float | None:
        """Return None: the unit has a single setpoint, not a range."""
        return None

    @property
    def target_temperature_low(self) -> float | None:
        """Return None: the unit has a single setpoint, not a range."""
        return None

    @property
    def fan_mode(self) -> str | None:
        """Return the current fan speed label."""
        speed = self._device_data.get("speed_state", SPEED_AUTO)
        return self._optimistic_get("fan_mode", _SPEED_TO_FAN.get(int(speed), "auto"))

    @property
    def swing_mode(self) -> str | None:
        """Return whether the vertical slats are swinging."""
        slat = self._device_data.get("slats_vertical_1", 0)
        return self._optimistic_get("swing_mode", "swing" if int(slat) == 9 else "off")

    async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
        """Turn the unit off, or switch it on into the requested mode."""
        # Resolved before the try block below. An unsupported mode is a caller
        # error, not a cloud failure, and raising it inside the try would let
        # the blanket handler re-wrap it as "Failed to set HVAC mode:
        # Unsupported HVAC mode: ...".
        mode: int | None = None
        if hvac_mode != HVACMode.OFF:
            mode = _HVAC_TO_MODE.get(hvac_mode)
            if mode is None:
                raise HomeAssistantError(f"Unsupported HVAC mode: {hvac_mode}")

        installation_id = self._installation_id
        async with self._get_device_lock():
            try:
                if hvac_mode == HVACMode.OFF:
                    await self.coordinator.client.async_send_machine_event(
                        installation_id, self._command_mac, "power", False
                    )
                    self._optimistic_set(
                        "power", False, device_key="power", device_value=False
                    )
                    self._optimistic_set(
                        "hvac_mode",
                        HVACMode.OFF,
                        device_key="power",
                        device_value=False,
                    )
                else:
                    await self.coordinator.client.async_send_machine_event(
                        installation_id, self._command_mac, "power", True
                    )
                    await self.coordinator.client.async_send_machine_event(
                        installation_id, self._command_mac, "mode", mode
                    )
                    self._optimistic_set(
                        "power", True, device_key="power", device_value=True
                    )
                    self._optimistic_set(
                        "hvac_mode",
                        hvac_mode,
                        device_key="mode",
                        device_value=mode,
                    )
            except Exception as err:
                raise HomeAssistantError(f"Failed to set HVAC mode: {err}") from err

        self._schedule_refresh()
        self.async_write_ha_state()

    async def async_set_temperature(self, **kwargs: Any) -> None:
        """Set the target temperature, optionally changing mode first."""
        hvac_mode = kwargs.get("hvac_mode")
        if hvac_mode is not None:
            await self.async_set_hvac_mode(hvac_mode)

        temperature = kwargs.get(ATTR_TEMPERATURE)
        if temperature is None:
            return

        target_mode = hvac_mode or self.hvac_mode
        if target_mode in _NO_TARGET_TEMP_MODES:
            raise HomeAssistantError(
                f"Target temperature is not supported in {target_mode} mode"
            )

        installation_id = self._installation_id
        requested_mode_code = _HVAC_TO_MODE.get(target_mode)

        async with self._get_device_lock():
            # A mode change made moments ago (either above, or by a separate
            # set_hvac_mode call from the UI) may not have landed in the cloud
            # yet. Give it a bounded window to be echoed back before writing the
            # setpoint, so the setpoint is not applied against the old mode.
            if (
                requested_mode_code is not None
                and self._device_data.get("mode") != requested_mode_code
            ):
                await self._async_wait_for_device_value("mode", requested_mode_code)

            property_name = self._writable_temperature_property_for_mode(target_mode)
            device_temp = self._to_device_temperature(float(temperature))
            try:
                await self.coordinator.client.async_send_machine_event(
                    installation_id, self._command_mac, property_name, device_temp
                )
            except Exception as err:
                raise HomeAssistantError(f"Failed to set temperature: {err}") from err

        self._optimistic_set(
            "target_temp",
            float(temperature),
            device_key=property_name,
            device_value=device_temp,
        )
        self._schedule_refresh()
        self.async_write_ha_state()

    async def async_set_fan_mode(self, fan_mode: str) -> None:
        """Set the fan speed."""
        speed = _FAN_TO_SPEED.get(fan_mode)
        if speed is None:
            raise HomeAssistantError(f"Unsupported fan mode: {fan_mode}")
        if speed not in available_fan_speeds(self._device_data):
            raise HomeAssistantError(f"Fan mode not supported by device: {fan_mode}")

        installation_id = self._installation_id
        async with self._get_device_lock():
            try:
                await self.coordinator.client.async_send_machine_event(
                    installation_id, self._command_mac, "speed_state", speed
                )
            except Exception as err:
                raise HomeAssistantError(f"Failed to set fan mode: {err}") from err

        self._optimistic_set(
            "fan_mode", fan_mode, device_key="speed_state", device_value=speed
        )
        self._schedule_refresh()
        self.async_write_ha_state()

    async def async_set_swing_mode(self, swing_mode: str) -> None:
        """Start or stop the vertical slat swing."""
        if swing_mode not in {"off", "swing"}:
            raise HomeAssistantError(f"Unsupported swing mode: {swing_mode}")
        if not supports_swing(self._device_data):
            raise HomeAssistantError("Swing mode not supported by device")

        installation_id = self._installation_id
        slat = 9 if swing_mode == "swing" else 0
        async with self._get_device_lock():
            try:
                await self.coordinator.client.async_send_machine_event(
                    installation_id, self._command_mac, "slats_vertical_1", slat
                )
            except Exception as err:
                raise HomeAssistantError(f"Failed to set swing mode: {err}") from err

        self._optimistic_set(
            "swing_mode", swing_mode, device_key="slats_vertical_1", device_value=slat
        )
        self._schedule_refresh()
        self.async_write_ha_state()

    async def async_turn_on(self) -> None:
        """Turn the unit on into auto mode."""
        await self.async_set_hvac_mode(HVACMode.AUTO)

    async def async_turn_off(self) -> None:
        """Turn the unit off."""
        await self.async_set_hvac_mode(HVACMode.OFF)

    @property
    def _installation_id(self) -> str:
        installation_id = str(self._device_data.get("_installation_id") or "").strip()
        if not installation_id:
            raise HomeAssistantError("Missing installation id for device")
        return installation_id

    def _temperature_property_for_mode(self, hvac_mode: HVACMode) -> str:
        if hvac_mode not in {HVACMode.AUTO, HVACMode.HEAT, HVACMode.COOL}:
            raise HomeAssistantError(
                f"Target temperature is not supported in {hvac_mode} mode"
            )
        requested = {
            HVACMode.AUTO: DEVICE_MODE_AUTO,
            HVACMode.COOL: DEVICE_MODE_COOL,
            HVACMode.HEAT: DEVICE_MODE_HEAT,
        }[hvac_mode]
        key = target_temperature_key(requested)
        if key is None:
            raise HomeAssistantError(
                f"Target temperature is not supported in {hvac_mode} mode"
            )
        return key

    def _writable_temperature_property_for_mode(self, hvac_mode: HVACMode) -> str:
        # The key must follow the mode being requested, not whatever mode the
        # cached device payload still reports; that payload lags behind a mode
        # change and would send e.g. a heat setpoint to setpoint_air_cool.
        requested = self._temperature_property_for_mode(hvac_mode)
        preferred = writable_target_temperature_key(
            self._device_data, _HVAC_TO_MODE.get(hvac_mode)
        )
        return preferred or requested

    def _to_device_temperature(self, temperature_c: float) -> float | int:
        return to_device_temperature(temperature_c, self._device_data.get("units"))