coordinator.py

Go to the documentation of this file.

"""The Aprilaire coordinator."""
 
from __future__ import annotations
 
from collections.abc import Awaitable, Callable
import logging
from typing import Any
 
import pyaprilaire.client
from pyaprilaire.const import MODELS, Attribute, FunctionalDomain
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import BaseDataUpdateCoordinatorProtocol
 
from .const import DOMAIN
 
RECONNECT_INTERVAL = 60 * 60
RETRY_CONNECTION_INTERVAL = 10
WAIT_TIMEOUT = 30
 
_LOGGER = logging.getLogger(__name__)
 
type AprilaireConfigEntry = ConfigEntry[AprilaireCoordinator]
 
 
class AprilaireCoordinator(BaseDataUpdateCoordinatorProtocol):
    """Coordinator for interacting with the thermostat."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        unique_id: str | None,
        host: str,
        port: int,
    ) -> None:
        """Initialize the coordinator."""
 
        self.hasshass = hass
        self.unique_idunique_id = unique_id
        self.datadata: dict[str, Any] = {}
 
        self._listeners: dict[CALLBACK_TYPE, tuple[CALLBACK_TYPE, object | None]] = {}
 
        self.clientclient = pyaprilaire.client.AprilaireClient(
            host,
            port,
            self.async_set_updated_dataasync_set_updated_data,
            _LOGGER,
            RECONNECT_INTERVAL,
            RETRY_CONNECTION_INTERVAL,
        )
 
        if hasattr(self.clientclient, "data") and self.clientclient.data:
            self.datadata = self.clientclient.data
 
    @callback
    def async_add_listener(
        self, update_callback: CALLBACK_TYPE, context: Any = None
    ) -> Callable[[], None]:
        """Listen for data updates."""
 
        @callback
        def remove_listener() -> None:
            """Remove update listener."""
            self._listeners.pop(remove_listener)
 
        self._listeners[remove_listener] = (update_callback, context)
 
        return remove_listener
 
    @callback
    def async_update_listeners(self) -> None:
        """Update all registered listeners."""
        for update_callback, _ in list(self._listeners.values()):
            update_callback()
 
    def async_set_updated_data(self, data: Any) -> None:
        """Manually update data, notify listeners and reset refresh interval."""
 
        old_device_info = self.create_device_infocreate_device_info(self.datadata)
 
        self.datadata = self.datadata | data
 
        self.async_update_listenersasync_update_listeners()
 
        new_device_info = self.create_device_infocreate_device_info(data)
 
        if (
            old_device_info is not None
            and new_device_info is not None
            and old_device_info != new_device_info
        ):
            device_registry = dr.async_get(self.hasshass)
 
            device = device_registry.async_get_device(old_device_info["identifiers"])
 
            if device is not None:
                new_device_info.pop("identifiers", None)
                new_device_info.pop("connections", None)
 
                device_registry.async_update_device(
                    device_id=device.id,
                    **new_device_info,  # type: ignore[misc]
                )
 
    async def start_listen(self):
        """Start listening for data."""
        await self.clientclient.start_listen()
 
    def stop_listen(self):
        """Stop listening for data."""
        self.clientclient.stop_listen()
 
    async def wait_for_ready(
        self, ready_callback: Callable[[bool], Awaitable[None]]
    ) -> bool:
        """Wait for the client to be ready."""
 
        if not self.datadata or Attribute.MAC_ADDRESS not in self.datadata:
            data = await self.clientclient.wait_for_response(
                FunctionalDomain.IDENTIFICATION, 2, WAIT_TIMEOUT
            )
 
            if not data or Attribute.MAC_ADDRESS not in data:
                _LOGGER.error("Missing MAC address")
                await ready_callback(False)
 
                return False
 
        if not self.datadata or Attribute.NAME not in self.datadata:
            await self.clientclient.wait_for_response(
                FunctionalDomain.IDENTIFICATION, 4, WAIT_TIMEOUT
            )
 
        if not self.datadata or Attribute.THERMOSTAT_MODES not in self.datadata:
            await self.clientclient.wait_for_response(
                FunctionalDomain.CONTROL, 7, WAIT_TIMEOUT
            )
 
        if (
            not self.datadata
            or Attribute.INDOOR_TEMPERATURE_CONTROLLING_SENSOR_STATUS not in self.datadata
        ):
            await self.clientclient.wait_for_response(
                FunctionalDomain.SENSORS, 2, WAIT_TIMEOUT
            )
 
        await ready_callback(True)
 
        return True
 
    @property
    def device_name(self) -> str:
        """Get the name of the thermostat."""
 
        return self.create_device_namecreate_device_name(self.datadata)
 
    def create_device_name(self, data: dict[str, Any] | None) -> str:
        """Create the name of the thermostat."""
 
        name = data.get(Attribute.NAME) if data else None
 
        return name if name else "Aprilaire"
 
    def get_hw_version(self, data: dict[str, Any]) -> str:
        """Get the hardware version."""
 
        if hardware_revision := data.get(Attribute.HARDWARE_REVISION):
            return (
                f"Rev. {chr(hardware_revision)}"
                if hardware_revision > ord("A")
                else str(hardware_revision)
            )
 
        return "Unknown"
 
    @property
    def device_info(self) -> DeviceInfo | None:
        """Get the device info for the thermostat."""
        return self.create_device_infocreate_device_info(self.datadata)
 
    def create_device_info(self, data: dict[str, Any]) -> DeviceInfo | None:
        """Create the device info for the thermostat."""
 
        if data is None or Attribute.MAC_ADDRESS not in data or self.unique_idunique_id is None:
            return None
 
        device_info = DeviceInfo(
            identifiers={(DOMAIN, self.unique_idunique_id)},
            name=self.create_device_namecreate_device_name(data),
            manufacturer="Aprilaire",
        )
 
        model_number = data.get(Attribute.MODEL_NUMBER)
        if model_number is not None:
            device_info["model"] = MODELS.get(model_number, f"Unknown ({model_number})")
 
        device_info["hw_version"] = self.get_hw_versionget_hw_version(data)
 
        firmware_major_revision = data.get(Attribute.FIRMWARE_MAJOR_REVISION)
        firmware_minor_revision = data.get(Attribute.FIRMWARE_MINOR_REVISION)
        if firmware_major_revision is not None:
            device_info["sw_version"] = (
                str(firmware_major_revision)
                if firmware_minor_revision is None
                else f"{firmware_major_revision}.{firmware_minor_revision:02}"
            )
 
        return device_info