From e889b8084124506f988544ffba896e099d2f5d7c Mon Sep 17 00:00:00 2001 From: Florian Bezannier Date: Fri, 12 Mar 2021 19:03:39 +0100 Subject: [PATCH] code clean up --- ChargeControl.py | 25 +++++++++++++++---------- app_decoder.py | 4 ++-- ecomix.py | 2 +- otp/Otp.py | 11 ++++++----- requirements.txt | 1 + web/app.py | 2 +- web/db.py | 11 +++++++---- 7 files changed, 33 insertions(+), 23 deletions(-) diff --git a/ChargeControl.py b/ChargeControl.py index f876978..e66b308 100644 --- a/ChargeControl.py +++ b/ChargeControl.py @@ -34,6 +34,9 @@ class ChargeControl: if self._next_stop_hour < datetime.now(): self._next_stop_hour += timedelta(days=1) + def get_stop_hour(self): + return self._stop_hour + def process(self): now = datetime.now() vehicle_status = self.psacc.vehicles_list.get_car_by_vin(self.vin).status @@ -84,22 +87,24 @@ class ChargeControl: return chd -class ChargeControls: +class ChargeControls(dict): def __init__(self): - self.list: dict = {} - self._confighash = None + super().__init__() + self._config_hash = None def save_config(self, name="charge_config.json", force=False): chd = {} - for el in self.list.values(): - chd[el.vin] = {"percentage_threshold": el.percentage_threshold, "stop_hour": el._stop_hour} + chargeControl: ChargeControl + for chargeControl in self.values(): + chd[chargeControl.vin] = {"percentage_threshold": chargeControl.percentage_threshold, + "stop_hour": chargeControl.get_stop_hour()} config_str = json.dumps(chd, sort_keys=True, indent=4).encode('utf-8') new_hash = md5(config_str).hexdigest() - if force or self._confighash != new_hash: + if force or self._config_hash != new_hash: with open(name, "wb") as f: f.write(config_str) - self._confighash = new_hash + self._config_hash = new_hash logger.info("save config change") @staticmethod @@ -109,15 +114,15 @@ class ChargeControls: chd = json.loads(config_str) charge_control_list = ChargeControls() for vin, el in chd.items(): - charge_control_list.list[vin] = ChargeControl(psacc, vin, **el) + charge_control_list[vin] = ChargeControl(psacc, vin, **el) return charge_control_list def get(self, vin) -> ChargeControl: try: - return self.list[vin] + return self[vin] except KeyError: return None def start(self): - for charge_control in self.list.values(): + for charge_control in self.values(): charge_control.psacc.info_callback.append(charge_control.process) diff --git a/app_decoder.py b/app_decoder.py index 9893424..7538b8f 100755 --- a/app_decoder.py +++ b/app_decoder.py @@ -50,8 +50,8 @@ def find_preferences_xml(): def save_key_to_pem(pfx_data, pfx_password): - private_key, certificate, additional_certificates = pkcs12.load_key_and_certificates(pfx_data, - bytes.fromhex(pfx_password), default_backend()) + private_key, certificate = pkcs12.load_key_and_certificates(pfx_data, + bytes.fromhex(pfx_password), default_backend())[:2] try: os.mkdir("certs") except FileExistsError: diff --git a/ecomix.py b/ecomix.py index abea48d..f609d03 100644 --- a/ecomix.py +++ b/ecomix.py @@ -1,6 +1,6 @@ from datetime import datetime from statistics import mean, StatisticsError -import xml.etree.ElementTree as ElT +import xml.etree.cElementTree as ElT import requests import reverse_geocode diff --git a/otp/Otp.py b/otp/Otp.py index e357acf..83c8266 100644 --- a/otp/Otp.py +++ b/otp/Otp.py @@ -93,7 +93,7 @@ class Otp: def init(self, Kfact=None, Kiw=None, pinmode=None): self.Kfact = Kfact self.pinmode = pinmode - self.Kiw = self.decode_oeap(Kiw, self.Kfact) + self.Kiw = self.decode_oaep(Kiw, self.Kfact) key = RSA.construct((int(self.Kiw, 16), Otp.exponent)) self.cipher = oaep.new(key, hashAlgo=Hash.SHA256) @@ -124,7 +124,8 @@ class Otp: "R1": hashlib.sha256(R1.encode("utf-8")).hexdigest(), "R2": hashlib.sha256(R2.encode("utf-8")).hexdigest()} - def decode_oeap(self, enc, key): + @staticmethod + def decode_oaep(enc, key): modulus = int(key, 16) key = RSA.construct((modulus, Otp.exponent)) cipher = oaep.new(key, hashAlgo=Hash.SHA256) @@ -221,7 +222,7 @@ class Otp: self.challenge = xml["challenge"] self.action = "synchro" - res = self.decode_oeap(xml["ms_key"], self.Kfact) + res = self.decode_oaep(xml["ms_key"], self.Kfact) temp_key = RSA.construct((int(res, 16), self.exponent)) temp_cipher = oaep.new(temp_key, hashAlgo=Hash.SHA256) if random_bytes is None: @@ -286,8 +287,8 @@ def save_otp(obj): def load_otp(): try: - with open("otp.bin", 'rb') as input: - return pickle.load(input) + with open("otp.bin", 'rb') as input_file: + return pickle.load(input_file) except: logger.debug(traceback.format_exc()) return None diff --git a/requirements.txt b/requirements.txt index 2e64fe0..15421cf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ paho-mqtt>=1.5.0 dash>=1.18.0 plotly>=4 cryptography>=3.0 +Werkzeug>=1.0.0 pandas oauth2_client requests diff --git a/web/app.py b/web/app.py index aaed20e..4c4a663 100644 --- a/web/app.py +++ b/web/app.py @@ -8,7 +8,7 @@ import locale from werkzeug import run_simple try: from werkzeug.middleware.dispatcher import DispatcherMiddleware -except: +except ImportError: from werkzeug import DispatcherMiddleware from ChargeControl import ChargeControls diff --git a/web/db.py b/web/db.py index 7d99f96..150777d 100644 --- a/web/db.py +++ b/web/db.py @@ -1,8 +1,11 @@ import sqlite3 from datetime import datetime -import pytz +from types import GenericAlias -callback_fct = None +import pytz +from typing import Callable + +callback_fct = Callable[[],None] default_db_file = 'info.db' @@ -11,7 +14,7 @@ def convert_datetime(st): def update_callback(): - if callback_fct is not None: + if callback_fct is not GenericAlias: callback_fct() @@ -23,7 +26,7 @@ def get_db(db_file=default_db_file): "latitude REAL, mileage REAL, level INTEGER, level_fuel INTEGER, moving BOOLEAN, temperature INTEGER);") try: conn.execute("ALTER TABLE position ADD level_fuel INTEGER;") - except: + except sqlite3.OperationalError: pass conn.execute("CREATE TABLE IF NOT EXISTS battery (start_at DATETIME PRIMARY KEY,stop_at DATETIME,VIN TEXT, " "start_level INTEGER, end_level INTEGER, co2 INTEGER, kw INTEGER);")