Add Lock platform to wallbox (#68414)

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

homeassistant/components/wallbox/__init__.py

@@ -22,6 +22,7 @@ from ...helpers.entity import DeviceInfo
 from .const import (
     CONF_CURRENT_VERSION_KEY,
     CONF_DATA_KEY,
+    CONF_LOCKED_UNLOCKED_KEY,
     CONF_MAX_CHARGING_CURRENT_KEY,
     CONF_NAME_KEY,
     CONF_PART_NUMBER_KEY,
@@ -33,7 +34,7 @@ from .const import (
 
 _LOGGER = logging.getLogger(__name__)
 
-PLATFORMS = [Platform.SENSOR, Platform.NUMBER]
+PLATFORMS = [Platform.SENSOR, Platform.NUMBER, Platform.LOCK]
 UPDATE_INTERVAL = 30
 
 
@@ -70,6 +71,10 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 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:
@@ -78,12 +83,19 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             data[CONF_MAX_CHARGING_CURRENT_KEY] = data[CONF_DATA_KEY][
                 CONF_MAX_CHARGING_CURRENT_KEY
             ]
+            data[CONF_LOCKED_UNLOCKED_KEY] = data[CONF_DATA_KEY][
+                CONF_LOCKED_UNLOCKED_KEY
+            ]
 
             return data
 
         except requests.exceptions.HTTPError as wallbox_connection_error:
             raise ConnectionError 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:
@@ -101,14 +113,23 @@ class WallboxCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         )
         await self.async_request_refresh()
 
-    async def _async_update_data(self) -> dict[str, Any]:
+    def _set_lock_unlock(self, lock: bool) -> None:
-        """Get new sensor data for Wallbox component."""
+        """Set wallbox to locked or unlocked."""
-        data = await self.hass.async_add_executor_job(self._get_data)
+        try:
-        return data
+            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_validate_input(self) -> None:
+    async def async_set_lock_unlock(self, lock: bool) -> None:
-        """Get new sensor data for Wallbox component."""
+        """Set wallbox to locked or unlocked."""
-        await self.hass.async_add_executor_job(self._validate)
+        await self.hass.async_add_executor_job(self._set_lock_unlock, lock)
+        await self.async_request_refresh()
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

homeassistant/components/wallbox/const.py

@@ -18,6 +18,7 @@ CONF_PART_NUMBER_KEY = "part_number"
 CONF_SOFTWARE_KEY = "software"
 CONF_MAX_AVAILABLE_POWER_KEY = "max_available_power"
 CONF_MAX_CHARGING_CURRENT_KEY = "max_charging_current"
+CONF_LOCKED_UNLOCKED_KEY = "locked"
 CONF_NAME_KEY = "name"
 CONF_STATE_OF_CHARGE_KEY = "state_of_charge"
 CONF_STATUS_DESCRIPTION_KEY = "status_description"

homeassistant/components/wallbox/lock.py

@@ -0,0 +1,76 @@
+"""Home Assistant component for accessing the Wallbox Portal API. The lock component creates a lock entity."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.lock import LockEntity, LockEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import InvalidAuth, WallboxCoordinator, WallboxEntity
+from .const import (
+    CONF_DATA_KEY,
+    CONF_LOCKED_UNLOCKED_KEY,
+    CONF_SERIAL_NUMBER_KEY,
+    DOMAIN,
+)
+
+LOCK_TYPES: dict[str, LockEntityDescription] = {
+    CONF_LOCKED_UNLOCKED_KEY: LockEntityDescription(
+        key=CONF_LOCKED_UNLOCKED_KEY,
+        name="Locked/Unlocked",
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Create wallbox lock entities in HASS."""
+    coordinator: WallboxCoordinator = hass.data[DOMAIN][entry.entry_id]
+    # Check if the user is authorized to lock, if so, add lock component
+    try:
+        await coordinator.async_set_lock_unlock(
+            coordinator.data[CONF_LOCKED_UNLOCKED_KEY]
+        )
+    except InvalidAuth:
+        return
+
+    async_add_entities(
+        [
+            WallboxLock(coordinator, entry, description)
+            for ent in coordinator.data
+            if (description := LOCK_TYPES.get(ent))
+        ]
+    )
+
+
+class WallboxLock(WallboxEntity, LockEntity):
+    """Representation of a wallbox lock."""
+
+    def __init__(
+        self,
+        coordinator: WallboxCoordinator,
+        entry: ConfigEntry,
+        description: LockEntityDescription,
+    ) -> None:
+        """Initialize a Wallbox lock."""
+
+        super().__init__(coordinator)
+        self.entity_description = description
+        self._attr_name = f"{entry.title} {description.name}"
+        self._attr_unique_id = f"{description.key}-{coordinator.data[CONF_DATA_KEY][CONF_SERIAL_NUMBER_KEY]}"
+
+    @property
+    def is_locked(self) -> bool:
+        """Return the status of the lock."""
+        return self.coordinator.data[CONF_LOCKED_UNLOCKED_KEY]  # type: ignore[no-any-return]
+
+    async def async_lock(self, **kwargs: Any) -> None:
+        """Lock charger."""
+        await self.coordinator.async_set_lock_unlock(True)
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Unlock charger."""
+        await self.coordinator.async_set_lock_unlock(False)

tests/components/wallbox/__init__.py

@@ -12,6 +12,7 @@ from homeassistant.components.wallbox.const import (
     CONF_CHARGING_SPEED_KEY,
     CONF_CURRENT_VERSION_KEY,
     CONF_DATA_KEY,
+    CONF_LOCKED_UNLOCKED_KEY,
     CONF_MAX_AVAILABLE_POWER_KEY,
     CONF_MAX_CHARGING_CURRENT_KEY,
     CONF_NAME_KEY,
@@ -38,6 +39,7 @@ test_response = json.loads(
             CONF_NAME_KEY: "WallboxName",
             CONF_DATA_KEY: {
                 CONF_MAX_CHARGING_CURRENT_KEY: 24,
+                CONF_LOCKED_UNLOCKED_KEY: False,
                 CONF_SERIAL_NUMBER_KEY: "20000",
                 CONF_PART_NUMBER_KEY: "PLP1-0-2-4-9-002-E",
                 CONF_SOFTWARE_KEY: {CONF_CURRENT_VERSION_KEY: "5.5.10"},

tests/components/wallbox/const.py

@@ -6,6 +6,7 @@ CONF_ERROR = "error"
 CONF_STATUS = "status"
 
 CONF_MOCK_NUMBER_ENTITY_ID = "number.mock_title_max_charging_current"
+CONF_MOCK_LOCK_ENTITY_ID = "lock.mock_title_locked_unlocked"
 CONF_MOCK_SENSOR_CHARGING_SPEED_ID = "sensor.mock_title_charging_speed"
 CONF_MOCK_SENSOR_CHARGING_POWER_ID = "sensor.mock_title_charging_power"
 CONF_MOCK_SENSOR_MAX_AVAILABLE_POWER = "sensor.mock_title_max_available_power"

tests/components/wallbox/test_lock.py

@@ -0,0 +1,129 @@
+"""Test Wallbox Lock component."""
+import json
+
+import pytest
+import requests_mock
+
+from homeassistant.components.lock import SERVICE_LOCK, SERVICE_UNLOCK
+from homeassistant.components.wallbox import CONF_LOCKED_UNLOCKED_KEY
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import HomeAssistant
+
+from tests.components.wallbox import (
+    entry,
+    setup_integration,
+    setup_integration_read_only,
+)
+from tests.components.wallbox.const import (
+    CONF_ERROR,
+    CONF_JWT,
+    CONF_MOCK_LOCK_ENTITY_ID,
+    CONF_STATUS,
+    CONF_TTL,
+    CONF_USER_ID,
+)
+
+authorisation_response = json.loads(
+    json.dumps(
+        {
+            CONF_JWT: "fakekeyhere",
+            CONF_USER_ID: 12345,
+            CONF_TTL: 145656758,
+            CONF_ERROR: "false",
+            CONF_STATUS: 200,
+        }
+    )
+)
+
+
+async def test_wallbox_lock_class(hass: HomeAssistant):
+    """Test wallbox lock class."""
+
+    await setup_integration(hass)
+
+    state = hass.states.get(CONF_MOCK_LOCK_ENTITY_ID)
+    assert state
+    assert state.state == "unlocked"
+
+    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.put(
+            "https://api.wall-box.com/v2/charger/12345",
+            json=json.loads(json.dumps({CONF_LOCKED_UNLOCKED_KEY: False})),
+            status_code=200,
+        )
+
+        await hass.services.async_call(
+            "lock",
+            SERVICE_LOCK,
+            {
+                ATTR_ENTITY_ID: CONF_MOCK_LOCK_ENTITY_ID,
+            },
+            blocking=True,
+        )
+
+        await hass.services.async_call(
+            "lock",
+            SERVICE_UNLOCK,
+            {
+                ATTR_ENTITY_ID: CONF_MOCK_LOCK_ENTITY_ID,
+            },
+            blocking=True,
+        )
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+async def test_wallbox_lock_class_connection_error(hass: HomeAssistant):
+    """Test wallbox lock 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.put(
+            "https://api.wall-box.com/v2/charger/12345",
+            json=json.loads(json.dumps({CONF_LOCKED_UNLOCKED_KEY: False})),
+            status_code=404,
+        )
+
+        with pytest.raises(ConnectionError):
+            await hass.services.async_call(
+                "lock",
+                SERVICE_LOCK,
+                {
+                    ATTR_ENTITY_ID: CONF_MOCK_LOCK_ENTITY_ID,
+                },
+                blocking=True,
+            )
+        with pytest.raises(ConnectionError):
+            await hass.services.async_call(
+                "lock",
+                SERVICE_UNLOCK,
+                {
+                    ATTR_ENTITY_ID: CONF_MOCK_LOCK_ENTITY_ID,
+                },
+                blocking=True,
+            )
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+async def test_wallbox_lock_class_authentication_error(hass: HomeAssistant):
+    """Test wallbox lock not loaded on authentication error."""
+
+    await setup_integration_read_only(hass)
+
+    state = hass.states.get(CONF_MOCK_LOCK_ENTITY_ID)
+
+    assert state is None
+
+    await hass.config_entries.async_unload(entry.entry_id)