From 47b8af3ee0fc756f8bd904d7001ccc1e4de8a062 Mon Sep 17 00:00:00 2001 From: Florian Bezannier Date: Sun, 30 May 2021 17:42:55 +0200 Subject: [PATCH] webui to connect to PSA API --- app_decoder.py | 192 ++++++++++++++++++++------------------------ libs/car_model.py | 11 +-- libs/charging.py | 4 +- libs/config.py | 101 +++++++++++++++++++++++ my_psacc.py | 28 ++++--- mylogger.py | 4 +- otp/otp.py | 16 ++-- server.py | 86 ++------------------ test/test_unit.py | 4 +- web/app.py | 23 ++---- web/config_views.py | 183 +++++++++++++++++++++++++++++++++++++++++ web/views.py | 141 ++++++++++++++++++-------------- 12 files changed, 509 insertions(+), 284 deletions(-) create mode 100644 libs/config.py create mode 100644 web/config_views.py diff --git a/app_decoder.py b/app_decoder.py index 825a208..752f659 100755 --- a/app_decoder.py +++ b/app_decoder.py @@ -2,12 +2,10 @@ import json import os import traceback -from sys import argv -import sys -from getpass import getpass +from urllib import request +from os import path from androguard.core.bytecodes.apk import APK - import requests from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.serialization import pkcs12 @@ -23,6 +21,8 @@ BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", " "com.psa.mym.myvauxhall": {"realm": "clientsB2CVauxhall", "brand_code": "VX", "app_name": "MyVauxhall"} } +DOWNLOAD_URL = "https://github.com/flobz/psa_apk/raw/main/" + def save_key_to_pem(pfx_data, pfx_password): private_key, certificate = pkcs12.load_key_and_certificates(pfx_data, @@ -40,111 +40,97 @@ def save_key_to_pem(pfx_data, pfx_password): encryption_algorithm=serialization.NoEncryption())) -current_dir = os.getcwd() -script_dir = dir_path = os.path.dirname(os.path.realpath(__file__)) -if sys.version_info < (3, 6): - raise RuntimeError("This application requires Python 3.6+") +def firstLaunchConfig(package_name, client_email, client_password, country_code, # pylint: disable=too-many-locals + config_prefix=""): + filename = package_name.split(".")[-1]+".apk" + if not path.exists(filename): + request.urlretrieve(DOWNLOAD_URL+filename, filename) + a = APK(filename) + package_name = a.get_package() + resources = a.get_android_resources() # .get_strings_resources() + client_id = resources.get_string(package_name, "PSA_API_CLIENT_ID_PROD")[1] + client_secret = resources.get_string(package_name, "PSA_API_CLIENT_SECRET_PROD")[1] + HOST_BRANDID_PROD = resources.get_string(package_name, "HOST_BRANDID_PROD")[1] + REMOTE_REFRESH_TOKEN = None + ## Get Customer id + site_code = BRAND[package_name]["brand_code"] + "_" + country_code + "_ESP" + try: + res = requests.post(HOST_BRANDID_PROD + "/GetAccessToken", + headers={ + "Connection": "Keep-Alive", + "Content-Type": "application/json", + "User-Agent": "okhttp/2.3.0" + }, + params={"jsonRequest": json.dumps( + {"siteCode": site_code, "culture": "fr-FR", "action": "authenticate", + "fields": {"USR_EMAIL": {"value": client_email}, + "USR_PASSWORD": {"value": client_password}} + } + )} + ) -if not argv[1].endswith(".apk"): - print("No apk given") - sys.exit(1) -print("APK loading...") -a = APK(argv[1]) -package_name = a.get_package() -resources = a.get_android_resources() # .get_strings_resources() -client_id = resources.get_string(package_name, "PSA_API_CLIENT_ID_PROD")[1] -client_secret = resources.get_string(package_name, "PSA_API_CLIENT_SECRET_PROD")[1] -HOST_BRANDID_PROD = resources.get_string(package_name, "HOST_BRANDID_PROD")[1] -pfx_cert = a.get_file("assets/MWPMYMA1.pfx") -REMOTE_REFRESH_TOKEN = None -print("APK loaded !") + token = res.json()["accessToken"] + except: # pylint: disable=bare-except + msg = traceback.format_exc() + f"\nHOST_BRANDID : {HOST_BRANDID_PROD} sitecode: {site_code}" + try: + msg += res.text + except: + pass + raise Exception(msg) -client_email = input(f"{BRAND[package_name]['app_name']} email: ") -client_password = getpass(f"{BRAND[package_name]['app_name']} password: ") + try: + res2 = requests.post( + f"https://mw-{BRAND[package_name]['brand_code'].lower()}-m2c.mym.awsmpsa.com/api/v1/" + f"user?culture=fr_FR&width=1080&v=1.27.0", + data=json.dumps({"site_code": site_code, "ticket": token}), + headers={ + "Connection": "Keep-Alive", + "Content-Type": "application/json;charset=UTF-8", + "Source-Agent": "App-Android", + "Token": token, + "User-Agent": "okhttp/4.8.0", + "Version": "1.27.0" + }, + cert=("certs/public.pem", "certs/private.pem"), + ) -country_code = input("What is your country code ? (ex: FR, GB, DE, ES...)\n") + res_dict = res2.json()["success"] + customer_id = BRAND[package_name]["brand_code"] + "-" + res_dict["id"] + except: # pylint: disable=bare-except + msg = traceback.format_exc() + try: + msg += res2.text + except: + pass + Exception(msg) -## Get Customer id -site_code = BRAND[package_name]["brand_code"] + "_" + country_code + "_ESP" -try: - res = requests.post(HOST_BRANDID_PROD + "/GetAccessToken", - headers={ - "Connection": "Keep-Alive", - "Content-Type": "application/json", - "User-Agent": "okhttp/2.3.0" - }, - params={"jsonRequest": json.dumps( - {"siteCode": site_code, "culture": "fr-FR", "action": "authenticate", - "fields": {"USR_EMAIL": {"value": client_email}, - "USR_PASSWORD": {"value": client_password}} - } - )} - ) + # Psacc + psacc = MyPSACC(None, client_id, client_secret, REMOTE_REFRESH_TOKEN, customer_id, BRAND[package_name]["realm"], + country_code) + psacc.connect(client_email, client_password) - token = res.json()["accessToken"] -except: # pylint: disable=bare-except - traceback.print_exc() - print(f"HOST_BRANDID : {HOST_BRANDID_PROD} sitecode: {site_code}") - print(res.text) - sys.exit(1) + psacc.save_config(name=config_prefix + "config.json") + res = psacc.get_vehicles() + print(f"\nYour vehicles: {res}") -save_key_to_pem(pfx_cert, "") + if len(res) == 0: + Exception("No vehicle in your account is compatible with this API, you vehicle is probably too old...") -try: - res2 = requests.post( - f"https://mw-{BRAND[package_name]['brand_code'].lower()}-m2c.mym.awsmpsa.com/api/v1/" - f"user?culture=fr_FR&width=1080&v=1.27.0", - data=json.dumps({"site_code": site_code, "ticket": token}), - headers={ - "Connection": "Keep-Alive", - "Content-Type": "application/json;charset=UTF-8", - "Source-Agent": "App-Android", - "Token": token, - "User-Agent": "okhttp/4.8.0", - "Version": "1.27.0" - }, - cert=("certs/public.pem", "certs/private.pem"), - ) + for vehicle in res_dict["vehicles"]: + car = psacc.vehicles_list.get_car_by_vin(vehicle["vin"]) + if car is not None and "short_label" in vehicle and car.label == "unknown": + car.label = vehicle["short_label"].split(" ")[-1] # remove new, nouvelle, neu word.... + car.set_energy_capacity() + else: + print("Warning: Can't get car model for please check cars.json") + psacc.vehicles_list.save_cars() - res_dict = res2.json()["success"] - customer_id = BRAND[package_name]["brand_code"] + "-" + res_dict["id"] + print(f"\nYour vehicles: {res}") -except: # pylint: disable=bare-except - traceback.print_exc() - print(res2.text) - sys.exit(1) - -# Psacc - -psacc = MyPSACC(None, client_id, client_secret, REMOTE_REFRESH_TOKEN, customer_id, BRAND[package_name]["realm"], - country_code) -psacc.connect(client_email, client_password) - -os.chdir(current_dir) -psacc.save_config(name="test.json") -res = psacc.get_vehicles() - -print(f"\nYour vehicles: {res}") - -if len(res) == 0: - print("No vehicle in your account is compatible with this API, you vehicle is probably too old...") - sys.exit(1) - -for vehicle in res_dict["vehicles"]: - car = psacc.vehicles_list.get_car_by_vin(vehicle["vin"]) - if "short_label" in vehicle and car.label == "unknown": - car.label = vehicle["short_label"].split(" ")[-1] # remove new, nouvelle, neu word.... - car.set_energy_capacity() - else: - print("Warning: Can't get car model please check cars.json") -psacc.vehicles_list.save_cars() - - -# Charge control -charge_controls = ChargeControls("charge_config1.json") -for vehicle in res: - chc = ChargeControl(psacc, vehicle.vin, 100, [0, 0]) - charge_controls[vehicle.vin] = chc -charge_controls.save_config() - -print("Success !!!") + # Charge control + charge_controls = ChargeControls(config_prefix + "charge_config.json") + for vehicle in res: + chc = ChargeControl(psacc, vehicle.vin, 100, [0, 0]) + charge_controls[vehicle.vin] = chc + charge_controls.save_config() + return "Success !!!" diff --git a/libs/car_model.py b/libs/car_model.py index 39c23df..80e963f 100644 --- a/libs/car_model.py +++ b/libs/car_model.py @@ -27,11 +27,12 @@ class CarModel: @staticmethod def find_model_by_vin(vin): - for carmodel in carmodels: - if carmodel.match(vin): - return carmodel - logger.warning("Can't get car model, please report an issue on github with your car model" - " and first ten letter of your VIN : %s", vin[:10]) + if vin != "vin": + for carmodel in carmodels: + if carmodel.match(vin): + return carmodel + logger.warning("Can't get car model, please report an issue on github with your car model" + " and first ten letter of your VIN : %s", vin[:10]) return CarModel("unknown", DEFAULT_BATTERY_POWER, DEFAULT_FUEL_CAPACITY) @staticmethod diff --git a/libs/charging.py b/libs/charging.py index 519e08c..9b432ec 100644 --- a/libs/charging.py +++ b/libs/charging.py @@ -39,13 +39,13 @@ class Charging: Database.clean_battery(conn) @staticmethod - def record_charging(car, charging_status, charge_date: datetime, level, latitude, # pylint: disable=too-many-locals + def record_charging(car, charging_status, charge_date: datetime, level, latitude, # pylint: disable=too-many-locals longitude, country_code, charging_mode, charging_rate, autonomy): conn = Database.get_db() charge_date = charge_date.replace(microsecond=0) if charging_status == "InProgress": stop_at, start_at = conn.execute("SELECT stop_at, start_at FROM battery WHERE VIN=? ORDER BY start_at " - "DESC limit 1", (car.vin,)).fetchone() or [False, None] + "DESC limit 1", (car.vin,)).fetchone() or [False, None] try: conn.execute("INSERT INTO battery_curve(start_at,VIN,date,level,rate,autonomy) VALUES(?,?,?,?,?,?)", (start_at, car.vin, charge_date, level, charging_rate, autonomy)) diff --git a/libs/config.py b/libs/config.py new file mode 100644 index 0000000..d407fcb --- /dev/null +++ b/libs/config.py @@ -0,0 +1,101 @@ +import argparse +import atexit +import threading +from os import environ, path + +from oauth2_client.credentials_manager import OAuthError + +from charge_control import ChargeControls +from libs.charging import Charging +from libs.elec_price import ElecPrice +from my_psacc import MyPSACC +from mylogger import logger, my_logger +from otp.otp import CONFIG_NAME as OTP_CONFIG_NAME + +DEFAULT_NAME = "config.json" + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument("-f", "--config", help="config file, default file: config.json", default="config.json") + 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 or name') + 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") + parser.add_argument("-r", "--record", help="save vehicle data to db", action='store_true') + parser.add_argument("-R", "--refresh", help="refresh vehicles status every x min", type=int) + parser.add_argument("-m", "--mail", default=environ.get('USER_EMAIL', None), help="set the email address") + parser.add_argument("-P", "--password", default=environ.get('USER_PASSWORD', None), help="set the password") + parser.add_argument("--remote-disable", help="disable remote control", action='store_true') + parser.add_argument("--offline", help="offline limited mode", action='store_true') + parser.add_argument("--web-conf", help="ignore if config files not existing yet", action='store_true') + parser.add_argument("-b", "--base-path", help="base path for web app", default="/") + return parser.parse_args() + + +class Singleton(type): + _instances = {} + + def __call__(cls, *args, **kwargs): + if cls not in cls._instances: + cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) + return cls._instances[cls] + + +class Config(metaclass=Singleton): + def __init__(self): + self.args = parse_args() + self.myp: MyPSACC + self.chc: ChargeControls + self.config_name = DEFAULT_NAME + self.is_good: bool = False + + def start_remote_control(self): + if self.args.remote_disable: + logger.info("mqtt disabled") + elif not self.args.web_conf or path.exists(OTP_CONFIG_NAME): + if self.myp.mqtt_client is not None: + self.myp.mqtt_client.disconnect() + self.myp.start_mqtt() + if self.args.charge_control: + Config.chc = ChargeControls.load_config(self.myp, name=self.args.charge_control) + Config.chc.init() + self.myp.start_refresh_thread() + + def load_app(self) -> bool: + my_logger(handler_level=self.args.debug) + if self.args.config: + self.config_name = self.args.config + if path.exists(self.config_name): + self.myp = MyPSACC.load_config(name=self.config_name) + elif self.args.web_conf: + return False + else: + raise FileNotFoundError(self.config_name) + atexit.register(self.save_config) + self.myp.set_record(self.args.record) + Charging.elec_price = ElecPrice.read_config() + if self.args.offline: + logger.info("offline mode") + else: + try: + self.myp.refresh_token() + except OAuthError: + if self.args.mail and self.args.password: + self.myp.connect(self.args.mail, self.args.password) + else: + logger.error("Please reconnect by going to config web page") + logger.info(str(self.myp.get_vehicles())) + if self.args.refresh: + self.myp.info_refresh_rate = self.args.refresh * 60 + self.myp.start_refresh_thread() + self.start_remote_control() + self.save_config() + self.is_good = True + return True + + def save_config(self): + self.myp.save_config(self.config_name) + threading.Timer(30, self.save_config).start() diff --git a/my_psacc.py b/my_psacc.py index 4f146e7..0c0c149 100644 --- a/my_psacc.py +++ b/my_psacc.py @@ -38,6 +38,7 @@ realm_info = { "app_name": "MyVauxhall"} } + AUTHORIZE_SERVICE = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize" REMOTE_URL = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id=" SCOPE = ['openid profile'] @@ -105,6 +106,7 @@ class MyPSACC: self.set_proxies(proxies) self.config_file = DEFAULT_CONFIG_FILENAME Ecomix.co2_signal_key = co2_signal_api + self.refresh_thread = None def get_app_name(self): return realm_info[self.realm]['app_name'] @@ -153,7 +155,7 @@ class MyPSACC: car.status = res return res - def refresh_vehicle_info(self): + def __refresh_vehicle_info(self): if self.info_refresh_rate is not None: while True: sleep(self.info_refresh_rate) @@ -163,6 +165,12 @@ class MyPSACC: for callback in self.info_callback: callback() + def start_refresh_thread(self): + if self.refresh_thread is None: + self.refresh_thread = threading.Thread(target=self.__refresh_vehicle_info) + self.refresh_thread.setDaemon(True) + self.refresh_thread.start() + # monitor doesn't seem to work def new_monitor(self, vin, body): res = self.manager.post("https://api.groupe-psa.com/connectedcar/v4/user/vehicles/" + @@ -232,7 +240,7 @@ class MyPSACC: try: if self.remote_refresh_token is None: logger.error("remote_refresh_token isn't defined") - self.load_otp(force_new=True) + return None res = self.manager.post(REMOTE_URL + self.client_id, json={"grant_type": "refresh_token", "refresh_token": self.remote_refresh_token}, headers=self.headers) @@ -305,14 +313,14 @@ class MyPSACC: self.mqtt_client = mqtt.Client(clean_session=True, protocol=mqtt.MQTTv311) if environ.get("MQTT_LOG", "0") == "1": self.mqtt_client.enable_logger(logger=logger) - self.refresh_remote_token() - self.mqtt_client.tls_set_context() - self.mqtt_client.on_connect = self.__on_mqtt_connect - self.mqtt_client.on_message = self.__on_mqtt_message - self.mqtt_client.on_disconnect = self._on_mqtt_disconnect - self.mqtt_client.connect(MQTT_SERVER, 8885, 60) - self.mqtt_client.loop_start() - self.__keep_mqtt() + if self.refresh_remote_token(): + self.mqtt_client.tls_set_context() + self.mqtt_client.on_connect = self.__on_mqtt_connect + self.mqtt_client.on_message = self.__on_mqtt_message + self.mqtt_client.on_disconnect = self._on_mqtt_disconnect + self.mqtt_client.connect(MQTT_SERVER, 8885, 60) + self.mqtt_client.loop_start() + self.__keep_mqtt() return self.mqtt_client.is_connected() def __keep_mqtt(self): # avoid token expiration diff --git a/mylogger.py b/mylogger.py index 5a957d8..b329906 100644 --- a/mylogger.py +++ b/mylogger.py @@ -1,6 +1,8 @@ import logging from logging.handlers import RotatingFileHandler +LOG_FILE = 'activity.log' + DEBUG_LEVELV_NUM = 9 logging.addLevelName(DEBUG_LEVELV_NUM, "DEBUGV") @@ -28,7 +30,7 @@ logging.setLoggerClass(CustomLogger) logger = logging.getLogger("log") -def my_logger(file='activity.log', handler_level=logging.INFO): +def my_logger(file=LOG_FILE, handler_level=logging.INFO): global logger logger.setLevel(handler_level) formatter = logging.Formatter('%(asctime)s :: %(levelname)s :: %(message)s') diff --git a/otp/otp.py b/otp/otp.py index 17d6e21..38bf097 100644 --- a/otp/otp.py +++ b/otp/otp.py @@ -17,6 +17,8 @@ from .load import IWData # pylint: disable=too-many-instance-attributes,invalid-name +CONFIG_NAME = "otp.bin" + def etree_to_dict(t): d = {t.tag: {} if t.attrib else None} @@ -255,7 +257,7 @@ class Otp: password = self.data.iwK1 + ":" + str(self.defi) + ":" + self.data.iwsecval res = bytes(hashlib.sha256(password.encode("utf-8")).digest()) nb = ((int.from_bytes(res[:4], byteorder="big") & 0xfffffff) * 1024) + ( - int.from_bytes(res[4:8], byteorder="big") & 1023) + int.from_bytes(res[4:8], byteorder="big") & 1023) otp = number_to_base36(nb) return otp @@ -313,7 +315,7 @@ class RenameUnpickler(pickle.Unpickler): return super().find_class(renamed_module, name) -def load_otp(filename="otp.bin"): +def load_otp(filename=CONFIG_NAME): try: with open(filename, 'rb') as input_file: try: @@ -325,13 +327,17 @@ def load_otp(filename="otp.bin"): return None -def new_otp_session(old_otp_session: Otp = None): +def new_otp_session(old_otp_session: Otp = None, smscode=None, codepin=None): if old_otp_session is None: otp = Otp("bb8e981582b0f31353108fb020bead1c") else: otp = Otp("bb8e981582b0f31353108fb020bead1c", device_id=old_otp_session.device_id) - otp.smsCode = input("What is the code you just received by SMS ?") - otp.codepin = input("What is your app pin code ?") + if smscode is None: + otp.smsCode = input("What is the code you just received by SMS ?") + otp.codepin = input("What is your app pin code ?") + else: + otp.smsCode = smscode + otp.codepin = codepin otp.activation_start() otp.activation_finalyze() save_otp(otp) diff --git a/server.py b/server.py index 75d346c..d458cc1 100755 --- a/server.py +++ b/server.py @@ -1,91 +1,19 @@ #!/usr/bin/env python3 -import atexit +# pylint: disable=wrong-import-position import sys -from os import environ from threading import Thread -from getpass import getpass -import argparse - -from oauth2_client.credentials_manager import OAuthError - +if sys.version_info < (3, 6): + raise RuntimeError("This application requires Python 3.6+") import web.app -from charge_control import ChargeControls -from libs.charging import Charging -from libs.elec_price import ElecPrice -from mylogger import my_logger +from libs.config import Config from mylogger import logger -from my_psacc import MyPSACC -from libs.utils import is_port_in_use - -CONFIG_NAME = "config.json" - - -def parse_args(): - parser = argparse.ArgumentParser() - parser.add_argument("-f", "--config", help="config file, default file: config.json", type=argparse.FileType('r'), - default="config.json") - 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 or name') - 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") - parser.add_argument("-r", "--record", help="save vehicle data to db", action='store_true') - parser.add_argument("-R", "--refresh", help="refresh vehicles status every x min", type=int) - parser.add_argument("-m", "--mail", default=environ.get('USER_EMAIL', None), help="set the email address") - parser.add_argument("-P", "--password", default=environ.get('USER_PASSWORD', None), help="set the password") - parser.add_argument("--remote-disable", help="disable remote control", action='store_true') - parser.add_argument("--offline", help="offline limited mode", action='store_true') - parser.add_argument("-b", "--base-path", help="base path for web app", default="/") - return parser.parse_args() - # noqa: MC0001 if __name__ == "__main__": - if sys.version_info < (3, 6): - raise RuntimeError("This application requires Python 3.6+") - args = parse_args() - my_logger(handler_level=args.debug) - if is_port_in_use(args.listen, int(args.port)): - logger.error(" Address already in use") - sys.exit(1) - logger.info("server start") - if args.config: - CONFIG_NAME = args.config.name - web.app.myp = MyPSACC.load_config(name=CONFIG_NAME) - atexit.register(web.app.myp.save_config) - web.app.myp.set_record(args.record) - Charging.elec_price = ElecPrice.read_config() - if args.offline: - logger.info("offline mode") - else: - try: - web.app.myp.refresh_token() - except OAuthError: - if args.mail and args.password: - client_email = args.mail - client_password = args.password - else: - client_email = input(f"{web.app.myp.get_app_name()} email: ") - client_password = getpass(f"{web.app.myp.get_app_name()} password: ") - web.app.myp.connect(client_email, client_password) - logger.info(str(web.app.myp.get_vehicles())) - if args.remote_disable: - logger.info("mqtt disabled") - else: - web.app.myp.start_mqtt() - if args.refresh or args.charge_control: - if args.refresh: - web.app.myp.info_refresh_rate = args.refresh * 60 - if args.charge_control: - web.app.chc = ChargeControls.load_config(web.app.myp, name=args.charge_control) - web.app.chc.init() - t2 = Thread(target=web.app.myp.refresh_vehicle_info) - t2.setDaemon(True) - t2.start() - - web.app.save_config(web.app.myp, CONFIG_NAME) + conf = Config() + conf.load_app() + args = conf.args t1 = Thread(target=web.app.start_app, args=["My car info", args.base_path, logger.level < 20, args.listen, int(args.port)]) t1.setDaemon(True) diff --git a/test/test_unit.py b/test/test_unit.py index a792ed6..51ae418 100644 --- a/test/test_unit.py +++ b/test/test_unit.py @@ -3,6 +3,8 @@ import json import os import unittest from datetime import datetime, timedelta + +import libs.config from psa_connectedcar import ApiClient import psa_connectedcar as psacc import reverse_geocode @@ -67,7 +69,7 @@ class TestUnit(unittest.TestCase): myp.abrp.abrp_enable_vin.add(car.vin) res = myp.get_vehicle_info(myp.vehicles_list[0].vin) myp.abrp.call(car, 22.1) - myp.save_config() + libs.config.save_config() assert isinstance(get_temp(str(latitude), str(longitude), myp.weather_api), float) def test_car_model(self): diff --git a/web/app.py b/web/app.py index df3d5a0..ac8cb6f 100644 --- a/web/app.py +++ b/web/app.py @@ -1,4 +1,3 @@ -import threading import locale import dash @@ -13,18 +12,13 @@ try: except ImportError: from werkzeug import DispatcherMiddleware -from charge_control import ChargeControls from mylogger import logger -from my_psacc import MyPSACC +import importlib # pylint: disable=invalid-name app = None dash_app = None dispatcher = None -# noinspection PyTypeChecker -myp: MyPSACC = None -# noinspection PyTypeChecker -chc: ChargeControls = None def start_app(*args, **kwargs): @@ -32,7 +26,7 @@ def start_app(*args, **kwargs): def config_flask(title, base_path, debug: bool, host, port, reloader=False, # pylint: disable=too-many-arguments - unminified=False): + unminified=False, view="web.views"): global app, dash_app, dispatcher reload_view = app is not None app = Flask(__name__) @@ -54,20 +48,15 @@ def config_flask(title, base_path, debug: bool, host, port, reloader=False, # p application = DispatcherMiddleware(Flask('dummy_app'), {base_path: app}) requests_pathname_prefix = base_path + "/" dash_app = dash.Dash(external_stylesheets=[dbc.themes.BOOTSTRAP], external_scripts=locale_url, title=title, - server=app, requests_pathname_prefix=requests_pathname_prefix) + server=app, requests_pathname_prefix=requests_pathname_prefix, + suppress_callback_exceptions=True) dash_app.enable_dev_tools(reloader) # keep this line - import web.views # pylint: disable=import-outside-toplevel + importlib.import_module(view) if reload_view: - import importlib # pylint: disable=import-outside-toplevel - importlib.reload(web.views) + importlib.reload(view) return {"hostname": host, "port": port, "application": application, "use_reloader": reloader, "use_debugger": debug} def run(config): return run_simple(**config) - - -def save_config(my_peugeot: MyPSACC, name): - my_peugeot.save_config(name) - threading.Timer(30, save_config, args=[my_peugeot, name]).start() diff --git a/web/config_views.py b/web/config_views.py new file mode 100644 index 0000000..b454240 --- /dev/null +++ b/web/config_views.py @@ -0,0 +1,183 @@ +from dash import callback_context +from dash.exceptions import PreventUpdate +from flask import request + +from app_decoder import firstLaunchConfig +from libs.config import Config +from mylogger import LOG_FILE +from otp.otp import new_otp_session +from web.app import dash_app +import dash_bootstrap_components as dbc +from dash.dependencies import Output, Input, State +import dash_core_components as dcc +import dash_html_components as html + +config = Config() +config_layout = dbc.Row(dbc.Col(className="col-md-12 col-lg-2 ml-2", children=[ + html.H2('Config'), + dbc.Form([ + dbc.FormGroup([ + dbc.Label("Car Brand", html_for="psa-app"), + dcc.Dropdown( + id="psa-app", + options=[ + {"label": "Peugeot", "value": "com.psa.mym.mypeugeot"}, + {"label": "Opel", "value": "com.psa.mym.myopel"}, + {"label": "Cirtroën", "value": "com.psa.mym.citroen"}, + {"label": "DS", "value": "com.psa.mym.myds"}, + {"label": "Vauxhall", "value": "com.psa.mym.myvauxhll"} + ], + )]), + dbc.FormGroup( + [ + dbc.Label("Email", html_for="psa-email"), + dbc.Input(type="email", id="psa-email", placeholder="Enter email"), + dbc.FormText( + "PSA account email", + color="secondary", + ), + ] + ), + dbc.FormGroup( + [ + dbc.Label("Password", html_for="psa-password"), + dbc.Input( + type="password", + id="psa-password", + placeholder="Enter password", + ), + dbc.FormText( + "PSA account password", + color="secondary", + ), + ] + ), + dbc.FormGroup( + [ + dbc.Label("Country code", html_for="countrycode"), + dbc.Input( + type="text", + id="psa-countrycode", + placeholder="Enter your country code", + ), + dbc.FormText( + "Example: FR for FRANCE or EN for England", + color="secondary", + ) + ] + ), + dbc.FormGroup([ + dbc.Button("Submit", color="primary", id="submit-form"), + dbc.FormText( + "After submit be patient it can take some time...", + color="secondary", + ), + dcc.Loading( + id="loading-2", + children=[html.Div([html.Div(id="form_result")])], + type="circle", + )] + ), + ])])) + +config_otp_layout = dbc.Row(dbc.Col(className="col-md-12 col-lg-2 ml-2", children=[ + html.H2('Config OTP'), + dbc.Form([ + dbc.FormGroup([ + dbc.Label("Click to receive a code by SMS", html_for="ask-sms"), + dbc.Button("Send SMS", color="info", id="ask-sms"), + html.Div(id="sms-demand-result", className="mt-2") + ]), + dbc.FormGroup( + [ + dbc.Label("Write the code you just received by SMS", html_for="psa-email"), + dbc.Input(type="text", id="psa-code", placeholder="Enter code"), + ] + ), + dbc.FormGroup( + [ + dbc.Label("Enter your code PIN", html_for="psa-pin"), + dbc.Input( + type="password", + id="psa-pin", + placeholder="Enter codepin", + ), + dbc.FormText( + "It's a digit password", + color="secondary", + ), + ] + ), + dbc.Button("Submit", color="primary", id="finish-otp"), + html.Div(id="opt-result") + ])])) + + +def log_layout(): + with open(LOG_FILE, "r") as f: + log_text = f.read() + return html.H3(children=["Log:", dbc.Textarea( + valid=True, + bs_size="sm", + className="mt-3", + style={"height": "80vh"}, + placeholder="Log", + contentEditable=False, + value=log_text + )]) + + +@dash_app.callback( + Output("form_result", "children"), + Input("submit-form", "n_clicks"), + State("psa-app", "value"), + State("psa-email", "value"), + State("psa-password", "value"), + State("psa-countrycode", "value")) +def connectPSA(n_clicks, app_name, email, password, countrycode): # pylint: disable=unused-argument + ctx = callback_context + if ctx.triggered: + try: + res = firstLaunchConfig(app_name, email, password, countrycode) + config.load_app() + return dbc.Alert([res, html.A(" Go to otp config", href=request.url_root + "config_otp")], color="success") + except Exception as e: + res = str(e) + return dbc.Alert(res, color="danger") + else: + return "" + raise PreventUpdate() + + +@dash_app.callback( + Output("sms-demand-result", "children"), + Input("ask-sms", "n_clicks")) +def askCode(n_clicks): # pylint: disable=unused-argument + ctx = callback_context + if ctx.triggered: + try: + config.myp.get_sms_otp_code() + return dbc.Alert("Sms sent", color="success") + except Exception as e: + res = str(e) + return dbc.Alert(res, color="danger") + raise PreventUpdate() + +@dash_app.callback( + Output("opt-result", "children"), + Input("finish-otp", "n_clicks"), + State("psa-pin", "value"), + State("psa-code", "value")) +def finishOtp(n_clicks, code_pin, sms_code): # pylint: disable=unused-argument + ctx = callback_context + if ctx.triggered: + try: + otp_session = new_otp_session(smscode=sms_code, codepin=code_pin) + config.myp.otp = otp_session + Config().start_remote_control() + return dbc.Alert(["OTP config finish !!! ", html.A("Go to home", href=request.url_root)], + color="success") + except Exception as e: + res = str(e) + return dbc.Alert(res, color="danger") + raise PreventUpdate() \ No newline at end of file diff --git a/web/views.py b/web/views.py index 8a0917a..fcd2143 100644 --- a/web/views.py +++ b/web/views.py @@ -18,13 +18,15 @@ from trip import Trips from libs.charging import Charging from web import figures -from web.app import app, dash_app, myp, chc +from web.app import app, dash_app from web.db import Database +from web.config_views import config_layout, config_otp_layout, log_layout from web.utils import diff_dashtable, dash_date_to_datetime # pylint: disable=invalid-name from web.figurefilter import FigureFilter from web.utils import create_card +from libs.config import Config RESPONSE = "-response" EMPTY_DIV = "empty-div" @@ -34,6 +36,21 @@ CALLBACK_CREATED = False trips: Trips = Trips() chargings: List[dict] min_date = max_date = min_millis = max_millis = step = marks = cached_layout = None +CONFIG = Config() + + +@dash_app.callback(Output('page-content', 'children'), + [Input('url', 'pathname')]) +def display_page(pathname): + if pathname == "/config": + return config_layout + if pathname == "/log": + return log_layout() + if not CONFIG.is_good: + return dcc.Location(pathname="/config", id="config_redirect") + if pathname == "/config_otp": + return config_otp_layout + return serve_layout() def create_callback(): # noqa: MC0001 @@ -68,7 +85,7 @@ def create_callback(): # noqa: MC0001 is_open = False if active_cell is not None and active_cell["column_id"] in ["start_level", "end_level"] and not is_open: row = data[active_cell["row"]] - return figures.get_battery_curve_fig(row, myp.vehicles_list[0]), True + return figures.get_battery_curve_fig(row, CONFIG.myp.vehicles_list[0]), True return "", False @dash_app.callback([Output("tab_trips_popup_graph", "children"), Output("tab_trips_popup", "is_open"), ], @@ -91,17 +108,17 @@ def create_callback(): # noqa: MC0001 def update_abrp(div_id, value): vin = div_id["vin"] if value: - myp.abrp.abrp_enable_vin.add(vin) + CONFIG.myp.abrp.abrp_enable_vin.add(vin) else: - myp.abrp.abrp_enable_vin.discard(vin) - myp.save_config() + CONFIG.myp.abrp.abrp_enable_vin.discard(vin) + CONFIG.myp.save_config() return " " @app.route('/get_vehicles') def get_vehicules(): response = app.response_class( - response=json.dumps(myp.get_vehicles(), default=lambda car: car.to_dict()), + response=json.dumps(CONFIG.myp.get_vehicles(), default=lambda car: car.to_dict()), status=200, mimetype='application/json' ) @@ -112,7 +129,7 @@ def get_vehicules(): def get_vehicle_info(vin): from_cache = int(request.args.get('from_cache', 0)) == 1 response = app.response_class( - response=json.dumps(myp.get_vehicle_info(vin, from_cache).to_dict(), default=str), + response=json.dumps(CONFIG.myp.get_vehicle_info(vin, from_cache).to_dict(), default=str), status=200, mimetype='application/json' ) @@ -136,27 +153,27 @@ def get_style(): @app.route('/charge_now//') def charge_now(vin, charge): - return jsonify(myp.charge_now(vin, charge != 0)) + return jsonify(CONFIG.myp.charge_now(vin, charge != 0)) @app.route('/charge_hour') def change_charge_hour(): - return jsonify(myp.change_charge_hour(request.form['vin'], request.form['hour'], request.form['minute'])) + return jsonify(CONFIG.myp.change_charge_hour(request.form['vin'], request.form['hour'], request.form['minute'])) @app.route('/wakeup/') def wakeup(vin): - return jsonify(myp.wakeup(vin)) + return jsonify(CONFIG.myp.wakeup(vin)) @app.route('/preconditioning//') def preconditioning(vin, activate): - return jsonify(myp.preconditioning(vin, activate)) + return jsonify(CONFIG.myp.preconditioning(vin, activate)) @app.route('/position/') def get_position(vin): - res = myp.get_vehicle_info(vin) + res = CONFIG.myp.get_vehicle_info(vin) try: coordinates = res.last_position.geometry.coordinates except AttributeError: @@ -176,14 +193,14 @@ def get_position(vin): def get_charge_control(): logger.info(request) vin = request.args['vin'] - charge_control = chc.get(vin) + charge_control = CONFIG.chc.get(vin) if charge_control is None: return jsonify("error: VIN not in list") if 'hour' in request.args and 'minute' in request.args: charge_control.set_stop_hour([int(request.args["hour"]), int(request.args["minute"])]) if 'percentage' in request.args: charge_control.percentage_threshold = int(request.args['percentage']) - chc.save_config() + CONFIG.chc.save_config() return jsonify(charge_control.get_dict()) @@ -199,12 +216,12 @@ def abrp(): token = request.args.get('token', None) if vin is not None and enable is not None: if enable == '1': - myp.abrp.abrp_enable_vin.add(vin) + CONFIG.myp.abrp.abrp_enable_vin.add(vin) else: - myp.abrp.abrp_enable_vin.discard(vin) + CONFIG.myp.abrp.abrp_enable_vin.discard(vin) if token is not None: - myp.abrp.token = token - return jsonify(dict(myp.abrp)) + CONFIG.myp.abrp.token = token + return jsonify(dict(CONFIG.myp.abrp)) @app.after_request @@ -222,49 +239,50 @@ def update_trips(): conn.close() min_date = None max_date = None - car = myp.vehicles_list[0] # todo handle multiple car - 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): - logger.debug("No trips yet") - figures.get_figures(Car("vin","vid","brand")) - try: - chargings = Charging.get_chargings() - assert len(chargings) > 0 - 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) - min_millis = web.utils.unix_time_millis(min_date) - max_millis = web.utils.unix_time_millis(max_date) - step = (max_millis - min_millis) / 100 - marks = web.utils.get_marks_from_start_end(min_date, max_date) - cached_layout = None # force regenerate layout - figures.get_figures(car) - except (ValueError, IndexError): - logger.error("update_trips (slider): %s", exc_info=True) - except AttributeError: - logger.debug("position table is probably empty :", exc_info=True) + if CONFIG.is_good: + car = CONFIG.myp.vehicles_list[0] # todo handle multiple car + 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): + logger.debug("No trips yet") + figures.get_figures(Car("vin", "vid", "brand")) + try: + chargings = Charging.get_chargings() + assert len(chargings) > 0 + 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) + min_millis = web.utils.unix_time_millis(min_date) + max_millis = web.utils.unix_time_millis(max_date) + step = (max_millis - min_millis) / 100 + marks = web.utils.get_marks_from_start_end(min_date, max_date) + cached_layout = None # force regenerate layout + figures.get_figures(car) + except (ValueError, IndexError): + logger.error("update_trips (slider): %s", exc_info=True) + except AttributeError: + logger.debug("position table is probably empty :", exc_info=True) return def __get_control_tabs(): tabs = [] - for car in myp.vehicles_list: + for car in CONFIG.myp.vehicles_list: if car.label is None: label = car.vin else: @@ -273,7 +291,7 @@ def __get_control_tabs(): tabs.append(dbc.Tab(label=label, id="tab-" + car.vin, children=[ daq.ToggleSwitch( id={'role': ABRP_SWITCH, 'vin': car.vin}, - value=car.vin in myp.abrp.abrp_enable_vin, + value=car.vin in CONFIG.myp.abrp.abrp_enable_vin, label="Send data to ABRP" ), html.Div(id={'role': ABRP_SWITCH + RESPONSE, 'vin': car.vin}) @@ -311,7 +329,7 @@ def serve_layout(): fig_filter.src = {"trips": trips.get_trips_as_dict(), "chargings": chargings} fig_filter.set_clientside_callback(dash_app) create_callback() - except (IndexError, TypeError, NameError, AssertionError, NameError): + except (IndexError, TypeError, NameError, AssertionError, NameError, AttributeError): summary_tab = figures.ERROR_DIV maps = figures.ERROR_DIV logger.warning("Failed to generate figure, there is probably not enough data yet", exc_info_debug=True) @@ -320,8 +338,8 @@ def serve_layout(): data_div = html.Div([ *fig_filter.get_store(), - range_slider, html.Div([ + range_slider, dbc.Tabs([ dbc.Tab(label="Summary", tab_id="summary", children=summary_tab), dbc.Tab(label="Trips", tab_id="trips", id="tab_trips", @@ -368,7 +386,7 @@ def serve_layout(): html.Div(id=EMPTY_DIV), html.Div(id=EMPTY_DIV + "1") ])]) - cached_layout = dbc.Container(fluid=True, children=[html.H1('My car info'), data_div]) + cached_layout = data_div return cached_layout @@ -379,4 +397,5 @@ try: except (IndexError, TypeError): logger.debug("Failed to get trips, there is probably not enough data yet:", exc_info=True) -dash_app.layout = serve_layout +dash_app.layout = dbc.Container(fluid=True, children=[dcc.Location(id='url', refresh=False), + html.H1('My car info'), html.Div(id='page-content')])