move api call to pypolestar, split api call #14

custom_components/polestar_api/__init__.py

@@ -7,7 +7,9 @@
 from aiohttp import ClientConnectionError
 from async_timeout import timeout
 
-from .polestar_api import PolestarApi
+from .pypolestar.polestar import PolestarApi
+from .polestar import Polestar
+
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
@@ -41,7 +43,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
     conf = config_entry.data
 
     _LOGGER.debug("async_setup_entry: %s", config_entry)
-    polestarApi = PolestarApi(
+    polestarApi = Polestar(
         hass, conf[CONF_USERNAME], conf[CONF_PASSWORD])
     await polestarApi.init()
 
@@ -72,7 +74,7 @@ async def polestar_setup(hass: HomeAssistant, name: str, username: str, password
 
     try:
         with timeout(TIMEOUT):
-            device = PolestarApi(hass, name, username, password)
+            device = Polestar(hass, name, username, password)
             await device.init()
     except asyncio.TimeoutError:
         _LOGGER.debug("Connection to %s timed out", name)

custom_components/polestar_api/config_flow.py

@@ -5,7 +5,7 @@
 from aiohttp import ClientError
 from async_timeout import timeout
 import voluptuous as vol
-from .polestar_api import PolestarApi
+from .polestar import Polestar
 
 from homeassistant import config_entries
 from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
@@ -37,7 +37,7 @@ async def _create_device(self, username: str, password: str) -> None:
         """Create device."""
 
         try:
-            device = PolestarApi(
+            device = Polestar(
                 self.hass,
                 username,
                 password)

custom_components/polestar_api/const.py

@@ -1,11 +1,4 @@
 DOMAIN = "polestar_api"
 TIMEOUT = 90
 
-
-ACCESS_TOKEN_MANAGER_ID = "JWTh4Yf0b"
-GRANT_TYPE = "password"
-AUTHORIZATION = "Basic aDRZZjBiOlU4WWtTYlZsNnh3c2c1WVFxWmZyZ1ZtSWFEcGhPc3kxUENhVXNpY1F0bzNUUjVrd2FKc2U0QVpkZ2ZJZmNMeXc="
-
-HEADER_AUTHORIZATION = "authorization"
-
 CACHE_TIME = 30

custom_components/polestar_api/entity.py

@@ -1,6 +1,6 @@
 import logging
 
-from .polestar_api import PolestarApi
+from .polestar import Polestar
 
 from .const import DOMAIN as POLESTAR_API_DOMAIN
 from homeassistant.helpers.entity import DeviceInfo, Entity
@@ -10,7 +10,7 @@
 
 class PolestarEntity(Entity):
 
-    def __init__(self, device: PolestarApi) -> None:
+    def __init__(self, device: Polestar) -> None:
         """Initialize the Polestar entity."""
         self._device = device
 

custom_components/polestar_api/polestar.py

@@ -1,47 +1,58 @@
-from datetime import timedelta
-import json
+
 import logging
 
+from .pypolestar.polestar import PolestarApi
+
+
+from .const import (
+    CACHE_TIME,
+)
+
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
 from urllib3 import disable_warnings
 
 from homeassistant.core import HomeAssistant
-from homeassistant.util import Throttle
-from .polestar_api import PolestarApi
-
 
 POST_HEADER_JSON = {"Content-Type": "application/json"}
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class Polestar:
-
-    QUERY_PAYLOAD = '{"query": "{ me { homes { electricVehicles {id name shortName lastSeen lastSeenText isAlive hasNoSmartChargingCapability imgUrl schedule {isEnabled isSuspended localTimeTo minBatteryLevel} batteryText chargingText consumptionText consumptionUnitText energyCostUnitText chargeRightAwayButton chargeRightAwayAlert {imgUrl title description okText cancelText}backgroundStyle energyDealCallToAction{text url redirectUrlStartsWith link action enabled} settingsScreen{settings {key value valueType valueIsArray isReadOnly inputOptions{type title description pickerOptions {values postFix} rangeOptions{max min step defaultValue displayText displayTextPlural} selectOptions {value title description imgUrl iconName isRecommendedOption} textFieldOptions{imgUrl format placeholder} timeOptions{doNotSetATimeText}}} settingsLayout{uid type title description valueText imgUrl iconName isUpdated isEnabled callToAction {text url redirectUrlStartsWith link action enabled} childItems{uid type title description valueText imgUrl iconName isUpdated isEnabled callToAction {text url redirectUrlStartsWith link action enabled} settingKey settingKeyForIsHidden} settingKey settingKeyForIsHidden}} settingsButtonText settingsButton  {text url redirectUrlStartsWith link action enabled}enterPincode message {id title description style iconName iconSrc callToAction {text url redirectUrlStartsWith link action enabled} dismissButtonText} scheduleSuspendedText faqUrl battery { percent percentColor isCharging chargeLimit}}}}}"}'
-
     def __init__(self,
                  hass: HomeAssistant,
-                 raw_data: str,
-                 polestar_api: PolestarApi) -> None:
-
-        ev_id = raw_data.get("id").replace("-", "")[:8]
-        ev_name = raw_data.get("name")
-        self.id = ev_id
-        self.name = ev_name
-        self.raw_data = raw_data
-        self.polestar_api = polestar_api
+                 username: str,
+                 password: str
+                 ) -> None:
+        self.id = None
+        self.name = "Polestar "
+        self._session = async_get_clientsession(hass, verify_ssl=False)
+        self.polestarApi = PolestarApi(username, password)
+
+        self.vin = None
+        self.cache_data = {}
+        self.latest_call_code = None
+        self.updating = False
         disable_warnings()
 
-    async def init(self) -> None:
-        self.id = "polestar{}".format(self.name)
-        if self.name is None:
-            self.name = f"{self.info.identity} ({self.host})"
+    async def init(self):
+        await self.polestarApi.init()
+        vin = self.get_cache_data('getConsumerCarsV2', 'vin')
+        if vin:
+            # fill the vin and id in the constructor
+            self.vin = vin
+            self.id = vin[:8]
+            self.name = "Polestar " + vin[-4:]
+
+    def get_cache_data(self, query: str, field_name: str, skip_cache: bool = False):
+        return self.polestarApi.get_cache_data(query, field_name, skip_cache)
+
+    async def get_ev_data(self):
+        await self.polestarApi.get_ev_data(self.vin)
 
-    @property
-    def status(self) -> str:
-        return self._status
+    def get_latest_data(self, query: str, field_name: str):
+        return self.polestarApi.get_latest_data(query, field_name)
 
-    @Throttle(timedelta(seconds=10))
-    async def async_update(self) -> None:
-        self.raw_data = await self.polestar_api.get_ev_data()
+    def get_latest_call_code(self):
+        return self.polestarApi.latest_call_code

custom_components/polestar_api/pypolestar/__init__.py

@@ -0,0 +1 @@
+__version__ = "1.2.0"

custom_components/polestar_api/pypolestar/auth.py

@@ -0,0 +1,122 @@
+from datetime import datetime, timedelta
+import logging
+import json
+from urllib.parse import parse_qs, urlparse
+import aiohttp
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class PolestarAuth:
+    """ base class for Polestar authentication"""
+
+    def __init__(self, username: str, password: str) -> None:
+        self.username = username
+        self.password = password
+        self.access_token = None
+        self.refresh_token = None
+        self.token_expiry = None
+        self.latest_call_code = None
+        self.session = aiohttp.ClientSession()
+
+    async def get_token(self) -> None:
+        code = await self._get_code()
+        if code is None:
+            return
+
+        # get token
+        params = {
+            "query": "query getAuthToken($code: String!) { getAuthToken(code: $code) { id_token access_token refresh_token expires_in }}",
+            "operationName": "getAuthToken",
+            "variables": json.dumps({"code": code})
+        }
+
+        headers = {
+            "Content-Type": "application/json"
+        }
+        result = await self.session.get("https://pc-api.polestar.com/eu-north-1/auth/", params=params, headers=headers)
+        self.latest_call_code = result.status
+        if result.status != 200:
+            _LOGGER.error(f"Error getting token {result.status}")
+            return
+        resultData = await result.json()
+        _LOGGER.debug(resultData)
+
+        if resultData['data']:
+            self.access_token = resultData['data']['getAuthToken']['access_token']
+            self.refresh_token = resultData['data']['getAuthToken']['refresh_token']
+            self.token_expiry = datetime.now(
+            ) + timedelta(seconds=resultData['data']['getAuthToken']['expires_in'])
+            # ID Token
+
+        _LOGGER.debug(f"Response {self.access_token}")
+
+    async def _get_code(self) -> None:
+        resumePath = await self._get_resume_path()
+        parsed_url = urlparse(resumePath)
+        query_params = parse_qs(parsed_url.query)
+
+        # check if code is in query_params
+        if query_params.get('code'):
+            return query_params.get(('code'))[0]
+
+        # get the resumePath
+        if query_params.get('resumePath'):
+            resumePath = query_params.get(('resumePath'))[0]
+
+        if resumePath is None:
+            return
+
+        params = {
+            'client_id': 'polmystar'
+        }
+        data = {
+            'pf.username': self.username,
+            'pf.pass': self.password
+        }
+        result = await self.session.post(
+            f"https://polestarid.eu.polestar.com/as/{resumePath}/resume/as/authorization.ping",
+            params=params,
+            data=data
+        )
+        self.latest_call_code = result.status
+        if result.status != 200:
+            _LOGGER.error(f"Error getting code {result.status}")
+            return
+        # get the realUrl
+        url = result.url
+
+        parsed_url = urlparse(result.real_url.raw_path_qs)
+        query_params = parse_qs(parsed_url.query)
+
+        if not query_params.get('code'):
+            _LOGGER.error(f"Error getting code in {query_params}")
+            _LOGGER.warning("Check if username and password are correct")
+            return
+
+        code = query_params.get(('code'))[0]
+
+        # sign-in-callback
+        result = await self.session.get("https://www.polestar.com/sign-in-callback?code=" + code)
+        self.latest_call_code = result.status
+        if result.status != 200:
+            _LOGGER.error(f"Error getting code callback {result.status}")
+            return
+        # url encode the code
+        result = await self.session.get(url)
+        self.latest_call_code = result.status
+
+        return code
+
+    async def _get_resume_path(self):
+        # Get Resume Path
+        params = {
+            "response_type": "code",
+            "client_id": "polmystar",
+            "redirect_uri": "https://www.polestar.com/sign-in-callback"
+        }
+        result = await self.session.get("https://polestarid.eu.polestar.com/as/authorization.oauth2", params=params)
+        if result.status != 200:
+            _LOGGER.error(f"Error getting resume path {result.status}")
+            return
+        return result.real_url.raw_path_qs

custom_components/polestar_api/pypolestar/const.py

@@ -0,0 +1,5 @@
+CACHE_TIME = 30
+
+CAR_INFO_DATA = "getConsumerCarsV2"
+ODO_METER_DATA = "getOdometerData"
+BATTERY_DATA = "getBatteryData"