diff --git a/.prospector.yaml b/.prospector.yaml new file mode 100644 index 0000000..3a91988 --- /dev/null +++ b/.prospector.yaml @@ -0,0 +1,3 @@ +doc-warnings: false +ignore-paths: + - psa_connectedcar diff --git a/Car.py b/Car.py index 5106842..6e82724 100644 --- a/Car.py +++ b/Car.py @@ -29,7 +29,7 @@ class Car: self.battery_power = ENERGY_CAPACITY[self.label]["BATTERY_POWER"] self.fuel_capacity = ENERGY_CAPACITY[self.label]["FUEL_CAPACITY"] else: - logger.warn("Can't get car model please check cars.json") + logger.warning("Can't get car model please check cars.json") self.battery_power = DEFAULT_BATTERY_POWER self.fuel_capacity = DEFAULT_FUEL_CAPACITY @@ -48,11 +48,13 @@ class Cars(list): for car in self: if car.vin == vin: return car + return None def get_car_by_id(self, vehicle_id) -> Car: for car in self: if car.vehicle_id == vehicle_id: return car + return None def add(self, car: Car): if self.get_car_by_id(car.vehicle_id) is None: @@ -77,5 +79,5 @@ class Cars(list): with open(name, "r") as f: json_str = f.read() return Cars.from_json(json.loads(json_str)) - except: + except FileNotFoundError: return Cars() \ No newline at end of file diff --git a/ChargeControl.py b/ChargeControl.py index 0af5081..89eec24 100644 --- a/ChargeControl.py +++ b/ChargeControl.py @@ -56,7 +56,7 @@ class ChargeControl: if res is not None: status = res.get_energy('Electric').charging.status level = res.get_energy('Electric').level - logger.info(f"charging status of {self.vin} is {status}, battery level: {level}") + logger.info("charging status of %s is %s, battery level: %d", self.vin, status, level) if status == "InProgress": # force update if the car doesn't send info during 10 minutes last_update = res.get_energy('Electric').updated_at @@ -69,7 +69,7 @@ class ChargeControl: res = self.psacc.get_vehicle_info(self.vin) status = res.get_energy('Electric').charging.status if status == "InProgress": - logger.warn(f"retry to stop the charge of {self.vin}") + logger.warning("retry to stop the charge of %s", self.vin) self.psacc.charge_now(self.vin, False) self.retry_count += 1 if self._next_stop_hour is not None: @@ -79,7 +79,7 @@ class ChargeControl: else: self.retry_count = 0 else: - logger.error(f"error when get vehicle info of {self.vin}") + logger.error("error when get vehicle info of %s", self.vin) except: logger.error(traceback.format_exc()) self.thread = threading.Timer(periodicity, self.start) @@ -100,7 +100,7 @@ class ChargeControls: def save_config(self, name="charge_config.json", force=False): chd = {} - for key, el in self.list.items(): + for el in self.list.values(): chd[el.vin] = {"percentage_threshold": el.percentage_threshold, "stop_hour": el._stop_hour} config_str = json.dumps(chd, sort_keys=True, indent=4).encode('utf-8') new_hash = md5(config_str).hexdigest() @@ -112,8 +112,8 @@ class ChargeControls: def load_config(psacc: MyPSACC, name="charge_config.json"): with open(name, "r") as f: - str = f.read() - chd = json.loads(str) + config_str = f.read() + chd = json.loads(config_str) charge_control_list = ChargeControls() for vin, el in chd.items(): charge_control_list.list[vin] = ChargeControl(psacc, vin, **el) @@ -126,5 +126,5 @@ class ChargeControls: return None def start(self): - for vin, charge_control in self.list.items(): + for charge_control in self.list.values(): charge_control.start() diff --git a/MyPSACC.py b/MyPSACC.py index 90d8a99..6fd5422 100644 --- a/MyPSACC.py +++ b/MyPSACC.py @@ -26,6 +26,8 @@ from threading import Semaphore, Timer from functools import wraps from web.db import get_db +from geojson import Feature, Point, FeatureCollection +from geojson import dumps as geo_dumps oauhth_url = {"clientsB2CPeugeot": "https://idpcvs.peugeot.com/am/oauth2/access_token", "clientsB2CCitroen": "https://idpcvs.citroen.com/am/oauth2/access_token", @@ -43,6 +45,17 @@ MQTT_EVENT_TOPIC = "psa/RemoteServices/events/MPHRTServices/" MQTT_TOKEN_TTL = 890 +# add method to class Energy +def get_energy(self, energy_type): + for energy in self._energy: + if energy.type == energy_type: + return energy + return psac.models.energy.Energy(charging=psac.models.energy_charging.EnergyCharging()) + + +psac.models.status.Status.get_energy = get_energy + + def rate_limit(limit, every): def limit_decorator(fn): semaphore = Semaphore(limit) @@ -97,7 +110,7 @@ class OauthAPIClient(ApiClient): response_type=None, auth_settings=None, async_req=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None): - for x in range(0, 2): + for attempt in range(0, 2): try: if not async_req: return self._ApiClient__call_api(resource_path, method, @@ -106,16 +119,14 @@ class OauthAPIClient(ApiClient): response_type, auth_settings, _return_http_data_only, collection_formats, _preload_content, _request_timeout) - else: - thread = self.pool.apply_async(self.__call_api, (resource_path, - method, path_params, query_params, - header_params, body, - post_params, files, - response_type, auth_settings, - _return_http_data_only, - collection_formats, - _preload_content, _request_timeout)) - return thread + return self.pool.apply_async(self.__call_api, (resource_path, + method, path_params, query_params, + header_params, body, + post_params, files, + response_type, auth_settings, + _return_http_data_only, + collection_formats, + _preload_content, _request_timeout)) except ApiException as e: if e.reason == 'Unauthorized': self.configuration.refresh_callback() @@ -251,19 +262,19 @@ class MyPSACC: if not force and self.remote_token_last_update is not None: last_update: datetime = self.remote_token_last_update if (datetime.now() - last_update).total_seconds() < MQTT_TOKEN_TTL: - return + return None self.manager._refresh_token() res = self.manager.post(remote_url + self.client_id, json={"grant_type": "refresh_token", "refresh_token": self.remote_refresh_token}, headers=self.headers) data = res.json() - logger.debug(f"refresh_remote_token: {data}") + logger.debug("refresh_remote_token: %s", data) if "access_token" in data: self.remote_access_token = data["access_token"] self.remote_refresh_token = data["refresh_token"] self.remote_token_last_update = datetime.now() else: - logger.error(f"can't refresh_remote_token: {data}\n Create a new one") + logger.error("can't refresh_remote_token: %s\n Create a new one", data) self.remote_token_last_update = datetime.now() otp_code = self.getOtpCode() res = self.get_remote_access_token(otp_code) @@ -272,30 +283,30 @@ class MyPSACC: def on_mqtt_connect(self, client, userdata, rc, a): try: - logger.info("Connected with result code " + str(rc)) + logger.info("Connected with result code %s", rc) topics = [MQTT_RESP_TOPIC + self.customer_id + "/#"] for car in self.vehicles_list: topics.append(MQTT_EVENT_TOPIC + car.vin) for topic in topics: client.subscribe(topic) - logger.info("subscribe to " + topic) + logger.info("subscribe to %s", topic) except: logger.error(traceback.format_exc()) def on_mqtt_disconnect(self, client, userdata, rc): try: - logger.warn("Disconnected with result code " + str(rc)) + logger.warning("Disconnected with result code %d", rc) if rc == 1: self.refresh_remote_token(force=True) else: - logger.warn(mqtt.error_string(rc)) + logger.warning(mqtt.error_string(rc)) except: logger.error(traceback.format_exc()) def on_mqtt_message(self, client, userdata, msg): charge_not_detected = False try: - logger.info(f"mqtt msg {msg.topic} {str(msg.payload)}") + logger.info("mqtt msg %s %s", msg.topic, msg.payload) data = json.loads(msg.payload) if msg.topic.startswith(MQTT_RESP_TOPIC): if "return_code" in data: @@ -305,19 +316,22 @@ class MyPSACC: self.refresh_remote_token(force=True) logger.error("retry last request, token was expired") elif data["return_code"] == "300": - logger.error(f'{data["return_code"]}') + logger.error('%d', data["return_code"]) else: - logger.error(f'{data["return_code"]} : {data["reason"]}') + logger.error('%s : %s', data["return_code"], data["reason"]) else: logger.debug("mqtt msg hasn't return code") if msg.topic.startswith(MQTT_EVENT_TOPIC): if data["charging_state"]['remaining_time'] != 0 and data["charging_state"]['rate'] == 0: charge_not_detected = True elif msg.topic.endswith("/VehicleState"): - if data["resp_data"]["charging_state"]['remaining_time'] != 0 \ - and data["resp_data"]["charging_state"]['rate'] == 0: - charge_not_detected = True - self.precond_programs[data["vin"]] = data["resp_data"]["precond_state"]["programs"] + try: + if data["resp_data"]["charging_state"]['remaining_time'] != 0 \ + and data["resp_data"]["charging_state"]['rate'] == 0: + charge_not_detected = True + self.precond_programs[data["vin"]] = data["resp_data"]["precond_state"]["programs"] + except KeyError: + pass if charge_not_detected: # fix a psa server bug where charge beginning without status api being properly updated logger.info("charge begin") @@ -344,7 +358,7 @@ class MyPSACC: try: self.get_state(self.vehicles_list[0].vin) except: - logger.warn("keep_mqtt error") + logger.warning("keep_mqtt error") threading.Timer(timeout, self.__keep_mqtt).start() def mqtt_request(self, vin, req_parameters): @@ -372,7 +386,7 @@ class MyPSACC: return h, m except IndexError: logger.error(traceback.format_exc()) - logger.error(f"Can't get charge hour: {hour_str}") + logger.error("Can't get charge hour: %s", hour_str) def get_charge_status(self, vin): data = self.get_vehicle_info(vin) @@ -411,7 +425,7 @@ class MyPSACC: @rate_limit(3, 60 * 20) def wakeup(self, vin): - logger.info("ask wakeup to " + vin) + logger.info("ask wakeup to %s", vin) msg = self.mqtt_request(vin, {"action": "state"}) logger.info(msg) self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehCharge/state", msg) @@ -419,7 +433,7 @@ class MyPSACC: # get state from server by mqtt def get_state(self, vin): - logger.info("ask state to " + vin) + logger.info("ask state to %s", vin) msg = self.mqtt_request(vin, {"action": "state"}) logger.info(msg) self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehicleState", msg) @@ -469,8 +483,8 @@ class MyPSACC: @staticmethod def load_config(name="config.json"): with open(name, "r") as f: - str = f.read() - config = dict(**json.loads(str)) + config_str = f.read() + config = dict(**json.loads(config_str)) if "country_code" not in config: config["country_code"] = input("What is your country code ? (ex: FR, GB, DE, ES...)\n") return MyPSACC(**config) @@ -489,11 +503,12 @@ class MyPSACC: level_fuel = status.get_energy('Fuel').level moving = status.kinetic.moving logger.debug( - f"vin:{vin} longitude:{longitude} latitude:{latitude} date:{date} mileage:{mileage} level:{level} " - f"charging_status:{charging_status} charge_date:{charge_date} level_fuel:{level_fuel} moving:{moving}") + "vin:%s longitude:%s latitude:%s date:%s mileage:%s level:%s charging_status:%s charge_date:%s level_fuel:" + "%s moving:%s", vin, longitude, latitude, date, mileage, level, charging_status, charge_date, level_fuel, + moving) conn = get_db() if mileage == 0: # fix a bug of the api - logger.error(f"The api return a wrong mileage for {vin} : {mileage}") + logger.error("The api return a wrong mileage for %s : %f", vin, mileage) else: if conn.execute("SELECT Timestamp from position where Timestamp=?", (date,)).fetchone() is None: temp = None @@ -505,26 +520,27 @@ class MyPSACC: "appid": self.weather_api, "units": "metric"}) temp = weather_rep.json()["current"]["temp"] - logger.debug(f"Temperature :{temp}c") - except Exception as e: - logger.error(f"Unable to get temperature from openweathermap :{e}") + logger.debug("Temperature :%fc", temp) + except: + logger.error("Unable to get temperature from openweathermap :%s", temp) if level_fuel == 0: # fix fuel level not provided when car is off try: level_fuel = conn.execute( "SELECT level_fuel FROM position WHERE level_fuel>0 AND VIN=? ORDER BY Timestamp DESC LIMIT 1", (vin,)).fetchone()[0] - logger.info(f"level_fuel fixed with last real value {level_fuel} for {vin}") + logger.info("level_fuel fixed with last real value %f for %s", level_fuel, vin) except TypeError: level_fuel = None - logger.info(f"level_fuel unfixed for {vin}") + logger.info("level_fuel unfixed for %s", vin) conn.execute( - "INSERT INTO position(Timestamp,VIN,longitude,latitude,mileage,level,level_fuel,moving,temperature) VALUES(?,?,?,?,?,?,?,?,?)", + "INSERT INTO position(Timestamp,VIN,longitude,latitude,mileage,level,level_fuel,moving,temperature) " + "VALUES(?,?,?,?,?,?,?,?,?)", (date, vin, longitude, latitude, mileage, level, level_fuel, moving, temp)) conn.commit() - logger.info(f"new position recorded for {vin}") + logger.info("new position recorded for %s", vin) res = conn.execute( "SELECT Timestamp,mileage,level from position ORDER BY Timestamp DESC LIMIT 3;").fetchall() # Clean DB @@ -564,13 +580,11 @@ class MyPSACC: except TypeError: logger.debug("battery table is empty") except: - logger.debug("Error when saving status " + traceback.format_exc()) + logger.debug("Error when saving status %s", traceback.format_exc()) conn.close() @staticmethod def get_recorded_position(): - from geojson import Feature, Point, FeatureCollection - from geojson import dumps as geo_dumps conn = get_db() res = conn.execute('SELECT * FROM position ORDER BY Timestamp') features_list = [] @@ -585,15 +599,15 @@ class MyPSACC: return geo_dumps(feature_collection, sort_keys=True) @staticmethod - def get_chargings(min=None, max=None): + def get_chargings(mini=None, maxi=None): conn = get_db() - if min is not None: - if max is not None: - res = conn.execute("select * from battery WHERE start_at>=? and start_at<=?", (min, max)).fetchall() + if mini is not None: + if maxi is not None: + res = conn.execute("select * from battery WHERE start_at>=? and start_at<=?", (mini, maxi)).fetchall() else: - res = conn.execute("select * from battery WHERE start_at>=?", (min,)).fetchall() - elif max is not None: - res = conn.execute("select * from battery WHERE start_at<=?", (max,)).fetchall() + res = conn.execute("select * from battery WHERE start_at>=?", (mini,)).fetchall() + elif maxi is not None: + res = conn.execute("select * from battery WHERE start_at<=?", (maxi,)).fetchall() else: res = conn.execute("select * from battery").fetchall() return tuple(map(dict, res)) @@ -607,14 +621,3 @@ class MyPeugeotEncoder(JSONEncoder): for el in ["client_id", "realm", "remote_refresh_token", "customer_id", "weather_api", "country_code"]: mpd[el] = data[el] return mpd - - -# add method to class Energy -def get_energy(self, energy_type): - for energy in self._energy: - if energy.type == energy_type: - return energy - return psac.models.energy.Energy(charging=psac.models.energy_charging.EnergyCharging()) - - -psac.models.status.Status.get_energy = get_energy diff --git a/Trip.py b/Trip.py index 88102b3..c242a07 100644 --- a/Trip.py +++ b/Trip.py @@ -8,6 +8,7 @@ from web.db import get_db DEBUG = False + class Points(): def __init__(self, latitude, longitude): self.latitude = latitude @@ -31,8 +32,8 @@ class Trip: self.duration = None self.mileage = None - def add_points(self, longitude, latitude): - self.positions.append(Points(longitude, latitude)) + def add_points(self, latitude, longitude): + self.positions.append(Points(latitude, longitude)) def get_consumption(self): return { @@ -75,7 +76,7 @@ class Trips(list): vehicles = conn.execute( "SELECT DISTINCT vin FROM position;").fetchall() for vin in vehicles: - vin = vin[0] # todo handle multiple vin + vin = vin[0] car = vehicles_list.get_car_by_vin(vin) battery_capacity = car.battery_power fuel_capacity = car.fuel_capacity @@ -88,8 +89,8 @@ class Trips(list): # res = list(map(dict,res)) for x in range(0, len(res) - 2): if DEBUG: - logger.debug( - f"{res[x]['Timestamp']} mileage:{res[x]['mileage']} level:{res[x]['level']} level_fuel:{res[x]['level_fuel']}") + logger.debug("%s mileage:%f level:%d level_fuel:%f", + {res[x]['Timestamp']}, res[x]['mileage'], res[x]['level'], res[x]['level_fuel']) next_el = res[x + 2] distance = end["mileage"] - start["mileage"] duration = (end["Timestamp"] - start["Timestamp"]).total_seconds() / 3600 @@ -110,19 +111,19 @@ class Trips(list): elif distance == 0 and charge > 2: restart_trip = True if DEBUG: - logger.debug(f"charge detected") + logger.debug("charge detected") elif speed_average < 0.2 and duration > 0.05: # think again if duration is really needed # end["mileage"] - start["mileage"] == 0 #or \ # (end["Timestamp"] - start["Timestamp"]).total_seconds() / 3600 > 10: # condition useless ??? restart_trip = True if DEBUG: - logger.debug(f"low speed detected") + logger.debug("low speed detected") if restart_trip: start = end tr = Trip() if DEBUG: - logger.debug(f"restart trip at {start['Timestamp']} {start['mileage']}km level:" - f"{start['level']} level_fuel:{start['level_fuel']}") + logger.debug("restart trip at %d %f km level:%f level_fuel:%f", start['Timestamp'], + start['mileage'], start['level'], start['level_fuel']) else: distance = next_el["mileage"] - end["mileage"] # km duration = (next_el["Timestamp"] - end["Timestamp"]).total_seconds() / 3600 @@ -139,31 +140,31 @@ class Trips(list): if refuel > 0: end_trip = True if DEBUG: - logger.debug(f"refuel detected") + logger.debug("refuel detected") elif distance == 0 and charge > 2: end_trip = True if DEBUG: - logger.debug(f"charge detected") + logger.debug("charge detected") elif speed_average < 0.2 and duration > 0.05: # (distance == 0 and duration > 0.08) or duration > 2 or # check the speed to handle missing point end_trip = True if DEBUG: - logger.debug(f"low speed detected") + logger.debug("low speed detected") elif duration > 2: end_trip = True if DEBUG: - logger.debug(f"too much time detected") + logger.debug("too much time detected") elif x == len(res) - 3: # last record detected # think if add point is needed end = next_el end_trip = True if DEBUG: - logger.debug(f"last position found") + logger.debug("last position found") if end_trip: if DEBUG: - logger.debug(f"stop trip at {end['Timestamp']} {end['mileage']}km level:{end['level']} " - f"level_fuel:{end['level_fuel']}") + logger.debug("stop trip at %s %fkm level:%f level_fuel:%f", + end['Timestamp'], end['mileage'], end['level'], end['level_fuel']) tr.distance = end["mileage"] - start["mileage"] # km if tr.distance > 0: tr.start_at = start["Timestamp"] @@ -181,20 +182,20 @@ class Trips(list): 2) # L/100 km tr.mileage = end["mileage"] if DEBUG: - logger.debug( - f"Trip: {tr.start_at} -> {tr.end_at} {tr.distance:.1f}km {tr.duration:.2f}h {tr.speed_average:.0f}km/h " - f"{tr.consumption:.2f}kWh {tr.consumption_km:.2f}kWh/100km {tr.consumption_fuel}L {tr.consumption_fuel_km}L/100km " - f"{tr.mileage:.1f}km") + logger.debug("Trip: %s %.1fkm %.2fh %.0fkm/h %.2fkWh %:.2fkWh/100km %sL " + "%sL/100km %:.1fkm", tr.start_at, tr.end_at, tr.distance, tr.duration, + tr.speed_average, tr.consumption, tr.consumption_km, + tr.consumption_fuel, tr.consumption_fuel_km, tr.mileage) # filter bad value if tr.consumption_km < 70 and ( tr.consumption_fuel_km == None or tr.consumption_fuel_km < 30): trips.append(tr) else: if DEBUG: - logger.debug(f"trip discarded") + logger.debug("trip discarded") start = next_el tr = Trip() else: tr.add_points(end["longitude"], end["latitude"]) end = next_el - return trips \ No newline at end of file + return trips diff --git a/app_decoder.py b/app_decoder.py index b8f3ab3..9893424 100755 --- a/app_decoder.py +++ b/app_decoder.py @@ -72,7 +72,7 @@ if sys.version_info < (3, 6): if not argv[1].endswith(".apk"): print("No apk given") - exit(1) + sys.exit(1) print("APK loading...") a = APK(argv[1]) package_name = a.get_package() @@ -110,7 +110,7 @@ except: traceback.print_exc() print(f"HOST_BRANDID : {HOST_BRANDID_PROD} sitecode: {site_code}") print(res.text) - exit(1) + sys.exit(1) save_key_to_pem(pfx_cert, "") @@ -135,7 +135,7 @@ try: except: traceback.print_exc() print(res2.text) - exit(1) + sys.exit(1) # Psacc @@ -150,7 +150,7 @@ res = psacc.get_vehicles() for vehicle in res_dict["vehicles"]: label = vehicle["short_label"].split(" ")[-1] car = psacc.vehicles_list.get_car_by_vin(vehicle["vin"]) - if car.label is "unknown": + if car.label == "unknown": car.label = label car.set_energy_capacity() psacc.vehicles_list.save_cars() diff --git a/ecomix.py b/ecomix.py index 38aa70a..abea48d 100644 --- a/ecomix.py +++ b/ecomix.py @@ -1,6 +1,6 @@ -from datetime import datetime, timedelta +from datetime import datetime from statistics import mean, StatisticsError -import xml.etree.ElementTree as ET +import xml.etree.ElementTree as ElT import requests import reverse_geocode @@ -18,7 +18,7 @@ class Ecomix: } ) - etree = ET.fromstring(res.text) + etree = ElT.fromstring(res.text) period_start = (start.hour + int(start.minute / 30)) * 4 period_end = (end.hour + int(end.minute / 30)) * 4 @@ -46,4 +46,3 @@ class Ecomix: else: co2_per_kw = None return co2_per_kw - diff --git a/otp/Otp.py b/otp/Otp.py index ebf67b9..e357acf 100644 --- a/otp/Otp.py +++ b/otp/Otp.py @@ -8,15 +8,14 @@ from Cryptodome.PublicKey import RSA from Cryptodome import Hash from math import ceil - - from collections import defaultdict -from xml.etree import cElementTree as ET +from xml.etree import cElementTree as ElT from otp import oaep from otp.load import IWData import pickle from MyLogger import logger + proxies = None @@ -58,13 +57,14 @@ def numberToBase36(n): class Otp: OTP_TWICE = 10 - OK=0 - kPub = "11" - exponent = int(kPub, 16) + OK = 0 + KPub = "11" + exponent = int(KPub, 16) ACTIVATE_MODE = "activate" OTP_MODE = "otp" MS_MODE = "ms" iw_host = "https://otp.mpsa.com" + def __init__(self, inweboAccessId): self.Kiw = None self.pinmode = None @@ -119,7 +119,7 @@ class Otp: R0 = self.challenge + ";" + iw + ";" + self.getSerial() R1 = self.challenge + ";" + iw + ";" + self.data.iwK1 - logger.debug(f"{R0}\n{R1}\n{R2}") + logger.debug("%s\n%s\n%s", R0, R1, R2) return {"R0": hashlib.sha256(R0.encode("utf-8")).hexdigest(), "R1": hashlib.sha256(R1.encode("utf-8")).hexdigest(), "R2": hashlib.sha256(R2.encode("utf-8")).hexdigest()} @@ -135,11 +135,11 @@ class Otp: for x in range(0, nb_block): if x == nb_block - 1: - max = len(enc_b) + maxi = len(enc_b) else: - max = (1 + x) * 128 - min = x * 128 - ciphertext = cipher.decrypt(enc_b[min:max]) + maxi = (1 + x) * 128 + mini = x * 128 + ciphertext = cipher.decrypt(enc_b[mini:maxi]) dec_string += ciphertext.hex() logger.debug(dec_string) return dec_string @@ -159,9 +159,8 @@ class Otp: try: raw_xml = raw_xml[raw_xml.index("?>") + 2:] if setup: - return etree_to_dict(ET.XML(raw_xml))["ActionSetup"] - else: - return etree_to_dict(ET.XML(raw_xml))["ActionFinalize"] + return etree_to_dict(ElT.XML(raw_xml))["ActionSetup"] + return etree_to_dict(ElT.XML(raw_xml))["ActionFinalize"] except: logger.debug(raw_xml) @@ -183,8 +182,7 @@ class Otp: elif self.mode == Otp.OTP_MODE: self.challenge = xml["challenge"] return True - else: - return False + return False def activation_finalyze(self, random_bytes=None): @@ -211,8 +209,7 @@ class Otp: if "J" in xml: logger.debug("Need another otp request") return Otp.OTP_TWICE - else: - return Otp.OK + return Otp.OK if "ms_n" not in xml or xml["ms_n"] == 0: logger.debug("no ms_n request needed") @@ -220,8 +217,7 @@ class Otp: if int(xml["ms_n"]) > 1: raise NotImplementedError - else: - ms_n = "0" + ms_n = "0" self.challenge = xml["challenge"] self.action = "synchro" @@ -263,7 +259,7 @@ class Otp: self.activation_start() self.activation_finalyze() otp_code = self._getOtpCode() - logger.debug(f"otp code: {otp_code}") + logger.debug("otp code: %s", otp_code) return otp_code def __getstate__(self): @@ -271,21 +267,23 @@ class Otp: del odict['cipher'] # don't pickle this return odict - def __setstate__(self, dict): - self.__dict__.update(dict) + def __setstate__(self, dict_param): + self.__dict__.update(dict_param) if self.Kiw is not None: - key = RSA.construct((int(self.Kiw, 16), Otp.exponent)) - self.cipher = oaep.new(key, hashAlgo=Hash.SHA256) + key = RSA.construct((int(self.Kiw, 16), Otp.exponent)) + self.cipher = oaep.new(key, hashAlgo=Hash.SHA256) def encode_oeap(text, key): cipher = oaep.new(bytes.fromhex(key), hashAlgo=Hash.SHA256) return cipher.encrypt(text) + def save_otp(obj): with open("otp.bin", 'wb') as output: pickle.dump(obj, output) + def load_otp(): try: with open("otp.bin", 'rb') as input: @@ -294,6 +292,7 @@ def load_otp(): logger.debug(traceback.format_exc()) return None + def new_otp_session(): otp = Otp("bb8e981582b0f31353108fb020bead1c") otp.smsCode = input("What is the code you just received by SMS ?") @@ -301,4 +300,4 @@ def new_otp_session(): otp.activation_start() otp.activation_finalyze() save_otp(otp) - return otp \ No newline at end of file + return otp diff --git a/otp/Tokenizer.py b/otp/Tokenizer.py index 4ab35bf..a08a5f8 100644 --- a/otp/Tokenizer.py +++ b/otp/Tokenizer.py @@ -1,16 +1,13 @@ -from locale import atoi - - -class Tokenizer(object): - def __init__(self, str, delimiter="&&"): - self.s:str = str +class Tokenizer: + def __init__(self, tokens, delimiter="&&"): + self.s: str = tokens self.delimiter = delimiter self.currentIndex = 0 def nextToken(self): if self.currentIndex >= len(self.s): return "" - indexOf = self.currentIndex+self.s[self.currentIndex:].index(self.delimiter) + indexOf = self.currentIndex + self.s[self.currentIndex:].index(self.delimiter) if indexOf == -1: substring = self.s[self.currentIndex:] self.currentIndex = self.s.length() @@ -18,24 +15,13 @@ class Tokenizer(object): substring2 = self.s[self.currentIndex:indexOf] self.currentIndex = indexOf + len(self.delimiter) - #print(f"{substring2} index:{self.currentIndex}") return substring2 def nextTokenI(self): token = self.nextToken() if token == "": return 0 - else: - return int(token, 16) + return int(token, 16) def hasMoreTokens(self): - return self.currentIndex < len(self.s) - -# a="0.2.11&&&&&&0&&0&&0&&9f13ba238fbabba08e85d93638e98ef5e48682a9d3e5bc325c3dd6fac8199a6ce09e9b4f373aa6a75a905c3d690f6e3335d1e8e5b748ecec3020a794149033f6ada6896db6d73b8d43b8365bbe15b9ac66f49d4e684a3628f1e9f3deda0c4e24aba771946e6085b92c5ad312477152acf8db01e6aea4b409d5ac1a05c2fd4e95&&0&&&&&&&&&&&&0&&0&&0&&0&&0&&0&&0&&&&&&&&0&&0&&0&&0&&0&&2.0.0&&http://m.inwebo.com/&&" -# t=Tokenizer(a) -# self = t -# t.nextToken() -# t.nextToken() -# t.currentIndex -# t.nextTokenI() -# atoi("") \ No newline at end of file + return self.currentIndex < len(self.s) \ No newline at end of file diff --git a/otp/load.py b/otp/load.py index 8aebeee..dd22711 100644 --- a/otp/load.py +++ b/otp/load.py @@ -6,15 +6,20 @@ from Cryptodome.Cipher import AES from otp.Tokenizer import Tokenizer -default_token="0.2.11&&&&&&0&&0&&0&&9f13ba238fbabba08e85d93638e98ef5e48682a9d3e5bc325c3dd6fac8199a6ce09e9b4f373aa6a75a905c3d690f6e3335d1e8e5b748ecec3020a794149033f6ada6896db6d73b8d43b8365bbe15b9ac66f49d4e684a3628f1e9f3deda0c4e24aba771946e6085b92c5ad312477152acf8db01e6aea4b409d5ac1a05c2fd4e95&&0&&&&&&&&&&&&0&&0&&0&&0&&0&&0&&0&&&&&&&&0&&0&&0&&0&&0&&2.0.0&&http://m.inwebo.com/&&" -default_version="529" +default_token = "0.2.11&&&&&&0&&0&&0&&9f13ba238fbabba08e85d93638e98ef5e48682a9d3e5bc325c3dd6fac8199a6ce09e9b4f373aa6a" \ + "75a905c3d690f6e3335d1e8e5b748ecec3020a794149033f6ada6896db6d73b8d43b8365bbe15b9ac66f49d4e684a3628f1e" \ + "9f3deda0c4e24aba771946e6085b92c5ad312477152acf8db01e6aea4b409d5ac1a05c2fd4e95&&0&&&&&&&&&&&&0&&0&&0&" \ + "&0&&0&&0&&0&&&&&&&&0&&0&&0&&0&&0&&2.0.0&&http://m.inwebo.com/&&" +default_version = "529" -def filterLoad(string:str): + +def filterLoad(string: str): return string.replace("&", "&") -class IWData(object): - def __init__(self,IW): - self.IW=IW + +class IWData: + def __init__(self, IW): + self.IW = IW self.tokenizer = Tokenizer(default_token) self.tokenizer.nextToken() self.load1xx(int(default_version), self.tokenizer) @@ -46,16 +51,16 @@ class IWData(object): self.iwH = tokenizer.nextToken() nextTokenI = tokenizer.nextTokenI() self.iwsrvn = nextTokenI - self.iwsrvid = [None]*(nextTokenI) - self.iwsrvname = [None]*(nextTokenI) - self.iwsrvlogo = [None]*(nextTokenI) - self.iwsrvurl = [None]*(nextTokenI) - self.iwsrvonlineotp = [None]*(nextTokenI) + self.iwsrvid = [None] * (nextTokenI) + self.iwsrvname = [None] * (nextTokenI) + self.iwsrvlogo = [None] * (nextTokenI) + self.iwsrvurl = [None] * (nextTokenI) + self.iwsrvonlineotp = [None] * (nextTokenI) if self.IW.isMac: - self.iwsrvconnected = [None]*(self.iwsrvn) + self.iwsrvconnected = [None] * (self.iwsrvn) j2 = self.iwsrvn - self.iwsrvsecure = [None]*(j2) - self.iwsrvksc = [None]*(j2) + self.iwsrvsecure = [None] * (j2) + self.iwsrvksc = [None] * (j2) i = 0 while i < self.iwsrvn: self.iwsrvid[i] = tokenizer.nextToken() @@ -63,12 +68,12 @@ class IWData(object): self.iwsrvlogo[i] = filterLoad(tokenizer.nextToken()) if self.IW.isMac: self.iwsrvconnected[i] = tokenizer.nextTokenI() - if j>515: + if j > 515: i2 = 1 - elif j ==515: - i2=0 + elif j == 515: + i2 = 0 else: - i2=-1 + i2 = -1 if i2 < 0 or self.IW.isMac: self.iwsrvurl[i] = "" else: @@ -85,8 +90,8 @@ class IWData(object): i += 1 nextTokenI2 = tokenizer.nextTokenI() self.iwsecn = nextTokenI2 - self.iwsecid = [None]*(nextTokenI2) - self.iwsecval = [None]*(nextTokenI2) + self.iwsecid = [None] * (nextTokenI2) + self.iwsecval = [None] * (nextTokenI2) i3 = 0 while ((i3)) < self.iwsecn: self.iwsecid[i3] = tokenizer.nextToken() @@ -94,15 +99,15 @@ class IWData(object): i3 += 1 self.iwmsgn = tokenizer.nextTokenI() self.iwmsgtime = tokenizer.nextTokenI() - self.iwmsgid ="" - self.iwmsgtitle ="" + self.iwmsgid = "" + self.iwmsgtitle = "" self.iwmsgcontent = "" self.iwmsgack = "" i4 = 0 while i4 < self.iwmsgn: - self.iwmsgid+= tokenizer.nextToken() - self.iwmsgtitle+= filterLoad(tokenizer.nextToken()) - self.iwmsgcontent+=filterLoad(tokenizer.nextToken()) + self.iwmsgid += tokenizer.nextToken() + self.iwmsgtitle += filterLoad(tokenizer.nextToken()) + self.iwmsgcontent += filterLoad(tokenizer.nextToken()) self.iwmsgack += tokenizer.nextTokenI() i4 += 1 self.iwmajorversion = tokenizer.nextTokenI() @@ -111,9 +116,8 @@ class IWData(object): self.mustupgrade = False self.datatouch = 0 - def synchro(self, ixml:dict, key): + def synchro(self, ixml: dict, key): aes_cipher = AES.new(bytes.fromhex(key), AES.MODE_ECB) - """ generated source for method synchro """ value = ixml.get("id") if value is not None and len(value) > 0: self.iwid = value @@ -162,14 +166,14 @@ class IWData(object): self.iwsrvksc = ixml.get("s_ksc") self.iwsrvsecure = ixml.get("s_secure") self.iwsrvurl = ixml.get("s_url") - self.iwsrvonlineotp = ixml.get("s_onlineotp") + self.iwsrvonlineotp = ixml.get("s_onlineotp") self.IW.synchroJustDone = 1 value = ixml.get("m_n") if value is not None and len(value) > 0: self.iwmsgtime = int(time()) - self.iwmsgn = ixml.get("m_n") - self.iwmsgid = ixml.get("m_id") - self.iwmsgtitle = ixml.get("m_title") - self.iwmsgcontent = ixml.get("m_content") - self.iwmsgack = ixml.get("m_ack") - self.datatouch = 1 \ No newline at end of file + self.iwmsgn = ixml.get("m_n") + self.iwmsgid = ixml.get("m_id") + self.iwmsgtitle = ixml.get("m_title") + self.iwmsgcontent = ixml.get("m_content") + self.iwmsgack = ixml.get("m_ack") + self.datatouch = 1 diff --git a/otp/oaep.py b/otp/oaep.py index 7229711..7cf14ea 100644 --- a/otp/oaep.py +++ b/otp/oaep.py @@ -80,5 +80,4 @@ def new(key, hashAlgo=None, mgfunc=None, label=b'', randfunc=None): def notrandom(x): if x == 32: return b'\xf56\xccL`\x8a\x97l\nX0\xf4\x11\x9a\x0e\xce\x99K^\xe6\xcbU\xf3W+It"\xf5\x84\x1d\xe6' - else: - return None + return None diff --git a/server.py b/server.py index 7b25c2a..e255434 100755 --- a/server.py +++ b/server.py @@ -19,7 +19,8 @@ parser = argparse.ArgumentParser() def parse_args(): parser = argparse.ArgumentParser() parser.add_argument("-f", "--config", help="config file, default file: config.json", type=argparse.FileType('r')) - parser.add_argument("-c", "--charge-control", help="enable charge control, default charge_config.json", const="charge_config.json", nargs='?', metavar='charge config file') + parser.add_argument("-c", "--charge-control", help="enable charge control, default charge_config.json", + const="charge_config.json", nargs='?', metavar='charge config file') parser.add_argument("-d", "--debug", help="enable debug", const=10, default=20, nargs='?', metavar='Debug level number') parser.add_argument("-l", "--listen", help="change server listen address", default="127.0.0.1", metavar="IP") parser.add_argument("-p", "--port", help="change server listen port", default="5000") diff --git a/web/app.py b/web/app.py index e857e71..54985f6 100644 --- a/web/app.py +++ b/web/app.py @@ -27,7 +27,7 @@ def start_app(title, base_path, debug: bool, host, port): locale_url = [f"https://cdn.plot.ly/plotly-locale-{lang}-latest.js"] except: locale_url = None - logger.warn("Can't get language") + logger.warning("Can't get language") app = Flask(__name__) app.config["DEBUG"] = debug if base_path == "/": @@ -44,7 +44,7 @@ def start_app(title, base_path, debug: bool, host, port): myp = None -chc: ChargeControls = None +chc: ChargeControls def save_config(my_peugeot: MyPSACC): diff --git a/web/callback.py b/web/callback.py index ab5ccd6..b3c7c93 100644 --- a/web/callback.py +++ b/web/callback.py @@ -16,8 +16,8 @@ from web import figures from web.app import app, dash_app, myp, chc import web.db -trips = None -chargings = None +trips: Trips +chargings: dict @dash_app.callback(Output('trips_map', 'figure'), @@ -28,14 +28,14 @@ chargings = None Output('tab_battery', 'children'), Input('date-slider', 'value')) def display_value(value): - min = datetime.fromtimestamp(value[0], tz=timezone.utc) - max = datetime.fromtimestamp(value[1], tz=timezone.utc) + mini = datetime.fromtimestamp(value[0], tz=timezone.utc) + maxi = datetime.fromtimestamp(value[1], tz=timezone.utc) filtered_trips = [] for trip in trips: - if min <= trip.start_at <= max: + if mini <= trip.start_at <= maxi: filtered_trips.append(trip) - filtered_chargings = MyPSACC.get_chargings(min,max) - figures.get_figures(filtered_trips,filtered_chargings) + filtered_chargings = MyPSACC.get_chargings(mini, maxi) + figures.get_figures(filtered_trips, filtered_chargings) consumption = "Average consumption: {:.1f} kWh/100km".format(float(figures.consumption_df.mean(numeric_only=True))) return figures.trips_map, figures.consumption_fig, figures.consumption_fig_by_speed, consumption, figures.table_fig, figures.battery_info @@ -78,16 +78,16 @@ def preconditioning(vin, activate): @app.route('/position/') def get_position(vin): res = myp.get_vehicle_info(vin) - coordinates=res.last_position.geometry.coordinates - if len(coordinates)==3: # altitude is not always availlable + coordinates = res.last_position.geometry.coordinates + if len(coordinates) == 3: # altitude is not always availlable longitude, latitude, altitude = coordinates return jsonify( - {"longitude": longitude, "latitude": latitude, "altitude": altitude, "url": f"http://maps.google.com/maps?q={latitude},{longitude}"}) - else: - longitude, latitude = coordinates - return jsonify( - {"longitude": longitude, "latitude": latitude, "url": f"http://maps.google.com/maps?q={latitude},{longitude}"}) - + {"longitude": longitude, "latitude": latitude, "altitude": altitude, + "url": f"http://maps.google.com/maps?q={latitude},{longitude}"}) + longitude, latitude = coordinates + return jsonify( + {"longitude": longitude, "latitude": latitude, + "url": f"http://maps.google.com/maps?q={latitude},{longitude}"}) # Set a battery threshold and schedule an hour to stop the charge @@ -125,7 +125,8 @@ def update_trips(): trips = Trips.get_trips(myp.vehicles_list) chargings = MyPSACC.get_chargings() except: - logger.error("update_trips: " + traceback.format_exc()) + logger.error("update_trips: %s", traceback.format_exc()) + try: web.db.callback_fct = update_trips @@ -163,9 +164,9 @@ try: ), html.Div(id="tab-content", className="p-4"), ])]) -except (IndexError, TypeError) as e: +except (IndexError, TypeError): logger.debug("Failed to generate figure, there is probably not enough data yet") - data_div = dbc.Alert("No data to show", color="danger") + data_div = dbc.Alert("No data to show, there is probably no trips recorded yet", color="danger") except: logger.error("Failed to generate figure, there is probably not enough data yet") diff --git a/web/db.py b/web/db.py index f877e98..7d99f96 100644 --- a/web/db.py +++ b/web/db.py @@ -4,6 +4,8 @@ import pytz callback_fct = None default_db_file = 'info.db' + + def convert_datetime(st): return datetime.strptime(st.decode("utf-8"), "%Y-%m-%d %H:%M:%S+00:00").replace(tzinfo=pytz.UTC) @@ -11,7 +13,7 @@ def convert_datetime(st): def update_callback(): if callback_fct is not None: callback_fct() - return + def get_db(db_file=default_db_file): sqlite3.register_converter("DATETIME", convert_datetime) diff --git a/web/figures.py b/web/figures.py index beb2eb8..c005958 100644 --- a/web/figures.py +++ b/web/figures.py @@ -38,6 +38,7 @@ def get_marks_from_start_end(start, end): for date in result: marks[unix_time_millis(date)] = str(date.strftime(date_f)) return marks + return None consumption_fig = None