Skip to content

soc_bmwbc: improve error handling; set timestamp 0 after new captcha #2478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LKuemmel merged 1 commit into from
Jun 24, 2025
Merged

soc_bmwbc: improve error handling; set timestamp 0 after new captcha #2478

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 52 additions & 10 deletions packages/modules/vehicles/bmwbc/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,10 @@
from datetime import datetime
from copy import deepcopy
from threading import Lock
# from uuid import uuid4
import os.path
import shutil
import time

from helpermodules.utils.error_handling import ImportErrorContext
with ImportErrorContext():
Expand All @@ -21,6 +23,7 @@
from bimmer_connected.account import MyBMWAccount
from bimmer_connected.api.regions import Regions
from bimmer_connected.utils import MyBMWJSONEncoder
from bimmer_connected.models import MyBMWAPIError

from modules.common.component_state import CarState
from modules.common.store import RAMDISK_PATH
Expand Down Expand Up @@ -205,13 +208,15 @@ async def _fetch_soc(self,
if self._store[user_id]['captcha_token'] != captcha_token:
# invalidate current refresh and access token to force new login
log.debug("new captcha token configured - invalidate stored token set")
self._new_captcha = True
self._store[user_id]['expires_at'] = None
self._store[user_id]['access_token'] = None
self._store[user_id]['refresh_token'] = None
self._store[user_id]['session_id'] = None
self._store[user_id]['gcid'] = None
else:
log.debug("captcha token unchanged")
self._new_captcha = False

if user_id not in self._auth:
if self._store[user_id]['expires_at'] is not None and \
Expand All @@ -227,7 +232,8 @@ async def _fetch_soc(self,
Regions.REST_OF_WORLD,
refresh_token=self._store[user_id]['refresh_token'],
access_token=self._store[user_id]['access_token'],
expires_at=expires_at)
expires_at=expires_at,
hcaptcha_token=captcha_token)
else:
# no token, authenticate via user_id, password and captcha_token
log.info("authenticate via userid, password, captcha token")
Expand All @@ -252,7 +258,7 @@ async def _fetch_soc(self,
else:
log.debug("# Reuse _clconf instance")

# instantiate account object of not existent yet
# instantiate account object if not existent yet
if user_id not in self._account:
log.debug("# Create _account instance")
# user, password and region already set in BMWAuthentication/ClientConfiguration!
Expand All @@ -273,6 +279,7 @@ async def _fetch_soc(self,
if user_id not in self._last_reload:
self._last_reload[user_id] = 0
if self._now > self._last_reload[user_id] + 5 * 60:
# self._auth[user_id].session_id = str(uuid4()) # experimental to avoid error 408: reset session_id
# experimental: login when expires_at is reached to force token refresh
if self._store[user_id]['expires_at'] is not None:
expires_at = datetime.fromisoformat(self._store[user_id]['expires_at'])
Expand All @@ -287,9 +294,36 @@ async def _fetch_soc(self,
"/" + self._auth[user_id].refresh_token)

# get vehicle list - needs to be called async
log.info(self._mode + ": reload vehicles data")
await self._account[user_id].get_vehicles()
self._last_reload[user_id] = datetime.timestamp(datetime.now())
_loop = 5 # 5 retries
while _loop > 0:
log.info(self._mode + ": reload vehicles data/" + str(_loop))
try:
await self._account[user_id].get_vehicles()
except MyBMWAPIError as err:
log.info(self._mode + ": get_vehicles err= " + str(err.__class__) + ": " + str(err))
_err = -1
if 'Internal Server Error' in str(err):
log.info(self._mode + ": get_vehicles : Internal Server Error")
_err = 500
if 'Request Timeout' in str(err):
log.info(self._mode + ": get_vehicles : Request Timeout")
_err = 408
log.info(self._mode + ": get_vehicles : _err=" + str(_err))
time.sleep(10) # experimental: sleep for 10 secs
# log.info(self._mode + ": get_vehicles exception " + str(err))
log.info("# before except login:" + str(self._auth[user_id].expires_at) +
"/" + self._auth[user_id].refresh_token)
await self._auth[user_id].login()
log.info("# after except login:" + str(self._auth[user_id].expires_at) +
"/" + self._auth[user_id].refresh_token)
await self._account[user_id].get_vehicles()
_loop = _loop - 1
except Exception as err:
log.error("bmwbc.fetch_soc: get_vehicles Error, vehicle_id: " +
vehicle_id + f" {err=}, {type(err)=}")
raise err
self._last_reload[user_id] = datetime.timestamp(datetime.now())
_loop = 0

# get vehicle data for vin
vehicle = self._account[user_id].get_vehicle(vin)
Expand All @@ -305,9 +339,15 @@ async def _fetch_soc(self,
soc = int(state['electricChargingState']['chargingLevelPercent'])
range = float(state['electricChargingState']['range'])
lastUpdatedAt = state['lastUpdatedAt']
# convert lastUpdatedAt to soc_timestamp
soc_tsdtZ = datetime.strptime(lastUpdatedAt, ts_fmt + "Z")
soc_tsX = datetime.timestamp(soc_tsdtZ)
if self._new_captcha:
# after new captcha use system timestamp
# soc_tsX = datetime.timestamp(datetime.now())
# after new captcha use timestamp 0 (19700101)
soc_tsX = 0
else:
# convert lastUpdatedAt to soc_timestamp
soc_tsdtZ = datetime.strptime(lastUpdatedAt, ts_fmt + "Z")
soc_tsX = datetime.timestamp(soc_tsdtZ)

# save the vehicle data for further analysis if required
dump_json(respd, replyFilePrefix + vehicle_id)
Expand All @@ -333,15 +373,17 @@ async def _fetch_soc(self,
"/" + self._auth[user_id].refresh_token)

except Exception as err:
log.error("bmwbc.fetch_soc: requestData Error, vehicle_id: " + vehicle_id + f" {err=}, {type(err)=}")
# log.error("bmwbc.fetch_soc: requestData Error, vehicle_id: " + vehicle_id + f" {err=}, {type(err)=}")
log.error("bmwbc.fetch_soc: requestData Error, vehicle_id: " + str(vehicle_id))
# self._auth = None
self._auth.pop(user_id, None)
self._clconf.pop(user_id, None)
self._account.pop(user_id, None)
soc = 0
range = 0.0
soc_tsX = datetime.timestamp(datetime.now())
raise RequestFailed("SoC Request failed:\n" + str(err))
# raise RequestFailed("SoC Request failed:\n" + str(err))
raise Exception("SoC Request failed") from err
return soc, range, soc_tsX


Expand Down