Wallbox set maximum charging current as number (torbennehmer#165)

Based on the discussion in issue
torbennehmer#47 and requested by
@Wolfgang-03,
This PR implements the functionality to set the maximum charging current
for a wallbox via a number Entity (Which allows sliders without
helpers).

Under the hood, this PR brings the following features:
- Introduces E3DCNumber which allows to call methods on-change. (Will be
handy for setting other E3DC values in the future)
- Reads & respects the hard upper and lower current limits as stored in
the wallbox. These limits can only be set by the E3DC electrician during
installation, so this is now more fail-safe compared to the previously
hardcoded limits.

For reference, see the two videos:
First one shows how by changing the number, the sensor which reads back
from the E3DC the set current updates.

https://github.com/user-attachments/assets/595fbdda-e6d0-4a9e-b422-128d929c4a8b

2nd one shows how Number as well as Sensor update, when changing the max
charging current from somewhere else
(In this case i had a 2nd HA instance running on another machine and
changed from there max charging current.

https://github.com/user-attachments/assets/919f8ad6-c7d0-4afc-b6b9-88569befd37b

custom_components/e3dc_rscp/const.py

@@ -11,12 +11,12 @@
 SERVICE_SET_POWER_LIMITS = "set_power_limits"
 SERVICE_MANUAL_CHARGE = "manual_charge"
 SERVICE_SET_WALLBOX_MAX_CHARGE_CURRENT = "set_wallbox_max_charge_current"
-MAX_CHARGE_CURRENT = 32  # Maximum allowed wallbox charging current in Amperes
 MAX_WALLBOXES_POSSIBLE = 8 # 8 is the maximum according to RSCP Specification
 
 PLATFORMS: list[Platform] = [
     Platform.BINARY_SENSOR,
     Platform.SENSOR,
     Platform.SWITCH,
     Platform.BUTTON,
+    Platform.NUMBER
 ]

custom_components/e3dc_rscp/coordinator.py

@@ -3,7 +3,7 @@
 from datetime import timedelta, datetime
 import logging
 from time import time
-from typing import Any
+from typing import Any, TypedDict
 import pytz
 import re
 
@@ -18,14 +18,24 @@
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 from homeassistant.components.sensor import SensorStateClass
 
-from .const import DOMAIN, MAX_CHARGE_CURRENT, MAX_WALLBOXES_POSSIBLE
+from .const import DOMAIN, MAX_WALLBOXES_POSSIBLE
 
 from .e3dc_proxy import E3DCProxy
 
 _LOGGER = logging.getLogger(__name__)
 _STAT_REFRESH_INTERVAL = 60
 
 
+
+class E3DCWallbox(TypedDict):
+    """E3DC Wallbox, keeps general information, attributes and identity data for an individual wallbox."""
+
+    index: int
+    key: str
+    deviceInfo: DeviceInfo
+    lowerCurrentLimit: int
+    upperCurrentLimit: int
+
 class E3DCCoordinator(DataUpdateCoordinator[dict[str, Any]]):
     """E3DC Coordinator, fetches all relevant data and provides proxies for all service calls."""
 
@@ -38,7 +48,7 @@ def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
         self._sw_version: str = ""
         self._update_guard_powersettings: bool = False
         self._update_guard_wallboxsettings: bool = False
-        self._wallboxes: list[dict[str, str | int]] = []
+        self._wallboxes: list[E3DCWallbox] = []
         self._timezone_offset: int = 0
         self._next_stat_update: float = 0
 
@@ -115,26 +125,45 @@ async def async_identify_wallboxes(self, hass: HomeAssistant):
                     configuration_url="https://my.e3dc.com/",
                 )
 
-                wallbox = {
+                wallbox: E3DCWallbox = {
                     "index": wallbox_index,
                     "key": unique_id,
-                    "deviceInfo": deviceInfo
+                    "deviceInfo": deviceInfo,
+                    "lowerCurrentLimit": request_data["lowerCurrentLimit"],
+                    "upperCurrentLimit": request_data["upperCurrentLimit"]
                 }
                 self.wallboxes.append(wallbox)
             else:
                 _LOGGER.debug("No Wallbox with index %s has been found", wallbox_index)
 
     # Getter for _wallboxes
     @property
-    def wallboxes(self) -> list[dict[str, str | int]]:
+    def wallboxes(self) -> list[E3DCWallbox]:
         """Get the list of wallboxes."""
         return self._wallboxes
 
-    # Setter for _wallboxes
-    @wallboxes.setter
-    def wallboxes(self, value: list[dict[str, str | int]]) -> None:
-        """Set the list of wallboxes."""
-        self._wallboxes = value
+    # Setter for individual wallbox values
+    def setWallboxValue(self, index: int, key: str, value: Any) -> None:
+        """Set the value for a specific key in a wallbox identified by its index."""
+        for wallbox in self._wallboxes:
+            if wallbox['index'] == index:
+                wallbox[key] = value
+                _LOGGER.debug(f"Set {key} to {value} for wallbox with index {index}")
+                return
+        raise ValueError(f"Wallbox with index {index} not found")
+
+    # Getter for individual wallbox values
+    def getWallboxValue(self, index: int, key: str) -> Any:
+        """Get the value for a specific key in a wallbox identified by its index."""
+        for wallbox in self._wallboxes:
+            if wallbox['index'] == index:
+                value = wallbox.get(key)
+                if value is not None:
+                    _LOGGER.debug(f"Got {key} value {value} for wallbox with index {index}")
+                    return value
+                else:
+                    raise KeyError(f"Key {key} not found in wallbox with index {index}")
+        raise ValueError(f"Wallbox with index {index} not found")
 
     async def _async_connect_additional_powermeters(self):
         """Identify the installed powermeters and reconnect to E3DC with this config."""
@@ -534,9 +563,16 @@ async def async_set_wallbox_max_charge_current(self, current: int | None, wallbo
                 "async_set_wallbox_max_charge_current must be called with a valid wallbox id."
             )
 
-        if current > MAX_CHARGE_CURRENT:
-            _LOGGER.warning("Limiting current to %s", MAX_CHARGE_CURRENT)
-            current = MAX_CHARGE_CURRENT
+        upperCurrentLimit = self.getWallboxValue(wallbox_index, "upperCurrentLimit")
+        if current > upperCurrentLimit:
+            _LOGGER.warning("Requested Wallbox current of %s is too high. Limiting current to %s", current, upperCurrentLimit)
+            current = upperCurrentLimit
+
+        lowerCurrentLimit = self.getWallboxValue(wallbox_index, "lowerCurrentLimit")
+        if current < lowerCurrentLimit:
+            _LOGGER.warning("Requested Wallbox current of %s is too low. Limiting current to %s", current, lowerCurrentLimit)
+            current = lowerCurrentLimit
+
 
         _LOGGER.debug("Setting wallbox max charge current to %s", current)
 

custom_components/e3dc_rscp/e3dc_proxy.py

@@ -14,7 +14,7 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
 
-from .const import CONF_RSCPKEY, MAX_CHARGE_CURRENT
+from .const import CONF_RSCPKEY
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -174,6 +174,8 @@ def get_wallbox_identification_data(self, wallbox_index: int) -> dict[str, Any]:
                     ("WB_REQ_DEVICE_NAME", "None", None),
                     ("WB_REQ_SERIAL", "None", None),
                     ("WB_REQ_WALLBOX_TYPE", "None", None),
+                    ("WB_REQ_LOWER_CURRENT_LIMIT", "None", None),
+                    ("WB_REQ_UPPER_CURRENT_LIMIT", "None", None)
                 ],
             ),
             keepAlive=True,
