coordinator.py

Go to the documentation of this file.

"""DataUpdateCoordinator for the wallbox integration."""
 
from __future__ import annotations
 
from collections.abc import Callable
from datetime import timedelta
from http import HTTPStatus
import logging
from typing import Any, Concatenate
 
import requests
from wallbox import Wallbox
 
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
from .const import (
    CHARGER_CURRENCY_KEY,
    CHARGER_DATA_KEY,
    CHARGER_ENERGY_PRICE_KEY,
    CHARGER_FEATURES_KEY,
    CHARGER_LOCKED_UNLOCKED_KEY,
    CHARGER_MAX_CHARGING_CURRENT_KEY,
    CHARGER_MAX_ICP_CURRENT_KEY,
    CHARGER_PLAN_KEY,
    CHARGER_POWER_BOOST_KEY,
    CHARGER_STATUS_DESCRIPTION_KEY,
    CHARGER_STATUS_ID_KEY,
    CODE_KEY,
    DOMAIN,
    UPDATE_INTERVAL,
    ChargerStatus,
)
 
_LOGGER = logging.getLogger(__name__)
 
# Translation of StatusId based on Wallbox portal code:
# https://my.wallbox.com/src/utilities/charger/chargerStatuses.js
CHARGER_STATUS: dict[int, ChargerStatus] = {
    0: ChargerStatus.DISCONNECTED,
    14: ChargerStatus.ERROR,
    15: ChargerStatus.ERROR,
    161: ChargerStatus.READY,
    162: ChargerStatus.READY,
    163: ChargerStatus.DISCONNECTED,
    164: ChargerStatus.WAITING,
    165: ChargerStatus.LOCKED,
    166: ChargerStatus.UPDATING,
    177: ChargerStatus.SCHEDULED,
    178: ChargerStatus.PAUSED,
    179: ChargerStatus.SCHEDULED,
    180: ChargerStatus.WAITING_FOR_CAR,
    181: ChargerStatus.WAITING_FOR_CAR,
    182: ChargerStatus.PAUSED,
    183: ChargerStatus.WAITING_IN_QUEUE_POWER_SHARING,
    184: ChargerStatus.WAITING_IN_QUEUE_POWER_SHARING,
    185: ChargerStatus.WAITING_IN_QUEUE_POWER_BOOST,
    186: ChargerStatus.WAITING_IN_QUEUE_POWER_BOOST,
    187: ChargerStatus.WAITING_MID_FAILED,
    188: ChargerStatus.WAITING_MID_SAFETY,
    189: ChargerStatus.WAITING_IN_QUEUE_ECO_SMART,
    193: ChargerStatus.CHARGING,
    194: ChargerStatus.CHARGING,
    195: ChargerStatus.CHARGING,
    196: ChargerStatus.DISCHARGING,
    209: ChargerStatus.LOCKED,
    210: ChargerStatus.LOCKED_CAR_CONNECTED,
}
 
 
def _require_authentication[_WallboxCoordinatorT: WallboxCoordinator, **_P](
    func: Callable[Concatenate[_WallboxCoordinatorT, _P], Any],
) -> Callable[Concatenate[_WallboxCoordinatorT, _P], Any]:
    """Authenticate with decorator using Wallbox API."""
 
    def require_authentication(
        self: _WallboxCoordinatorT, *args: _P.args, **kwargs: _P.kwargs
    ) -> Any:
        """Authenticate using Wallbox API."""
        try:
            self.authenticate()
            return func(self, *args, **kwargs)
        except requests.exceptions.HTTPError as wallbox_connection_error:
            if wallbox_connection_error.response.status_code == HTTPStatus.FORBIDDEN:
                raise ConfigEntryAuthFailed from wallbox_connection_error
            raise ConnectionError from wallbox_connection_error
 
    return require_authentication
 
 
def _validate(wallbox: Wallbox) -> None:
    """Authenticate using Wallbox API."""
    try:
        wallbox.authenticate()
    except requests.exceptions.HTTPError as wallbox_connection_error:
        if wallbox_connection_error.response.status_code == 403:
            raise InvalidAuth from wallbox_connection_error
        raise ConnectionError from wallbox_connection_error
 
 
async def async_validate_input(hass: HomeAssistant, wallbox: Wallbox) -> None:
    """Get new sensor data for Wallbox component."""
    await hass.async_add_executor_job(_validate, wallbox)
 
 
