improved satbility on session expiration

arielle.py

@@ -2,76 +2,85 @@ import os.path
 import time
 import pprint
 
+import credentials
 from wci import wci
 from credentials import user, password
 import json
-from folder_mgr.folder_mgr import get_date, date_has_changed, to_date_str, to_time_str, FolderMgr
+from folder_mgr.folder_mgr import get_date, date_has_changed, to_date_str, to_time_str, FolderMgr, create_timestamp
 
 INTERVAL = 5*60
 
+domains = [
+	"access",
+	"charging",
+	"climatisation",
+	"climatisationTimers",
+	"fuelStatus",
+	"userCapabilities",
+	"vehicleLights",
+	"vehicleHealthInspection",
+	"oilLevel",
+	"measurements"
+]
+
+
 # Values after training
 # Convert and write JSON object to file
 
 
 def main():
-	w = wci.WeConnectId(user, password)
+	expired_token_for_test = 'eyJraWQiOiI0ODEyODgzZi05Y2FiLTQwMWMtYTI5OC0wZmEyMTA5Y2ViY2EiLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiIxMDQyNmRjZC1mYzljLTRkODctODBmYy0yMDIyNzdmYzhkYzIiLCJhdWQiOiJhMjRmYmE2My0zNGIzLTRkNDMtYjE4MS05NDIxMTFlNmJkYThAYXBwc192dy1kaWxhYl9jb20iLCJzY3AiOiJvcGVuaWQgcHJvZmlsZSBiYWRnZSBjYXJzIGRlYWxlcnMgdmluIiwiYWF0IjoiaWRlbnRpdHlraXQiLCJpc3MiOiJodHRwczovL2lkZW50aXR5LnZ3Z3JvdXAuaW8iLCJqdHQiOiJhY2Nlc3NfdG9rZW4iLCJleHAiOjE3Mjg5ODQzMjAsImlhdCI6MTcyODk4MDcyMCwibGVlIjpbIlZPTEtTV0FHRU4iXSwianRpIjoiYWQ0MDhkMDEtYmQxMy00YWZkLWIyMWEtYmZiMTE5NTFhOGE3In0.r78z00BS0Je7mjUcukmrkuSiPO9F8me-bLs3ZZF6BokhPX8uqu_2j7I3Pu-8C1xwFBb-YDs6rr0iiVUeOeiKM4O-jOik8ny57sMKdYJ_FHZNo-GLMA_WPppNaGWkM8HojIxEW4mrO3Zqd9f_vDrTAYukG4WyULbxq_OH6c6yQdmEx4M1PGisVB0DI6DG0Dl4748BCJMA_TyNvKagQHZuRhT0ma7kj_GVSqlGGtW0erwPBwTHY_X_wDitngod2iUw9nRtli-HkJKqgER82jSu2k5b36ePglsRqx6k4ls8Tiz6ykgcjuuW4_zdsfUNuhBElVd1QwCzolQ4e7IzHbVaRZbub8tlisrHZMskGJmNJeAUv9xZMJ8o6YNyAz__WPvP6iubrCIixSKOThv-YLXa0cFkCPmoaLR9eeTy6sWFsFvslTdD-a0zrrdP8ais4g5Cxao6ph3NV1xCgkEldImI3n1hxeMLefe6sqfoZ6Ogz-EDVPBxS9qpazPWN6EQyyYLXQZA0s6p5hcrjVq7pD-5yXjeFhCrymzizaNm0JhVF_L36F3D5a3s8tnQ2kaMfdK6jdHQyauxL-8SHfAgk_py33XOcgGQkp4aOXN3pm7UhutQ8fAS9kMoquqikuxTAu-UlWAYRhDUaypab1F51f50eK84s1jpgWDAQ1aL7PIgwj8'
-
+	w = wci.WeConnectId(user, password, expired_token_for_test)
-	vehicles = w.get('/vehicles')
+	vehicles = w.get('/vehicles')['data']
 	print('list of vehicles:')
-	for key in vehicles:
-		print(key)
-
-	vin = vehicles['data'][0]['vin']
-
 	pprint.pprint(vehicles)
-	domains = [
+
-		"access",
+	# Print vins
-		"charging",
+	for vehicle in vehicles:
-		"climatisation",
+		vin = vehicle['vin']
-		"climatisationTimers",
+		print(f'vin: {vin}')
-		"fuelStatus",
+
-		"userCapabilities",
+	vin = credentials.vin
-		"vehicleLights",
+
-		"vehicleHealthInspection",
+	# Create request url
-		"oilLevel",
-		"measurements"
-	]
 	endpoint = '/vehicles/' + vin + '/selectivestatus?jobs='
 	for domain in domains:
 		endpoint += domain + ','
 	endpoint = endpoint[:-1]
