From b0807101b65c87c8d8463fbc84327e46f0e6de56 Mon Sep 17 00:00:00 2001 From: Flock82 <81387035+Flock82@users.noreply.github.com> Date: Wed, 12 Jul 2023 00:33:36 +0200 Subject: [PATCH 1/6] Create Kia-folder --- packages/modules/vehicles/kia/__init__.py | 1 + 1 file changed, 1 insertion(+) create mode 100644 packages/modules/vehicles/kia/__init__.py diff --git a/packages/modules/vehicles/kia/__init__.py b/packages/modules/vehicles/kia/__init__.py new file mode 100644 index 0000000000..8b13789179 --- /dev/null +++ b/packages/modules/vehicles/kia/__init__.py @@ -0,0 +1 @@ + From 80f90b67d1299f4c1e170401b28a49d2810abb52 Mon Sep 17 00:00:00 2001 From: Flock82 <81387035+Flock82@users.noreply.github.com> Date: Wed, 12 Jul 2023 00:35:40 +0200 Subject: [PATCH 2/6] Initial version --- packages/modules/vehicles/kia/api.py | 758 ++++++++++++++++++++++++ packages/modules/vehicles/kia/config.py | 19 + packages/modules/vehicles/kia/soc.py | 43 ++ 3 files changed, 820 insertions(+) create mode 100644 packages/modules/vehicles/kia/api.py create mode 100644 packages/modules/vehicles/kia/config.py create mode 100644 packages/modules/vehicles/kia/soc.py diff --git a/packages/modules/vehicles/kia/api.py b/packages/modules/vehicles/kia/api.py new file mode 100644 index 0000000000..e421609dc5 --- /dev/null +++ b/packages/modules/vehicles/kia/api.py @@ -0,0 +1,758 @@ +#!/usr/bin/env python3 + +import base64 +import json +import uuid +import hashlib +import requests +import urllib.parse as urlparse +from urllib.parse import parse_qs +import time +from typing import Union + +import logging +from modules.common.component_state import CarState +from modules.common.store import RAMDISK_PATH + +log = logging.getLogger(__name__) + +#---------- constants ---------- +def getString(param_id: str, brand: str) -> str: + if brand == 'kia': + if param_id == "app_id": + paramStr = "1518dd6b-2759-4995-9ae5-c9ad4a9ddad1" + elif param_id == "client_id": + paramStr = "fdc85c00-0a2f-4c64-bcb4-2cfb1500730a" + elif param_id == "auth_client_id": + paramStr = "572e0304-5f8d-4b4c-9dd5-41aa84eed160" + elif param_id == "gcm_sender_id": + paramStr = "cF5o4DiiQkaw5wsAkLzYIS:APA91bFB59MltBMK29zI0U2llq7khbB2jELkNFKMfBCH6KlCPL16pz_dG0fZ4ncvFn1IMT8nfojb83JyLiT_skBTXtClHhDCKeRbyPy3yQjCVRC3zTZt--wI7vv4jD9aknhHhiQsoZoU" + elif param_id == "basic_token": + paramStr = "Basic ZmRjODVjMDAtMGEyZi00YzY0LWJjYjQtMmNmYjE1MDA3MzBhOnNlY3JldA==" + elif param_id == "stamp_pad": + paramStr = "C0B4D5C7089D987F027C96015929C70FA9D2B2AA99530CFD017E4B243C4BA5C5DED96DEB128EEB5DD3963DFC12432C9073EF" + elif param_id == "host": + paramStr = "prd.eu-ccapi.kia.com:8080" + elif param_id == "base_url": + paramStr = "https://prd.eu-ccapi.kia.com:8080" + else: + raise RuntimeError + + if brand == 'hyundai': + if param_id == "app_id": + paramStr = "1eba27d2-9a5b-4eba-8ec7-97eb6c62fb51" + elif param_id == "client_id": + paramStr = "6d477c38-3ca4-4cf3-9557-2a1929a94654" + elif param_id == "auth_client_id": + paramStr = "64621b96-0f0d-11ec-82a8-0242ac130003" + elif param_id == "gcm_sender_id": + paramStr = "414998006775" + elif param_id == "basic_token": + paramStr = "Basic NmQ0NzdjMzgtM2NhNC00Y2YzLTk1NTctMmExOTI5YTk0NjU0OktVeTQ5WHhQekxwTHVvSzB4aEJDNzdXNlZYaG10UVI5aVFobUlGampvWTRJcHhzVg==" + elif param_id == "stamp_pad": + paramStr = "445B6846AFEF0D726646776865A650C9AEF98E51A474DCB7EC9B1B67D29C66EAAEF621CA02522A0B80A8087F7A3A7BB0F71B" + elif param_id == "host": + paramStr = "prd.eu-ccapi.hyundai.com:8080" + elif param_id == "base_url": + paramStr = "https://prd.eu-ccapi.hyundai.com:8080" + else: + raise RuntimeError + + return paramStr + +def getBrand(vin: str) -> str: + #Determinate brand based on VIN + try: + if vin[:2] == 'KN' or vin[:3] == 'U5Y' or vin[:3] == 'U6Z': + brand = "kia" + elif vin[:3] == 'KMH' or vin[:3] == 'TMA': + brand = "hyundai" + else: + brand = "" + soclogging.logDebug(2, "Vehicle WMI unknown") + raise RuntimeError + except Exception as err: + log.error("kia.getBrand: VIN error", exc_info = True) + raise + + return brand + +#---------- stamp generation ---------- +def getStamp(brand: str) -> str: + try: + #Set App-ID and App-ID specific XOR-pad + app_id = getString("app_id", brand) + secret_ba = bytearray.fromhex(getString("stamp_pad", brand)) + + # Timestamp in ms as string + nowStr = str(int(time.time()) * 1000) + + # Combine plaintext and convert to bytearray + plaintext = app_id + ":" + nowStr + plaintext_ba = bytearray(plaintext.encode()) + + # XOR plaintext and key + stamp_ba = bytes(a ^ b for (a, b) in zip(plaintext_ba, secret_ba)) + + # Convert result to base64-string + stamp_b64_ba = base64.b64encode(stamp_ba) + stamp = stamp_b64_ba.decode() + + except Exception as err: + log.error("kia.getStamp: stamp error", exc_info = True) + raise + + return stamp + +#---------- HTTP functions ---------- TBD! +last_cookies = {} +last_url = "" + +def getHTTP(url: str = "", headers: str = "", cookies: str = "", timeout: int = 30, allow_redirects: bool = True) -> str: + global last_cookies + global last_url + + try: + response = requests.get(url, headers=headers, cookies=cookies, timeout=timeout, allow_redirects=allow_redirects) + except Exception as err: + log.error("kia.getHTTP: HTTP error", exc_info = True) + raise + + if response.status_code == 200 or response.status_code == 204: + last_cookies = response.cookies.get_dict() + return response.text + elif response.status_code == 302: + return response.headers['Location'] + else: + try: + response_dict = json.loads(response.text) + if response.status_code == 400 or response.status_code == 408 or response.status_code == 503: + error_string = "[" + response_dict['resCode'] + "] " + response_dict['resMsg'] + else: + error_string = "[" + response_dict['errCode'] + "] " + response_dict['errMsg'] + except: + error_string = "[XXXX] Unidentified Error" + " " + response.text + + log.debug("kia.getHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) + raise RuntimeError + + return "" + + +def putHTTP(url: str = "", data: Union[str, dict] = "", headers: str = "", cookies: str = "", timeout: int = 30) -> str: + try: + if isinstance(data, dict): + response = requests.put(url, json=data, headers=headers, cookies=cookies, timeout=timeout) + else: + response = requests.put(url, data=data, headers=headers, cookies=cookies, timeout=timeout) + except Exception as err: + log.error("kia.putHTTP: HTTP error", exc_info = True) + raise + + if response.status_code == 200 or response.status_code == 204: + return response.text + else: + try: + response_dict = json.loads(response.text) + if response.status_code == 408: + error_string = "[" + response_dict['resCode'] + "] " + response_dict['resMsg'] + else: + error_string = "[" + response_dict['errCode'] + "] " + response_dict['errMsg'] + except: + error_string = "[XXXX] Unidentified Error" + + log.debug("kia.putHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) + raise RuntimeError + + return "" + +def deleteHTTP(url: str = "", headers: str = "", cookies: str = "", timeout: int = 30) -> None: + try: + response = requests.delete(url, headers=headers, cookies=cookies, timeout=timeout) + except Exception as err: + log.error("kia.putHTTP: HTTP error", exc_info = True) + raise + + return + + +def postHTTP(url: str = "", data: Union[str, dict] = "", headers: str = "", cookies: str = "", timeout: int = 30, allow_redirects: bool = True) -> str: + global last_cookies + global last_url + + try: + if isinstance(data, dict): + response = requests.post(url, json=data, headers=headers, cookies=cookies, + timeout=timeout, allow_redirects=allow_redirects) + else: + response = requests.post(url, data=data, headers=headers, cookies=cookies, + timeout=timeout, allow_redirects=allow_redirects) + except Exception as err: + log.error("kia.postHTTP: HTTP error", exc_info = True) + raise + + if response.status_code == 200 or response.status_code == 204: + last_url = response.url + return response.text + elif response.status_code == 302: + return response.headers['Location'] + else: + try: + response_dict = json.loads(response.text) + if response.status_code == 408: + error_string = "[" + response_dict['resCode'] + "] " + response_dict['resMsg'] + else: + error_string = "[" + response_dict['errCode'] + "] " + response_dict['errMsg'] + except: + error_string = "[XXXX] Unidentified Error" + + log.debug("kia.postHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) + raise RuntimeError + + return "" + + +def getHTTPCookies(url: str = "") -> dict: + try: + session = requests.Session() + response = session.get(url) + except Exception as err: + log.error("kia.getHTTPCookies: HTTP error", exc_info = True) + raise + + if response.status_code == 200: + cookies = session.cookies.get_dict() + else: + log.debug("kia.getHTTPCookies: Receiving cookies failed, StatusCode: " + str(response.status_code)) + raise RuntimeError + + return cookies + +#---------- token management ---------- +def getUserHash(user_id: str, password: str) -> str: + try: + account = user_id + ':' + password + hash = hashlib.md5(account.encode()).hexdigest() + except Exception as err: + log.error("kia.getUserHash: hash error", exc_info = True) + raise + + return hash + +def loadToken(user_id: str, password: str, vehicle: int) -> dict: + try: + token_file = str(RAMDISK_PATH) + "/soc_kia_vehicle" + str(vehicle) + "_token" + with open(token_file, 'r', encoding='utf-8') as f: + token = json.loads(f.read()) + except Exception as err: + log.debug("kia.loadToken: token file error: ", exc_info = True) + token = { + "userHash": "" + } + pass + + try: + if token["userHash"] != getUserHash(user_id, password): + log.debug("kia.loadToken: account data changed") + token = { + "userHash": "", + "deviceId": "", + "accessToken": "", + "gcmVehicleId": "", + "refreshToken": "", + "gcmClientId": "", + "tokenType": "" + } + except Exception as err: + log.error("kia.loadToken: token error", exc_info = True) + raise + + return token + +def saveToken(user_id: str, password: str, vehicle: int, token: dict) -> None: + try: + token["userHash"] = getUserHash(user_id, password) + token_file = str(RAMDISK_PATH) + "/soc_kia_vehicle" + str(vehicle) + "_token" + with open(token_file, 'w', encoding='utf-8') as f: + f.write(json.dumps(token)) + except Exception as err: + log.debug("kia.saveToken: token could not be saved", exc_info = True) + raise + +#---------- authentication ---------- +def getCookies(brand: str) -> dict: + log.info("Kia/Hyundai: Create Login-session") + + try: + url = getString("base_url", brand) + '/api/v1/user/oauth2/authorize?response_type=code&state=test&client_id=' + \ + getString("client_id", brand) + '&redirect_uri=' + getString("base_url", brand) + \ + '/api/v1/user/oauth2/redirect' + cookies = getHTTPCookies(url) + + url = getString("base_url", brand) + '/api/v1/user/session' + getHTTP(url=url, cookies=cookies) + + url = getString("base_url", brand) + '/api/v1/user/language' + headers = {'Content-type': 'application/json'} + data = {"lang": "en"} + response = postHTTP(url=url, data=data, headers=headers, cookies=cookies) + + url = getString("base_url", brand) + '/api/v1/user/session' + deleteHTTP(url=url, cookies=cookies) + except Exception as err: + log.debug("kia.getCookies: ", exc_info = True) + raise + + return cookies + +def getDeviceId(brand: str) -> dict: + log.info("Kia/Hyundai: Requesting device ids") + + token = {} + + try: + token["gcmClientId"] = str(uuid.uuid4()) + token["gcmVehicleId"] = str(uuid.uuid4()) + + url = getString("base_url", brand) + '/api/v1/spa/notifications/register' + data = {"pushRegId": getString("gcm_sender_id", brand), "pushType": "GCM", "uuid": str(uuid.uuid4())} + headers = { + 'Authorization': '', + 'Ccsp-Device-Id': '', + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'Offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Content-type': 'application/json; charset=UTF-8', + 'Content-Length': str(len(data)), + 'Host': getString("host", brand), + 'Connection': 'Keep-Alive', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = postHTTP(url=url, data=data, headers=headers) + + response_dict = json.loads(response) + token["deviceId"] = response_dict['resMsg']['deviceId'] + + log.debug("Kia/Hyundai: DeviceId = " + token["deviceId"][:8] + "[...]") + except Exception as err: + log.error("kia.getDeviceId: Request device id failed") + log.debug("kia.getDeviceId: Request device id failed: " + response, exc_info = True) + raise + + + log.info("Sending VersionInfo") + try: + url = getString("base_url", brand) + '/api/v1/spa/devices/version' + + data = { + "teleType": 'none', + "appVer": '2.1.9', + "buildVer": '7.1.2', + "phoneType": 'SM-G988N', + "osType": 'android', + "osVer": '7.1.2' + } + headers = { + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'Offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Content-type': 'application/json; charset=UTF-8', + 'Content-Length': str(len(data)), + 'Host': getString("host", brand), + 'Connection': 'Keep-Alive', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = postHTTP(url=url, data=data, headers=headers) + except Exception as err: + log.error("kia.getDeviceId: Set version info failed") + log.debug("kia.getDeviceId: Set version info failed: " + response, exc_info = True) + raise + + token["userHash"] = "" + token["tokenType"] = "" + token["accessToken"] = "" + token["refreshToken"] = "" + + return token + +def getAuthCode(username: str, password: str, brand: str, cookies: dict) -> str: + global last_cookies + + log.info("Kia/Hyundai: Sending username/password") + + try: + url = getString("base_url", brand) + '/api/v1/user/integrationinfo' + headers = {'Content-type': 'application/json'} + response = getHTTP(url=url, headers=headers, cookies=cookies) + + response_dict = json.loads(response) + user_id = response_dict['userId'] + service_id = response_dict['serviceId'] + log.debug("Kia/Hyundai: UserId = " + user_id[:8] + "[...]") + log.debug("Kia/Hyundai: ServiceId = " + service_id[:8] + "[...]") + + url = 'https://eu-account.' + brand + '.com/auth/realms/eu' + brand + 'idm/protocol/openid-connect/auth?client_id=' + \ + getString("auth_client_id", brand) + '&scope=openid%20profile%20email%20phone&response_type=code&hkid_session_reset=true&redirect_uri=' + \ + getString("base_url", brand) + '/api/v1/user/integration/redirect/login&ui_locales=en&state=' + service_id + ':' + user_id + headers = {} + response = getHTTP(url=url, headers=headers, cookies=cookies) + + left = response.find('action="') + 8 + right = response.find('"', left) + url = response[left:right].replace('&', '&') + data = urlparse.urlencode({'username': username, 'password': password, 'credentialId': ''}) + headers = { + 'Content-type': 'application/x-www-form-urlencoded', + 'User-Agent': 'Mozilla/5.0 (Linux; Android 4.1.1; Galaxy Nexus Build/JRO03C) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.166 Mobile Safari/535.19_CCS_APP_AOS' + } + cookies['AUTH_SESSION_ID'] = last_cookies['AUTH_SESSION_ID'] + response = postHTTP(url=url, data=data, headers=headers, cookies=cookies, allow_redirects=False) + + url = response + response = getHTTP(url=url, cookies=cookies, allow_redirects=True) + + url = getString("base_url", brand) + '/api/v1/user/silentsignin' + headers = {'Content-type': 'text/plain;charset=UTF-8'} + data = {'intUserId': ""} + response = postHTTP(url=url, data=data, headers=headers, cookies=cookies, allow_redirects=False) + response_dict = json.loads(response) + response_url = response_dict['redirectUrl'] + parsed = urlparse.urlparse(response_url) + auth_code = ''.join(parse_qs(parsed.query)['code']) + except Exception as err: + log.error("kia.getAuthCode: Login failed") + log.debug("kia.getAuthCode: Login failed: " + response, exc_info = True) + raise + + log.debug("Kia/Hyundai: AuthCode = " + auth_code[:8] + "[...]") + + return auth_code + +def getAuthToken(auth_code: str, token: dict, brand: str) -> dict: + log.info("Kia/Hyundai: Requesting access token") + + try: + url = getString("base_url", brand) + '/api/v1/user/oauth2/token' + data = 'client_id=' +getString("client_id", brand) +'&grant_type=authorization_code&code=' + \ + auth_code +'&redirect_uri=' + getString("base_url", brand) + '%2Fapi%2Fv1%2Fuser%2Foauth2%2Fredirect' + headers = { + 'Authorization': getString("basic_token", brand), + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'Offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Content-type': 'application/x-www-form-urlencoded', + 'Content-Length': str(len(data)), + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = postHTTP(url=url, headers=headers, data=data) + + access_token = json.loads(response) + token["tokenType"] = access_token["token_type"] + token["accessToken"] = access_token["access_token"] + token["refreshToken"] = access_token["refresh_token"] + + except Exception as err: + log.error("kia.getAuthToken: Login failed") + log.debug("kia.getAuthToken: Login failed: " + response, exc_info = True) + raise + + log.debug("Kia/Hyundai: AuthToken = " + token["accessToken"][:8] + "[...]") + + return token + +def registerDevice(token: dict, brand: str) -> None: + log.info("Kia/Hyundai: Registering DeviceId") + + try: + url = getString("base_url", brand) + '/api/v1/spa/notifications/' + token["deviceId"] + '/register' + data = {} + headers = { + 'Authorization': token["tokenType"] + ' ' + token["accessToken"], + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'Offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Content-Length': '0', + 'Host': getString("host", brand), + 'Connection': 'Keep-Alive', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = postHTTP(url=url, data=data, headers=headers) + except Exception as err: + log.error("kia.getAuthToken: Login failed") + log.debug("kia.getAuthToken: Login failed: " + response, exc_info = True) + raise + + return + +def requestToken(user_id: str, password: str, brand: str) -> dict: + log.info("Kia/Hyundai: Token request starting") + + try: + cookies = getCookies(brand) + token = getDeviceId(brand) + auth_code = getAuthCode(user_id, password, brand, cookies) + token = getAuthToken(auth_code, token, brand) + token["userHash"] = getUserHash(user_id, password) + registerDevice(token, brand) + except Exception as err: + log.error("kia.requestToken: Login failed") + log.debug("kia.requestToken: ", exc_info = True) + raise + + return token + +def refreshToken(token: dict, brand: str) -> dict: + log.info("Kia/Hyundai: Token refresh starting") + + try: + url = getString("base_url", brand) + '/api/v1/user/oauth2/token' + data = 'client_id='+ getString("client_id", brand) +'&grant_type=refresh_token&refresh_token='+ token["refreshToken"] +'&redirect_uri=' + \ + getString("base_url", brand) + '%2Fapi%2Fv1%2Fuser%2Foauth2%2Fredirect' + headers = { + 'Authorization': getString("basic_token", brand), + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'Offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Content-type': 'application/x-www-form-urlencoded', + 'Content-Length': str(len(data)), + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + + response = postHTTP(url=url, headers=headers, data=data) + + token_new = json.loads(response) + token["tokenType"] = token_new["token_type"] + token["accessToken"] = token_new["access_token"] + + except Exception as err: + log.error("kia.refreshToken: refresh token failed") + log.debug("kia.refreshToken: refresh token error: " + response, exc_info = True) + raise + + log.debug("kia.refreshToken: New access token = " + token["accessToken"][:8] + "...") + + return token + +def getControlToken(pin: str, token: dict, brand: str) -> str: + log.info("Kia/Hyundai: Sending PIN") + + try: + url = getString("base_url", brand) + '/api/v1/user/pin' + data = {"deviceId": token["deviceId"], "pin": pin} + headers = { + 'Authorization': token["tokenType"] + ' ' + token["accessToken"], + 'Content-type': 'application/json;charset=UTF-8', + 'Content-Length': str(len(data)), + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12' + } + response = putHTTP(url=url, data=data, headers=headers) + + response_dict = json.loads(response) + control_token = 'Bearer ' + response_dict['controlToken'] + except Exception as err: + log.error("kia.getControlToken: Sending PIN failed") + log.debug("kia.getControlToken: Sending PIN error: " + response, exc_info = True) + raise + + log.debug("kia.refreshToken: control token = " + control_token[7:15] + "...") + + return control_token + +#---------- API functions ---------- +def getVehicleId(vin: str, token: dict, brand: str) -> str: + log.info("Kia/Hyundai: Requesting vehicle list") + + try: + url = getString("base_url", brand) + '/api/v1/spa/vehicles' + headers = { + 'Authorization': token["tokenType"] + ' ' + token["accessToken"], + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = getHTTP(url=url, headers=headers) + + vehicle_id = "" + response_dict = json.loads(response) + for vehicle in response_dict['resMsg']['vehicles']: + if vehicle['vin'] == vin: + vehicle_id = vehicle['vehicleId'] + vehicle_name = vehicle['nickname'] + + if vehicle_id == "": + log.error("Kia/Hyundai: VIN " + vin + " unknown") + raise + + except Exception as err: + log.error("kia.getVehicleId: failed") + log.debug("kia.getVehicleId: error: " + response, exc_info = True) + raise + + log.debug("kia.getVehicleId: VehicleId = " + vehicle_id[:8] + "... (" + vehicle_name + ")") + + return vehicle_id + +def doPrewakeup(vehicle_id: str, token: dict, brand: str) -> None: + log.info("Kia/Hyundai: Triggering Pre-Wakeup") + + try: + url = getString("base_url", brand) + '/api/v1/spa/vehicles/' + vehicle_id + '/control/engine' + data = {"action": "prewakeup", "deviceId": token["deviceId"]} + headers = { + 'Authorization': token["tokenType"] + ' ' + token["accessToken"], + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'offset': '2', + 'Content-Type': 'application/json; charset=UTF-8', + 'Content-Length': str(len(data)), + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = postHTTP(url=url, data=data, headers=headers, timeout=125) + except Exception as err: + log.error("kia.doPrewakeup: failed") + log.debug("kia.doPrewakeup: error: " + response, exc_info = True) + raise + + return + +def getStatusFull(vehicle_id: str, control_token: str, token: dict, brand: str) -> CarState: + log.info("Kia/Hyundai: Triggering Update") + + try: + url = getString("base_url", brand) + '/api/v2/spa/vehicles/' + vehicle_id + '/ccs2/carstatus' + headers = { + 'Authorization': control_token, + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'offset': '2', + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = getHTTP(url=url, headers=headers, timeout=125) + except Exception as err: + log.error("kia.getStatusFull: triggering update failed") + log.debug("kia.getStatusFull: triggering update error: " + response, exc_info = True) + raise + + log.info("Kia/Hyundai: Waiting 130 seconds") + time.sleep(130) + + log.info("Kia/Hyundai: Receiving status") + + try: + url = getString("base_url", brand) + '/api/v1/spa/vehicles/' + vehicle_id + '/ccs2/carstatus/latest' + headers = { + 'Authorization': token["tokenType"] + ' ' + token["accessToken"], + 'Ccsp-Device-Id': token["deviceId"], + 'Ccsp-Service-Id': getString("client_id", brand), + 'Ccsp-Application-Id': getString("app_id", brand), + 'Clientid': token["gcmClientId"], + 'Vehicleid': token["gcmVehicleId"], + 'Ccuccs2protocolsupport': '0', + 'Offset': '2', + 'Host': getString("host", brand), + 'Connection': 'close', + 'Accept-Encoding': 'gzip, deflate', + 'User-Agent': 'okhttp/3.12.12', + 'Stamp': getStamp(brand) + } + response = getHTTP(url=url, headers=headers) + response_dict = json.loads(response) + + soc = int(response_dict['resMsg']['state']['Vehicle']['Green']['BatteryManagement']['BatteryRemain']['Ratio']) + range = float(response_dict['resMsg']['state']['Vehicle']['Drivetrain']['FuelSystem']['DTE']['Total']) + + except Exception as err: + log.error("kia.getStatusFull: receiving update failed") + log.debug("kia.getStatusFull: receiving update error: " + response, exc_info = True) + raise + + return CarState(soc, range) + +#---------- main function ---------- + +def fetch_soc(user_id: str, password: str, pin: str, vin: str, vehicle: int) -> CarState: + + log.info("Kia/Hyundai: Update starting") + + try: + brand = getBrand(vin) + token = loadToken(user_id, password, vehicle) + if token["accessToken"] == "": + token = requestToken(user_id, password, brand) + else: + token = refreshToken(token, brand) + saveToken(user_id, password, vehicle, token) + except Exception as err: + log.error("kia.fetch_soc: login error, vehicle: " + str(vehicle)) + log.debug("kia.fetch_soc: ", exc_info = True) + raise + + try: + vehicle_id = getVehicleId(vin, token, brand) + doPrewakeup(vehicle_id, token, brand) + control_token = getControlToken(pin, token, brand) + soc_state = getStatusFull(vehicle_id, control_token, token, brand) + except Exception as err: + log.error("kia.fetch_soc: API error, vehicle: " + str(vehicle)) + log.debug("kia.fetch_soc: ", exc_info = True) + raise + + return soc_state diff --git a/packages/modules/vehicles/kia/config.py b/packages/modules/vehicles/kia/config.py new file mode 100644 index 0000000000..dd6710ebc1 --- /dev/null +++ b/packages/modules/vehicles/kia/config.py @@ -0,0 +1,19 @@ +from typing import Optional + + +class KIAConfiguration: + def __init__(self, user_id: Optional[str] = None, password: Optional[str] = None, pin: Optional[str] = None, vin: Optional[str] = None): + self.user_id = user_id + self.password = password + self.pin = pin + self.vin = vin + + +class KIA: + def __init__(self, + name: str = "KIA / Hyundai (experimental)", + type: str = "kia", + configuration: KIAConfiguration = None) -> None: + self.name = name + self.type = type + self.configuration = configuration or KIAConfiguration() diff --git a/packages/modules/vehicles/kia/soc.py b/packages/modules/vehicles/kia/soc.py new file mode 100644 index 0000000000..07cafe312f --- /dev/null +++ b/packages/modules/vehicles/kia/soc.py @@ -0,0 +1,43 @@ +from typing import List + +import logging + +from helpermodules.cli import run_using_positional_cli_args +from modules.common import store +from modules.common.abstract_device import DeviceDescriptor +from modules.common.abstract_soc import SocUpdateData +from modules.common.component_state import CarState +from modules.common.configurable_vehicle import ConfigurableVehicle +from modules.vehicles.kia import api +from modules.vehicles.kia.config import KIA, KIAConfiguration + + +log = logging.getLogger(__name__) + + +def create_vehicle(vehicle_config: KIA, vehicle: int): + def updater(soc_update_data: SocUpdateData) -> CarState: + return api.fetch_soc( + vehicle_config.configuration.user_id, + vehicle_config.configuration.password, + vehicle_config.configuration.pin, + vehicle_config.configuration.vin, + vehicle) + return ConfigurableVehicle(vehicle_config=vehicle_config, component_updater=updater, vehicle=vehicle) + + +def kia_update(user_id: str, password: str, vin: str, charge_point: int): + vehicle_config = KIA(configuration=KIAConfiguration(charge_point, user_id, password, pin, vin)) + store.get_car_value_store(charge_point).store.set(api.fetch_soc( + vehicle_config.configuration.user_id, + vehicle_config.configuration.password, + vehicle_config.configuration.pin, + vehicle_config.configuration.vin, + charge_point)) + + +def main(argv: List[str]): + run_using_positional_cli_args(kia_update, argv) + + +device_descriptor = DeviceDescriptor(configuration_factory=KIA) From 76bf97adfa8f2ce93cf6bf5fc318d8987e72b784 Mon Sep 17 00:00:00 2001 From: Flock82 <81387035+Flock82@users.noreply.github.com> Date: Wed, 12 Jul 2023 21:36:07 +0200 Subject: [PATCH 3/6] Flake8 --- packages/modules/vehicles/kia/api.py | 456 +++++++++++++++--------- packages/modules/vehicles/kia/config.py | 4 +- 2 files changed, 282 insertions(+), 178 deletions(-) diff --git a/packages/modules/vehicles/kia/api.py b/packages/modules/vehicles/kia/api.py index e421609dc5..ae842d9035 100644 --- a/packages/modules/vehicles/kia/api.py +++ b/packages/modules/vehicles/kia/api.py @@ -16,7 +16,9 @@ log = logging.getLogger(__name__) -#---------- constants ---------- +# ---------- constants ---------- + + def getString(param_id: str, brand: str) -> str: if brand == 'kia': if param_id == "app_id": @@ -26,18 +28,23 @@ def getString(param_id: str, brand: str) -> str: elif param_id == "auth_client_id": paramStr = "572e0304-5f8d-4b4c-9dd5-41aa84eed160" elif param_id == "gcm_sender_id": - paramStr = "cF5o4DiiQkaw5wsAkLzYIS:APA91bFB59MltBMK29zI0U2llq7khbB2jELkNFKMfBCH6KlCPL16pz_dG0fZ4ncvFn1IMT8nfojb83JyLiT_skBTXtClHhDCKeRbyPy3yQjCVRC3zTZt--wI7vv4jD9aknhHhiQsoZoU" + paramStr = "cF5o4DiiQkaw5wsAkLzYIS:APA91bFB59MltBMK29zI0U2llq" + \ + "7khbB2jELkNFKMfBCH6KlCPL16pz_dG0fZ4ncvFn1IMT8nfoj" + \ + "b83JyLiT_skBTXtClHhDCKeRbyPy3yQjCVRC3zTZt--wI7vv4" + \ + "jD9aknhHhiQsoZoU" elif param_id == "basic_token": - paramStr = "Basic ZmRjODVjMDAtMGEyZi00YzY0LWJjYjQtMmNmYjE1MDA3MzBhOnNlY3JldA==" + paramStr = "Basic ZmRjODVjMDAtMGEyZi00YzY0LWJjYjQtMmNmYjE1MDA" + \ + "3MzBhOnNlY3JldA==" elif param_id == "stamp_pad": - paramStr = "C0B4D5C7089D987F027C96015929C70FA9D2B2AA99530CFD017E4B243C4BA5C5DED96DEB128EEB5DD3963DFC12432C9073EF" + paramStr = "C0B4D5C7089D987F027C96015929C70FA9D2B2AA99530CFD0" + \ + "17E4B243C4BA5C5DED96DEB128EEB5DD3963DFC12432C9073EF" elif param_id == "host": paramStr = "prd.eu-ccapi.kia.com:8080" elif param_id == "base_url": paramStr = "https://prd.eu-ccapi.kia.com:8080" else: raise RuntimeError - + if brand == 'hyundai': if param_id == "app_id": paramStr = "1eba27d2-9a5b-4eba-8ec7-97eb6c62fb51" @@ -48,20 +55,24 @@ def getString(param_id: str, brand: str) -> str: elif param_id == "gcm_sender_id": paramStr = "414998006775" elif param_id == "basic_token": - paramStr = "Basic NmQ0NzdjMzgtM2NhNC00Y2YzLTk1NTctMmExOTI5YTk0NjU0OktVeTQ5WHhQekxwTHVvSzB4aEJDNzdXNlZYaG10UVI5aVFobUlGampvWTRJcHhzVg==" + paramStr = "Basic NmQ0NzdjMzgtM2NhNC00Y2YzLTk1NTctMmExOTI5YTk" + \ + "0NjU0OktVeTQ5WHhQekxwTHVvSzB4aEJDNzdXNlZYaG10UVI5" + \ + "aVFobUlGampvWTRJcHhzVg==" elif param_id == "stamp_pad": - paramStr = "445B6846AFEF0D726646776865A650C9AEF98E51A474DCB7EC9B1B67D29C66EAAEF621CA02522A0B80A8087F7A3A7BB0F71B" + paramStr = "445B6846AFEF0D726646776865A650C9AEF98E51A474DCB7E" + \ + "C9B1B67D29C66EAAEF621CA02522A0B80A8087F7A3A7BB0F71B" elif param_id == "host": paramStr = "prd.eu-ccapi.hyundai.com:8080" elif param_id == "base_url": paramStr = "https://prd.eu-ccapi.hyundai.com:8080" else: raise RuntimeError - + return paramStr + def getBrand(vin: str) -> str: - #Determinate brand based on VIN + # Determinate brand based on VIN try: if vin[:2] == 'KN' or vin[:3] == 'U5Y' or vin[:3] == 'U6Z': brand = "kia" @@ -69,53 +80,61 @@ def getBrand(vin: str) -> str: brand = "hyundai" else: brand = "" - soclogging.logDebug(2, "Vehicle WMI unknown") + log.error("kia.getBrand: Vehicle WMI unknown") raise RuntimeError - except Exception as err: - log.error("kia.getBrand: VIN error", exc_info = True) + except Exception: + log.error("kia.getBrand: VIN error", exc_info=True) raise return brand - -#---------- stamp generation ---------- + +# ---------- stamp generation ---------- + + def getStamp(brand: str) -> str: try: - #Set App-ID and App-ID specific XOR-pad + # Set App-ID and App-ID specific XOR-pad app_id = getString("app_id", brand) secret_ba = bytearray.fromhex(getString("stamp_pad", brand)) - + # Timestamp in ms as string nowStr = str(int(time.time()) * 1000) - + # Combine plaintext and convert to bytearray plaintext = app_id + ":" + nowStr plaintext_ba = bytearray(plaintext.encode()) - + # XOR plaintext and key stamp_ba = bytes(a ^ b for (a, b) in zip(plaintext_ba, secret_ba)) - + # Convert result to base64-string stamp_b64_ba = base64.b64encode(stamp_ba) stamp = stamp_b64_ba.decode() - - except Exception as err: - log.error("kia.getStamp: stamp error", exc_info = True) + + except Exception: + log.error("kia.getStamp: stamp error", exc_info=True) raise return stamp -#---------- HTTP functions ---------- TBD! +# ---------- HTTP functions ---------- + + last_cookies = {} last_url = "" -def getHTTP(url: str = "", headers: str = "", cookies: str = "", timeout: int = 30, allow_redirects: bool = True) -> str: + +def getHTTP(url: str = "", headers: str = "", cookies: str = "", + timeout: int = 30, allow_redirects: bool = True) -> str: global last_cookies global last_url try: - response = requests.get(url, headers=headers, cookies=cookies, timeout=timeout, allow_redirects=allow_redirects) - except Exception as err: - log.error("kia.getHTTP: HTTP error", exc_info = True) + response = requests.get(url, headers=headers, + cookies=cookies, timeout=timeout, + allow_redirects=allow_redirects) + except Exception: + log.error("kia.getHTTP: HTTP error", exc_info=True) raise if response.status_code == 200 or response.status_code == 204: @@ -126,27 +145,35 @@ def getHTTP(url: str = "", headers: str = "", cookies: str = "", timeout: int = else: try: response_dict = json.loads(response.text) - if response.status_code == 400 or response.status_code == 408 or response.status_code == 503: - error_string = "[" + response_dict['resCode'] + "] " + response_dict['resMsg'] + if response.status_code == 400 or \ + response.status_code == 408 or \ + response.status_code == 503: + error_string = "[" + response_dict['resCode'] + "] " + \ + response_dict['resMsg'] else: - error_string = "[" + response_dict['errCode'] + "] " + response_dict['errMsg'] - except: + error_string = "[" + response_dict['errCode'] + "] " + \ + response_dict['errMsg'] + except Exception: error_string = "[XXXX] Unidentified Error" + " " + response.text - - log.debug("kia.getHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) + + log.debug("kia.getHTTP:Request failed, StatusCode: " + + str(response.status_code) + ', Error: ' + error_string) raise RuntimeError return "" -def putHTTP(url: str = "", data: Union[str, dict] = "", headers: str = "", cookies: str = "", timeout: int = 30) -> str: +def putHTTP(url: str = "", data: Union[str, dict] = "", + headers: str = "", cookies: str = "", timeout: int = 30) -> str: try: if isinstance(data, dict): - response = requests.put(url, json=data, headers=headers, cookies=cookies, timeout=timeout) + response = requests.put(url, json=data, headers=headers, + cookies=cookies, timeout=timeout) else: - response = requests.put(url, data=data, headers=headers, cookies=cookies, timeout=timeout) - except Exception as err: - log.error("kia.putHTTP: HTTP error", exc_info = True) + response = requests.put(url, data=data, headers=headers, + cookies=cookies, timeout=timeout) + except Exception: + log.error("kia.putHTTP: HTTP error", exc_info=True) raise if response.status_code == 200 or response.status_code == 204: @@ -155,40 +182,50 @@ def putHTTP(url: str = "", data: Union[str, dict] = "", headers: str = "", cooki try: response_dict = json.loads(response.text) if response.status_code == 408: - error_string = "[" + response_dict['resCode'] + "] " + response_dict['resMsg'] + error_string = "[" + response_dict['resCode'] + "] " + \ + response_dict['resMsg'] else: - error_string = "[" + response_dict['errCode'] + "] " + response_dict['errMsg'] - except: + error_string = "[" + response_dict['errCode'] + "] " + \ + response_dict['errMsg'] + except Exception: error_string = "[XXXX] Unidentified Error" - log.debug("kia.putHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) + log.debug("kia.putHTTP:Request failed, StatusCode: " + + str(response.status_code) + ', Error: ' + error_string) raise RuntimeError return "" - -def deleteHTTP(url: str = "", headers: str = "", cookies: str = "", timeout: int = 30) -> None: + + +def deleteHTTP(url: str = "", headers: str = "", cookies: str = "", + timeout: int = 30) -> None: try: - response = requests.delete(url, headers=headers, cookies=cookies, timeout=timeout) - except Exception as err: - log.error("kia.putHTTP: HTTP error", exc_info = True) + response = requests.delete(url, headers=headers, cookies=cookies, + timeout=timeout) + except Exception: + log.error("kia.putHTTP: HTTP error: " + response, exc_info=True) raise return -def postHTTP(url: str = "", data: Union[str, dict] = "", headers: str = "", cookies: str = "", timeout: int = 30, allow_redirects: bool = True) -> str: +def postHTTP(url: str = "", data: Union[str, dict] = "", + headers: str = "", cookies: str = "", timeout: int = 30, + allow_redirects: bool = True) -> str: global last_cookies global last_url try: if isinstance(data, dict): - response = requests.post(url, json=data, headers=headers, cookies=cookies, - timeout=timeout, allow_redirects=allow_redirects) + response = requests.post(url, json=data, headers=headers, + cookies=cookies, timeout=timeout, + allow_redirects=allow_redirects) else: - response = requests.post(url, data=data, headers=headers, cookies=cookies, - timeout=timeout, allow_redirects=allow_redirects) - except Exception as err: - log.error("kia.postHTTP: HTTP error", exc_info = True) + response = requests.post(url, data=data, headers=headers, + cookies=cookies, timeout=timeout, + allow_redirects=allow_redirects) + except Exception: + log.error("kia.postHTTP: HTTP error", exc_info=True) raise if response.status_code == 200 or response.status_code == 204: @@ -200,13 +237,16 @@ def postHTTP(url: str = "", data: Union[str, dict] = "", headers: str = "", cook try: response_dict = json.loads(response.text) if response.status_code == 408: - error_string = "[" + response_dict['resCode'] + "] " + response_dict['resMsg'] + error_string = "[" + response_dict['resCode'] + "] " + \ + response_dict['resMsg'] else: - error_string = "[" + response_dict['errCode'] + "] " + response_dict['errMsg'] - except: + error_string = "[" + response_dict['errCode'] + "] " + \ + response_dict['errMsg'] + except Exception: error_string = "[XXXX] Unidentified Error" - log.debug("kia.postHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) + log.debug("kia.postHTTP:Request failed, StatusCode: " + + str(response.status_code) + ', Error: ' + error_string) raise RuntimeError return "" @@ -216,41 +256,46 @@ def getHTTPCookies(url: str = "") -> dict: try: session = requests.Session() response = session.get(url) - except Exception as err: - log.error("kia.getHTTPCookies: HTTP error", exc_info = True) + except Exception: + log.error("kia.getHTTPCookies: HTTP error", exc_info=True) raise if response.status_code == 200: cookies = session.cookies.get_dict() else: - log.debug("kia.getHTTPCookies: Receiving cookies failed, StatusCode: " + str(response.status_code)) + log.debug("kia.getHTTPCookies: Receiving cookies failed, StatusCode:" + + " " + str(response.status_code)) raise RuntimeError return cookies - -#---------- token management ---------- + +# ---------- token management ---------- + + def getUserHash(user_id: str, password: str) -> str: try: account = user_id + ':' + password hash = hashlib.md5(account.encode()).hexdigest() - except Exception as err: - log.error("kia.getUserHash: hash error", exc_info = True) + except Exception: + log.error("kia.getUserHash: hash error", exc_info=True) raise return hash + def loadToken(user_id: str, password: str, vehicle: int) -> dict: try: - token_file = str(RAMDISK_PATH) + "/soc_kia_vehicle" + str(vehicle) + "_token" + token_file = str(RAMDISK_PATH) + "/soc_kia_vehicle" + \ + str(vehicle) + "_token" with open(token_file, 'r', encoding='utf-8') as f: token = json.loads(f.read()) - except Exception as err: - log.debug("kia.loadToken: token file error: ", exc_info = True) + except Exception: + log.debug("kia.loadToken: token file error: ", exc_info=True) token = { "userHash": "" } pass - + try: if token["userHash"] != getUserHash(user_id, password): log.debug("kia.loadToken: account data changed") @@ -263,59 +308,72 @@ def loadToken(user_id: str, password: str, vehicle: int) -> dict: "gcmClientId": "", "tokenType": "" } - except Exception as err: - log.error("kia.loadToken: token error", exc_info = True) + except Exception: + log.error("kia.loadToken: token error", exc_info=True) raise - + return token + def saveToken(user_id: str, password: str, vehicle: int, token: dict) -> None: try: token["userHash"] = getUserHash(user_id, password) - token_file = str(RAMDISK_PATH) + "/soc_kia_vehicle" + str(vehicle) + "_token" + token_file = str(RAMDISK_PATH) + "/soc_kia_vehicle" + \ + str(vehicle) + "_token" with open(token_file, 'w', encoding='utf-8') as f: f.write(json.dumps(token)) - except Exception as err: - log.debug("kia.saveToken: token could not be saved", exc_info = True) + except Exception: + log.debug("kia.saveToken: token could not be saved", exc_info=True) raise -#---------- authentication ---------- +# ---------- authentication ---------- + + def getCookies(brand: str) -> dict: log.info("Kia/Hyundai: Create Login-session") try: - url = getString("base_url", brand) + '/api/v1/user/oauth2/authorize?response_type=code&state=test&client_id=' + \ - getString("client_id", brand) + '&redirect_uri=' + getString("base_url", brand) + \ - '/api/v1/user/oauth2/redirect' + url = getString("base_url", brand) + \ + '/api/v1/user/oauth2/authorize?' + \ + 'response_type=code&state=test&client_id=' + \ + getString("client_id", brand) + '&redirect_uri=' + \ + getString("base_url", brand) + '/api/v1/user/oauth2/redirect' cookies = getHTTPCookies(url) - + url = getString("base_url", brand) + '/api/v1/user/session' getHTTP(url=url, cookies=cookies) url = getString("base_url", brand) + '/api/v1/user/language' headers = {'Content-type': 'application/json'} data = {"lang": "en"} - response = postHTTP(url=url, data=data, headers=headers, cookies=cookies) - + response = postHTTP(url=url, data=data, + headers=headers, cookies=cookies) + url = getString("base_url", brand) + '/api/v1/user/session' deleteHTTP(url=url, cookies=cookies) - except Exception as err: - log.debug("kia.getCookies: ", exc_info = True) + except Exception: + log.debug("kia.getCookies: " + response, exc_info=True) raise - + return cookies + def getDeviceId(brand: str) -> dict: log.info("Kia/Hyundai: Requesting device ids") - + token = {} - + try: token["gcmClientId"] = str(uuid.uuid4()) token["gcmVehicleId"] = str(uuid.uuid4()) - - url = getString("base_url", brand) + '/api/v1/spa/notifications/register' - data = {"pushRegId": getString("gcm_sender_id", brand), "pushType": "GCM", "uuid": str(uuid.uuid4())} + + url = getString("base_url", brand) + \ + '/api/v1/spa/notifications/register' + data = { + "pushRegId": getString("gcm_sender_id", brand), + "pushType": "GCM", + "uuid": str(uuid.uuid4()) + } headers = { 'Authorization': '', 'Ccsp-Device-Id': '', @@ -337,14 +395,14 @@ def getDeviceId(brand: str) -> dict: response_dict = json.loads(response) token["deviceId"] = response_dict['resMsg']['deviceId'] - + log.debug("Kia/Hyundai: DeviceId = " + token["deviceId"][:8] + "[...]") - except Exception as err: + except Exception: log.error("kia.getDeviceId: Request device id failed") - log.debug("kia.getDeviceId: Request device id failed: " + response, exc_info = True) - raise + log.debug("kia.getDeviceId: Request device id failed: " + + response, exc_info=True) + raise - log.info("Sending VersionInfo") try: url = getString("base_url", brand) + '/api/v1/spa/devices/version' @@ -373,12 +431,13 @@ def getDeviceId(brand: str) -> dict: 'User-Agent': 'okhttp/3.12.12', 'Stamp': getStamp(brand) } - response = postHTTP(url=url, data=data, headers=headers) - except Exception as err: + response = postHTTP(url=url, data=data, headers=headers) + except Exception: log.error("kia.getDeviceId: Set version info failed") - log.debug("kia.getDeviceId: Set version info failed: " + response, exc_info = True) - raise - + log.debug("kia.getDeviceId: Set version info failed: " + + response, exc_info=True) + raise + token["userHash"] = "" token["tokenType"] = "" token["accessToken"] = "" @@ -386,9 +445,11 @@ def getDeviceId(brand: str) -> dict: return token -def getAuthCode(username: str, password: str, brand: str, cookies: dict) -> str: + +def getAuthCode(username: str, password: str, brand: str, + cookies: dict) -> str: global last_cookies - + log.info("Kia/Hyundai: Sending username/password") try: @@ -402,50 +463,64 @@ def getAuthCode(username: str, password: str, brand: str, cookies: dict) -> str: log.debug("Kia/Hyundai: UserId = " + user_id[:8] + "[...]") log.debug("Kia/Hyundai: ServiceId = " + service_id[:8] + "[...]") - url = 'https://eu-account.' + brand + '.com/auth/realms/eu' + brand + 'idm/protocol/openid-connect/auth?client_id=' + \ - getString("auth_client_id", brand) + '&scope=openid%20profile%20email%20phone&response_type=code&hkid_session_reset=true&redirect_uri=' + \ - getString("base_url", brand) + '/api/v1/user/integration/redirect/login&ui_locales=en&state=' + service_id + ':' + user_id + url = 'https://eu-account.' + brand + '.com/auth/realms/eu' + brand + \ + 'idm/protocol/openid-connect/auth?client_id=' + \ + getString("auth_client_id", brand) + \ + '&scope=openid%20profile%20email%20phone&response_type=code&' + \ + 'hkid_session_reset=true&redirect_uri=' + \ + getString("base_url", brand) + '/api/v1/user/integration/' + \ + 'redirect/login&ui_locales=en&state=' + \ + service_id + ':' + user_id headers = {} response = getHTTP(url=url, headers=headers, cookies=cookies) left = response.find('action="') + 8 right = response.find('"', left) url = response[left:right].replace('&', '&') - data = urlparse.urlencode({'username': username, 'password': password, 'credentialId': ''}) + data = urlparse.urlencode({'username': username, 'password': password, + 'credentialId': ''}) headers = { 'Content-type': 'application/x-www-form-urlencoded', - 'User-Agent': 'Mozilla/5.0 (Linux; Android 4.1.1; Galaxy Nexus Build/JRO03C) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.166 Mobile Safari/535.19_CCS_APP_AOS' + 'User-Agent': 'Mozilla/5.0 (Linux; Android 4.1.1; Galaxy Nexus ' + + 'Build/JRO03C) AppleWebKit/535.19 (KHTML, ' + + 'like Gecko) Chrome/18.0.1025.166 Mobile ' + + 'Safari/535.19_CCS_APP_AOS' } cookies['AUTH_SESSION_ID'] = last_cookies['AUTH_SESSION_ID'] - response = postHTTP(url=url, data=data, headers=headers, cookies=cookies, allow_redirects=False) - + response = postHTTP(url=url, data=data, headers=headers, + cookies=cookies, allow_redirects=False) + url = response response = getHTTP(url=url, cookies=cookies, allow_redirects=True) - + url = getString("base_url", brand) + '/api/v1/user/silentsignin' headers = {'Content-type': 'text/plain;charset=UTF-8'} data = {'intUserId': ""} - response = postHTTP(url=url, data=data, headers=headers, cookies=cookies, allow_redirects=False) + response = postHTTP(url=url, data=data, headers=headers, + cookies=cookies, allow_redirects=False) response_dict = json.loads(response) response_url = response_dict['redirectUrl'] parsed = urlparse.urlparse(response_url) auth_code = ''.join(parse_qs(parsed.query)['code']) - except Exception as err: + except Exception: log.error("kia.getAuthCode: Login failed") - log.debug("kia.getAuthCode: Login failed: " + response, exc_info = True) - raise + log.debug("kia.getAuthCode: Login failed: " + response, exc_info=True) + raise log.debug("Kia/Hyundai: AuthCode = " + auth_code[:8] + "[...]") return auth_code + def getAuthToken(auth_code: str, token: dict, brand: str) -> dict: log.info("Kia/Hyundai: Requesting access token") try: url = getString("base_url", brand) + '/api/v1/user/oauth2/token' - data = 'client_id=' +getString("client_id", brand) +'&grant_type=authorization_code&code=' + \ - auth_code +'&redirect_uri=' + getString("base_url", brand) + '%2Fapi%2Fv1%2Fuser%2Foauth2%2Fredirect' + data = 'client_id=' + getString("client_id", brand) + \ + '&grant_type=authorization_code&code=' + \ + auth_code + '&redirect_uri=' + getString("base_url", brand) + \ + '%2Fapi%2Fv1%2Fuser%2Foauth2%2Fredirect' headers = { 'Authorization': getString("basic_token", brand), 'Ccsp-Device-Id': token["deviceId"], @@ -464,26 +539,28 @@ def getAuthToken(auth_code: str, token: dict, brand: str) -> dict: 'Stamp': getStamp(brand) } response = postHTTP(url=url, headers=headers, data=data) - + access_token = json.loads(response) token["tokenType"] = access_token["token_type"] token["accessToken"] = access_token["access_token"] token["refreshToken"] = access_token["refresh_token"] - - except Exception as err: + + except Exception: log.error("kia.getAuthToken: Login failed") - log.debug("kia.getAuthToken: Login failed: " + response, exc_info = True) - raise + log.debug("kia.getAuthToken: Login failed: " + response, exc_info=True) + raise log.debug("Kia/Hyundai: AuthToken = " + token["accessToken"][:8] + "[...]") return token - + + def registerDevice(token: dict, brand: str) -> None: - log.info("Kia/Hyundai: Registering DeviceId") - + log.info("Kia/Hyundai: Registering DeviceId") + try: - url = getString("base_url", brand) + '/api/v1/spa/notifications/' + token["deviceId"] + '/register' + url = getString("base_url", brand) + '/api/v1/spa/notifications/' + \ + token["deviceId"] + '/register' data = {} headers = { 'Authorization': token["tokenType"] + ' ' + token["accessToken"], @@ -502,16 +579,17 @@ def registerDevice(token: dict, brand: str) -> None: 'Stamp': getStamp(brand) } response = postHTTP(url=url, data=data, headers=headers) - except Exception as err: + except Exception: log.error("kia.getAuthToken: Login failed") - log.debug("kia.getAuthToken: Login failed: " + response, exc_info = True) - raise + log.debug("kia.getAuthToken: Login failed: " + response, exc_info=True) + raise return - + + def requestToken(user_id: str, password: str, brand: str) -> dict: log.info("Kia/Hyundai: Token request starting") - + try: cookies = getCookies(brand) token = getDeviceId(brand) @@ -519,20 +597,24 @@ def requestToken(user_id: str, password: str, brand: str) -> dict: token = getAuthToken(auth_code, token, brand) token["userHash"] = getUserHash(user_id, password) registerDevice(token, brand) - except Exception as err: + except Exception: log.error("kia.requestToken: Login failed") - log.debug("kia.requestToken: ", exc_info = True) + log.debug("kia.requestToken: ", exc_info=True) raise - + return token - + + def refreshToken(token: dict, brand: str) -> dict: log.info("Kia/Hyundai: Token refresh starting") - try: + try: url = getString("base_url", brand) + '/api/v1/user/oauth2/token' - data = 'client_id='+ getString("client_id", brand) +'&grant_type=refresh_token&refresh_token='+ token["refreshToken"] +'&redirect_uri=' + \ - getString("base_url", brand) + '%2Fapi%2Fv1%2Fuser%2Foauth2%2Fredirect' + data = 'client_id=' + getString("client_id", brand) + \ + '&grant_type=refresh_token&refresh_token=' + \ + token["refreshToken"] + \ + '&redirect_uri=' + getString("base_url", brand) + \ + '%2Fapi%2Fv1%2Fuser%2Foauth2%2Fredirect' headers = { 'Authorization': getString("basic_token", brand), 'Ccsp-Device-Id': token["deviceId"], @@ -550,22 +632,25 @@ def refreshToken(token: dict, brand: str) -> dict: 'User-Agent': 'okhttp/3.12.12', 'Stamp': getStamp(brand) } - + response = postHTTP(url=url, headers=headers, data=data) - + token_new = json.loads(response) token["tokenType"] = token_new["token_type"] token["accessToken"] = token_new["access_token"] - except Exception as err: + except Exception: log.error("kia.refreshToken: refresh token failed") - log.debug("kia.refreshToken: refresh token error: " + response, exc_info = True) + log.debug("kia.refreshToken: refresh token error: " + + response, exc_info=True) raise - log.debug("kia.refreshToken: New access token = " + token["accessToken"][:8] + "...") - + log.debug("kia.refreshToken: New access token = " + + token["accessToken"][:8] + "...") + return token + def getControlToken(pin: str, token: dict, brand: str) -> str: log.info("Kia/Hyundai: Sending PIN") @@ -585,16 +670,20 @@ def getControlToken(pin: str, token: dict, brand: str) -> str: response_dict = json.loads(response) control_token = 'Bearer ' + response_dict['controlToken'] - except Exception as err: + except Exception: log.error("kia.getControlToken: Sending PIN failed") - log.debug("kia.getControlToken: Sending PIN error: " + response, exc_info = True) + log.debug("kia.getControlToken: Sending PIN error: " + + response, exc_info=True) raise - - log.debug("kia.refreshToken: control token = " + control_token[7:15] + "...") + + log.debug("kia.refreshToken: control token = " + + control_token[7:15] + "...") return control_token - -#---------- API functions ---------- + +# ---------- API functions ---------- + + def getVehicleId(vin: str, token: dict, brand: str) -> str: log.info("Kia/Hyundai: Requesting vehicle list") @@ -616,32 +705,35 @@ def getVehicleId(vin: str, token: dict, brand: str) -> str: 'Stamp': getStamp(brand) } response = getHTTP(url=url, headers=headers) - + vehicle_id = "" response_dict = json.loads(response) for vehicle in response_dict['resMsg']['vehicles']: if vehicle['vin'] == vin: vehicle_id = vehicle['vehicleId'] vehicle_name = vehicle['nickname'] - + if vehicle_id == "": log.error("Kia/Hyundai: VIN " + vin + " unknown") raise - - except Exception as err: + + except Exception: log.error("kia.getVehicleId: failed") - log.debug("kia.getVehicleId: error: " + response, exc_info = True) + log.debug("kia.getVehicleId: error: " + response, exc_info=True) raise - log.debug("kia.getVehicleId: VehicleId = " + vehicle_id[:8] + "... (" + vehicle_name + ")") + log.debug("kia.getVehicleId: VehicleId = " + vehicle_id[:8] + + "... (" + vehicle_name + ")") return vehicle_id + def doPrewakeup(vehicle_id: str, token: dict, brand: str) -> None: log.info("Kia/Hyundai: Triggering Pre-Wakeup") try: - url = getString("base_url", brand) + '/api/v1/spa/vehicles/' + vehicle_id + '/control/engine' + url = getString("base_url", brand) + '/api/v1/spa/vehicles/' +\ + vehicle_id + '/control/engine' data = {"action": "prewakeup", "deviceId": token["deviceId"]} headers = { 'Authorization': token["tokenType"] + ' ' + token["accessToken"], @@ -658,18 +750,21 @@ def doPrewakeup(vehicle_id: str, token: dict, brand: str) -> None: 'Stamp': getStamp(brand) } response = postHTTP(url=url, data=data, headers=headers, timeout=125) - except Exception as err: + except Exception: log.error("kia.doPrewakeup: failed") - log.debug("kia.doPrewakeup: error: " + response, exc_info = True) + log.debug("kia.doPrewakeup: error: " + response, exc_info=True) raise return -def getStatusFull(vehicle_id: str, control_token: str, token: dict, brand: str) -> CarState: + +def getStatusFull(vehicle_id: str, control_token: str, + token: dict, brand: str) -> CarState: log.info("Kia/Hyundai: Triggering Update") try: - url = getString("base_url", brand) + '/api/v2/spa/vehicles/' + vehicle_id + '/ccs2/carstatus' + url = getString("base_url", brand) + '/api/v2/spa/vehicles/' + \ + vehicle_id + '/ccs2/carstatus' headers = { 'Authorization': control_token, 'Ccsp-Device-Id': token["deviceId"], @@ -686,9 +781,10 @@ def getStatusFull(vehicle_id: str, control_token: str, token: dict, brand: str) 'Stamp': getStamp(brand) } response = getHTTP(url=url, headers=headers, timeout=125) - except Exception as err: + except Exception: log.error("kia.getStatusFull: triggering update failed") - log.debug("kia.getStatusFull: triggering update error: " + response, exc_info = True) + log.debug("kia.getStatusFull: triggering update error: " + + response, exc_info=True) raise log.info("Kia/Hyundai: Waiting 130 seconds") @@ -697,7 +793,8 @@ def getStatusFull(vehicle_id: str, control_token: str, token: dict, brand: str) log.info("Kia/Hyundai: Receiving status") try: - url = getString("base_url", brand) + '/api/v1/spa/vehicles/' + vehicle_id + '/ccs2/carstatus/latest' + url = getString("base_url", brand) + '/api/v1/spa/vehicles/' + \ + vehicle_id + '/ccs2/carstatus/latest' headers = { 'Authorization': token["tokenType"] + ' ' + token["accessToken"], 'Ccsp-Device-Id': token["deviceId"], @@ -715,20 +812,25 @@ def getStatusFull(vehicle_id: str, control_token: str, token: dict, brand: str) } response = getHTTP(url=url, headers=headers) response_dict = json.loads(response) - - soc = int(response_dict['resMsg']['state']['Vehicle']['Green']['BatteryManagement']['BatteryRemain']['Ratio']) - range = float(response_dict['resMsg']['state']['Vehicle']['Drivetrain']['FuelSystem']['DTE']['Total']) - - except Exception as err: + + soc = int(response_dict['resMsg']['state']['Vehicle']['Green'] + ['BatteryManagement']['BatteryRemain']['Ratio']) + range = float(response_dict['resMsg']['state']['Vehicle']['Drivetrain'] + ['FuelSystem']['DTE']['Total']) + + except Exception: log.error("kia.getStatusFull: receiving update failed") - log.debug("kia.getStatusFull: receiving update error: " + response, exc_info = True) + log.debug("kia.getStatusFull: receiving update error: " + + response, exc_info=True) raise return CarState(soc, range) - -#---------- main function ---------- -def fetch_soc(user_id: str, password: str, pin: str, vin: str, vehicle: int) -> CarState: +# ---------- main function ---------- + + +def fetch_soc(user_id: str, password: str, pin: str, + vin: str, vehicle: int) -> CarState: log.info("Kia/Hyundai: Update starting") @@ -740,19 +842,19 @@ def fetch_soc(user_id: str, password: str, pin: str, vin: str, vehicle: int) -> else: token = refreshToken(token, brand) saveToken(user_id, password, vehicle, token) - except Exception as err: + except Exception: log.error("kia.fetch_soc: login error, vehicle: " + str(vehicle)) - log.debug("kia.fetch_soc: ", exc_info = True) + log.debug("kia.fetch_soc: ", exc_info=True) raise - + try: vehicle_id = getVehicleId(vin, token, brand) doPrewakeup(vehicle_id, token, brand) control_token = getControlToken(pin, token, brand) soc_state = getStatusFull(vehicle_id, control_token, token, brand) - except Exception as err: + except Exception: log.error("kia.fetch_soc: API error, vehicle: " + str(vehicle)) - log.debug("kia.fetch_soc: ", exc_info = True) + log.debug("kia.fetch_soc: ", exc_info=True) raise return soc_state diff --git a/packages/modules/vehicles/kia/config.py b/packages/modules/vehicles/kia/config.py index dd6710ebc1..7c6c370ca0 100644 --- a/packages/modules/vehicles/kia/config.py +++ b/packages/modules/vehicles/kia/config.py @@ -2,7 +2,9 @@ class KIAConfiguration: - def __init__(self, user_id: Optional[str] = None, password: Optional[str] = None, pin: Optional[str] = None, vin: Optional[str] = None): + def __init__(self, user_id: Optional[str] = None, password: Optional[str] = None, + pin: Optional[str] = None, vin: Optional[str] = None): + self.user_id = user_id self.password = password self.pin = pin From 4f1cf88099b30a193a298ca3fd5f90740ceceff7 Mon Sep 17 00:00:00 2001 From: Flock82 <81387035+Flock82@users.noreply.github.com> Date: Wed, 12 Jul 2023 21:39:26 +0200 Subject: [PATCH 4/6] Flake8 --- packages/modules/vehicles/kia/config.py | 4 ++-- packages/modules/vehicles/kia/soc.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/modules/vehicles/kia/config.py b/packages/modules/vehicles/kia/config.py index 7c6c370ca0..8277209c3e 100644 --- a/packages/modules/vehicles/kia/config.py +++ b/packages/modules/vehicles/kia/config.py @@ -3,8 +3,8 @@ class KIAConfiguration: def __init__(self, user_id: Optional[str] = None, password: Optional[str] = None, - pin: Optional[str] = None, vin: Optional[str] = None): - + pin: Optional[str] = None, vin: Optional[str] = None): + self.user_id = user_id self.password = password self.pin = pin diff --git a/packages/modules/vehicles/kia/soc.py b/packages/modules/vehicles/kia/soc.py index 07cafe312f..c00a04051e 100644 --- a/packages/modules/vehicles/kia/soc.py +++ b/packages/modules/vehicles/kia/soc.py @@ -26,7 +26,7 @@ def updater(soc_update_data: SocUpdateData) -> CarState: return ConfigurableVehicle(vehicle_config=vehicle_config, component_updater=updater, vehicle=vehicle) -def kia_update(user_id: str, password: str, vin: str, charge_point: int): +def kia_update(user_id: str, password: str, pin: str, vin: str, charge_point: int): vehicle_config = KIA(configuration=KIAConfiguration(charge_point, user_id, password, pin, vin)) store.get_car_value_store(charge_point).store.set(api.fetch_soc( vehicle_config.configuration.user_id, From 041eced73c5da090cd0c29a6eae64b39e9c16ae7 Mon Sep 17 00:00:00 2001 From: Flock82 <81387035+Flock82@users.noreply.github.com> Date: Wed, 12 Jul 2023 21:43:02 +0200 Subject: [PATCH 5/6] Flake8 --- packages/modules/vehicles/kia/__init__.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/modules/vehicles/kia/__init__.py b/packages/modules/vehicles/kia/__init__.py index 8b13789179..e69de29bb2 100644 --- a/packages/modules/vehicles/kia/__init__.py +++ b/packages/modules/vehicles/kia/__init__.py @@ -1 +0,0 @@ - From 1eb92155a0465d4bdebd612f1a6e84cf3279f5b3 Mon Sep 17 00:00:00 2001 From: Flock82 <81387035+Flock82@users.noreply.github.com> Date: Thu, 27 Jul 2023 09:29:41 +0200 Subject: [PATCH 6/6] Logging adjusted --- packages/modules/vehicles/kia/api.py | 89 ++++++++++++---------------- 1 file changed, 39 insertions(+), 50 deletions(-) diff --git a/packages/modules/vehicles/kia/api.py b/packages/modules/vehicles/kia/api.py index ae842d9035..a82a109c5c 100644 --- a/packages/modules/vehicles/kia/api.py +++ b/packages/modules/vehicles/kia/api.py @@ -83,7 +83,7 @@ def getBrand(vin: str) -> str: log.error("kia.getBrand: Vehicle WMI unknown") raise RuntimeError except Exception: - log.error("kia.getBrand: VIN error", exc_info=True) + log.exception("kia.getBrand: VIN error") raise return brand @@ -112,7 +112,7 @@ def getStamp(brand: str) -> str: stamp = stamp_b64_ba.decode() except Exception: - log.error("kia.getStamp: stamp error", exc_info=True) + log.exception("kia.getStamp: stamp error") raise return stamp @@ -134,7 +134,7 @@ def getHTTP(url: str = "", headers: str = "", cookies: str = "", cookies=cookies, timeout=timeout, allow_redirects=allow_redirects) except Exception: - log.error("kia.getHTTP: HTTP error", exc_info=True) + log.exception("kia.getHTTP: HTTP error") raise if response.status_code == 200 or response.status_code == 204: @@ -155,8 +155,9 @@ def getHTTP(url: str = "", headers: str = "", cookies: str = "", response_dict['errMsg'] except Exception: error_string = "[XXXX] Unidentified Error" + " " + response.text + log.exception("kia.getHTTP:Request failed") - log.debug("kia.getHTTP:Request failed, StatusCode: " + + log.error("kia.getHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) raise RuntimeError @@ -173,7 +174,7 @@ def putHTTP(url: str = "", data: Union[str, dict] = "", response = requests.put(url, data=data, headers=headers, cookies=cookies, timeout=timeout) except Exception: - log.error("kia.putHTTP: HTTP error", exc_info=True) + log.exception("kia.putHTTP: HTTP error") raise if response.status_code == 200 or response.status_code == 204: @@ -189,8 +190,9 @@ def putHTTP(url: str = "", data: Union[str, dict] = "", response_dict['errMsg'] except Exception: error_string = "[XXXX] Unidentified Error" + log.exception("kia.putHTTP:Request failed") - log.debug("kia.putHTTP:Request failed, StatusCode: " + + log.error("kia.putHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) raise RuntimeError @@ -203,7 +205,7 @@ def deleteHTTP(url: str = "", headers: str = "", cookies: str = "", response = requests.delete(url, headers=headers, cookies=cookies, timeout=timeout) except Exception: - log.error("kia.putHTTP: HTTP error: " + response, exc_info=True) + log.exception("kia.deleteHTTP: HTTP error: " + response) raise return @@ -225,7 +227,7 @@ def postHTTP(url: str = "", data: Union[str, dict] = "", cookies=cookies, timeout=timeout, allow_redirects=allow_redirects) except Exception: - log.error("kia.postHTTP: HTTP error", exc_info=True) + log.exception("kia.postHTTP: HTTP error") raise if response.status_code == 200 or response.status_code == 204: @@ -244,8 +246,9 @@ def postHTTP(url: str = "", data: Union[str, dict] = "", response_dict['errMsg'] except Exception: error_string = "[XXXX] Unidentified Error" + log.exception("kia.postHTTP:Request failed") - log.debug("kia.postHTTP:Request failed, StatusCode: " + + log.error("kia.postHTTP:Request failed, StatusCode: " + str(response.status_code) + ', Error: ' + error_string) raise RuntimeError @@ -257,13 +260,13 @@ def getHTTPCookies(url: str = "") -> dict: session = requests.Session() response = session.get(url) except Exception: - log.error("kia.getHTTPCookies: HTTP error", exc_info=True) + log.exception("kia.getHTTPCookies: HTTP error") raise if response.status_code == 200: cookies = session.cookies.get_dict() else: - log.debug("kia.getHTTPCookies: Receiving cookies failed, StatusCode:" + + log.error("kia.getHTTPCookies: Receiving cookies failed, StatusCode:" + " " + str(response.status_code)) raise RuntimeError @@ -277,7 +280,7 @@ def getUserHash(user_id: str, password: str) -> str: account = user_id + ':' + password hash = hashlib.md5(account.encode()).hexdigest() except Exception: - log.error("kia.getUserHash: hash error", exc_info=True) + log.exception("kia.getUserHash: hash error") raise return hash @@ -290,7 +293,7 @@ def loadToken(user_id: str, password: str, vehicle: int) -> dict: with open(token_file, 'r', encoding='utf-8') as f: token = json.loads(f.read()) except Exception: - log.debug("kia.loadToken: token file error: ", exc_info=True) + log.exception("kia.loadToken: token file error: ") token = { "userHash": "" } @@ -309,7 +312,7 @@ def loadToken(user_id: str, password: str, vehicle: int) -> dict: "tokenType": "" } except Exception: - log.error("kia.loadToken: token error", exc_info=True) + log.exception("kia.loadToken: token error") raise return token @@ -323,7 +326,7 @@ def saveToken(user_id: str, password: str, vehicle: int, token: dict) -> None: with open(token_file, 'w', encoding='utf-8') as f: f.write(json.dumps(token)) except Exception: - log.debug("kia.saveToken: token could not be saved", exc_info=True) + log.exception("kia.saveToken: token could not be saved") raise # ---------- authentication ---------- @@ -352,7 +355,7 @@ def getCookies(brand: str) -> dict: url = getString("base_url", brand) + '/api/v1/user/session' deleteHTTP(url=url, cookies=cookies) except Exception: - log.debug("kia.getCookies: " + response, exc_info=True) + log.exception("kia.getCookies: " + response) raise return cookies @@ -398,9 +401,8 @@ def getDeviceId(brand: str) -> dict: log.debug("Kia/Hyundai: DeviceId = " + token["deviceId"][:8] + "[...]") except Exception: - log.error("kia.getDeviceId: Request device id failed") - log.debug("kia.getDeviceId: Request device id failed: " + - response, exc_info=True) + log.exception("kia.getDeviceId: Request device id failed: " + + response) raise log.info("Sending VersionInfo") @@ -433,9 +435,8 @@ def getDeviceId(brand: str) -> dict: } response = postHTTP(url=url, data=data, headers=headers) except Exception: - log.error("kia.getDeviceId: Set version info failed") - log.debug("kia.getDeviceId: Set version info failed: " + - response, exc_info=True) + log.exception("kia.getDeviceId: Set version info failed: " + + response) raise token["userHash"] = "" @@ -503,8 +504,7 @@ def getAuthCode(username: str, password: str, brand: str, parsed = urlparse.urlparse(response_url) auth_code = ''.join(parse_qs(parsed.query)['code']) except Exception: - log.error("kia.getAuthCode: Login failed") - log.debug("kia.getAuthCode: Login failed: " + response, exc_info=True) + log.exception("kia.getAuthCode: Login failed: " + response) raise log.debug("Kia/Hyundai: AuthCode = " + auth_code[:8] + "[...]") @@ -546,8 +546,7 @@ def getAuthToken(auth_code: str, token: dict, brand: str) -> dict: token["refreshToken"] = access_token["refresh_token"] except Exception: - log.error("kia.getAuthToken: Login failed") - log.debug("kia.getAuthToken: Login failed: " + response, exc_info=True) + log.exception("kia.getAuthToken: Login failed: " + response) raise log.debug("Kia/Hyundai: AuthToken = " + token["accessToken"][:8] + "[...]") @@ -580,8 +579,7 @@ def registerDevice(token: dict, brand: str) -> None: } response = postHTTP(url=url, data=data, headers=headers) except Exception: - log.error("kia.getAuthToken: Login failed") - log.debug("kia.getAuthToken: Login failed: " + response, exc_info=True) + log.exception("kia.getAuthToken: Login failed: " + response) raise return @@ -598,8 +596,7 @@ def requestToken(user_id: str, password: str, brand: str) -> dict: token["userHash"] = getUserHash(user_id, password) registerDevice(token, brand) except Exception: - log.error("kia.requestToken: Login failed") - log.debug("kia.requestToken: ", exc_info=True) + log.exception("kia.requestToken: ") raise return token @@ -640,9 +637,8 @@ def refreshToken(token: dict, brand: str) -> dict: token["accessToken"] = token_new["access_token"] except Exception: - log.error("kia.refreshToken: refresh token failed") - log.debug("kia.refreshToken: refresh token error: " + - response, exc_info=True) + log.exception("kia.refreshToken: refresh token error: " + + response) raise log.debug("kia.refreshToken: New access token = " + @@ -671,9 +667,8 @@ def getControlToken(pin: str, token: dict, brand: str) -> str: response_dict = json.loads(response) control_token = 'Bearer ' + response_dict['controlToken'] except Exception: - log.error("kia.getControlToken: Sending PIN failed") - log.debug("kia.getControlToken: Sending PIN error: " + - response, exc_info=True) + log.exception("kia.getControlToken: Sending PIN error: " + + response) raise log.debug("kia.refreshToken: control token = " + @@ -718,8 +713,7 @@ def getVehicleId(vin: str, token: dict, brand: str) -> str: raise except Exception: - log.error("kia.getVehicleId: failed") - log.debug("kia.getVehicleId: error: " + response, exc_info=True) + log.exception("kia.getVehicleId: error: " + response) raise log.debug("kia.getVehicleId: VehicleId = " + vehicle_id[:8] + @@ -751,8 +745,7 @@ def doPrewakeup(vehicle_id: str, token: dict, brand: str) -> None: } response = postHTTP(url=url, data=data, headers=headers, timeout=125) except Exception: - log.error("kia.doPrewakeup: failed") - log.debug("kia.doPrewakeup: error: " + response, exc_info=True) + log.exception("kia.doPrewakeup: error: " + response) raise return @@ -782,9 +775,8 @@ def getStatusFull(vehicle_id: str, control_token: str, } response = getHTTP(url=url, headers=headers, timeout=125) except Exception: - log.error("kia.getStatusFull: triggering update failed") - log.debug("kia.getStatusFull: triggering update error: " + - response, exc_info=True) + log.exception("kia.getStatusFull: triggering update error: " + + response) raise log.info("Kia/Hyundai: Waiting 130 seconds") @@ -819,9 +811,8 @@ def getStatusFull(vehicle_id: str, control_token: str, ['FuelSystem']['DTE']['Total']) except Exception: - log.error("kia.getStatusFull: receiving update failed") - log.debug("kia.getStatusFull: receiving update error: " + - response, exc_info=True) + log.exception("kia.getStatusFull: receiving update error: " + + response) raise return CarState(soc, range) @@ -843,8 +834,7 @@ def fetch_soc(user_id: str, password: str, pin: str, token = refreshToken(token, brand) saveToken(user_id, password, vehicle, token) except Exception: - log.error("kia.fetch_soc: login error, vehicle: " + str(vehicle)) - log.debug("kia.fetch_soc: ", exc_info=True) + log.exception("kia.fetch_soc: ") raise try: @@ -853,8 +843,7 @@ def fetch_soc(user_id: str, password: str, pin: str, control_token = getControlToken(pin, token, brand) soc_state = getStatusFull(vehicle_id, control_token, token, brand) except Exception: - log.error("kia.fetch_soc: API error, vehicle: " + str(vehicle)) - log.debug("kia.fetch_soc: ", exc_info=True) + log.exception("kia.fetch_soc: ") raise return soc_state