coordinator.py

Go to the documentation of this file.

"""Coordinator for the Moehlenhoff Alpha2."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
 
import aiohttp
from moehlenhoff_alpha2 import Alpha2Base
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
_LOGGER = logging.getLogger(__name__)
 
UPDATE_INTERVAL = timedelta(seconds=60)
 
 
class Alpha2BaseCoordinator(DataUpdateCoordinator[dict[str, dict]]):
    """Keep the base instance in one place and centralize the update."""
 
    def __init__(self, hass: HomeAssistant, base: Alpha2Base) -> None:
        """Initialize Alpha2Base data updater."""
        self.basebase = base
        super().__init__(
            hass=hass,
            logger=_LOGGER,
            name="alpha2_base",
            update_interval=UPDATE_INTERVAL,
        )
 
    async def _async_update_data(self) -> dict[str, dict[str, dict]]:
        """Fetch the latest data from the source."""
        await self.basebase.update_data()
        return {
            "heat_areas": {ha["ID"]: ha for ha in self.basebase.heat_areas if ha.get("ID")},
            "heat_controls": {
                hc["ID"]: hc for hc in self.basebase.heat_controls if hc.get("ID")
            },
            "io_devices": {io["ID"]: io for io in self.basebase.io_devices if io.get("ID")},
        }
 
    def get_cooling(self) -> bool:
        """Return if cooling mode is enabled."""
        return self.basebase.cooling
 
    async def async_set_cooling(self, enabled: bool) -> None:
        """Enable or disable cooling mode."""
        await self.basebase.set_cooling(enabled)
        self.async_update_listenersasync_update_listeners()
 
    async def async_set_target_temperature(
        self, heat_area_id: str, target_temperature: float
    ) -> None:
        """Set the target temperature of the given heat area."""
        _LOGGER.debug(
            "Setting target temperature of heat area %s to %0.1f",
            heat_area_id,
            target_temperature,
        )
 
        update_data = {"T_TARGET": target_temperature}
        is_cooling = self.get_coolingget_cooling()
        heat_area_mode = self.datadata["heat_areas"][heat_area_id]["HEATAREA_MODE"]
        if heat_area_mode == 1:
            if is_cooling:
                update_data["T_COOL_DAY"] = target_temperature
            else:
                update_data["T_HEAT_DAY"] = target_temperature
        elif heat_area_mode == 2:
            if is_cooling:
                update_data["T_COOL_NIGHT"] = target_temperature
            else:
                update_data["T_HEAT_NIGHT"] = target_temperature
 
        try:
            await self.basebase.update_heat_area(heat_area_id, update_data)
        except aiohttp.ClientError as http_err:
            raise HomeAssistantError(
                "Failed to set target temperature, communication error with alpha2 base"
            ) from http_err
        self.datadata["heat_areas"][heat_area_id].update(update_data)
        self.async_update_listenersasync_update_listeners()
 
    async def async_set_heat_area_mode(
        self, heat_area_id: str, heat_area_mode: int
    ) -> None:
        """Set the mode of the given heat area."""
        # HEATAREA_MODE: 0=Auto, 1=Tag, 2=Nacht
        if heat_area_mode not in (0, 1, 2):
            raise ValueError(f"Invalid heat area mode: {heat_area_mode}")
        _LOGGER.debug(
            "Setting mode of heat area %s to %d",
            heat_area_id,
            heat_area_mode,
        )
        try:
            await self.basebase.update_heat_area(
                heat_area_id, {"HEATAREA_MODE": heat_area_mode}
            )
        except aiohttp.ClientError as http_err:
            raise HomeAssistantError(
                "Failed to set heat area mode, communication error with alpha2 base"
            ) from http_err
 
        self.datadata["heat_areas"][heat_area_id]["HEATAREA_MODE"] = heat_area_mode
        is_cooling = self.get_coolingget_cooling()
        if heat_area_mode == 1:
            if is_cooling:
                self.datadata["heat_areas"][heat_area_id]["T_TARGET"] = self.datadata[
                    "heat_areas"
                ][heat_area_id]["T_COOL_DAY"]
            else:
                self.datadata["heat_areas"][heat_area_id]["T_TARGET"] = self.datadata[
                    "heat_areas"
                ][heat_area_id]["T_HEAT_DAY"]
        elif heat_area_mode == 2:
            if is_cooling:
                self.datadata["heat_areas"][heat_area_id]["T_TARGET"] = self.datadata[
                    "heat_areas"
                ][heat_area_id]["T_COOL_NIGHT"]
            else:
                self.datadata["heat_areas"][heat_area_id]["T_TARGET"] = self.datadata[
                    "heat_areas"
                ][heat_area_id]["T_HEAT_NIGHT"]
 
        self.async_update_listenersasync_update_listeners()