From a39d7e611463775608fc92e1beca736dc900ec60 Mon Sep 17 00:00:00 2001 From: Tuen Lee Date: Mon, 10 Jul 2023 01:05:06 +0200 Subject: [PATCH] cleanup some codes --- custom_components/alfen_wallbox/alfen.py | 68 +++++++++++-------- .../alfen_wallbox/config_flow.py | 2 +- custom_components/alfen_wallbox/const.py | 2 + custom_components/alfen_wallbox/sensor.py | 21 +++--- 4 files changed, 53 insertions(+), 40 deletions(-) diff --git a/custom_components/alfen_wallbox/alfen.py b/custom_components/alfen_wallbox/alfen.py index 55c0bd5..ea86bbb 100644 --- a/custom_components/alfen_wallbox/alfen.py +++ b/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) diff --git a/custom_components/alfen_wallbox/config_flow.py b/custom_components/alfen_wallbox/config_flow.py index c7a328c..703a2af 100644 --- a/custom_components/alfen_wallbox/config_flow.py +++ b/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(): diff --git a/custom_components/alfen_wallbox/const.py b/custom_components/alfen_wallbox/const.py index 290edcf..51775d2 100644 --- a/custom_components/alfen_wallbox/const.py +++ b/custom_components/alfen_wallbox/const.py @@ -28,6 +28,8 @@ CAT_TEMP = "temp" CAT_OCPP = "ocpp" +COMMAND_REBOOT = "reboot" + TIMEOUT = 60 diff --git a/custom_components/alfen_wallbox/sensor.py b/custom_components/alfen_wallbox/sensor.py index c1e557e..22cd0f3 100644 --- a/custom_components/alfen_wallbox/sensor.py +++ b/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