cleanup some codes

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/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