-#	endpoint = '/vehicles/' + vehicles['data'][0]['vin'] + f'/selectivestatus?jobs=access,charging,climatisation,climatisationTimers,fuelStatus,userCapabilities,vehicleLights,vehicleHealthInspection,oilLevel,measurements'
 	print(f'Endpoint: {endpoint}')
 
-	data_list = []
+	# Init collecting records
-	_date_last = {}
 	f_mgr = FolderMgr()
+	record_id = 0
+	record_list = []
+	_date_last = {}
 
-	_date_start = get_date()
+	# Collecting loop
 	while True:
 		start = time.time()
 		_date = get_date()
 		if date_has_changed(_date_last, _date):
-			data_list = []
+			record_id = 0
+			record_list = []
 
 		try:
+			timestamp = create_timestamp()
 			data = w.get(endpoint)
-			print(data)
+			record_list.append({'info': {'record_id': record_id, 'timestamp': timestamp}, 'data': data})
-			data_list.append(data)
 			folder = f_mgr.update(f"results/{vin}", _date)
-			filename = f"vehicle_{vin}_{to_date_str(_date_start)}_{to_time_str(_date_start)}.json"
+			filename = f"vehicle_{vin}.json"
 			_path = os.path.join(folder, filename)
 			with open(_path, "w") as fp:
-				json.dump(data_list, fp, indent=4)
+				json.dump(record_list, fp, indent=4)
 
-		except Exception:
+		except Exception as e:
-			pass
+			print(f"Exception: {e}")
 
 		stop = time.time()
-		diff = stop - start
+		diff = int(stop - start)
 		time.sleep(max(0, INTERVAL - diff))
 		_date_last = _date
-
+		record_id += 1
 
 
 if __name__ == '__main__':

credentials.py

@@ -3,3 +3,4 @@
 
 user = 'alex.thoni@gmx.de'
 password = 'Arielle2024!'
+vin = 'WVWZZZE1ZMP010760'

folder_mgr/folder_mgr.py

@@ -1,4 +1,5 @@
 import time
+import datetime
 import os
 
 
@@ -23,6 +24,12 @@ def to_date_str(_date: dict):
 	return f"{_date['year']:04d}-{_date['month']:02d}-{_date['day']:02d}"
 
 
+def create_timestamp():
+	now = datetime.datetime.now(datetime.UTC)
+	res = now.strftime('%Y-%m-%dT%H:%M:%S') + ('.%03dZ' % (now.microsecond / 1000))
+	return res
+
+
 def date_has_changed(d1: dict, d2: dict, fields=['year', 'month', 'day']):
 	count = 0
 

wci/wci.py

@@ -8,11 +8,13 @@ import string
 BASE_URL = 'https://emea.bff.cariad.digital/vehicle/v1'
 LOGIN_URL= 'https://emea.bff.cariad.digital/user-login/v1'
 
+
 def generate_url():
     nonce = ''.join(random.choices(string.ascii_uppercase + string.ascii_lowercase + string.digits, k=16))
 
     return LOGIN_URL + '/authorize?nonce=' + nonce + '&redirect_uri=weconnect://authenticated'
 
+
 class WeConnectId:
     def __init__(self, email_address, password, access_token=None):
         self._email_address = email_address
@@ -59,8 +61,7 @@ class WeConnectId:
                 'password': self._password,
             }
 
-            access_token = None
+            self._access_token = None
-
             try:
                 r = self._session.post(
                     'https://identity.vwgroup.io/signin-service/v1/' + some_data['clientLegalEntityModel']['clientId'] + '/login/authenticate',
@@ -71,6 +72,8 @@ class WeConnectId:
                 # We expect a redirect to 'weconnect://authenticated', which requests doesn't understand
                 self._access_token = re.search('access_token=([^&$]+)', str(e)).group(1)
 
+        print(f"AccessToken: {self._access_token}")
+
         self._session.headers.update({
             'Authorization': 'Bearer ' + self._access_token,
         })
@@ -81,9 +84,11 @@ class WeConnectId:
     def get(self, endpoint):
         r = self._session.get(BASE_URL + endpoint)
 
-        if r.status_code == 401:
+        if r.status_code == 400 or r.status_code == 401:
+            print(f"Error: {r.status_code}:{r.text}")
+            print(f"Restart session")
+            self._setup_session()
             self._sign_in(True)
-
+            return self.get(endpoint)
-            r = self._session.get(BASE_URL + endpoint)
 
         return r.json()