Add Switch platform to Wallbox (#70584)

Co-authored-by: Marc Mueller <30130371+cdce8p@users.noreply.github.com>

homeassistant/components/wallbox/__init__.py

@@ -32,44 +32,45 @@ from .const import (
     CHARGER_STATUS_ID_KEY,
     CONF_STATION,
     DOMAIN,
+    ChargerStatus,
 )
 
 _LOGGER = logging.getLogger(__name__)
 
-PLATFORMS = [Platform.SENSOR, Platform.NUMBER, Platform.LOCK]
+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, str] = {
+CHARGER_STATUS: dict[int, ChargerStatus] = {
-    0: "Disconnected",
+    0: ChargerStatus.DISCONNECTED,
-    14: "Error",
+    14: ChargerStatus.ERROR,
-    15: "Error",
+    15: ChargerStatus.ERROR,
-    161: "Ready",
+    161: ChargerStatus.READY,
-    162: "Ready",
+    162: ChargerStatus.READY,
-    163: "Disconnected",
+    163: ChargerStatus.DISCONNECTED,
-    164: "Waiting",
+    164: ChargerStatus.WAITING,
-    165: "Locked",
+    165: ChargerStatus.LOCKED,
-    166: "Updating",
+    166: ChargerStatus.UPDATING,
-    177: "Scheduled",
+    177: ChargerStatus.SCHEDULED,
-    178: "Paused",
+    178: ChargerStatus.PAUSED,
-    179: "Scheduled",
+    179: ChargerStatus.SCHEDULED,
-    180: "Waiting for car demand",
+    180: ChargerStatus.WAITING_FOR_CAR,
-    181: "Waiting for car demand",
+    181: ChargerStatus.WAITING_FOR_CAR,
-    182: "Paused",
+    182: ChargerStatus.PAUSED,
-    183: "Waiting in queue by Power Sharing",
+    183: ChargerStatus.WAITING_IN_QUEUE_POWER_SHARING,
-    184: "Waiting in queue by Power Sharing",
+    184: ChargerStatus.WAITING_IN_QUEUE_POWER_SHARING,
-    185: "Waiting in queue by Power Boost",
+    185: ChargerStatus.WAITING_IN_QUEUE_POWER_BOOST,
-    186: "Waiting in queue by Power Boost",
+    186: ChargerStatus.WAITING_IN_QUEUE_POWER_BOOST,
-    187: "Waiting MID failed",
+    187: ChargerStatus.WAITING_MID_FAILED,
-    188: "Waiting MID safety margin exceeded",
+    188: ChargerStatus.WAITING_MID_SAFETY,
-    189: "Waiting in queue by Eco-Smart",
+    189: ChargerStatus.WAITING_IN_QUEUE_ECO_SMART,
-    193: "Charging",
+    193: ChargerStatus.CHARGING,
-    194: "Charging",
+    194: ChargerStatus.CHARGING,
-    195: "Charging",
+    195: ChargerStatus.CHARGING,
-    196: "Discharging",
+    196: ChargerStatus.DISCHARGING,
-    209: "Locked",
+    209: ChargerStatus.LOCKED,
-    210: "Locked",
+    210: ChargerStatus.LOCKED,
 }
 
 
@@ -122,7 +123,7 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 CHARGER_LOCKED_UNLOCKED_KEY
             ]
             data[CHARGER_STATUS_DESCRIPTION_KEY] = CHARGER_STATUS.get(
-                data[CHARGER_STATUS_ID_KEY], "Unknown"
+                data[CHARGER_STATUS_ID_KEY], ChargerStatus.UNKNOWN
             )
 
             return data
@@ -169,6 +170,24 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         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:
     """Set up Wallbox from a config entry."""

homeassistant/components/wallbox/const.py

@@ -1,4 +1,5 @@
 """Constants for the Wallbox integration."""
+from homeassistant.backports.enum import StrEnum
 
 DOMAIN = "wallbox"
 
@@ -18,9 +19,32 @@ CHARGER_PART_NUMBER_KEY = "part_number"
 CHARGER_SOFTWARE_KEY = "software"
 CHARGER_MAX_AVAILABLE_POWER_KEY = "max_available_power"
 CHARGER_MAX_CHARGING_CURRENT_KEY = "max_charging_current"
+CHARGER_PAUSE_RESUME_KEY = "paused"
 CHARGER_LOCKED_UNLOCKED_KEY = "locked"
 CHARGER_NAME_KEY = "name"
 CHARGER_STATE_OF_CHARGE_KEY = "state_of_charge"
 CHARGER_STATUS_ID_KEY = "status_id"
 CHARGER_STATUS_DESCRIPTION_KEY = "status_description"
 CHARGER_CONNECTIONS = "connections"
