Merge pull request #5 from leeyuentuen/dev

dev to master

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:

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])

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

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,

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()

custom_components/polestar_api/polestar_api.py

@@ -1,8 +1,10 @@
 from datetime import datetime, timedelta
+import json
 import logging
 from .const import (
     ACCESS_TOKEN_MANAGER_ID,
     AUTHORIZATION,
+    CACHE_TIME,
     GRANT_TYPE,
     HEADER_AUTHORIZATION,
     HEADER_VCC_API_KEY
@@ -21,8 +23,6 @@
 
 
 class PolestarApi:
-    QUERY_PAYLOAD = ""
-
     def __init__(self,
                  hass: HomeAssistant,
                  username: str,
@@ -40,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={
@@ -74,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
@@ -88,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)
@@ -124,6 +123,7 @@ async def get_data(self, path, reponse_path=None):
         resp = await response.json(content_type=None)
 
         _LOGGER.debug(f"Response {resp}")
+
         data = resp['data']
 
         # add cache_data[path]

custom_components/polestar_api/sensor.py

@@ -1,11 +1,9 @@
-import json
+from datetime import datetime, timedelta
 import logging
 from typing import Final
 from dataclasses import dataclass
-from datetime import timedelta
 
-from .const import MAX_CHARGE_RANGE
-from .entity import TibberEVEntity
+from .entity import PolestarEntity
 
 from homeassistant.helpers.typing import StateType
 
@@ -24,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
@@ -47,6 +45,7 @@ class PolestarSensorDescriptionMixin:
     round_digits: int | None
     unit: str | None
     response_path: str | None
+    max_value: int | None
 
 
 @dataclass
@@ -69,16 +68,40 @@ class PolestarSensorDescription(
 }
 
 
-TIBBER_SENSOR_TYPES: Final[tuple[PolestarSensorDescription, ...]] = (
+POLESTAR_SENSOR_TYPES: Final[tuple[PolestarSensorDescription, ...]] = (
+    PolestarSensorDescription(
+        key="estimate_full_charge_range",
+        name="Est. full charge range",
+        icon="mdi:map-marker-distance",
+        path="{vin}/recharge-status",
+        response_path=None,
+        unit='km',
+        round_digits=None,
+        state_class=SensorStateClass.MEASUREMENT,
+        device_class=SensorDeviceClass.DISTANCE,
+        max_value=None
+    ),
     PolestarSensorDescription(
         key="battery_charge_level",
         name="Battery level",
         path="{vin}/recharge-status",
         response_path="batteryChargeLevel.value",
         unit=PERCENTAGE,
-        round_digits=1,
+        round_digits=0,
         state_class=SensorStateClass.MEASUREMENT,
         device_class=SensorDeviceClass.BATTERY,
+        max_value=None,
+    ),
+    PolestarSensorDescription(
+        key="last_updated",
+        name="Last updated",
+        path="{vin}/recharge-status",
+        response_path="batteryChargeLevel.timestamp",
+        unit=None,
+        round_digits=None,
+        state_class=SensorStateClass.MEASUREMENT,
+        device_class=SensorDeviceClass.TIMESTAMP,
+        max_value=None,
     ),
     PolestarSensorDescription(
         key="electric_range",
@@ -90,6 +113,7 @@ class PolestarSensorDescription(
         round_digits=None,
         state_class=SensorStateClass.MEASUREMENT,
         device_class=SensorDeviceClass.DISTANCE,
+        max_value=570,  # prevent spike value, and this should be the max range of polestar
     ),
     PolestarSensorDescription(
         key="estimated_charging_time",
@@ -99,6 +123,19 @@ class PolestarSensorDescription(
         response_path="estimatedChargingTime.value",
         unit='Minutes',
         round_digits=None,
+        max_value=None,
+    ),
+    PolestarSensorDescription(
+        key="estimated_fully_charged_time",
+        name="Fully charged time",
+        icon="mdi:battery-clock",
+        path="{vin}/recharge-status",
+        response_path="estimatedChargingTime.value",
+        unit=None,
+        round_digits=None,
+        state_class=SensorStateClass.MEASUREMENT,
+        device_class=SensorDeviceClass.DURATION,
+        max_value=None,
     ),
     PolestarSensorDescription(
         key="charging_connection_status",
@@ -108,6 +145,8 @@ class PolestarSensorDescription(
         response_path="chargingConnectionStatus.value",
         unit=None,
         round_digits=None,
+        state_class=SensorStateClass.MEASUREMENT,
+        max_value=None,
     ),
     PolestarSensorDescription(
         key="charging_system_status",
@@ -117,6 +156,7 @@ class PolestarSensorDescription(
         response_path="chargingSystemStatus.value",
         unit=None,
         round_digits=None,
+        max_value=None,
     ),
 
 )
@@ -137,18 +177,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
@@ -159,10 +199,11 @@ def __init__(self,
         """Initialize the sensor."""
         super().__init__(device)
         self._device = device
-        # get the first 8 character of the id
-        unique_id = device.vin[:8]
+        # get the last 4 character of the id
+        unique_id = device.vin[-4:]
+        self.entity_id = f"{POLESTAR_API_DOMAIN}.'polestar_'.{unique_id}_{description.key}"
         self._attr_name = f"{description.name}"
-        self._attr_unique_id = f"{unique_id}-{description.key}"
+        self._attr_unique_id = f"polestar_{unique_id}-{description.key}"
         self.value = None
         self.description = description
 
@@ -212,27 +253,71 @@ 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', ''))
+
+        # prevent exponentianal value, we only give state value that is lower than the max value
+        if self.entity_description.max_value is not None:
+            if isinstance(self._attr_native_value, str):
+                self._attr_native_value = int(self._attr_native_value)
+            if self._attr_native_value > self.entity_description.max_value:
+                return None
+
+        # Custom state for estimated_fully_charged_time
+        if self.entity_description.key == 'estimated_fully_charged_time':
+            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)
+
+        if self.entity_description.key == 'estimate_full_charge_range':
+            battery_level = self._device.get_cache_data(
+                self.entity_description.path, 'batteryChargeLevel.value')
+            estimate_range = self._device.get_cache_data(
+                self.entity_description.path, 'electricRange.value')
+
+            if battery_level is None or estimate_range is None:
+                return None
+
+            if battery_level is False or estimate_range is False:
+                return None
+
+            battery_level = int(battery_level.replace('.0', ''))
+            estimate_range = int(estimate_range)
+
+            return round(estimate_range / battery_level * 100)
+
         return self._attr_native_value
 
     @property
     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