github/psa_car_controller
1
0
Fork 0
mirror of https://github.com/flobz/psa_car_controller.git synced 2026-08-22 09:26:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: remove assert

Browse Source
This commit is contained in:
Florian Bezannier
2026-05-20 18:58:51 +02:00
committed by Florian BEZANNIER
parent 9f47f053eb
commit 515498fa4e
9 changed files with 58 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
psa_car_controller/psa/connected_car_api/rest.py
+5 -2
View File
@@ -127,8 +127,11 @@ class RESTClientObject(object):
(connection, read) timeouts.
"""
method = method.upper()
assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT',
'PATCH', 'OPTIONS']
allowed_methods = ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH', 'OPTIONS']
if method not in allowed_methods:
raise ValueError(
f"Invalid HTTP method: {method}. Must be one of {allowed_methods}"
)
if post_params and body:
raise ValueError(
psa_car_controller/psa/oauth.py
+2 -1
View File
@@ -51,7 +51,8 @@ class OpenIdCredentialManager(CredentialManager):
code_challenge=code_challenge, code_challenge_method="S256")
def connect_with_code(self, code: str):
assert len(code) == 36, "Invalid code length"
if len(code) != 36:
raise ValueError(f"Invalid code length: {len(code)} (expected 36)")
self._token_request({"grant_type": 'authorization_code', "code": code,
"redirect_uri": self.redirect_uri, "code_verifier": self.code_verifier},
False)
psa_car_controller/psa/otp/otp.py
+20 -20
View File
@@ -210,14 +210,14 @@ class Otp:
params.update(R)
xml = self.request(params)
if xml["err"] != "OK":
raise ConfigException(f"Error during activation: {xml}")
raise ConfigException(f"Error during activation finalize: {xml}")
self.data.synchro(xml, self.generate_kma(self.codepin))
if self.mode == Otp.OTP_MODE:
try:
self.defi = str(xml["defi"])
except KeyError:
raise ConfigException from KeyError
raise ConfigException("Missing 'defi' in response") from KeyError
if "J" in xml:
logger.debug("Need another otp request")
return Otp.OTP_TWICE
@@ -228,7 +228,7 @@ class Otp:
return Otp.OK
if int(xml["ms_n"]) > 1:
raise NotImplementedError
raise NotImplementedError("Multiple ms_n not supported")
ms_n = "0"
self.challenge = xml["challenge"]
@@ -251,6 +251,8 @@ class Otp:
req_param.update({"id": self.data.iwid, "lastsync": self.data.iwTsync, "ms_n": 1})
req_param.update(self.get_r())
xml = self.request(req_param)
if xml["err"] != "OK":
raise ConfigException(f"Error during MS_MODE activation finalize: {xml}")
self.data.synchro(xml, self.generate_kma(self.codepin))
return Otp.OK
@@ -265,18 +267,17 @@ class Otp:
def get_otp_code(self):
self.mode = Otp.OTP_MODE
otp_code = None
try:
if self.activation_start():
res = self.activation_finalyze()
if res == Otp.OTP_TWICE:
self.mode = Otp.OTP_MODE
self.activation_start()
assert self.activation_finalyze() == Otp.OK
otp_code = self._get_otp_code()
assert otp_code is not None
logger.debug("otp code: %s", otp_code)
except AssertionError as e:
raise ConfigException("Can't get otp code") from e
if self.activation_start():
res = self.activation_finalyze()
if res == Otp.OTP_TWICE:
self.mode = Otp.OTP_MODE
self.activation_start()
if self.activation_finalyze() != Otp.OK:
raise ConfigException("Can't get otp code (second attempt failed)")
otp_code = self._get_otp_code()
if otp_code is None:
raise ConfigException("Can't get otp code (code is None)")
logger.debug("otp code: %s", otp_code)
return otp_code
def __getstate__(self):
@@ -332,8 +333,7 @@ def new_otp_session(smscode, codepin, old_otp_session: Otp = None, ):
otp = Otp("bb8e981582b0f31353108fb020bead1c", device_id=old_otp_session.device_id)
otp.smsCode = smscode
otp.codepin = codepin
if otp.activation_start():
otp.activation_finalyze()
save_otp(otp)
return otp
return None
otp.activation_start()
otp.activation_finalyze()
save_otp(otp)
return otp
psa_car_controller/psacc/application/ecomix.py
+7 -3
View File
@@ -64,7 +64,9 @@ class Ecomix:
try:
now = datetime.utcnow().replace(tzinfo=UTC)
country_code = Ecomix.get_country(latitude, longitude, country_code_default)
assert country_code is not None
if country_code is None:
logger.warning("Can't find country for %s %s", latitude, longitude)
return False
if country_code not in Ecomix._cache:
Ecomix._cache[country_code] = []
elif len(Ecomix._cache[country_code]) > 0 and \
@@ -79,10 +81,12 @@ class Ecomix:
data = res.json()
value = data["carbonIntensity"]
assert isinstance(value, numbers.Number)
if not isinstance(value, numbers.Number):
logger.error("carbonIntensity invalid value: '%s'", value)
return False
Ecomix._cache[country_code].append([now, value])
return True
except (AssertionError, NameError, KeyError):
except (NameError, KeyError):
logger.debug("ecomix:", exc_info=True)
return False
else:
psa_car_controller/psacc/model/charge.py
+2 -1
View File
@@ -16,7 +16,8 @@ class Charge:
# pylint: disable=too-many-arguments,too-many-positional-arguments
def __init__(self, start_at: datetime, stop_at: datetime = None, vin=None, start_level=None, end_level=None,
co2=None, kw=None, price=None, charging_mode=None, mileage=None):
assert isinstance(start_at, datetime)
if not isinstance(start_at, datetime):
raise TypeError(f"start_at must be a datetime object, got {type(start_at)}")
self.charging_mode: ChargingMode = ChargingMode(charging_mode)
self.start_at = start_at
self.stop_at = stop_at
psa_car_controller/psacc/repository/db.py
+5 -3
View File
@@ -151,9 +151,11 @@ class Database:
@staticmethod
def get_db(db_file=None, force_new_conn=False) -> CustomSqliteConnection:
assert sqlite3.threadsafety == 3, \
"SQLite is not in serialized mode (sqlite3.threadsafety != 3). " + \
"Upgrade python to python3.11"
if sqlite3.threadsafety != 3:
raise RuntimeError(
"SQLite is not in serialized mode (sqlite3.threadsafety != 3). "
"Please upgrade python to python3.11 or ensure sqlite3 is compiled with threading support."
)
with Database.__thread_lock:
if not Database.__conn or force_new_conn:
if db_file is None:
psa_car_controller/psacc/repository/trips.py
+3 -1
View File
@@ -64,7 +64,9 @@ class Trips(list):
'AND mileage IS NOT NULL AND Timestamp IS NOT NULL ORDER BY Timestamp', (vin,)).fetchall()
if len(res) > 1:
car = vehicles_list.get_car_by_vin(vin)
assert car is not None
if car is None:
logger.warning("Car with vin %s not found in vehicles list", vin)
continue
trip_parser = TripParser(car)
start = res[0]
end = res[1]
psa_car_controller/web/tools/utils.py
+2 -1
View File
@@ -72,7 +72,8 @@ def create_card(card: dict):
def diff_dashtable(data, data_previous, row_id_name="row_id"):
df, df_previous = DataFrame(data=data), DataFrame(data_previous)
for _df in [df, df_previous]:
assert row_id_name in _df.columns
if row_id_name not in _df.columns:
raise ValueError(f"Column '{row_id_name}' not found in dataframes")
_df = _df.set_index(row_id_name)
mask = df.ne(df_previous)
df_diff = df[mask].dropna(how="all", axis="columns").dropna(how="all", axis="rows")
psa_car_controller/web/view/views.py
+12 -12
View File
@@ -199,26 +199,26 @@ def update_trips():
try:
trips_by_vin = Trips.get_trips(Cars([car]))
trips = trips_by_vin[car.vin]
assert len(trips) > 0
min_date = trips[0].start_at
max_date = trips[-1].start_at
figures.get_figures(trips[0].car)
except (AssertionError, KeyError):
if len(trips) > 0:
min_date = trips[0].start_at
max_date = trips[-1].start_at
figures.get_figures(trips[0].car)
except (KeyError, IndexError):
logger.debug("No trips yet")
figures.get_figures(Car("vin", "vid", "brand"))
try:
chargings = Charging.get_chargings()
assert len(chargings) > 0
figures.get_figures(Car("vin", "vid", "brand"))
chargings = Charging.get_chargings()
if len(chargings) == 0:
logger.debug("No chargings yet")
if min_date is None:
return
else:
if min_date:
min_date = min(min_date, chargings[0]["start_at"])
max_date = max(max_date, chargings[-1]["start_at"])
else:
min_date = chargings[0]["start_at"]
max_date = chargings[-1]["start_at"]
except AssertionError:
logger.debug("No chargings yet")
if min_date is None:
return
# update for slider
try:
logger.debug("min_date:%s - max_date:%s", min_date, max_date)