jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-20 03:37:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add solar charging options to Wallbox integration (#139286)

Browse Source 
Co-authored-by: Abílio Costa <abmantis@users.noreply.github.com>
This commit is contained in:
Joris Drenth 2025-05-20 15:19:48 +02:00 committed by GitHub
parent 010b4f6b15
commit fc62bc5fc1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 415 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
homeassistant/components/wallbox/__init__.py
View File

@ -12,7 +12,13 @@ from homeassistant.exceptions import ConfigEntryAuthFailed
from .const import DOMAIN, UPDATE_INTERVAL
from .coordinator import InvalidAuth, WallboxCoordinator, async_validate_input
PLATFORMS = [Platform.LOCK, Platform.NUMBER, Platform.SENSOR, Platform.SWITCH]
PLATFORMS = [
Platform.LOCK,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

11
homeassistant/components/wallbox/const.py
View File

@ -38,6 +38,9 @@ CHARGER_STATE_OF_CHARGE_KEY = "state_of_charge"
CHARGER_STATUS_ID_KEY = "status_id"
CHARGER_STATUS_DESCRIPTION_KEY = "status_description"
CHARGER_CONNECTIONS = "connections"
CHARGER_ECO_SMART_KEY = "ecosmart"
CHARGER_ECO_SMART_STATUS_KEY = "enabled"
CHARGER_ECO_SMART_MODE_KEY = "mode"
class ChargerStatus(StrEnum):
@ -61,3 +64,11 @@ class ChargerStatus(StrEnum):
WAITING_MID_SAFETY = "Waiting MID safety margin exceeded"
WAITING_IN_QUEUE_ECO_SMART = "Waiting in queue by Eco-Smart"
UNKNOWN = "Unknown"
class EcoSmartMode(StrEnum):
"""Charger Eco mode select options."""
OFF = "off"
ECO_MODE = "eco_mode"
FULL_SOLAR = "full_solar"

36
homeassistant/components/wallbox/coordinator.py
View File

@ -19,6 +19,9 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import (
CHARGER_CURRENCY_KEY,
CHARGER_DATA_KEY,
CHARGER_ECO_SMART_KEY,
CHARGER_ECO_SMART_MODE_KEY,
CHARGER_ECO_SMART_STATUS_KEY,
CHARGER_ENERGY_PRICE_KEY,
CHARGER_FEATURES_KEY,
CHARGER_LOCKED_UNLOCKED_KEY,
@ -33,6 +36,7 @@ from .const import (
DOMAIN,
UPDATE_INTERVAL,
ChargerStatus,
EcoSmartMode,
)
_LOGGER = logging.getLogger(__name__)
@ -160,6 +164,21 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
data[CHARGER_STATUS_DESCRIPTION_KEY] = CHARGER_STATUS.get(
data[CHARGER_STATUS_ID_KEY], ChargerStatus.UNKNOWN
)
# Set current solar charging mode
eco_smart_enabled = data[CHARGER_DATA_KEY][CHARGER_ECO_SMART_KEY][
CHARGER_ECO_SMART_STATUS_KEY
]
eco_smart_mode = data[CHARGER_DATA_KEY][CHARGER_ECO_SMART_KEY][
CHARGER_ECO_SMART_MODE_KEY
]
if eco_smart_enabled is False:
data[CHARGER_ECO_SMART_KEY] = EcoSmartMode.OFF
elif eco_smart_mode == 0:
data[CHARGER_ECO_SMART_KEY] = EcoSmartMode.ECO_MODE
elif eco_smart_mode == 1:
data[CHARGER_ECO_SMART_KEY] = EcoSmartMode.FULL_SOLAR
return data
async def _async_update_data(self) -> dict[str, Any]:
@ -241,6 +260,23 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
await self.hass.async_add_executor_job(self._pause_charger, pause)
await self.async_request_refresh()
@_require_authentication
def _set_eco_smart(self, option: str) -> None:
"""Set wallbox solar charging mode."""
if option == EcoSmartMode.ECO_MODE:
self._wallbox.enableEcoSmart(self._station, 0)
elif option == EcoSmartMode.FULL_SOLAR:
self._wallbox.enableEcoSmart(self._station, 1)
else:
self._wallbox.disableEcoSmart(self._station)
async def async_set_eco_smart(self, option: str) -> None:
"""Set wallbox solar charging mode."""
await self.hass.async_add_executor_job(self._set_eco_smart, option)
await self.async_request_refresh()
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

5
homeassistant/components/wallbox/icons.json
View File

@ -1,5 +1,10 @@
{
"entity": {
"select": {
"ecosmart": {
"default": "mdi:solar-power"
}
},
"sensor": {
"charging_speed": {
"default": "mdi:speedometer"

105
homeassistant/components/wallbox/select.py Normal file
View File

@ -0,0 +1,105 @@
"""Home Assistant component for accessing the Wallbox Portal API. The switch component creates a switch entity."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from requests import HTTPError
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
CHARGER_DATA_KEY,
CHARGER_ECO_SMART_KEY,
CHARGER_FEATURES_KEY,
CHARGER_PLAN_KEY,
CHARGER_POWER_BOOST_KEY,
CHARGER_SERIAL_NUMBER_KEY,
DOMAIN,
EcoSmartMode,
)
from .coordinator import WallboxCoordinator
from .entity import WallboxEntity
@dataclass(frozen=True, kw_only=True)
class WallboxSelectEntityDescription(SelectEntityDescription):
"""Describes Wallbox select entity."""
current_option_fn: Callable[[WallboxCoordinator], str | None]
select_option_fn: Callable[[WallboxCoordinator, str], Awaitable[None]]
supported_fn: Callable[[WallboxCoordinator], bool]
SELECT_TYPES: dict[str, WallboxSelectEntityDescription] = {
CHARGER_ECO_SMART_KEY: WallboxSelectEntityDescription(
key=CHARGER_ECO_SMART_KEY,
translation_key=CHARGER_ECO_SMART_KEY,
options=[
EcoSmartMode.OFF,
EcoSmartMode.ECO_MODE,
EcoSmartMode.FULL_SOLAR,
],
select_option_fn=lambda coordinator, mode: coordinator.async_set_eco_smart(
mode
),
current_option_fn=lambda coordinator: coordinator.data[CHARGER_ECO_SMART_KEY],
supported_fn=lambda coordinator: coordinator.data[CHARGER_DATA_KEY][
CHARGER_PLAN_KEY
][CHARGER_FEATURES_KEY].count(CHARGER_POWER_BOOST_KEY),
)
}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Create wallbox select entities in HASS."""
coordinator: WallboxCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
WallboxSelect(coordinator, description)
for ent in coordinator.data
if (
(description := SELECT_TYPES.get(ent))
and description.supported_fn(coordinator)
)
)
class WallboxSelect(WallboxEntity, SelectEntity):
"""Representation of the Wallbox portal."""
entity_description: WallboxSelectEntityDescription
def __init__(
self,
coordinator: WallboxCoordinator,
description: WallboxSelectEntityDescription,
) -> None:
"""Initialize a Wallbox select entity."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{description.key}-{coordinator.data[CHARGER_DATA_KEY][CHARGER_SERIAL_NUMBER_KEY]}"
@property
def current_option(self) -> str | None:
"""Return an option."""
return self.entity_description.current_option_fn(self.coordinator)
async def async_select_option(self, option: str) -> None:
"""Handle the selection of an option."""
try:
await self.entity_description.select_option_fn(self.coordinator, option)
except (ConnectionError, HTTPError) as e:
raise HomeAssistantError(
translation_key="api_failed", translation_domain=DOMAIN
) from e
await self.coordinator.async_request_refresh()

15
homeassistant/components/wallbox/strings.json
View File

@ -91,6 +91,21 @@
"pause_resume": {
"name": "Pause/resume"
}
},
"select": {
"ecosmart": {
"name": "Solar charging",
"state": {
"off": "[%key:common::state::off%]",
"eco_mode": "Eco mode",
"full_solar": "Full solar"
}
}
}
},
"exceptions": {
"api_failed": {
"message": "Error communicating with Wallbox API"
}
}
}

113
tests/components/wallbox/__init__.py
View File

@ -2,6 +2,7 @@
from http import HTTPStatus
import requests
import requests_mock
from homeassistant.components.wallbox.const import (
@ -12,6 +13,9 @@ from homeassistant.components.wallbox.const import (
CHARGER_CURRENCY_KEY,
CHARGER_CURRENT_VERSION_KEY,
CHARGER_DATA_KEY,
CHARGER_ECO_SMART_KEY,
CHARGER_ECO_SMART_MODE_KEY,
CHARGER_ECO_SMART_STATUS_KEY,
CHARGER_ENERGY_PRICE_KEY,
CHARGER_FEATURES_KEY,
CHARGER_LOCKED_UNLOCKED_KEY,
@ -50,6 +54,10 @@ test_response = {
CHARGER_CURRENCY_KEY: {"code": "EUR/kWh"},
CHARGER_MAX_ICP_CURRENT_KEY: 20,
CHARGER_PLAN_KEY: {CHARGER_FEATURES_KEY: [CHARGER_POWER_BOOST_KEY]},
CHARGER_ECO_SMART_KEY: {
CHARGER_ECO_SMART_STATUS_KEY: False,
CHARGER_ECO_SMART_MODE_KEY: 0,
},
},
}
@ -71,9 +79,89 @@ test_response_bidir = {
CHARGER_CURRENCY_KEY: {"code": "EUR/kWh"},
CHARGER_MAX_ICP_CURRENT_KEY: 20,
CHARGER_PLAN_KEY: {CHARGER_FEATURES_KEY: [CHARGER_POWER_BOOST_KEY]},
CHARGER_ECO_SMART_KEY: {
CHARGER_ECO_SMART_STATUS_KEY: False,
CHARGER_ECO_SMART_MODE_KEY: 0,
},
},
}
test_response_eco_mode = {
CHARGER_CHARGING_POWER_KEY: 0,
CHARGER_STATUS_ID_KEY: 193,
CHARGER_MAX_AVAILABLE_POWER_KEY: 25.0,
CHARGER_CHARGING_SPEED_KEY: 0,
CHARGER_ADDED_RANGE_KEY: 150,
CHARGER_ADDED_ENERGY_KEY: 44.697,
CHARGER_NAME_KEY: "WallboxName",
CHARGER_DATA_KEY: {
CHARGER_MAX_CHARGING_CURRENT_KEY: 24,
CHARGER_ENERGY_PRICE_KEY: 0.4,
CHARGER_LOCKED_UNLOCKED_KEY: False,
CHARGER_SERIAL_NUMBER_KEY: "20000",
CHARGER_PART_NUMBER_KEY: "PLP1-0-2-4-9-002-E",
CHARGER_SOFTWARE_KEY: {CHARGER_CURRENT_VERSION_KEY: "5.5.10"},
CHARGER_CURRENCY_KEY: {"code": "EUR/kWh"},
CHARGER_MAX_ICP_CURRENT_KEY: 20,
CHARGER_PLAN_KEY: {CHARGER_FEATURES_KEY: [CHARGER_POWER_BOOST_KEY]},
CHARGER_ECO_SMART_KEY: {
CHARGER_ECO_SMART_STATUS_KEY: True,
CHARGER_ECO_SMART_MODE_KEY: 0,
},
},
}
test_response_full_solar = {
CHARGER_CHARGING_POWER_KEY: 0,
CHARGER_STATUS_ID_KEY: 193,
CHARGER_MAX_AVAILABLE_POWER_KEY: 25.0,
CHARGER_CHARGING_SPEED_KEY: 0,
CHARGER_ADDED_RANGE_KEY: 150,
CHARGER_ADDED_ENERGY_KEY: 44.697,
CHARGER_NAME_KEY: "WallboxName",
CHARGER_DATA_KEY: {
CHARGER_MAX_CHARGING_CURRENT_KEY: 24,
CHARGER_ENERGY_PRICE_KEY: 0.4,
CHARGER_LOCKED_UNLOCKED_KEY: False,
CHARGER_SERIAL_NUMBER_KEY: "20000",
CHARGER_PART_NUMBER_KEY: "PLP1-0-2-4-9-002-E",
CHARGER_SOFTWARE_KEY: {CHARGER_CURRENT_VERSION_KEY: "5.5.10"},
CHARGER_CURRENCY_KEY: {"code": "EUR/kWh"},
CHARGER_MAX_ICP_CURRENT_KEY: 20,
CHARGER_PLAN_KEY: {CHARGER_FEATURES_KEY: [CHARGER_POWER_BOOST_KEY]},
CHARGER_ECO_SMART_KEY: {
CHARGER_ECO_SMART_STATUS_KEY: True,
CHARGER_ECO_SMART_MODE_KEY: 1,
},
},
}
test_response_no_power_boost = {
CHARGER_CHARGING_POWER_KEY: 0,
CHARGER_STATUS_ID_KEY: 193,
CHARGER_MAX_AVAILABLE_POWER_KEY: 25.0,
CHARGER_CHARGING_SPEED_KEY: 0,
CHARGER_ADDED_RANGE_KEY: 150,
CHARGER_ADDED_ENERGY_KEY: 44.697,
CHARGER_NAME_KEY: "WallboxName",
CHARGER_DATA_KEY: {
CHARGER_MAX_CHARGING_CURRENT_KEY: 24,
CHARGER_ENERGY_PRICE_KEY: 0.4,
CHARGER_LOCKED_UNLOCKED_KEY: False,
CHARGER_SERIAL_NUMBER_KEY: "20000",
CHARGER_PART_NUMBER_KEY: "PLP1-0-2-4-9-002-E",
CHARGER_SOFTWARE_KEY: {CHARGER_CURRENT_VERSION_KEY: "5.5.10"},
CHARGER_CURRENCY_KEY: {"code": "EUR/kWh"},
CHARGER_MAX_ICP_CURRENT_KEY: 20,
CHARGER_PLAN_KEY: {CHARGER_FEATURES_KEY: []},
},
}
http_404_error = requests.exceptions.HTTPError()
http_404_error.response = requests.Response()
http_404_error.response.status_code = HTTPStatus.NOT_FOUND
authorisation_response = {
"data": {
@ -128,6 +216,31 @@ async def setup_integration(hass: HomeAssistant, entry: MockConfigEntry) -> None
await hass.async_block_till_done()
async def setup_integration_select(
hass: HomeAssistant, entry: MockConfigEntry, response
) -> None:
"""Test wallbox sensor class setup."""
with requests_mock.Mocker() as mock_request:
mock_request.get(
"https://user-api.wall-box.com/users/signin",
json=authorisation_response,
status_code=HTTPStatus.OK,
)
mock_request.get(
"https://api.wall-box.com/chargers/status/12345",
json=response,
status_code=HTTPStatus.OK,
)
mock_request.put(
"https://api.wall-box.com/v2/charger/12345",
json={CHARGER_MAX_CHARGING_CURRENT_KEY: 20},
status_code=HTTPStatus.OK,
)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
async def setup_integration_bidir(hass: HomeAssistant, entry: MockConfigEntry) -> None:
"""Test wallbox sensor class setup."""
with requests_mock.Mocker() as mock_request:

1
tests/components/wallbox/const.py
View File

@ -15,3 +15,4 @@ MOCK_SENSOR_CHARGING_SPEED_ID = "sensor.wallbox_wallboxname_charging_speed"
MOCK_SENSOR_CHARGING_POWER_ID = "sensor.wallbox_wallboxname_charging_power"
MOCK_SENSOR_MAX_AVAILABLE_POWER = "sensor.wallbox_wallboxname_max_available_power"
MOCK_SWITCH_ENTITY_ID = "switch.wallbox_wallboxname_pause_resume"
MOCK_SELECT_ENTITY_ID = "select.wallbox_wallboxname_solar_charging"

122
tests/components/wallbox/test_select.py Normal file
View File

@ -0,0 +1,122 @@
"""Test Wallbox Select component."""
from unittest.mock import Mock, patch
import pytest
from homeassistant.components.select import (
ATTR_OPTION,
DOMAIN as SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
)
from homeassistant.components.wallbox.const import CHARGER_STATUS_ID_KEY, EcoSmartMode
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant, HomeAssistantError
from . import (
authorisation_response,
http_404_error,
setup_integration_select,
test_response,
test_response_eco_mode,
test_response_full_solar,
test_response_no_power_boost,
)
from .const import MOCK_SELECT_ENTITY_ID
from tests.common import MockConfigEntry
TEST_OPTIONS = [
(EcoSmartMode.OFF, test_response),
(EcoSmartMode.ECO_MODE, test_response_eco_mode),
(EcoSmartMode.FULL_SOLAR, test_response_full_solar),
]
@pytest.fixture
def mock_authenticate():
"""Fixture to patch Wallbox methods."""
with patch(
"homeassistant.components.wallbox.Wallbox.authenticate",
new=Mock(return_value=authorisation_response),
):
yield
@pytest.mark.parametrize(("mode", "response"), TEST_OPTIONS)
async def test_wallbox_select_solar_charging_class(
hass: HomeAssistant, entry: MockConfigEntry, mode, response, mock_authenticate
) -> None:
"""Test wallbox select class."""
with (
patch(
"homeassistant.components.wallbox.Wallbox.enableEcoSmart",
new=Mock(return_value={CHARGER_STATUS_ID_KEY: 193}),
),
patch(
"homeassistant.components.wallbox.Wallbox.disableEcoSmart",
new=Mock(return_value={CHARGER_STATUS_ID_KEY: 193}),
),
):
await setup_integration_select(hass, entry, response)
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: MOCK_SELECT_ENTITY_ID,
ATTR_OPTION: mode,
},
blocking=True,
)
state = hass.states.get(MOCK_SELECT_ENTITY_ID)
assert state.state == mode
async def test_wallbox_select_no_power_boost_class(
hass: HomeAssistant, entry: MockConfigEntry
) -> None:
"""Test wallbox select class."""
await setup_integration_select(hass, entry, test_response_no_power_boost)
state = hass.states.get(MOCK_SELECT_ENTITY_ID)
assert state is None
@pytest.mark.parametrize(("mode", "response"), TEST_OPTIONS)
@pytest.mark.parametrize("error", [http_404_error, ConnectionError])
async def test_wallbox_select_class_error(
hass: HomeAssistant,
entry: MockConfigEntry,
mode,
response,
error,
mock_authenticate,
) -> None:
"""Test wallbox select class connection error."""
await setup_integration_select(hass, entry, response)
with (
patch(
"homeassistant.components.wallbox.Wallbox.disableEcoSmart",
new=Mock(side_effect=error),
),
patch(
"homeassistant.components.wallbox.Wallbox.enableEcoSmart",
new=Mock(side_effect=error),
),
pytest.raises(HomeAssistantError, match="Error communicating with Wallbox API"),
):
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: MOCK_SELECT_ENTITY_ID,
ATTR_OPTION: mode,
},
blocking=True,
)