diff --git a/custom_components/polestar_api/__init__.py b/custom_components/polestar_api/__init__.py index 35f8647..ef00591 100644 --- a/custom_components/polestar_api/__init__.py +++ b/custom_components/polestar_api/__init__.py @@ -68,7 +68,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> return unload_ok -async def tibber_setup(hass: HomeAssistant, name: str, username: str, password: str) -> PolestarApi | None: +async def polestar_setup(hass: HomeAssistant, name: str, username: str, password: str) -> PolestarApi | None: """Create a Polestar instance only once.""" try: diff --git a/custom_components/polestar_api/config_flow.py b/custom_components/polestar_api/config_flow.py index ecb1691..978c2eb 100644 --- a/custom_components/polestar_api/config_flow.py +++ b/custom_components/polestar_api/config_flow.py @@ -35,7 +35,7 @@ async def _create_entry(self, username: str, password: str, vin: str, vcc_api_ke } ) - async def _create_device(self, username: str, password: str, vin: str, vcc_api_key: str): + async def _create_device(self, username: str, password: str, vin: str, vcc_api_key: str) -> None: """Create device.""" try: @@ -59,7 +59,7 @@ async def _create_device(self, username: str, password: str, vin: str, vcc_api_k return await self._create_entry(username, password, vin, vcc_api_key) - async def async_step_user(self, user_input=None): + async def async_step_user(self, user_input: dict = None) -> None: """User initiated config flow.""" if user_input is None: return self.async_show_form( @@ -72,6 +72,6 @@ async def async_step_user(self, user_input=None): ) return await self._create_device(user_input[CONF_USERNAME], user_input[CONF_PASSWORD], user_input[CONF_VIN], user_input[CONF_VCC_API_KEY]) - async def async_step_import(self, user_input): + async def async_step_import(self, user_input: dict) -> None: """Import a config entry.""" return await self._create_device(user_input[CONF_USERNAME], user_input[CONF_PASSWORD], user_input[CONF_VIN], user_input[CONF_VCC_API_KEY]) diff --git a/custom_components/polestar_api/const.py b/custom_components/polestar_api/const.py index ea20b05..8e60b2b 100644 --- a/custom_components/polestar_api/const.py +++ b/custom_components/polestar_api/const.py @@ -1,7 +1,6 @@ DOMAIN = "polestar_api" TIMEOUT = 90 -MAX_CHARGE_RANGE = 375 CONF_VIN = "vin" CONF_VCC_API_KEY = "vcc_api_key" @@ -12,3 +11,5 @@ HEADER_AUTHORIZATION = "authorization" HEADER_VCC_API_KEY = "vcc-api-key" + +CACHE_TIME = 15 diff --git a/custom_components/polestar_api/entity.py b/custom_components/polestar_api/entity.py index 5d6c485..694775a 100644 --- a/custom_components/polestar_api/entity.py +++ b/custom_components/polestar_api/entity.py @@ -1,21 +1,20 @@ -from datetime import timedelta import logging from .polestar import PolestarApi -from .const import DOMAIN as Tibber_EV_DOMAIN +from .const import DOMAIN as POLESTAR_API_DOMAIN from homeassistant.helpers.entity import DeviceInfo, Entity _LOGGER = logging.getLogger(__name__) -class TibberEVEntity(Entity): +class PolestarEntity(Entity): def __init__(self, device: PolestarApi) -> None: """Initialize the Polestar entity.""" self._device = device self._attr_device_info = DeviceInfo( - identifiers={(Tibber_EV_DOMAIN, self._device.name)}, + identifiers={(POLESTAR_API_DOMAIN, self._device.name)}, manufacturer="Polestar", model=None, name=device.name, diff --git a/custom_components/polestar_api/polestar.py b/custom_components/polestar_api/polestar.py index 43fa799..94f8e47 100644 --- a/custom_components/polestar_api/polestar.py +++ b/custom_components/polestar_api/polestar.py @@ -33,8 +33,8 @@ def __init__(self, self.polestar_api = polestar_api disable_warnings() - async def init(self): - self.id = "tibber_{}".format(self.name) + async def init(self) -> None: + self.id = "polestar{}".format(self.name) if self.name is None: self.name = f"{self.info.identity} ({self.host})" @@ -43,5 +43,5 @@ def status(self) -> str: return self._status @Throttle(timedelta(seconds=10)) - async def async_update(self): + async def async_update(self) -> None: self.raw_data = await self.polestar_api.get_ev_data() diff --git a/custom_components/polestar_api/polestar_api.py b/custom_components/polestar_api/polestar_api.py index a40d3bd..0dd6c43 100644 --- a/custom_components/polestar_api/polestar_api.py +++ b/custom_components/polestar_api/polestar_api.py @@ -4,6 +4,7 @@ from .const import ( ACCESS_TOKEN_MANAGER_ID, AUTHORIZATION, + CACHE_TIME, GRANT_TYPE, HEADER_AUTHORIZATION, HEADER_VCC_API_KEY @@ -22,8 +23,6 @@ class PolestarApi: - QUERY_PAYLOAD = "" - def __init__(self, hass: HomeAssistant, username: str, @@ -41,14 +40,13 @@ def __init__(self, self.refresh_token = None self.vin = vin self.vcc_api_key = vcc_api_key - # data and timestamp e.g. {'data': {}, 'timestamp': 1234567890} self.cache_data = None disable_warnings() async def init(self): await self.get_token() - async def get_token(self): + async def get_token(self) -> None: response = await self._session.post( url='https://volvoid.eu.volvocars.com/as/token.oauth2', data={ @@ -75,12 +73,12 @@ async def get_token(self): _LOGGER.debug(f"Response {self.access_token}") - def get_cache_data(self, path, reponse_path=None): + def get_cache_data(self, path: str, reponse_path: str = None) -> dict or bool or None: # replace the string {vin} with the actual vin path = path.replace('{vin}', self.vin) if self.cache_data and self.cache_data[path]: - if self.cache_data[path]['timestamp'] > datetime.now() - timedelta(seconds=15): + if self.cache_data[path]['timestamp'] > datetime.now() - timedelta(seconds=CACHE_TIME): data = self.cache_data[path]['data'] if data is None: return False @@ -89,7 +87,7 @@ def get_cache_data(self, path, reponse_path=None): data = data[key] return data - async def get_data(self, path, reponse_path=None): + async def get_data(self, path: str, reponse_path: str = None) -> dict or bool or None: path = path.replace('{vin}', self.vin) cache_data = self.get_cache_data(path, reponse_path) diff --git a/custom_components/polestar_api/sensor.py b/custom_components/polestar_api/sensor.py index f62242d..847d526 100644 --- a/custom_components/polestar_api/sensor.py +++ b/custom_components/polestar_api/sensor.py @@ -3,8 +3,7 @@ from typing import Final from dataclasses import dataclass -from .const import MAX_CHARGE_RANGE -from .entity import TibberEVEntity +from .entity import PolestarEntity from homeassistant.helpers.typing import StateType @@ -23,7 +22,7 @@ from homeassistant.helpers import entity_platform -from . import DOMAIN as TIBBER_EV_DOMAIN +from . import DOMAIN as POLESTAR_API_DOMAIN from .polestar import PolestarApi @@ -68,7 +67,7 @@ class PolestarSensorDescription( } -TIBBER_SENSOR_TYPES: Final[tuple[PolestarSensorDescription, ...]] = ( +POLESTAR_SENSOR_TYPES: Final[tuple[PolestarSensorDescription, ...]] = ( PolestarSensorDescription( key="battery_charge_level", name="Battery level", @@ -148,18 +147,18 @@ async def async_setup_entry( """Set up using config_entry.""" # get the device device: PolestarApi - device = hass.data[TIBBER_EV_DOMAIN][entry.entry_id] + device = hass.data[POLESTAR_API_DOMAIN][entry.entry_id] # put data in cache await device.get_data("{vin}/recharge-status") sensors = [ - PolestarSensor(device, description) for description in TIBBER_SENSOR_TYPES + PolestarSensor(device, description) for description in POLESTAR_SENSOR_TYPES ] async_add_entities(sensors) platform = entity_platform.current_platform.get() -class PolestarSensor(TibberEVEntity, SensorEntity): +class PolestarSensor(PolestarEntity, SensorEntity): """Representation of a Polestar Sensor.""" entity_description: PolestarSensorDescription @@ -223,24 +222,34 @@ def native_unit_of_measurement(self) -> str | None: @property def state(self) -> StateType: """Return the state of the sensor.""" + if self._attr_native_value is None: + return None + # parse the long text with a shorter one from the dict if self.entity_description.key == 'charging_connection_status': return ChargingConnectionStatusDict.get(self._attr_native_value, self._attr_native_value) if self.entity_description.key == 'charging_system_status': return ChargingSystemStatusDict.get(self._attr_native_value, self._attr_native_value) + # battery charge level contain ".0" at the end, this should be removed + if self.entity_description.key == 'battery_charge_level': + if isinstance(self._attr_native_value, str): + self._attr_native_value = int( + self._attr_native_value.replace('.0', '')) + # Custom state for estimated_fully_charged_time if self.entity_description.key == 'estimated_fully_charged_time': - if self._attr_native_value is not None: - value = int(self._attr_native_value) - if value > 0: - return datetime.now().replace(second=0, microsecond=0) + timedelta(minutes=round(value)) + value = int(self._attr_native_value) + if value > 0: + return datetime.now().replace(second=0, microsecond=0) + timedelta(minutes=round(value)) return 'Not charging' # round the value if self.entity_description.round_digits is not None: - if self._attr_native_value is not None: - return round(float(self._attr_native_value), self.entity_description.round_digits) + # if the value is integer, remove the decimal + if self.entity_description.round_digits == 0 and isinstance(self._attr_native_value, int): + return int(self._attr_native_value) + return round(float(self._attr_native_value), self.entity_description.round_digits) return self._attr_native_value @property @@ -248,9 +257,11 @@ def unit_of_measurement(self) -> str: """Return the unit of measurement.""" return self.entity_description.unit - async def async_update(self): + async def async_update(self) -> None: """Get the latest data and updates the states.""" data = await self._device.get_data(self.entity_description.path, self.entity_description.response_path) - if data is not None: - self._attr_native_value = data - self.value = data + if data is None: + return + + self._attr_native_value = data + self.value = data