@@ -203,6 +205,14 @@ def get_wallbox_identification_data(self, wallbox_index: int) -> dict[str, Any]:
         if wallbox_type is not None:
             outObj["wallboxType"] = rscpFindTagIndex(wallbox_type, "WB_WALLBOX_TYPE")
 
+        lower_current_limit = rscpFindTag(req, "WB_LOWER_CURRENT_LIMIT")
+        if lower_current_limit is not None:
+            outObj["lowerCurrentLimit"] = rscpFindTagIndex(lower_current_limit, "WB_LOWER_CURRENT_LIMIT")
+
+        upper_current_limit = rscpFindTag(req, "WB_UPPER_CURRENT_LIMIT")
+        if upper_current_limit is not None:
+            outObj["upperCurrentLimit"] = rscpFindTagIndex(upper_current_limit, "WB_UPPER_CURRENT_LIMIT")
+
         return outObj
 
 
@@ -356,10 +366,7 @@ def set_wallbox_max_charge_current(
             False if error
 
         """
-
-        if max_charge_current > MAX_CHARGE_CURRENT:
-            _LOGGER.warning("Limiting max_charge_current to %s", MAX_CHARGE_CURRENT)
-            max_charge_current = MAX_CHARGE_CURRENT
+        _LOGGER.debug("Wallbox %s: Setting max_charge_current to %s", wallbox_index, max_charge_current)
 
         return self.e3dc.set_wallbox_max_charge_current(
             max_charge_current=max_charge_current, wbIndex=wallbox_index, keepAlive=True

custom_components/e3dc_rscp/number.py

@@ -0,0 +1,110 @@
+"""E3DC Number platform."""
+
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+import logging
+from typing import Any
+
+from homeassistant.components.number import (
+    NumberDeviceClass,
+    NumberEntity,
+    NumberEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.const import EntityCategory
+
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import E3DCCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+@dataclass
+class E3DCNumberEntityDescription(NumberEntityDescription):
+    """Derived helper for advanced configs."""
+
+    async_set_native_value_action: Callable[
+        [E3DCCoordinator, float, int], Coroutine[Any, Any, bool]
+    ] | None = None
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Initialize Number Platform."""
+    assert isinstance(entry.unique_id, str)
+    coordinator: E3DCCoordinator = hass.data[DOMAIN][entry.unique_id]
+    entities: list[E3DCNumber] = []
+
+    # Add Number descriptions for wallboxes
+    for wallbox in coordinator.wallboxes:
+        unique_id = list(wallbox["deviceInfo"]["identifiers"])[0][1]
+        wallbox_key = wallbox["key"]
+
+        wallbox_charge_current_limit_description = E3DCNumberEntityDescription(
+            key=f"{wallbox_key}-max-charge-current",
+            translation_key="wallbox-max-charge-current",
+            name="Wallbox Max Charge Current",
+            icon="mdi:current-ac",
+            native_min_value=wallbox["lowerCurrentLimit"],
+            native_max_value=wallbox["upperCurrentLimit"],
+            native_step=1,
+            device_class=NumberDeviceClass.CURRENT,
+            entity_category=EntityCategory.CONFIG,
+            native_unit_of_measurement="A",
+            async_set_native_value_action=lambda coordinator, value, index=wallbox["index"]: coordinator.async_set_wallbox_max_charge_current(int(value), index),
+        )
+        entities.append(E3DCNumber(coordinator, wallbox_charge_current_limit_description, unique_id, wallbox["deviceInfo"]))
+
+    async_add_entities(entities)
+
+
+class E3DCNumber(CoordinatorEntity, NumberEntity):
+    """Custom E3DC Number Implementation."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: E3DCCoordinator,
+        description: E3DCNumberEntityDescription,
+        uid: str,
+        device_info: DeviceInfo | None = None
+    ) -> None:
+        """Initialize the Number."""
+        super().__init__(coordinator)
+        self.coordinator: E3DCCoordinator = coordinator
+        self.entity_description: E3DCNumberEntityDescription = description
+        self._attr_value = self.coordinator.data.get(self.entity_description.key)
+        self._attr_unique_id = f"{uid}_{description.key}"
+        if device_info is not None:
+            self._deviceInfo = device_info
+        else:
+            self._deviceInfo = self.coordinator.device_info()
+
+    @property
+    def native_value(self):
+        """Return the current value."""
+        return self._attr_value
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Process coordinator updates."""
+        self._attr_value = self.coordinator.data.get(self.entity_description.key)
+        self.async_write_ha_state()
+
+    async def async_set_native_value(self, value: float) -> None:
+        """Set the number value asynchronously."""
+        if self.entity_description.async_set_native_value_action is not None:
+            self._attr_value = value
+            self.async_write_ha_state()
+            await self.entity_description.async_set_native_value_action(self.coordinator, value)
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return the device information."""
+        return self._deviceInfo

custom_components/e3dc_rscp/strings.json

@@ -270,6 +270,11 @@
       "wallbox-toggle-wallbox-phases": {
         "name": "Phases"
       }
+    },
+    "number": {
+      "wallbox-max-charge-current": {
+        "name": "Max charge current"
+      }
     }
   },
   "services": {

custom_components/e3dc_rscp/translations/en.json

@@ -270,6 +270,11 @@
             "wallbox-toggle-wallbox-phases": {
                 "name": "Phases"
             }
+        },
+        "number": {
+            "wallbox-max-charge-current": {
+                "name": "Max charge current"
+            }
         }
     },
     "services": {