diff --git a/custom_components/evse_load_balancer/chargers/__init__.py b/custom_components/evse_load_balancer/chargers/__init__.py index 3ef25fd..b536900 100644 --- a/custom_components/evse_load_balancer/chargers/__init__.py +++ b/custom_components/evse_load_balancer/chargers/__init__.py @@ -11,6 +11,7 @@ from .easee_charger import EaseeCharger from .keba_charger import KebaCharger from .lektrico_charger import LektricoCharger +from .wallbox_charger import WallboxCharger from .zaptec_charger import ZaptecCharger if TYPE_CHECKING: @@ -34,6 +35,7 @@ async def charger_factory( ZaptecCharger, KebaCharger, LektricoCharger, + WallboxCharger, ]: if charger_cls.is_charger_device(device): return charger_cls(hass, config_entry, device) diff --git a/custom_components/evse_load_balancer/chargers/wallbox_charger.py b/custom_components/evse_load_balancer/chargers/wallbox_charger.py new file mode 100644 index 0000000..0510a3d --- /dev/null +++ b/custom_components/evse_load_balancer/chargers/wallbox_charger.py @@ -0,0 +1,180 @@ +"""Wallbox Charger implementation.""" + +import logging + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceEntry + +from ..const import CHARGER_DOMAIN_WALLBOX, Phase # noqa: TID252 +from ..ha_device import HaDevice # noqa: TID252 +from .charger import Charger, PhaseMode + +_LOGGER = logging.getLogger(__name__) + + +class WallboxEntityMap: + """ + Map Wallbox entities to their respective translation keys. + + https://github.com/home-assistant/core/blob/dev/homeassistant/components/wallbox/number.py + https://github.com/home-assistant/core/blob/dev/homeassistant/components/wallbox/sensor.py + """ + + DynamicChargerLimit = "maximum_charging_current" + MaxChargerLimit = "max_available_power" + Status = "status_description" + + +class WallboxStatusMap: + """ + Normalized status values used by the adapter. + + Values mirror the Wallbox integration `ChargerStatus` strings. + https://github.com/home-assistant/core/blob/dev/homeassistant/components/wallbox/const.py + """ + + Charging = "Charging" + Discharging = "Discharging" + Paused = "Paused" + Scheduled = "Scheduled" + WaitingForCarDemand = "Waiting for car demand" + Waiting = "Waiting" + Disconnected = "Disconnected" + Error = "Error" + Ready = "Ready" + Locked = "Locked" + LockedCarConnected = "Locked, car connected" + Updating = "Updating" + WaitingInQueuePowerSharing = "Waiting in queue by Power Sharing" + WaitingInQueuePowerBoost = "Waiting in queue by Power Boost" + WaitingMidFailed = "Waiting MID failed" + WaitingMidSafety = "Waiting MID safety margin exceeded" + WaitingInQueueEcoSmart = "Waiting in queue by Eco-Smart" + Unknown = "Unknown" + + +class WallboxCharger(HaDevice, Charger): + """Implementation of the Charger class for Wallbox chargers.""" + + def __init__( + self, hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry + ) -> None: + """Initialize the Wallbox charger.""" + HaDevice.__init__(self, hass, device_entry) + Charger.__init__(self, hass, config_entry, device_entry) + self.refresh_entities() + + @staticmethod + def is_charger_device(device: DeviceEntry) -> bool: + """Check if the given device is a Wallbox charger.""" + return any( + id_domain == CHARGER_DOMAIN_WALLBOX for id_domain, _ in device.identifiers + ) + + async def async_setup(self) -> None: + """Set up the charger (no-op).""" + + def set_phase_mode(self, mode: PhaseMode, _phase: Phase | None = None) -> None: + """Set the phase mode (no-op for Wallbox).""" + if mode not in PhaseMode: + msg = "Invalid mode. Must be 'single' or 'multi'." + raise ValueError(msg) + + def has_synced_phase_limits(self) -> bool: + """Wallbox number entity exposes a single max current value (global).""" + return True + + async def set_current_limit(self, limit: dict[Phase, int]) -> None: + """ + Set the charger current limit. + + The official Wallbox integration exposes the charging current as a + `number` entity. We update that entity using the common + `number.set_value` service with the entity_id and a numeric `value`. + """ + amps = int(min(limit.values())) + + entity_id = self._get_entity_id_by_translation_key( + WallboxEntityMap.DynamicChargerLimit + ) + + await self.hass.services.async_call( + domain="number", + service="set_value", + service_data={"entity_id": entity_id, "value": amps}, + blocking=True, + ) + + def get_current_limit(self) -> dict[Phase, int] | None: + """Return the currently configured charging limit (from the `number` entity).""" + state = self._get_entity_state_by_translation_key( + WallboxEntityMap.DynamicChargerLimit + ) + if state is None: + _LOGGER.warning( + "Wallbox dynamic charger limit not available for device %s", + self.device_entry.id, + ) + return None + try: + return dict.fromkeys(Phase, int(float(state))) + except (ValueError, TypeError): + _LOGGER.warning("Unable to parse Wallbox dynamic limit state: %s", state) + return None + + def get_max_current_limit(self) -> dict[Phase, int] | None: + """Return the configured maximum charging current.""" + state = self._get_entity_state_by_translation_key( + WallboxEntityMap.MaxChargerLimit + ) + if state is None: + _LOGGER.warning( + "Wallbox max charger limit not available for device %s", + self.device_entry.id, + ) + return None + try: + return dict.fromkeys(Phase, int(float(state))) + except (ValueError, TypeError): + _LOGGER.warning("Unable to parse Wallbox max limit state: %s", state) + return None + + def _get_status(self) -> str | None: + return self._get_entity_state_by_translation_key(WallboxEntityMap.Status) + + def car_connected(self) -> bool: + """Return whether a car is connected to the charger.""" + status = self._get_status() + return status in ( + WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.Paused, + WallboxStatusMap.Scheduled, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Waiting, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + ) + + def can_charge(self) -> bool: + """Return whether the charger can deliver charge.""" + status = self._get_status() + return status in ( + WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Paused, + ) + + def is_charging(self) -> bool: + """Return whether the charger is actively charging.""" + status = self._get_status() + return status == WallboxStatusMap.Charging + + async def async_unload(self) -> None: + """Unload the Wallbox charger (no-op).""" diff --git a/custom_components/evse_load_balancer/config_flow.py b/custom_components/evse_load_balancer/config_flow.py index 291dfd8..f5a3645 100644 --- a/custom_components/evse_load_balancer/config_flow.py +++ b/custom_components/evse_load_balancer/config_flow.py @@ -27,6 +27,7 @@ CHARGER_DOMAIN_EASEE, CHARGER_DOMAIN_KEBA, CHARGER_DOMAIN_LEKTRICO, + CHARGER_DOMAIN_WALLBOX, CHARGER_DOMAIN_ZAPTEC, CHARGER_MANUFACTURER_AMINA, DOMAIN, @@ -58,6 +59,7 @@ {"integration": CHARGER_DOMAIN_ZAPTEC}, {"integration": CHARGER_DOMAIN_KEBA}, {"integration": CHARGER_DOMAIN_LEKTRICO}, + {"integration": CHARGER_DOMAIN_WALLBOX}, { "integration": HA_INTEGRATION_DOMAIN_MQTT, "manufacturer": CHARGER_MANUFACTURER_AMINA, diff --git a/custom_components/evse_load_balancer/const.py b/custom_components/evse_load_balancer/const.py index 8dd828d..5fb7317 100644 --- a/custom_components/evse_load_balancer/const.py +++ b/custom_components/evse_load_balancer/const.py @@ -8,6 +8,7 @@ CHARGER_DOMAIN_ZAPTEC = "zaptec" CHARGER_DOMAIN_LEKTRICO = "lektrico" CHARGER_DOMAIN_KEBA = "keba" +CHARGER_DOMAIN_WALLBOX = "wallbox" HA_INTEGRATION_DOMAIN_MQTT = "mqtt" Z2M_DEVICE_IDENTIFIER_DOMAIN = "zigbee2mqtt" diff --git a/tests/chargers/test_wallbox_charger.py b/tests/chargers/test_wallbox_charger.py new file mode 100644 index 0000000..7c40168 --- /dev/null +++ b/tests/chargers/test_wallbox_charger.py @@ -0,0 +1,269 @@ +"""Tests for the Wallbox charger implementation.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from homeassistant.helpers.device_registry import DeviceEntry +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.evse_load_balancer.chargers.charger import PhaseMode +from custom_components.evse_load_balancer.chargers.wallbox_charger import ( + WallboxCharger, + WallboxEntityMap, + WallboxStatusMap, +) +from custom_components.evse_load_balancer.const import CHARGER_DOMAIN_WALLBOX, Phase + + +@pytest.fixture +def mock_hass(): + """Create a mock HomeAssistant instance for testing.""" + hass = MagicMock() + hass.services = MagicMock() + hass.services.async_call = AsyncMock() + return hass + + +@pytest.fixture +def mock_config_entry(): + """Create a mock ConfigEntry for the tests.""" + return MockConfigEntry( + domain="evse_load_balancer", + title="Wallbox Test Charger", + data={}, + unique_id="test_wallbox_charger", + ) + + +@pytest.fixture +def mock_device_entry(): + """Create a mock DeviceEntry object for testing.""" + device_entry = MagicMock(spec=DeviceEntry) + device_entry.id = "wallbox_123" + device_entry.identifiers = {(CHARGER_DOMAIN_WALLBOX, "test_charger")} + return device_entry + + +@pytest.fixture +def wallbox_charger(mock_hass, mock_config_entry, mock_device_entry): + """Create a WallboxCharger instance for testing.""" + with patch( + "custom_components.evse_load_balancer.chargers.wallbox_charger.WallboxCharger.refresh_entities" + ): + charger = WallboxCharger( + hass=mock_hass, config_entry=mock_config_entry, device_entry=mock_device_entry + ) + charger._get_entity_state_by_translation_key = MagicMock() + charger._get_entity_id_by_translation_key = MagicMock() + return charger + + +def test_is_charger_device_true(mock_device_entry): + """Test is_charger_device returns True for Wallbox devices.""" + assert WallboxCharger.is_charger_device(mock_device_entry) is True + + +def test_is_charger_device_false(): + """Test is_charger_device returns False for non-Wallbox devices.""" + device = MagicMock(spec=DeviceEntry) + device.identifiers = {("other_domain", "test_charger")} + assert WallboxCharger.is_charger_device(device) is False + + +async def test_set_current_limit(wallbox_charger, mock_hass): + """Test setting current limits on the Wallbox charger.""" + test_limits = {Phase.L1: 16, Phase.L2: 14, Phase.L3: 15} + entity_id = "number.wallbox_test_maximum_charging_current" + wallbox_charger._get_entity_id_by_translation_key.return_value = entity_id + + await wallbox_charger.set_current_limit(test_limits) + + mock_hass.services.async_call.assert_called_once_with( + domain="number", + service="set_value", + service_data={"entity_id": entity_id, "value": 14}, + blocking=True, + ) + + +def test_get_current_limit_success(wallbox_charger): + """Test retrieving the current limit when entity exists.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "16" + + result = wallbox_charger.get_current_limit() + assert result == {Phase.L1: 16, Phase.L2: 16, Phase.L3: 16} + wallbox_charger._get_entity_state_by_translation_key.assert_called_with( + WallboxEntityMap.DynamicChargerLimit + ) + + +def test_get_current_limit_float_value(wallbox_charger): + """Test retrieving the current limit when entity returns float string.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "16.5" + + result = wallbox_charger.get_current_limit() + assert result == {Phase.L1: 16, Phase.L2: 16, Phase.L3: 16} + + +def test_get_current_limit_missing(wallbox_charger): + """Test retrieving the current limit when entity doesn't exist.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = None + + result = wallbox_charger.get_current_limit() + assert result is None + + +def test_get_max_current_limit_success(wallbox_charger): + """Test retrieving the max current limit when entity exists.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "32" + + result = wallbox_charger.get_max_current_limit() + assert result == {Phase.L1: 32, Phase.L2: 32, Phase.L3: 32} + wallbox_charger._get_entity_state_by_translation_key.assert_called_with( + WallboxEntityMap.MaxChargerLimit + ) + + +def test_get_max_current_limit_float_value(wallbox_charger): + """Test retrieving the max current limit when entity returns float string.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = "32.0" + + result = wallbox_charger.get_max_current_limit() + assert result == {Phase.L1: 32, Phase.L2: 32, Phase.L3: 32} + + +def test_get_max_current_limit_missing(wallbox_charger): + """Test retrieving the max current limit when entity doesn't exist.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = None + + result = wallbox_charger.get_max_current_limit() + assert result is None + + +def test_has_synced_phase_limits(wallbox_charger): + """Test that Wallbox charger always has synced phase limits.""" + assert wallbox_charger.has_synced_phase_limits() is True + + +def test_car_connected_true(wallbox_charger): + """Test car_connected returns True for connected statuses.""" + for status in [ + WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.Paused, + WallboxStatusMap.Scheduled, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Waiting, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.car_connected() is True, f"Expected car_connected=True for {status}" + + +def test_car_connected_false(wallbox_charger): + """Test car_connected returns False for disconnected statuses.""" + for status in [ + WallboxStatusMap.Ready, + WallboxStatusMap.Disconnected, + WallboxStatusMap.Error, + WallboxStatusMap.Locked, + WallboxStatusMap.Updating, + WallboxStatusMap.Unknown, + None, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.car_connected() is False, f"Expected car_connected=False for {status}" + + +def test_can_charge_true(wallbox_charger): + """Test can_charge returns True for chargeable statuses.""" + for status in [ + WallboxStatusMap.Charging, + WallboxStatusMap.Discharging, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Paused, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.can_charge() is True, f"Expected can_charge=True for {status}" + + +def test_can_charge_false(wallbox_charger): + """Test can_charge returns False for non-chargeable statuses.""" + for status in [ + WallboxStatusMap.Ready, + WallboxStatusMap.Disconnected, + WallboxStatusMap.Error, + WallboxStatusMap.Locked, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.Scheduled, + WallboxStatusMap.Waiting, + WallboxStatusMap.Updating, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + WallboxStatusMap.Unknown, + None, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.can_charge() is False, f"Expected can_charge=False for {status}" + + +def test_is_charging_true(wallbox_charger): + """Test is_charging returns True when actively charging.""" + wallbox_charger._get_entity_state_by_translation_key.return_value = WallboxStatusMap.Charging + assert wallbox_charger.is_charging() is True + + +def test_is_charging_false(wallbox_charger): + """Test is_charging returns False for non-charging statuses.""" + for status in [ + WallboxStatusMap.Discharging, + WallboxStatusMap.Paused, + WallboxStatusMap.Scheduled, + WallboxStatusMap.WaitingForCarDemand, + WallboxStatusMap.Waiting, + WallboxStatusMap.Disconnected, + WallboxStatusMap.Error, + WallboxStatusMap.Ready, + WallboxStatusMap.Locked, + WallboxStatusMap.LockedCarConnected, + WallboxStatusMap.Updating, + WallboxStatusMap.WaitingInQueuePowerSharing, + WallboxStatusMap.WaitingInQueuePowerBoost, + WallboxStatusMap.WaitingMidFailed, + WallboxStatusMap.WaitingMidSafety, + WallboxStatusMap.WaitingInQueueEcoSmart, + WallboxStatusMap.Unknown, + None, + ]: + wallbox_charger._get_entity_state_by_translation_key.return_value = status + assert wallbox_charger.is_charging() is False, f"Expected is_charging=False for {status}" + + +def test_set_phase_mode_valid(wallbox_charger): + """Test set_phase_mode accepts valid PhaseMode values.""" + wallbox_charger.set_phase_mode(PhaseMode.SINGLE, Phase.L1) + wallbox_charger.set_phase_mode(PhaseMode.MULTI, Phase.L1) + + +def test_set_phase_mode_invalid(wallbox_charger): + """Test set_phase_mode rejects invalid values.""" + with pytest.raises(ValueError, match="Invalid mode"): + wallbox_charger.set_phase_mode("invalid_mode", Phase.L1) + + +async def test_async_setup(wallbox_charger): + """Test async_setup method (no-op).""" + await wallbox_charger.async_setup() + + +async def test_async_unload(wallbox_charger): + """Test async_unload method (no-op).""" + await wallbox_charger.async_unload()