Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Move wallbox coordinator to its own file #101577

Merged
merged 2 commits into from
Oct 7, 2023
Merged

Move wallbox coordinator to its own file #101577

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2544c39
Move wallbox coordinator to its own file
jpbede Oct 7, 2023
269159a
Fix imports in wallbox tests
jpbede Oct 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
188 changes: 3 additions & 185 deletions homeassistant/components/wallbox/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,195 +1,17 @@
"""The Wallbox integration."""
from __future__ import annotations

from datetime import timedelta
from http import HTTPStatus
import logging
from typing import Any

import requests
from wallbox import Wallbox

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from .const import (
CHARGER_CURRENCY_KEY,
CHARGER_DATA_KEY,
CHARGER_ENERGY_PRICE_KEY,
CHARGER_LOCKED_UNLOCKED_KEY,
CHARGER_MAX_CHARGING_CURRENT_KEY,
CHARGER_STATUS_DESCRIPTION_KEY,
CHARGER_STATUS_ID_KEY,
CODE_KEY,
CONF_STATION,
DOMAIN,
ChargerStatus,
)
from homeassistant.exceptions import ConfigEntryAuthFailed

_LOGGER = logging.getLogger(__name__)
from .const import CONF_STATION, DOMAIN, UPDATE_INTERVAL
from .coordinator import InvalidAuth, WallboxCoordinator

PLATFORMS = [Platform.SENSOR, Platform.NUMBER, Platform.LOCK, Platform.SWITCH]
UPDATE_INTERVAL = 30

# 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,
}


class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Wallbox Coordinator class."""

def __init__(self, station: str, wallbox: Wallbox, hass: HomeAssistant) -> None:
"""Initialize."""
self._station = station
self._wallbox = wallbox

super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=UPDATE_INTERVAL),
)

def _authenticate(self) -> None:
"""Authenticate using Wallbox API."""
try:
self._wallbox.authenticate()

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

def _validate(self) -> None:
"""Authenticate using Wallbox API."""
try:
self._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(self) -> None:
"""Get new sensor data for Wallbox component."""
await self.hass.async_add_executor_job(self._validate)

def _get_data(self) -> dict[str, Any]:
"""Get new sensor data for Wallbox component."""
try:
self._authenticate()
data: dict[str, Any] = self._wallbox.getChargerStatus(self._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
]
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
except (
ConnectionError,
requests.exceptions.HTTPError,
) as wallbox_connection_error:
raise UpdateFailed from wallbox_connection_error

async def _async_update_data(self) -> dict[str, Any]:
"""Get new sensor data for Wallbox component."""
return await self.hass.async_add_executor_job(self._get_data)

def _set_charging_current(self, charging_current: float) -> None:
"""Set maximum charging current for Wallbox."""
try:
self._authenticate()
self._wallbox.setMaxChargingCurrent(self._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 ConnectionError from wallbox_connection_error

async def async_set_charging_current(self, charging_current: float) -> None:
"""Set maximum charging current for Wallbox."""
await self.hass.async_add_executor_job(
self._set_charging_current, charging_current
)
await self.async_request_refresh()

def _set_lock_unlock(self, lock: bool) -> None:
"""Set wallbox to locked or unlocked."""
try:
self._authenticate()
if lock:
self._wallbox.lockCharger(self._station)
else:
self._wallbox.unlockCharger(self._station)
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_set_lock_unlock(self, lock: bool) -> None:
"""Set wallbox to locked or unlocked."""
await self.hass.async_add_executor_job(self._set_lock_unlock, lock)
await self.async_request_refresh()

def _pause_charger(self, pause: bool) -> None:
"""Set wallbox to pause or resume."""
try:
self._authenticate()
if pause:
self._wallbox.pauseChargingSession(self._station)
else:
self._wallbox.resumeChargingSession(self._station)
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_pause_charger(self, pause: bool) -> None:
"""Set wallbox to pause or resume."""
await self.hass.async_add_executor_job(self._pause_charger, pause)
await self.async_request_refresh()


async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
Expand Down Expand Up @@ -227,7 +49,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data[DOMAIN].pop(entry.entry_id)

return unload_ok


class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""
2 changes: 1 addition & 1 deletion homeassistant/components/wallbox/config_flow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.data_entry_flow import FlowResult

from . import InvalidAuth, WallboxCoordinator
from .const import CONF_STATION, DOMAIN
from .coordinator import InvalidAuth, WallboxCoordinator

COMPONENT_DOMAIN = DOMAIN

Expand Down
35 changes: 35 additions & 0 deletions homeassistant/components/wallbox/const.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from enum import StrEnum

DOMAIN = "wallbox"
UPDATE_INTERVAL = 30
jpbede marked this conversation as resolved.
Show resolved Hide resolved

BIDIRECTIONAL_MODEL_PREFIXES = ["QSX"]

Expand Down Expand Up @@ -55,3 +56,37 @@ class ChargerStatus(StrEnum):
WAITING_MID_SAFETY = "Waiting MID safety margin exceeded"
WAITING_IN_QUEUE_ECO_SMART = "Waiting in queue by Eco-Smart"
UNKNOWN = "Unknown"


# 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,
}
Loading