Merge pull request #17 from leeyuentuen/button-entity

Button entity + cleanup codes

custom_components/alfen_wallbox/__init__.py

@@ -17,6 +17,7 @@
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
 from .alfen import AlfenDevice
 
@@ -31,6 +32,7 @@
     Platform.BINARY_SENSOR,
     Platform.SWITCH,
     Platform.NUMBER,
+    Platform.BUTTON
 ]
 SCAN_INTERVAL = timedelta(seconds=60)
 
@@ -43,7 +45,7 @@ async def async_setup(hass: HomeAssistant, config: dict) -> bool:
     return True
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     conf = entry.data
     device = await alfen_setup(
         hass, conf[CONF_HOST], conf[CONF_NAME], conf[CONF_USERNAME], conf[CONF_PASSWORD]
@@ -61,7 +63,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
     return True
 
 
-async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
+async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Unload a config entry."""
     _LOGGER.debug("async_unload_entry: %s", config_entry)
 
@@ -74,10 +76,10 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
     return unload_ok
 
 
-async def alfen_setup(hass: HomeAssistant, host: str, name: str, username: str, password: str):
+async def alfen_setup(hass: HomeAssistant, host: str, name: str, username: str, password: str) -> AlfenDevice | None:
     """Create a Alfen instance only once."""
 
-    session = hass.helpers.aiohttp_client.async_get_clientsession()
+    session = async_get_clientsession(hass, verify_ssl=False)
     try:
         with timeout(TIMEOUT):
             device = AlfenDevice(host, name, session, username, password)

custom_components/alfen_wallbox/alfen.py

@@ -1,5 +1,5 @@
 import logging
-from aiohttp import ClientSession
+from aiohttp import ClientResponse, ClientSession
 
 import ssl
 from datetime import timedelta
@@ -48,11 +48,11 @@ async def init(self):
         await self.async_update()
 
     @property
-    def status(self):
+    def status(self) -> str:
         return self._status
 
     @property
-    def device_info(self):
+    def device_info(self) -> dict:
         """Return a device description for device registry."""
         return {
             "identifiers": {(DOMAIN, self.name)},
@@ -69,20 +69,17 @@ def _request(self, parameter_list):
     async def async_update(self):
         await self._do_update()
 
-    async def login(self):
-        response = await self._session.request(
-            ssl=self.ssl,
-            method=METHOD_POST,
-            headers=HEADER_JSON,
-            url=self.__get_url(LOGIN),
-            json={PARAM_USERNAME: self.username,
-                  PARAM_PASSWORD: self.password},
-        )
+    async def login(self) -> bool:
+        response = await self.request(
+            METHOD_POST,
+            HEADER_JSON,
+            LOGIN,
+            {PARAM_USERNAME: self.username, PARAM_PASSWORD: self.password})
 
         _LOGGER.debug(f"Login response {response}")
         return response.status == 200
 
-    async def logout(self):
+    async def logout(self) -> bool:
         response = await self._session.request(
             ssl=self.ssl,
             method=METHOD_POST,
@@ -92,14 +89,15 @@ async def logout(self):
         _LOGGER.debug(f"Logout response {response}")
         return response.status == 200
 
-    async def _update_value(self, api_param, value):
-        response = await self._session.request(
-            ssl=self.ssl,
-            method=METHOD_POST,
-            headers=POST_HEADER_JSON,
-            url=self.__get_url(PROP),
-            json={api_param: {ID: api_param, VALUE: value}},
+    async def _update_value(self, api_param, value) -> bool:
+
+        response = await self.request(
+            METHOD_POST,
+            POST_HEADER_JSON,
+            PROP,
+            {api_param: {ID: api_param, VALUE: value}}
         )
+
         _LOGGER.debug(f"Set {api_param} value {value} response {response}")
         return response.status == 200
 
@@ -186,12 +184,23 @@ async def reboot_wallbox(self):
         _LOGGER.debug(f"Reboot response {response}")
         await self.logout()
 
-    async def set_value(self, api_param, value):
+    async def request(self, method: str, headers: str, url_cmd: str, json=None) -> ClientResponse:
+        response = await self._session.request(
+            ssl=self.ssl,
+            method=method,
+            headers=headers,
+            url=self.__get_url(url_cmd),
+            json=json,
+        )
+        _LOGGER.debug(f"Request response {response}")
+        return response
+
+    async def set_value(self, api_param, value) -> bool:
 
         logged_in = await self.login()
         # if not logged in, we can't set the value, show error
         if not logged_in:
-            return self.async_abort(reason="Unable to authenticate to wallbox")
+            return None
 
         success = await self._update_value(api_param, value)
         await self.logout()
@@ -201,7 +210,8 @@ async def set_value(self, api_param, value):
                 if prop[ID] == api_param:
                     _LOGGER.debug(f"Set {api_param} value {value}")
                     prop[VALUE] = value
-                    break
+                    return True
+        return False
 
     async def get_value(self, api_param):
         await self.login()
@@ -211,7 +221,7 @@ async def get_value(self, api_param):
     async def set_current_limit(self, limit):
         _LOGGER.debug(f"Set current limit {limit}A")
         if limit > 32 | limit < 1:
-            return self.async_abort(reason="invalid_current_limit")
+            return None
         self.set_value("2129_0", limit)
 
     async def set_rfid_auth_mode(self, enabled):
@@ -226,9 +236,7 @@ async def set_rfid_auth_mode(self, enabled):
     async def set_current_phase(self, phase):
         _LOGGER.debug(f"Set current phase {phase}")
         if phase not in ('L1', 'L2', 'L3'):
-            return self.async_abort(
-                reason="invalid phase mapping allowed value: L1, L2, L3"
-            )
+            return None
         self.set_value("2069_0", phase)
 
     async def set_phase_switching(self, enabled):
@@ -243,16 +251,16 @@ async def set_phase_switching(self, enabled):
     async def set_green_share(self, value):
         _LOGGER.debug(f"Set green share value {value}%")
         if value < 0 | value > 100:
-            return self.async_abort(reason="invalid_value")
+            return None
         self.set_value("3280_2", value)
 
     async def set_comfort_power(self, value):
         _LOGGER.debug(f"Set Comfort Level {value}W")
         if value < 1400 | value > 5000:
-            return self.async_abort(reason="invalid_value")
+            return None
         self.set_value("3280_3", value)
 
-    def __get_url(self, action):
+    def __get_url(self, action) -> str:
         return "https://{}/api/{}".format(self.host, action)
 
 

custom_components/alfen_wallbox/button.py

@@ -0,0 +1,83 @@
+
+from dataclasses import dataclass
+import logging
+from typing import Final
+from .alfen import POST_HEADER_JSON, AlfenDevice
+from .entity import AlfenEntity
+
+from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from .const import CMD, METHOD_POST, PARAM_COMMAND, COMMAND_REBOOT
+
+from . import DOMAIN as ALFEN_DOMAIN
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class AlfenButtonDescriptionMixin:
+    """Define an entity description mixin for button entities."""
+
+    method: str
+    url_action: str
+    json_action: str
+
+
+@dataclass
+class AlfenButtonDescription(ButtonEntityDescription, AlfenButtonDescriptionMixin):
+    """Class to describe an Alfen button entity."""
+
+
+ALFEN_BUTTON_TYPES: Final[tuple[AlfenButtonDescription, ...]] = (
+    AlfenButtonDescription(
+        key="reboot_wallbox",
+        name="Reboot Wallbox",
+        method=METHOD_POST,
+        url_action=CMD,
+        json_action={PARAM_COMMAND: COMMAND_REBOOT},
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Alfen switch entities from a config entry."""
+    device = hass.data[ALFEN_DOMAIN][entry.entry_id]
+    buttons = [AlfenButton(device, description)
+               for description in ALFEN_BUTTON_TYPES]
+
+    async_add_entities(buttons)
+
+
+class AlfenButton(AlfenEntity, ButtonEntity):
+    """Representation of a Alfen button entity."""
+
+    def __init__(
+        self,
+        device: AlfenDevice,
+        description: AlfenButtonDescription,
+    ) -> None:
+        """Initialize the Alfen button entity."""
+        super().__init__(device)
+        self._device = device
+        self._attr_name = f"{device.name} {description.name}"
+        self._attr_unique_id = f"{device.id}-{description.key}"
+        self.entity_description = description
+
+    async def async_press(self) -> None:
+        """Press the button."""
+        await self._device.login()
+        response = await self._device.request(
+            self.entity_description.method,
+            POST_HEADER_JSON,
+            self.entity_description.url_action,
+            self.entity_description.json_action
+        )
+        await self._device.logout()
+        _LOGGER.debug("Response: %s", response)

custom_components/alfen_wallbox/config_flow.py

@@ -23,7 +23,7 @@ class FlowHandler(config_entries.ConfigFlow):
     VERSION = 1
     CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
 
-    async def _create_entry(self, host, name, username, password):
+    async def _create_entry(self, host, name, username, password) -> None:
         """Register new entry."""
         # Check if ip already is registered
         for entry in self._async_current_entries():

custom_components/alfen_wallbox/const.py

@@ -28,6 +28,8 @@
 CAT_TEMP = "temp"
 CAT_OCPP = "ocpp"
 
+COMMAND_REBOOT = "reboot"
+
 
 TIMEOUT = 60
 

custom_components/alfen_wallbox/sensor.py

@@ -4,6 +4,9 @@
 
 import voluptuous as vol
 
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.typing import StateType
+
 from .entity import AlfenEntity
 from homeassistant import const
 from homeassistant.config_entries import ConfigEntry
@@ -757,7 +760,7 @@ def __init__(self,
             self._attr_state_class = SensorStateClass.TOTAL_INCREASING
         self._async_update_attrs()
 
-    def _get_current_value(self):
+    def _get_current_value(self) -> StateType | None:
         """Get the current value."""
         for prop in self._device.properties:
             if prop[ID] == self.entity_description.api_param:
@@ -770,32 +773,32 @@ def _async_update_attrs(self) -> None:
         self._attr_native_value = self._get_current_value()
 
     @property
-    def unique_id(self):
+    def unique_id(self) -> str:
         """Return a unique ID."""
         return f"{self._device.id}-{self.entity_description.key}"
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Return the name of the sensor."""
         return self._attr_name
 
     @property
-    def icon(self):
+    def icon(self) -> str | None:
         """Return the icon of the sensor."""
         return self.entity_description.icon
 
     @property
-    def native_value(self):
+    def native_value(self) -> StateType:
         """Return the state of the sensor."""
         return round(self.state, 2)
 
     @property
-    def native_unit_of_measurement(self):
+    def native_unit_of_measurement(self) -> str | None:
         """Return the unit the value is expressed in."""
         return self.entity_description.unit
 
     @property
-    def state(self):
+    def state(self) -> StateType:
         """Return the state of the sensor."""
         for prop in self._device.properties:
             if prop[ID] == self.entity_description.api_param:
@@ -827,7 +830,7 @@ def state(self):
                 return prop[VALUE]
 
     @property
-    def unit_of_measurement(self):
+    def unit_of_measurement(self) -> str:
         """Return the unit of measurement."""
         return self.entity_description.unit
 
@@ -836,6 +839,6 @@ async def async_update(self):
         await self._device.async_update()
 
     @property
-    def device_info(self):
+    def device_info(self) -> DeviceInfo:
         """Return a device description for device registry."""
         return self._device.device_info