tado_connector.py

Go to the documentation of this file.

"""Tado Connector a class to store the data as an object."""
 
from datetime import datetime, timedelta
import logging
from typing import Any
 
from PyTado.interface import Tado
from requests import RequestException
 
from homeassistant.components.climate import PRESET_AWAY, PRESET_HOME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.util import Throttle
 
from .const import (
    INSIDE_TEMPERATURE_MEASUREMENT,
    PRESET_AUTO,
    SIGNAL_TADO_MOBILE_DEVICE_UPDATE_RECEIVED,
    SIGNAL_TADO_UPDATE_RECEIVED,
    TEMP_OFFSET,
)
 
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=4)
SCAN_INTERVAL = timedelta(minutes=5)
SCAN_MOBILE_DEVICE_INTERVAL = timedelta(seconds=30)
 
 
_LOGGER = logging.getLogger(__name__)
 
 
class TadoConnector:
    """An object to store the Tado data."""
 
    def __init__(
        self, hass: HomeAssistant, username: str, password: str, fallback: str
    ) -> None:
        """Initialize Tado Connector."""
        self.hasshass = hass
        self._username_username = username
        self._password_password = password
        self._fallback_fallback = fallback
 
        self.home_idhome_id: int = 0
        self.home_namehome_name = None
        self.tadotado = None
        self.zoneszones: list[dict[Any, Any]] = []
        self.devicesdevices: list[dict[Any, Any]] = []
        self.data: dict[str, dict] = {
            "device": {},
            "mobile_device": {},
            "weather": {},
            "geofence": {},
            "zone": {},
        }
 
    @property
    def fallback(self):
        """Return fallback flag to Smart Schedule."""
        return self._fallback_fallback
 
    def setup(self):
        """Connect to Tado and fetch the zones."""
        self.tadotado = Tado(self._username_username, self._password_password)
        # Load zones and devices
        self.zoneszones = self.tadotado.get_zones()
        self.devicesdevices = self.tadotado.get_devices()
        tado_home = self.tadotado.get_me()["homes"][0]
        self.home_idhome_id = tado_home["id"]
        self.home_namehome_name = tado_home["name"]
 
    def get_mobile_devices(self):
        """Return the Tado mobile devices."""
        return self.tadotado.get_mobile_devices()
 
    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        """Update the registered zones."""
        self.update_devicesupdate_devices()
        self.update_mobile_devicesupdate_mobile_devices()
        self.update_zonesupdate_zones()
        self.update_homeupdate_home()
 
    def update_mobile_devices(self) -> None:
        """Update the mobile devices."""
        try:
            mobile_devices = self.get_mobile_devicesget_mobile_devices()
        except RuntimeError:
            _LOGGER.error("Unable to connect to Tado while updating mobile devices")
            return
 
        if not mobile_devices:
            _LOGGER.debug("No linked mobile devices found for home ID %s", self.home_idhome_id)
            return
 
        # Errors are planned to be converted to exceptions
        # in PyTado library, so this can be removed
        if isinstance(mobile_devices, dict) and mobile_devices.get("errors"):
            _LOGGER.error(
                "Error for home ID %s while updating mobile devices: %s",
                self.home_idhome_id,
                mobile_devices["errors"],
            )
            return
 
        for mobile_device in mobile_devices:
            self.data["mobile_device"][mobile_device["id"]] = mobile_device
            _LOGGER.debug(
                "Dispatching update to %s mobile device: %s",
                self.home_idhome_id,
                mobile_device,
            )
 
        dispatcher_send(
            self.hasshass,
            SIGNAL_TADO_MOBILE_DEVICE_UPDATE_RECEIVED.format(self.home_idhome_id),
        )
 
    def update_devices(self):
        """Update the device data from Tado."""
        try:
            devices = self.tadotado.get_devices()
        except RuntimeError:
            _LOGGER.error("Unable to connect to Tado while updating devices")
            return
 
        if not devices:
            _LOGGER.debug("No linked devices found for home ID %s", self.home_idhome_id)
            return
 
        # Errors are planned to be converted to exceptions
        # in PyTado library, so this can be removed
        if isinstance(devices, dict) and devices.get("errors"):
            _LOGGER.error(
                "Error for home ID %s while updating devices: %s",
                self.home_idhome_id,
                devices["errors"],
            )
            return
 
        for device in devices:
            device_short_serial_no = device["shortSerialNo"]
            _LOGGER.debug("Updating device %s", device_short_serial_no)
            try:
                if (
                    INSIDE_TEMPERATURE_MEASUREMENT
                    in device["characteristics"]["capabilities"]
                ):
                    device[TEMP_OFFSET] = self.tadotado.get_device_info(
                        device_short_serial_no, TEMP_OFFSET
                    )
            except RuntimeError:
                _LOGGER.error(
                    "Unable to connect to Tado while updating device %s",
                    device_short_serial_no,
                )
                return
 
            self.data["device"][device_short_serial_no] = device
 
            _LOGGER.debug(
                "Dispatching update to %s device %s: %s",
                self.home_idhome_id,
                device_short_serial_no,
                device,
            )
            dispatcher_send(
                self.hasshass,
                SIGNAL_TADO_UPDATE_RECEIVED.format(
                    self.home_idhome_id, "device", device_short_serial_no
                ),
            )
 
    def update_zones(self):
        """Update the zone data from Tado."""
        try:
            zone_states = self.tadotado.get_zone_states()["zoneStates"]
        except RuntimeError:
            _LOGGER.error("Unable to connect to Tado while updating zones")
            return
 
        for zone in zone_states:
            self.update_zoneupdate_zone(int(zone))
 
    def update_zone(self, zone_id):
        """Update the internal data from Tado."""
        _LOGGER.debug("Updating zone %s", zone_id)
        try:
            data = self.tadotado.get_zone_state(zone_id)
        except RuntimeError:
            _LOGGER.error("Unable to connect to Tado while updating zone %s", zone_id)
            return
 
        self.data["zone"][zone_id] = data
 
        _LOGGER.debug(
            "Dispatching update to %s zone %s: %s",
            self.home_idhome_id,
            zone_id,
            data,
        )
        dispatcher_send(
            self.hasshass,
            SIGNAL_TADO_UPDATE_RECEIVED.format(self.home_idhome_id, "zone", zone_id),
        )
 
    def update_home(self):
        """Update the home data from Tado."""
        try:
            self.data["weather"] = self.tadotado.get_weather()
            self.data["geofence"] = self.tadotado.get_home_state()
            dispatcher_send(
                self.hasshass,
                SIGNAL_TADO_UPDATE_RECEIVED.format(self.home_idhome_id, "home", "data"),
            )
        except RuntimeError:
            _LOGGER.error(
                "Unable to connect to Tado while updating weather and geofence data"
            )
            return
 
    def get_capabilities(self, zone_id):
        """Return the capabilities of the devices."""
        return self.tadotado.get_capabilities(zone_id)
 
    def get_auto_geofencing_supported(self):
        """Return whether the Tado Home supports auto geofencing."""
        return self.tadotado.get_auto_geofencing_supported()
 
    def reset_zone_overlay(self, zone_id):
        """Reset the zone back to the default operation."""
        self.tadotado.reset_zone_overlay(zone_id)
        self.update_zoneupdate_zone(zone_id)
 
    def set_presence(
        self,
        presence=PRESET_HOME,
    ):
        """Set the presence to home, away or auto."""
        if presence == PRESET_AWAY:
            self.tadotado.set_away()
        elif presence == PRESET_HOME:
            self.tadotado.set_home()
        elif presence == PRESET_AUTO:
            self.tadotado.set_auto()
 
        # Update everything when changing modes
        self.update_zonesupdate_zones()
        self.update_homeupdate_home()
 
    def set_zone_overlay(
        self,
        zone_id=None,
        overlay_mode=None,
        temperature=None,
        duration=None,
        device_type="HEATING",
        mode=None,
        fan_speed=None,
        swing=None,
        fan_level=None,
        vertical_swing=None,
        horizontal_swing=None,
    ):
        """Set a zone overlay."""
        _LOGGER.debug(
            (
                "Set overlay for zone %s: overlay_mode=%s, temp=%s, duration=%s,"
                " type=%s, mode=%s fan_speed=%s swing=%s fan_level=%s vertical_swing=%s horizontal_swing=%s"
            ),
            zone_id,
            overlay_mode,
            temperature,
            duration,
            device_type,
            mode,
            fan_speed,
            swing,
            fan_level,
            vertical_swing,
            horizontal_swing,
        )
 
        try:
            self.tadotado.set_zone_overlay(
                zone_id,
                overlay_mode,
                temperature,
                duration,
                device_type,
                "ON",
                mode,
                fan_speed=fan_speed,
                swing=swing,
                fan_level=fan_level,
                vertical_swing=vertical_swing,
                horizontal_swing=horizontal_swing,
            )
 
        except RequestException as exc:
            _LOGGER.error("Could not set zone overlay: %s", exc)
 
        self.update_zoneupdate_zone(zone_id)
 
    def set_zone_off(self, zone_id, overlay_mode, device_type="HEATING"):
        """Set a zone to off."""
        try:
            self.tadotado.set_zone_overlay(
                zone_id, overlay_mode, None, None, device_type, "OFF"
            )
        except RequestException as exc:
            _LOGGER.error("Could not set zone overlay: %s", exc)
 
        self.update_zoneupdate_zone(zone_id)
 
    def set_temperature_offset(self, device_id, offset):
        """Set temperature offset of device."""
        try:
            self.tadotado.set_temp_offset(device_id, offset)
        except RequestException as exc:
            _LOGGER.error("Could not set temperature offset: %s", exc)
 
    def set_meter_reading(self, reading: int) -> dict[str, Any]:
        """Send meter reading to Tado."""
        dt: str = datetime.now().strftime("%Y-%m-%d")
        if self.tadotado is None:
            raise HomeAssistantError("Tado client is not initialized")
 
        try:
            return self.tadotado.set_eiq_meter_readings(date=dt, reading=reading)
        except RequestException as exc:
            raise HomeAssistantError("Could not set meter reading") from exc