coordinator.py

Go to the documentation of this file.

"""Tessie Data Coordinator."""
 
from datetime import timedelta
from http import HTTPStatus
import logging
from typing import Any
 
from aiohttp import ClientResponseError
from tesla_fleet_api import EnergySpecific
from tesla_fleet_api.exceptions import InvalidToken, MissingToken, TeslaFleetError
from tessie_api import get_state, get_status
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import TessieStatus
 
# This matches the update interval Tessie performs server side
TESSIE_SYNC_INTERVAL = 10
TESSIE_FLEET_API_SYNC_INTERVAL = timedelta(seconds=30)
 
_LOGGER = logging.getLogger(__name__)
 
 
def flatten(data: dict[str, Any], parent: str | None = None) -> dict[str, Any]:
    """Flatten the data structure."""
    result = {}
    for key, value in data.items():
        if parent:
            key = f"{parent}_{key}"
        if isinstance(value, dict):
            result.update(flatten(value, key))
        else:
            result[key] = value
    return result
 
 
class TessieStateUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Class to manage fetching data from the Tessie API."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        api_key: str,
        vin: str,
        data: dict[str, Any],
    ) -> None:
        """Initialize Tessie Data Update Coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name="Tessie",
            update_interval=timedelta(seconds=TESSIE_SYNC_INTERVAL),
        )
        self.api_keyapi_key = api_key
        self.vinvin = vin
        self.sessionsession = async_get_clientsession(hass)
        self.datadatadata = flatten(data)
 
    async def _async_update_data(self) -> dict[str, Any]:
        """Update vehicle data using Tessie API."""
        try:
            status = await get_status(
                session=self.sessionsession,
                api_key=self.api_keyapi_key,
                vin=self.vinvin,
            )
            if status["status"] == TessieStatus.ASLEEP:
                # Vehicle is asleep, no need to poll for data
                self.datadatadata["state"] = status["status"]
                return self.datadatadata
 
            vehicle = await get_state(
                session=self.sessionsession,
                api_key=self.api_keyapi_key,
                vin=self.vinvin,
                use_cache=True,
            )
        except ClientResponseError as e:
            if e.status == HTTPStatus.UNAUTHORIZED:
                # Auth Token is no longer valid
                raise ConfigEntryAuthFailed from e
            raise
 
        return flatten(vehicle)
 
 
class TessieEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Class to manage fetching energy site live status from the Tessie API."""
 
    def __init__(self, hass: HomeAssistant, api: EnergySpecific) -> None:
        """Initialize Tessie Energy Site Live coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name="Tessie Energy Site Live",
            update_interval=TESSIE_FLEET_API_SYNC_INTERVAL,
        )
        self.apiapi = api
 
    async def _async_update_data(self) -> dict[str, Any]:
        """Update energy site data using Tessie API."""
 
        try:
            data = (await self.apiapi.live_status())["response"]
        except (InvalidToken, MissingToken) as e:
            raise ConfigEntryAuthFailed from e
        except TeslaFleetError as e:
            raise UpdateFailed(e.message) from e
 
        # Convert Wall Connectors from array to dict
        data["wall_connectors"] = {
            wc["din"]: wc for wc in (data.get("wall_connectors") or [])
        }
 
        return data
 
 
class TessieEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Class to manage fetching energy site info from the Tessie API."""
 
    def __init__(self, hass: HomeAssistant, api: EnergySpecific) -> None:
        """Initialize Tessie Energy Info coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name="Tessie Energy Site Info",
            update_interval=TESSIE_FLEET_API_SYNC_INTERVAL,
        )
        self.apiapi = api
 
    async def _async_update_data(self) -> dict[str, Any]:
        """Update energy site data using Tessie API."""
 
        try:
            data = (await self.apiapi.site_info())["response"]
        except (InvalidToken, MissingToken) as e:
            raise ConfigEntryAuthFailed from e
        except TeslaFleetError as e:
            raise UpdateFailed(e.message) from e
 
        return flatten(data)