Skip to content

Commit

Permalink
Cleanup the code
Browse files Browse the repository at this point in the history
  • Loading branch information
Tuen Lee committed Nov 18, 2023
1 parent e42521c commit 2b052b7
Show file tree
Hide file tree
Showing 7 changed files with 45 additions and 36 deletions.
2 changes: 1 addition & 1 deletion custom_components/polestar_api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
6 changes: 3 additions & 3 deletions custom_components/polestar_api/config_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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(
Expand All @@ -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])
3 changes: 2 additions & 1 deletion custom_components/polestar_api/const.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
DOMAIN = "polestar_api"
TIMEOUT = 90

MAX_CHARGE_RANGE = 375

CONF_VIN = "vin"
CONF_VCC_API_KEY = "vcc_api_key"
Expand All @@ -12,3 +11,5 @@

HEADER_AUTHORIZATION = "authorization"
HEADER_VCC_API_KEY = "vcc-api-key"

CACHE_TIME = 15
7 changes: 3 additions & 4 deletions custom_components/polestar_api/entity.py
Original file line number Diff line number Diff line change
@@ -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,
Expand Down
6 changes: 3 additions & 3 deletions custom_components/polestar_api/polestar.py
Original file line number Diff line number Diff line change
Expand Up @@ -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})"

Expand All @@ -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()
12 changes: 5 additions & 7 deletions custom_components/polestar_api/polestar_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from .const import (
ACCESS_TOKEN_MANAGER_ID,
AUTHORIZATION,
CACHE_TIME,
GRANT_TYPE,
HEADER_AUTHORIZATION,
HEADER_VCC_API_KEY
Expand All @@ -22,8 +23,6 @@


class PolestarApi:
QUERY_PAYLOAD = ""

def __init__(self,
hass: HomeAssistant,
username: str,
Expand All @@ -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={
Expand All @@ -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
Expand All @@ -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)
Expand Down
45 changes: 28 additions & 17 deletions custom_components/polestar_api/sensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -223,34 +222,46 @@ 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
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

0 comments on commit 2b052b7

Please sign in to comment.