peter/volvo2mqtt
1
0
Fork 1
mirror of https://github.com/Dielee/volvo2mqtt.git synced 2026-08-13 11:19:52 +02:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Rework caching storage

Browse Source
This commit is contained in:
Linus Dietz
2023-06-24 10:07:51 +02:00
parent c7a883d781
commit 28721384f7
1 changed files with 22 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/volvo.py
+22 -43
View File
@@ -18,15 +18,8 @@ session.headers = {
token_expires_at: datetime token_expires_at: datetime
refresh_token = None refresh_token = None
vins = [] vins = []
recharge_cached_api_response = {}
recharge_api_last_update = {}
window_cached_api_response = {}
window_api_last_update = {}
door_cached_api_response = {}
door_api_last_update = {}
tyre_cached_api_response = {}
tyre_api_last_update = {}
supported_endpoints = {} supported_endpoints = {}
cached_requests = {}
def authorize(): def authorize():
@@ -177,43 +170,27 @@ def api_call(url, method, vin, sensor_id=None, force_update=False):
if url == RECHARGE_STATE_URL: if url == RECHARGE_STATE_URL:
# Minimize API calls for recharge state # Minimize API calls for recharge state
global recharge_cached_api_response, recharge_api_last_update response = cached_request(url, method, vin, force_update)
response, recharge_cached_api_response, recharge_api_last_update = cached_request(recharge_cached_api_response,
recharge_api_last_update,
url, method, vin,
force_update)
if response is None: if response is None:
# Exception catched while getting data from volvo api, doing nothing # Exception caught while getting data from volvo api, doing nothing
return None return None
elif url == WINDOWS_STATE_URL: elif url == WINDOWS_STATE_URL:
# Minimize API calls for window state # Minimize API calls for window state
global window_cached_api_response, window_api_last_update response = cached_request(url, method, vin, force_update)
response, recharge_cached_api_response, recharge_api_last_update = cached_request(window_cached_api_response,
window_api_last_update,
url, method, vin,
force_update)
if response is None: if response is None:
# Exception catched while getting data from volvo api, doing nothing # Exception caught while getting data from volvo api, doing nothing
return "" return ""
elif url == LOCK_STATE_URL: elif url == LOCK_STATE_URL:
# Minimize API calls for door state # Minimize API calls for door state
global door_cached_api_response, door_api_last_update response = cached_request(url, method, vin, force_update)
response, recharge_cached_api_response, recharge_api_last_update = cached_request(door_cached_api_response,
door_api_last_update,
url, method, vin,
force_update)
if response is None: if response is None:
# Exception catched while getting data from volvo api, doing nothing # Exception caught while getting data from volvo api, doing nothing
return "" return ""
elif url == TYRE_STATE_URL: elif url == TYRE_STATE_URL:
# Minimize API calls for tyre state # Minimize API calls for tyre state
global tyre_cached_api_response, tyre_api_last_update response = cached_request(url, method, vin, force_update)
response, recharge_cached_api_response, recharge_api_last_update = cached_request(tyre_cached_api_response,
tyre_api_last_update,
url, method, vin,
force_update)
if response is None: if response is None:
# Exception catched while getting data from volvo api, doing nothing # Exception caught while getting data from volvo api, doing nothing
return "" return ""
elif method == "GET": elif method == "GET":
print("Starting " + method + " call against " + url) print("Starting " + method + " call against " + url)
@@ -249,32 +226,34 @@ def api_call(url, method, vin, sensor_id=None, force_update=False):
return parse_api_data(data, sensor_id) return parse_api_data(data, sensor_id)
def cached_request(response_cache, update_cache, url, method, vin, force_update=False): def cached_request(url, method, vin, force_update=False):
if not vin in response_cache or force_update: global cached_requests
if not keys_exists(cached_requests, vin + "_" + url) or force_update:
# No API Data cached, get fresh data from API # No API Data cached, get fresh data from API
print("Starting " + method + " call against " + url) print("Starting " + method + " call against " + url)
try: try:
response = session.get(url.format(vin), timeout=15) response = session.get(url.format(vin), timeout=15)
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print("Error getting data: " + str(e)) print("Error getting data: " + str(e))
return None, None, None return None
response_cache[vin] = response
update_cache[vin] = datetime.now() data = {"response": response, "last_update": datetime.now()}
cached_requests[vin + "_" + url] = data
else: else:
if (datetime.now() - update_cache[vin]).total_seconds() >= settings["updateInterval"]: if (datetime.now() - cached_requests[vin + "_" + url]["last_update"]).total_seconds() >= settings["updateInterval"]:
# Old Data in Cache, updating # Old Data in Cache, updating
print("Starting " + method + " call against " + url) print("Starting " + method + " call against " + url)
try: try:
response = session.get(url.format(vin), timeout=15) response = session.get(url.format(vin), timeout=15)
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print("Error getting data: " + str(e)) print("Error getting data: " + str(e))
return None, None, None return None
response_cache[vin] = response data = {"response": response, "last_update": datetime.now()}
update_cache[vin] = datetime.now() cached_requests[vin + "_" + url] = data
else: else:
# Data is up do date, returning cached data # Data is up do date, returning cached data
response = response_cache[vin] response = cached_requests[vin + "_" + url]["response"]
return response, response_cache, update_cache return response
def parse_api_data(data, sensor_id=None): def parse_api_data(data, sensor_id=None):