+
+
+class ChargerStatus(StrEnum):
+    """Charger Status Description."""
+
+    CHARGING = "Charging"
+    DISCHARGING = "Discharging"
+    PAUSED = "Paused"
+    SCHEDULED = "Scheduled"
+    WAITING_FOR_CAR = "Waiting for car demand"
+    WAITING = "Waiting"
+    DISCONNECTED = "Disconnected"
+    ERROR = "Error"
+    READY = "Ready"
+    LOCKED = "Locked"
+    UPDATING = "Updating"
+    WAITING_IN_QUEUE_POWER_SHARING = "Waiting in queue by Power Sharing"
+    WAITING_IN_QUEUE_POWER_BOOST = "Waiting in queue by Power Boost"
+    WAITING_MID_FAILED = "Waiting MID failed"
+    WAITING_MID_SAFETY = "Waiting MID safety margin exceeded"
+    WAITING_IN_QUEUE_ECO_SMART = "Waiting in queue by Eco-Smart"
+    UNKNOWN = "Unknown"

homeassistant/components/wallbox/switch.py

@@ -0,0 +1,78 @@
+"""Home Assistant component for accessing the Wallbox Portal API. The switch component creates a switch entity."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import WallboxCoordinator, WallboxEntity
+from .const import (
+    CHARGER_DATA_KEY,
+    CHARGER_PAUSE_RESUME_KEY,
+    CHARGER_SERIAL_NUMBER_KEY,
+    CHARGER_STATUS_DESCRIPTION_KEY,
+    DOMAIN,
+    ChargerStatus,
+)
+
+SWITCH_TYPES: dict[str, SwitchEntityDescription] = {
+    CHARGER_PAUSE_RESUME_KEY: SwitchEntityDescription(
+        key=CHARGER_PAUSE_RESUME_KEY,
+        name="Pause/Resume",
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Create wallbox sensor entities in HASS."""
+    coordinator: WallboxCoordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities(
+        [WallboxSwitch(coordinator, entry, SWITCH_TYPES[CHARGER_PAUSE_RESUME_KEY])]
+    )
+
+
+class WallboxSwitch(WallboxEntity, SwitchEntity):
+    """Representation of the Wallbox portal."""
+
+    def __init__(
+        self,
+        coordinator: WallboxCoordinator,
+        entry: ConfigEntry,
+        description: SwitchEntityDescription,
+    ) -> None:
+        """Initialize a Wallbox switch."""
+        super().__init__(coordinator)
+        self.entity_description = description
+        self._attr_name = f"{entry.title} {description.name}"
+        self._attr_unique_id = f"{description.key}-{coordinator.data[CHARGER_DATA_KEY][CHARGER_SERIAL_NUMBER_KEY]}"
+
+    @property
+    def available(self) -> bool:
+        """Return the availability of the switch."""
+        return self.coordinator.data[CHARGER_STATUS_DESCRIPTION_KEY] in {
+            ChargerStatus.CHARGING,
+            ChargerStatus.PAUSED,
+            ChargerStatus.SCHEDULED,
+        }
+
+    @property
+    def is_on(self) -> bool:
+        """Return the status of pause/resume."""
+        return self.coordinator.data[CHARGER_STATUS_DESCRIPTION_KEY] in {
+            ChargerStatus.CHARGING,
+            ChargerStatus.WAITING_FOR_CAR,
+            ChargerStatus.WAITING,
+        }
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Pause charger."""
+        await self.coordinator.async_pause_charger(True)
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Resume charger."""
+        await self.coordinator.async_pause_charger(False)

tests/components/wallbox/__init__.py