class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Wallbox Coordinator class."""
 
    def __init__(self, station: str, wallbox: Wallbox, hass: HomeAssistant) -> None:
        """Initialize."""
        self._station_station = station
        self._wallbox_wallbox = wallbox
 
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=UPDATE_INTERVAL),
        )
 
    def authenticate(self) -> None:
        """Authenticate using Wallbox API."""
        self._wallbox_wallbox.authenticate()
 
    @_require_authentication
    def _get_data(self) -> dict[str, Any]:
        """Get new sensor data for Wallbox component."""
        data: dict[str, Any] = self._wallbox_wallbox.getChargerStatus(self._station_station)
        data[CHARGER_MAX_CHARGING_CURRENT_KEY] = data[CHARGER_DATA_KEY][
            CHARGER_MAX_CHARGING_CURRENT_KEY
        ]
        data[CHARGER_LOCKED_UNLOCKED_KEY] = data[CHARGER_DATA_KEY][
            CHARGER_LOCKED_UNLOCKED_KEY
        ]
        data[CHARGER_ENERGY_PRICE_KEY] = data[CHARGER_DATA_KEY][
            CHARGER_ENERGY_PRICE_KEY
        ]
        # Only show max_icp_current if power_boost is available in the wallbox unit:
        if (
            data[CHARGER_DATA_KEY].get(CHARGER_MAX_ICP_CURRENT_KEY, 0) > 0
            and CHARGER_POWER_BOOST_KEY
            in data[CHARGER_DATA_KEY][CHARGER_PLAN_KEY][CHARGER_FEATURES_KEY]
        ):
            data[CHARGER_MAX_ICP_CURRENT_KEY] = data[CHARGER_DATA_KEY][
                CHARGER_MAX_ICP_CURRENT_KEY
            ]
 
        data[CHARGER_CURRENCY_KEY] = (
            f"{data[CHARGER_DATA_KEY][CHARGER_CURRENCY_KEY][CODE_KEY]}/kWh"
        )
 
        data[CHARGER_STATUS_DESCRIPTION_KEY] = CHARGER_STATUS.get(
            data[CHARGER_STATUS_ID_KEY], ChargerStatus.UNKNOWN
        )
        return data
 
    async def _async_update_data(self) -> dict[str, Any]:
        """Get new sensor data for Wallbox component."""
        return await self.hasshass.async_add_executor_job(self._get_data_get_data)
 
    @_require_authentication
    def _set_charging_current(self, charging_current: float) -> None:
        """Set maximum charging current for Wallbox."""
        try:
            self._wallbox_wallbox.setMaxChargingCurrent(self._station_station, charging_current)
        except requests.exceptions.HTTPError as wallbox_connection_error:
            if wallbox_connection_error.response.status_code == 403:
                raise InvalidAuth from wallbox_connection_error
            raise
 
    async def async_set_charging_current(self, charging_current: float) -> None:
        """Set maximum charging current for Wallbox."""
        await self.hasshass.async_add_executor_job(
            self._set_charging_current_set_charging_current, charging_current
        )
        await self.async_request_refreshasync_request_refresh()
 
    @_require_authentication
    def _set_icp_current(self, icp_current: float) -> None:
        """Set maximum icp current for Wallbox."""
        try:
            self._wallbox_wallbox.setIcpMaxCurrent(self._station_station, icp_current)
        except requests.exceptions.HTTPError as wallbox_connection_error:
            if wallbox_connection_error.response.status_code == 403:
                raise InvalidAuth from wallbox_connection_error
            raise
 
    async def async_set_icp_current(self, icp_current: float) -> None:
        """Set maximum icp current for Wallbox."""
        await self.hasshass.async_add_executor_job(self._set_icp_current_set_icp_current, icp_current)
        await self.async_request_refreshasync_request_refresh()
 
    @_require_authentication
    def _set_energy_cost(self, energy_cost: float) -> None:
        """Set energy cost for Wallbox."""
 
        self._wallbox_wallbox.setEnergyCost(self._station_station, energy_cost)
 
    async def async_set_energy_cost(self, energy_cost: float) -> None:
        """Set energy cost for Wallbox."""
        await self.hasshass.async_add_executor_job(self._set_energy_cost_set_energy_cost, energy_cost)
        await self.async_request_refreshasync_request_refresh()
 
    @_require_authentication
    def _set_lock_unlock(self, lock: bool) -> None:
        """Set wallbox to locked or unlocked."""
        try:
            if lock:
                self._wallbox_wallbox.lockCharger(self._station_station)
            else:
                self._wallbox_wallbox.unlockCharger(self._station_station)
        except requests.exceptions.HTTPError as wallbox_connection_error:
            if wallbox_connection_error.response.status_code == 403:
                raise InvalidAuth from wallbox_connection_error
            raise
 
    async def async_set_lock_unlock(self, lock: bool) -> None:
        """Set wallbox to locked or unlocked."""
        await self.hasshass.async_add_executor_job(self._set_lock_unlock_set_lock_unlock, lock)
        await self.async_request_refreshasync_request_refresh()
 
    @_require_authentication
    def _pause_charger(self, pause: bool) -> None:
        """Set wallbox to pause or resume."""
 
        if pause:
            self._wallbox_wallbox.pauseChargingSession(self._station_station)
        else:
            self._wallbox_wallbox.resumeChargingSession(self._station_station)
 
    async def async_pause_charger(self, pause: bool) -> None:
        """Set wallbox to pause or resume."""
        await self.hasshass.async_add_executor_job(self._pause_charger_pause_charger, pause)
        await self.async_request_refreshasync_request_refresh()
 
 
class InvalidAuth(HomeAssistantError):
    """Error to indicate there is invalid auth."""