@@ -34,7 +34,7 @@ test_response = json.loads(
     json.dumps(
         {
             CHARGER_CHARGING_POWER_KEY: 0,
-            CHARGER_STATUS_ID_KEY: 161,
+            CHARGER_STATUS_ID_KEY: 193,
             CHARGER_MAX_AVAILABLE_POWER_KEY: 25.2,
             CHARGER_CHARGING_SPEED_KEY: 0,
             CHARGER_ADDED_RANGE_KEY: 150,

tests/components/wallbox/const.py

@@ -10,3 +10,4 @@ MOCK_LOCK_ENTITY_ID = "lock.mock_title_locked_unlocked"
 MOCK_SENSOR_CHARGING_SPEED_ID = "sensor.mock_title_charging_speed"
 MOCK_SENSOR_CHARGING_POWER_ID = "sensor.mock_title_charging_power"
 MOCK_SENSOR_MAX_AVAILABLE_POWER = "sensor.mock_title_max_available_power"
+MOCK_SWITCH_ENTITY_ID = "switch.mock_title_pause_resume"

tests/components/wallbox/test_switch.py

@@ -0,0 +1,153 @@
+"""Test Wallbox Lock component."""
+import json
+
+import pytest
+import requests_mock
+
+from homeassistant.components.switch import SERVICE_TURN_OFF, SERVICE_TURN_ON
+from homeassistant.components.wallbox import InvalidAuth
+from homeassistant.components.wallbox.const import CHARGER_STATUS_ID_KEY
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import HomeAssistant
+
+from tests.components.wallbox import entry, setup_integration
+from tests.components.wallbox.const import (
+    ERROR,
+    JWT,
+    MOCK_SWITCH_ENTITY_ID,
+    STATUS,
+    TTL,
+    USER_ID,
+)
+
+authorisation_response = json.loads(
+    json.dumps(
+        {
+            JWT: "fakekeyhere",
+            USER_ID: 12345,
+            TTL: 145656758,
+            ERROR: "false",
+            STATUS: 200,
+        }
+    )
+)
+
+
+async def test_wallbox_switch_class(hass: HomeAssistant) -> None:
+    """Test wallbox switch class."""
+
+    await setup_integration(hass)
+
+    state = hass.states.get(MOCK_SWITCH_ENTITY_ID)
+    assert state
+    assert state.state == "on"
+
+    with requests_mock.Mocker() as mock_request:
+        mock_request.get(
+            "https://api.wall-box.com/auth/token/user",
+            json=authorisation_response,
+            status_code=200,
+        )
+        mock_request.post(
+            "https://api.wall-box.com/v3/chargers/12345/remote-action",
+            json=json.loads(json.dumps({CHARGER_STATUS_ID_KEY: 193})),
+            status_code=200,
+        )
+
+        await hass.services.async_call(
+            "switch",
+            SERVICE_TURN_ON,
+            {
+                ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
+            },
+            blocking=True,
+        )
+
+        await hass.services.async_call(
+            "switch",
+            SERVICE_TURN_OFF,
+            {
+                ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
+            },
+            blocking=True,
+        )
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+async def test_wallbox_switch_class_connection_error(hass: HomeAssistant) -> None:
+    """Test wallbox switch class connection error."""
+
+    await setup_integration(hass)
+
+    with requests_mock.Mocker() as mock_request:
+        mock_request.get(
+            "https://api.wall-box.com/auth/token/user",
+            json=authorisation_response,
+            status_code=200,
+        )
+        mock_request.post(
+            "https://api.wall-box.com/v3/chargers/12345/remote-action",
+            json=json.loads(json.dumps({CHARGER_STATUS_ID_KEY: 193})),
+            status_code=404,
+        )
+
+        with pytest.raises(ConnectionError):
+            await hass.services.async_call(
+                "switch",
+                SERVICE_TURN_ON,
+                {
+                    ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
+                },
+                blocking=True,
+            )
+        with pytest.raises(ConnectionError):
+            await hass.services.async_call(
+                "switch",
+                SERVICE_TURN_OFF,
+                {
+                    ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
+                },
+                blocking=True,
+            )
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+async def test_wallbox_switch_class_authentication_error(hass: HomeAssistant) -> None:
+    """Test wallbox switch class connection error."""
+
+    await setup_integration(hass)
+
+    with requests_mock.Mocker() as mock_request:
+        mock_request.get(
+            "https://api.wall-box.com/auth/token/user",
+            json=authorisation_response,
+            status_code=200,
+        )
+        mock_request.post(
+            "https://api.wall-box.com/v3/chargers/12345/remote-action",
+            json=json.loads(json.dumps({CHARGER_STATUS_ID_KEY: 193})),
+            status_code=403,
+        )
+
+        with pytest.raises(InvalidAuth):
+            await hass.services.async_call(
+                "switch",
+                SERVICE_TURN_ON,
+                {
+                    ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
+                },
+                blocking=True,
+            )
+        with pytest.raises(InvalidAuth):
+            await hass.services.async_call(
+                "switch",
+                SERVICE_TURN_OFF,
+                {
+                    ATTR_ENTITY_ID: MOCK_SWITCH_ENTITY_ID,
+                },
+                blocking=True,
+            )
+
+    await hass.config_entries.async_unload(entry.entry_id)