github/psa_car_controller
1
0
Fork 0
mirror of https://github.com/flobz/psa_car_controller.git synced 2026-08-21 17:06:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

code clean

Browse Source
This commit is contained in:
Florian Bezannier
2021-04-21 11:44:46 +02:00
parent 98baca6601
commit 40a7f20d72
28 changed files with 382 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pre-commit-config.yaml
+1 -1
View File
@@ -1,5 +1,5 @@
repos:
- repo: https://github.com/PyCQA/prospector
rev: 1.3.0 # The version of Prospector to use, at least 1.1.7
rev: 1.3.1 # The version of Prospector to use, at least 1.1.7
hooks:
- id: prospector
.prospector.yaml
+12 -4
View File
@@ -1,7 +1,15 @@
doc-warnings: false
pylint:
disable:
- C0114
- C0111
ignore-paths:
- psa_connectedcar
pep8:
disable:
- E722
pyflakes:
disable:
- F401
pylint:
disable:
- fixme
- C0114
- C0115
- C0116
.pylintrc
+18
View File
@@ -0,0 +1,18 @@
[MASTER]
disable=
C0114,C0115,C0116,W0603,I0011,W0511
[FORMAT]
max-line-length=120
[BASIC]
good-names=i,
j,
k,
ex,
Run,
e,
f,
t,
x,
ip,
s
app_decoder.py
+15 -14
View File
@@ -2,6 +2,10 @@
import json
import os
import traceback
from sys import argv
import sys
import re
from getpass import getpass
from androguard.core.bytecodes.apk import APK
@@ -10,13 +14,8 @@ from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.hazmat.backends import default_backend
from getpass import getpass
from ChargeControl import ChargeControl, ChargeControls
from MyPSACC import MyPSACC
from sys import argv
import sys
import re
from charge_control import ChargeControl, ChargeControls
from my_psacc import MyPSACC
BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", "app_name": "MyOpel"},
"com.psa.mym.mypeugeot": {"realm": "clientsB2CPeugeot", "brand_code": "AP", "app_name": "MyPeugeot"},
@@ -85,7 +84,7 @@ 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
REMOTE_REFRESH_TOKEN = None
print("APK loaded !")
client_email = input(f"{BRAND[package_name]['app_name']} email: ")
@@ -105,12 +104,13 @@ try:
params={"jsonRequest": json.dumps(
{"siteCode": site_code, "culture": "fr-FR", "action": "authenticate",
"fields": {"USR_EMAIL": {"value": client_email},
"USR_PASSWORD": {"value": client_password}}})
}
"USR_PASSWORD": {"value": client_password}}
}
)}
)
token = res.json()["accessToken"]
except:
except: # pylint: disable=bare-except
traceback.print_exc()
print(f"HOST_BRANDID : {HOST_BRANDID_PROD} sitecode: {site_code}")
print(res.text)
@@ -120,7 +120,8 @@ save_key_to_pem(pfx_cert, "")
try:
res2 = requests.post(
f"https://mw-{BRAND[package_name]['brand_code'].lower()}-m2c.mym.awsmpsa.com/api/v1/user?culture=fr_FR&width=1080&v=1.27.0",
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",
@@ -136,14 +137,14 @@ try:
res_dict = res2.json()["success"]
customer_id = BRAND[package_name]["brand_code"] + "-" + res_dict["id"]
except:
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"],
psacc = MyPSACC(None, client_id, client_secret, REMOTE_REFRESH_TOKEN, customer_id, BRAND[package_name]["realm"],
country_code)
psacc.connect(client_email, client_password)
ChargeControl.py → charge_control.py
+8 -7
View File
@@ -8,8 +8,8 @@ from time import sleep
import pytz
from MyPSACC import MyPSACC
from MyLogger import logger
from my_psacc import MyPSACC
from mylogger import logger
DISCONNECTED = "Disconnected"
INPROGRESS = "InProgress"
@@ -93,7 +93,7 @@ class ChargeControl:
self.retry_count = 0
except AttributeError:
logger.error("Probably can't retrieve all information from API: %s", traceback.format_exc())
except:
except: # pylint: disable=bare-except
logger.error(traceback.format_exc())
def get_dict(self):
@@ -125,12 +125,12 @@ class ChargeControls(dict):
@staticmethod
def load_config(psacc: MyPSACC, name="charge_config.json"):
with open(name, "r") as f:
config_str = f.read()
with open(name, "r") as file:
config_str = file.read()
chd = json.loads(config_str)
charge_control_list = ChargeControls(name)
for vin, el in chd.items():
charge_control_list[vin] = ChargeControl(psacc, vin, **el)
for vin, params in chd.items():
charge_control_list[vin] = ChargeControl(psacc, vin, **params)
return charge_control_list
def get(self, vin) -> ChargeControl:
@@ -138,6 +138,7 @@ class ChargeControls(dict):
return self[vin]
except KeyError:
pass
return None
def init(self):
for charge_control in self.values():
ecomix.py
+10 -8
View File
@@ -1,12 +1,13 @@
from datetime import datetime, timedelta
from statistics import mean, StatisticsError
import xml.etree.cElementTree as ElT
import numbers
import traceback
import requests
import reverse_geocode
import traceback
import numbers
from MyLogger import logger
from mylogger import logger
CO2_SIGNAL_REQ_INTERVAL = 600
@@ -19,7 +20,8 @@ class Ecomix:
start_str = start.strftime("%d/%m/%Y")
end_str = end.strftime("%d/%m/%Y")
res = requests.get(
f"https://eco2mix.rte-france.com/curves/eco2mixWeb?type=co2&&dateDeb={start_str}&dateFin={end_str}&mode=NORM",
f"https://eco2mix.rte-france.com/curves/eco2mixWeb?type=co2&&dateDeb={start_str}"
f"&dateFin={end_str}&mode=NORM",
headers={
"Origin": "https://www.rte-france.com",
"Referer": "https://www.rte-france.com/eco2mix/les-emissions-de-co2-par-kwh-produit-en-france",
@@ -72,16 +74,16 @@ class Ecomix:
@staticmethod
def clean_cache():
max_date = datetime.now() - timedelta(days=1)
for country in Ecomix._cache.keys():
for country in Ecomix._cache:
Ecomix._cache[country][:] = [x for x in Ecomix._cache[country] if max_date < x[0]]
@staticmethod
def get_co2_from_signal_cache(start: datetime, end: datetime, country_code):
Ecomix.clean_cache()
co2_per_kw = []
for el in Ecomix._cache.get(country_code, []):
if start < el[0] < end:
co2_per_kw.append(el[1])
for row in Ecomix._cache.get(country_code, []):
if start < row[0] < end:
co2_per_kw.append(row[1])
if len(co2_per_kw) == 0:
return None
return mean(co2_per_kw)
Car.py → libs/car.py
+9 -9
View File
@@ -1,11 +1,11 @@
import json
from copy import copy
from MyLogger import logger
from mylogger import logger
from libs.car_model import CarModel
from libs.car_status import CarStatus
# pylint: disable=too-many-instance-attributes,too-many-arguments
class Car:
def __init__(self, vin, vehicle_id, brand, label=None, battery_power=None, fuel_capacity=None,
max_elec_consumption=None, max_fuel_consumption=None, abrp_name=None):
@@ -76,7 +76,7 @@ class Car:
class Cars(list):
def __init__(self, *args):
list.__init__(self, *args)
self.config_filename = "cars.json"
self.config_filename = "../cars.json"
def get_car_by_vin(self, vin) -> Car:
for car in self:
@@ -106,20 +106,20 @@ class Cars(list):
if name is None:
name = self.config_filename
config_str = json.dumps(self, default=lambda car: car.to_dict(), sort_keys=True, indent=4)
with open(name, "w") as f:
f.write(config_str)
with open(name, "w") as file:
file.write(config_str)
@staticmethod
def load_cars(name=None):
if name is None:
name = Cars().config_filename
try:
with open(name, "r") as f:
json_str = f.read()
with open(name, "r") as file:
json_str = file.read()
cars = Cars.from_json(json.loads(json_str))
cars.config_filename = name
cars.save_cars()
return cars
except (FileNotFoundError, TypeError) as e:
logger.debug(e)
except (FileNotFoundError, TypeError) as ex:
logger.debug(ex)
return Cars()
libs/car_model.py
+3 -1
View File
@@ -1,6 +1,6 @@
import re
from MyLogger import logger
from mylogger import logger
DEFAULT_BATTERY_POWER = 46
DEFAULT_FUEL_CAPACITY = 0
@@ -9,6 +9,7 @@ DEFAULT_MAX_FUEL_CONSUMPTION = 30
class CarModel:
# pylint: disable=too-many-arguments
def __init__(self, name, battery_power, fuel_capacity, abrp_name=None, reg=None,
max_elec_consumption=DEFAULT_MAX_ELEC_CONSUMPTION, max_fuel_consumption=DEFAULT_MAX_FUEL_CONSUMPTION):
self.name = name
@@ -42,6 +43,7 @@ class CarModel:
class ElecModel(CarModel):
# pylint: disable=too-many-arguments
def __init__(self, name, battery_power, abrp_name=None, reg=None,
max_elec_consumption=DEFAULT_MAX_ELEC_CONSUMPTION):
super().__init__(name, battery_power, 0, abrp_name, reg, max_elec_consumption, 0)
libs/car_status.py
+2 -1
View File
@@ -1,7 +1,8 @@
from MyLogger import logger
from mylogger import logger
from psa_connectedcar import Position, Geometry, PositionProperties, Kinetic, Energy, EnergyCharging, Status
# pylint: disable=too-many-arguments
class CarStatus(Status):
def __init__(self, embedded=None, links=None, battery=None, doors_state=None, energy=None, environment=None,
ignition=None, kinetic=None, last_position=None, preconditionning=None, privacy=None, safety=None,
libs/charging.py
+7 -6
View File
@@ -25,15 +25,16 @@ class Charging:
if elec_price.is_enable():
conn = get_db()
charge_list = list(map(dict, conn.execute("SELECT * FROM battery WHERE price IS NULL").fetchall()))
for el in charge_list:
el["price"] = elec_price.get_price(el["start_at"], el["stop_at"], el["kw"])
set_chargings_price(conn, el["start_at"], el["price"])
for charge in charge_list:
charge["price"] = elec_price.get_price(charge["start_at"], charge["stop_at"], charge["kw"])
set_chargings_price(conn, charge["start_at"], charge["price"])
conn.close()
# pylint: disable=too-many-arguments
@staticmethod
def update_chargings(conn, start_at, stop_at, level, co2_per_kw, kw, vin):
price = elec_price.get_price(start_at, stop_at, kw)
def update_chargings(conn, start_at, stop_at, level, co2_per_kw, consumption_kw, vin):
price = elec_price.get_price(start_at, stop_at, consumption_kw)
conn.execute(
"UPDATE battery set stop_at=?, end_level=?, co2=?, kw=?, price=? WHERE start_at=? and VIN=?",
(stop_at, level, co2_per_kw, kw, price, start_at, vin))
(stop_at, level, co2_per_kw, consumption_kw, price, start_at, vin))
clean_battery(conn)
libs/elec_price.py
+2 -1
View File
@@ -32,7 +32,8 @@ class ElecPrice:
for hours in value:
self.nights_hour.append(list(map(int, hours)))
def compare_hour(self, date: datetime, hour, minute):
@staticmethod
def compare_hour(date: datetime, hour, minute):
if date.hour < hour:
return False
if date.hour == hour and date.minute < minute:
libs/oauth.py
+77
View File
@@ -0,0 +1,77 @@
from http import HTTPStatus
from oauth2_client.credentials_manager import CredentialManager
from requests import Response
import psa_connectedcar as psac
from mylogger import logger
from psa_connectedcar import ApiClient
from psa_connectedcar.rest import ApiException
class OpenIdCredentialManager(CredentialManager):
def _grant_password_request_realm(self, login: str, password: str, realm: str) -> dict:
return dict(grant_type='password',
username=login,
scope=' '.join(self.service_information.scopes),
password=password, realm=realm)
def init_with_user_credentials_realm(self, login: str, password: str, realm: str):
self._token_request(self._grant_password_request_realm(login, password, realm), True)
@staticmethod
def _is_token_expired(response: Response) -> bool:
if response.status_code == HTTPStatus.UNAUTHORIZED.value:
logger.info("token expired, renew")
try:
json_data = response.json()
return json_data.get('moreInformation') == 'Token is invalid'
except ValueError:
return False
else:
return False
@property
def access_token(self):
return self._access_token
class Oauth2PSACCApiConfig(psac.Configuration):
def __init__(self):
super().__init__()
self.refresh_callback = None
def set_refresh_callback(self, callback):
self.refresh_callback = callback
class OauthAPIClient(ApiClient):
# pylint: disable=no-member,too-many-arguments
def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, async_req=None,
_return_http_data_only=None, collection_formats=None,
_preload_content=True, _request_timeout=None):
for _ in range(0, 2):
try:
if not async_req:
return self._ApiClient__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 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()
else:
raise e
MyPSACC.py → my_psacc.py
+56 -111
View File
@@ -4,29 +4,27 @@ import threading
import traceback
import uuid
from datetime import datetime
from http import HTTPStatus
from json import JSONEncoder
from hashlib import md5
from time import sleep
from oauth2_client.credentials_manager import CredentialManager, ServiceInformation
from oauth2_client.credentials_manager import ServiceInformation
import paho.mqtt.client as mqtt
from requests import Response
from geojson import Feature, Point, FeatureCollection
from geojson import dumps as geo_dumps
import psa_connectedcar as psac
from Car import Cars, Car
from ecomix import Ecomix
from libs.car import Cars, Car
from libs.charging import Charging
from otp.Otp import load_otp, new_otp_session, save_otp, ConfigException, Otp
from psa_connectedcar import ApiClient
from libs.oauth import OpenIdCredentialManager, Oauth2PSACCApiConfig, OauthAPIClient
from ecomix import Ecomix
from otp.otp import load_otp, new_otp_session, save_otp, ConfigException, Otp
from psa_connectedcar.rest import ApiException
from MyLogger import logger
from mylogger import logger
from utils import get_temp, rate_limit
from web.abrp import Abrp
from web.db import get_db, clean_position, get_last_temp
from geojson import Feature, Point, FeatureCollection
from geojson import dumps as geo_dumps
PSA_CORRELATION_DATE_FORMAT = "%Y%m%d%H%M%S%f"
PSA_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
@@ -40,9 +38,9 @@ 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="
scopes = ['openid profile']
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']
MQTT_SERVER = "mwa.mpsa.com"
MQTT_REQ_TOPIC = "psa/RemoteServices/from/cid/"
MQTT_RESP_TOPIC = "psa/RemoteServices/to/cid/"
@@ -52,65 +50,6 @@ CARS_FILE = "cars.json"
DEFAULT_CONFIG_FILENAME = "config.json"
class OpenIdCredentialManager(CredentialManager):
def _grant_password_request(self, login: str, password: str, realm: str) -> dict:
return dict(grant_type='password',
username=login,
scope=' '.join(self.service_information.scopes),
password=password, realm=realm)
def init_with_user_credentials(self, login: str, password: str, realm: str):
self._token_request(self._grant_password_request(login, password, realm), True)
@staticmethod
def _is_token_expired(response: Response) -> bool:
if response.status_code == HTTPStatus.UNAUTHORIZED.value:
logger.info("token expired, renew")
try:
json_data = response.json()
return json_data.get('moreInformation') == 'Token is invalid'
except ValueError:
return False
else:
return False
class Oauth2PSACCApiConfig(psac.Configuration):
def set_refresh_callback(self, callback):
self.refresh_callback = callback
class OauthAPIClient(ApiClient):
def call_api(self, resource_path, method,
path_params=None, query_params=None, header_params=None,
body=None, post_params=None, files=None,
response_type=None, auth_settings=None, async_req=None,
_return_http_data_only=None, collection_formats=None,
_preload_content=True, _request_timeout=None):
for _ in range(0, 2):
try:
if not async_req:
return self._ApiClient__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 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()
else:
raise e
def gen_correlation_id(date):
date_str = date.strftime(PSA_CORRELATION_DATE_FORMAT)[:-3]
uuid_str = str(uuid.uuid4()).replace("-", "")
@@ -118,22 +57,24 @@ def gen_correlation_id(date):
return correlation_id
# pylint: disable=too-many-instance-attributes,too-many-public-methods
class MyPSACC:
def connect(self, user, password):
self.manager.init_with_user_credentials(user, password, self.realm)
self.manager.init_with_user_credentials_realm(user, password, self.realm)
# pylint: disable=too-many-arguments
def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code,
proxies=None, weather_api=None, abrp=None, co2_signal_api=None):
self.realm = realm
self.service_information = ServiceInformation(authorize_service,
self.service_information = ServiceInformation(AUTHORIZE_SERVICE,
realm_info[self.realm]['oauth_url'],
client_id,
client_secret,
scopes, False)
SCOPE, False)
self.client_id = client_id
self.manager = OpenIdCredentialManager(self.service_information)
self.api_config = Oauth2PSACCApiConfig()
self.api_config.set_refresh_callback(self.manager._refresh_token)
self.api_config.set_refresh_callback(self.refresh_token)
self.manager.refresh_token = refresh_token
self.remote_refresh_token = remote_refresh_token
self.remote_access_token = None
@@ -169,11 +110,12 @@ class MyPSACC:
return realm_info[self.realm]['app_name']
def refresh_token(self):
# pylint: disable=protected-access
self.manager._refresh_token()
self.save_config()
def api(self) -> psac.VehiclesApi:
self.api_config.access_token = self.manager._access_token
self.api_config.access_token = self.manager.access_token
api_instance = psac.VehiclesApi(OauthAPIClient(self.api_config))
return api_instance
@@ -261,7 +203,7 @@ class MyPSACC:
return otp_code
def get_remote_access_token(self, password):
res = self.manager.post(remote_url + self.client_id,
res = self.manager.post(REMOTE_URL + self.client_id,
json={"grant_type": "password", "password": password},
headers=self.headers)
data = res.json()
@@ -278,7 +220,7 @@ class MyPSACC:
if self.remote_refresh_token is None:
logger.error("remote_refresh_token isn't defined")
self.load_otp(force_new=True)
res = self.manager.post(remote_url + self.client_id,
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()
@@ -296,8 +238,9 @@ class MyPSACC:
self.save_config()
return res
def on_mqtt_connect(self, client, userdata, rc, a):
logger.info("Connected with result code %s", rc)
# pylint: disable=unused-argument
def __on_mqtt_connect(self, client, userdata, result_code, _):
logger.info("Connected with result code %s", result_code)
topics = [MQTT_RESP_TOPIC + self.customer_id + "/#"]
for car in self.vehicles_list:
topics.append(MQTT_EVENT_TOPIC + car.vin)
@@ -305,14 +248,16 @@ class MyPSACC:
client.subscribe(topic)
logger.info("subscribe to %s", topic)
def on_mqtt_disconnect(self, client, userdata, rc):
logger.warning("Disconnected with result code %d", rc)
if rc == 1:
# pylint: disable=unused-argument
def _on_mqtt_disconnect(self, client, userdata, result_code):
logger.warning("Disconnected with result code %d", result_code)
if result_code == 1:
self.refresh_remote_token(force=True)
else:
logger.warning(mqtt.error_string(rc))
logger.warning(mqtt.error_string(result_code))
def on_mqtt_message(self, client, userdata, msg):
# pylint: disable=unused-argument
def __on_mqtt_message(self, client, userdata, msg):
try:
logger.info("mqtt msg %s %s", msg.topic, msg.payload)
data = json.loads(msg.payload)
@@ -345,9 +290,9 @@ class MyPSACC:
self.mqtt_client = mqtt.Client(clean_session=True, protocol=mqtt.MQTTv311)
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.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()
@@ -371,18 +316,18 @@ class MyPSACC:
return json.dumps(data)
def get_charge_hour(self, vin):
def __get_charge_hour(self, vin):
reg = r"PT([0-9]{1,2})H([0-9]{1,2})?"
data = self.get_vehicle_info(vin)
hour_str = data.get_energy('Electric').charging.next_delayed_time
try:
hour = re.findall(reg, hour_str)[0]
h = int(hour[0])
if hour[1] == '':
m = 0
hour_minute = re.findall(reg, hour_str)[0]
hour = int(hour_minute[0])
if hour_minute[1] == '':
minute = 0
else:
m = hour[1]
return h, m
minute = hour_minute[1]
return hour, minute
except IndexError:
logger.error(traceback.format_exc())
logger.error("Can't get charge hour: %s", hour_str)
@@ -392,15 +337,13 @@ class MyPSACC:
status = data.get_energy('Electric').charging.status
return status
def veh_charge_request(self, vin, hour, miinute, charge_type):
# todo consider actual state before change the hour
def __veh_charge_request(self, vin, hour, miinute, charge_type):
msg = self.mqtt_request(vin, {"program": {"hour": hour, "minute": miinute}, "type": charge_type})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehCharge", msg)
def change_charge_hour(self, vin, hour, miinute):
# todo consider actual state before change the hour
self.veh_charge_request(vin, hour, miinute, "delayed")
self.__veh_charge_request(vin, hour, miinute, "delayed")
return True
def charge_now(self, vin, now):
@@ -408,8 +351,8 @@ class MyPSACC:
charge_type = "immediate"
else:
charge_type = "delayed"
hour, minute = self.get_charge_hour(vin)
self.veh_charge_request(vin, hour, minute, charge_type)
hour, minute = self.__get_charge_hour(vin)
self.__veh_charge_request(vin, hour, minute, charge_type)
return True
def horn(self, vin, count):
@@ -513,16 +456,16 @@ class MyPSACC:
logger.debug("vin:%s longitude:%s latitude:%s date:%s mileage:%s level:%s charge_date:%s level_fuel:"
"%s moving:%s", car.vin, longitude, latitude, date, mileage, level, charge_date, level_fuel,
moving)
self.record_position(car.vin, mileage, latitude, longitude, date, level, level_fuel, moving)
self.__record_position(car.vin, mileage, latitude, longitude, date, level, level_fuel, moving)
self.abrp.call(car, get_last_temp(car.vin))
try:
charging_status = car.status.get_energy('Electric').charging.status
self.record_charging(car.vin, charging_status, charge_date, level, latitude, longitude)
self.__record_charging(car.vin, charging_status, charge_date, level, latitude, longitude)
logger.debug("charging_status:%s ", charging_status)
except AttributeError:
logger.error("charging status not available from api")
def record_position(self, vin, mileage, latitude, longitude, date, level, level_fuel, moving):
def __record_position(self, vin, mileage, latitude, longitude, date, level, level_fuel, moving):
conn = get_db()
if mileage == 0: # fix a bug of the api
logger.error("The api return a wrong mileage for %s : %f", vin, mileage)
@@ -551,7 +494,7 @@ class MyPSACC:
logger.debug("position already saved")
return False
def record_charging(self, vin, charging_status, charge_date, level, latitude, longitude):
def __record_charging(self, vin, charging_status, charge_date, level, latitude, longitude):
conn = get_db()
if charging_status == "InProgress":
try:
@@ -572,9 +515,9 @@ class MyPSACC:
if in_progress:
co2_per_kw = Ecomix.get_co2_per_kw(start_at, charge_date, latitude, longitude,
from_cache=self.co2_signal_api is not None)
kw = (level - start_level) / 100 * self.vehicles_list.get_car_by_vin(vin).battery_power
consumption_kw = (level - start_level) / 100 * self.vehicles_list.get_car_by_vin(vin).battery_power
Charging.update_chargings(conn, start_at, charge_date, level, co2_per_kw, kw, vin)
Charging.update_chargings(conn, start_at, charge_date, level, co2_per_kw, consumption_kw, vin)
conn.commit()
except TypeError:
logger.debug("battery table is empty")
@@ -602,12 +545,14 @@ class MyPSACC:
yield key, value
# pylint: disable=arguments-differ
class MyPeugeotEncoder(JSONEncoder):
def default(self, mp: MyPSACC):
data = dict(mp)
mpd = {"proxies": data["_proxies"], "refresh_token": mp.manager.refresh_token,
"client_secret": mp.service_information.client_secret, "abrp": dict(mp.abrp)}
for el in ["client_id", "realm", "remote_refresh_token", "customer_id", "weather_api", "country_code",
"co2_signal_api"]:
mpd[el] = data[el]
for param in ["client_id", "realm", "remote_refresh_token", "customer_id", "weather_api", "country_code",
"co2_signal_api"]:
mpd[param] = data[param]
return mpd
MyLogger.py → mylogger.py
+2 -2
View File
@@ -4,19 +4,19 @@ from logging.handlers import RotatingFileHandler
DEBUG_LEVELV_NUM = 9
logging.addLevelName(DEBUG_LEVELV_NUM, "DEBUGV")
def debugv(self, message, *args, **kws):
self.log(DEBUG_LEVELV_NUM, message, *args, **kws)
logging.Logger.debugv = debugv
# pylint: disable=invalid-name
logger = logging.getLogger("log")
def my_logger(file='activity.log', handler_level=logging.INFO):
global logger
#logger.setLevel(logging.DEBUG)
logger.setLevel(handler_level)
formatter = logging.Formatter('%(asctime)s :: %(levelname)s :: %(message)s')
file_handler = RotatingFileHandler(file, 'a', 1000000, 1, encoding='utf8')
otp/__init__.py
+1
View File
@@ -0,0 +1 @@
from .otp import Otp
otp/load.py
+15 -14
View File
@@ -4,25 +4,25 @@ from time import time
from Cryptodome.Cipher import AES
from otp.Tokenizer import Tokenizer
from .tokenizer import Tokenizer
default_token = "0.2.11&&&&&&0&&0&&0&&9f13ba238fbabba08e85d93638e98ef5e48682a9d3e5bc325c3dd6fac8199a6ce09e9b4f373aa6a" \
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"
DEFAULT_VERSION = "529"
def filterLoad(string: str):
def filter_load(string: str):
return string.replace("&amp;", "&")
# pylint: disable=invalid-name,too-many-instance-attributes,too-many-branches,too-many-statements
class IWData:
def __init__(self, IW):
self.IW = IW
self.tokenizer = Tokenizer(default_token)
self.tokenizer = Tokenizer(DEFAULT_TOKEN)
self.tokenizer.nextToken()
self.load1xx(int(default_version), self.tokenizer)
self.load1xx(int(DEFAULT_VERSION), self.tokenizer)
def load1xx(self, j, tokenizer):
self.iwid = tokenizer.nextToken()
@@ -64,8 +64,8 @@ class IWData:
i = 0
while i < self.iwsrvn:
self.iwsrvid[i] = tokenizer.nextToken()
self.iwsrvname[i] = filterLoad(tokenizer.nextToken())
self.iwsrvlogo[i] = filterLoad(tokenizer.nextToken())
self.iwsrvname[i] = filter_load(tokenizer.nextToken())
self.iwsrvlogo[i] = filter_load(tokenizer.nextToken())
if self.IW.isMac:
self.iwsrvconnected[i] = tokenizer.nextTokenI()
if j > 515:
@@ -77,7 +77,7 @@ class IWData:
if i2 < 0 or self.IW.isMac:
self.iwsrvurl[i] = ""
else:
self.iwsrvurl[i] = filterLoad(tokenizer.nextToken())
self.iwsrvurl[i] = filter_load(tokenizer.nextToken())
if j < 520 or self.IW.isMac:
self.iwsrvonlineotp[i] = 0
else:
@@ -106,16 +106,17 @@ class IWData:
i4 = 0
while i4 < self.iwmsgn:
self.iwmsgid += tokenizer.nextToken()
self.iwmsgtitle += filterLoad(tokenizer.nextToken())
self.iwmsgcontent += filterLoad(tokenizer.nextToken())
self.iwmsgtitle += filter_load(tokenizer.nextToken())
self.iwmsgcontent += filter_load(tokenizer.nextToken())
self.iwmsgack += tokenizer.nextTokenI()
i4 += 1
self.iwmajorversion = tokenizer.nextTokenI()
self.iwnewversion = filterLoad(tokenizer.nextToken())
self.iwnewversionurl = filterLoad(tokenizer.nextToken())
self.iwnewversion = filter_load(tokenizer.nextToken())
self.iwnewversionurl = filter_load(tokenizer.nextToken())
self.mustupgrade = False
self.datatouch = 0
# pylint: disable=attribute-defined-outside-init
def synchro(self, ixml: dict, key):
aes_cipher = AES.new(bytes.fromhex(key), AES.MODE_ECB)
value = ixml.get("id")
otp/oaep.py
+1
View File
@@ -7,6 +7,7 @@ from Cryptodome.Util.strxor import strxor
class MyOAEP(PKCS1OAEP_Cipher):
# pylint: disable=too-many-locals,invalid-name
def decrypt(self, ciphertext):
"""Decrypt a message with PKCS#1 OAEP.
otp/Otp.py → otp/otp.py
+9 -8
View File
@@ -1,21 +1,22 @@
import hashlib
import traceback
import pickle
from secrets import token_hex, token_bytes
from math import ceil
from collections import defaultdict
from xml.etree import cElementTree as ElT
import requests
from Cryptodome.Cipher import AES
from Cryptodome.PublicKey import RSA
from Cryptodome import Hash
from math import ceil
from collections import defaultdict
from xml.etree import cElementTree as ElT
from mylogger import logger
from otp import oaep
from otp.load import IWData
import pickle
from MyLogger import logger
from . import oaep
from .load import IWData
# pylint: disable=too-many-instance-attributes,invalid-name
def etree_to_dict(t):
d = {t.tag: {} if t.attrib else None}
@@ -254,7 +255,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
otp/Tokenizer.py → otp/tokenizer.py
+3 -2
View File
@@ -1,3 +1,4 @@
# pylint: disable=invalid-name
class Tokenizer:
def __init__(self, tokens, delimiter="&&"):
self.s: str = tokens
@@ -10,7 +11,7 @@ class Tokenizer:
index_of = self.currentIndex + self.s[self.currentIndex:].index(self.delimiter)
if index_of == -1:
substring = self.s[self.currentIndex:]
self.currentIndex = self.s.length()
self.currentIndex = len(self.s)
return substring
substring2 = self.s[self.currentIndex:index_of]
@@ -24,4 +25,4 @@ class Tokenizer:
return int(token, 16)
def hasMoreTokens(self):
return self.currentIndex < len(self.s)
return self.currentIndex < len(self.s)
server.py
+16 -24
View File
@@ -3,21 +3,21 @@ import atexit
import sys
from os import environ
from threading import Thread
from getpass import getpass
import argparse
from oauth2_client.credentials_manager import OAuthError
from getpass import getpass
import web.app
from ChargeControl import ChargeControls
from MyLogger import my_logger
import argparse
from MyLogger import logger
from MyPSACC import MyPSACC
from charge_control import ChargeControls
from mylogger import my_logger
from mylogger import logger
from my_psacc import MyPSACC
from utils import is_port_in_use
from web.app import start_app, save_config
parser = argparse.ArgumentParser()
CONFIG_NAME = "config.json"
def parse_args():
@@ -27,7 +27,7 @@ def parse_args():
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')
metavar='Debug level number', type=int)
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')
@@ -37,32 +37,24 @@ def parse_args():
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="/")
parser.parse_args()
return parser
return parser.parse_args()
# flake8: noqa: C901
if __name__ == "__main__":
if sys.version_info < (3, 6):
raise RuntimeError("This application requires Python 3.6+")
parser = parse_args()
args = parser.parse_args()
try:
args.debug = int(args.debug)
except ValueError:
pass
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")
exit(1)
sys.exit(1)
logger.info("server start")
if args.config:
config_name = args.config.name
else:
config_name = "config.json"
web.app.myp = MyPSACC.load_config(name=config_name)
CONFIG_NAME = args.config.name
web.app.myp = MyPSACC.load_config(name=CONFIG_NAME)
atexit.register(web.app.myp.save_config)
if args.record:
web.app.myp.set_record(True)
web.app.myp.set_record(args.record)
if args.offline:
logger.info("offline mode")
else:
@@ -91,7 +83,7 @@ if __name__ == "__main__":
t2.setDaemon(True)
t2.start()
save_config(web.app.myp, config_name)
save_config(web.app.myp, CONFIG_NAME)
t1 = Thread(target=start_app, args=["My car info", args.base_path, logger.level < 20, args.listen, int(args.port)])
t1.setDaemon(True)
t1.start()
Trip.py → trip.py
+35 -30
View File
@@ -6,13 +6,14 @@ from typing import List, Dict
from dateutil import tz
from geojson import Feature, FeatureCollection, MultiLineString
from Car import Cars, Car
from MyLogger import logger
from libs.car import Cars, Car
from mylogger import logger
from trip_parser import TripParser
from web.db import get_db
class Points():
class Points:
# pylint: disable= too-few-public-methods
def __init__(self, latitude, longitude):
self.latitude = latitude
self.longitude = longitude
@@ -22,6 +23,7 @@ class Points():
class Trip:
# pylint: disable= too-many-instance-attributes
def __init__(self):
self.start_at = None
self.end_at = None
@@ -105,21 +107,24 @@ class Trips(list):
def get_long_trips(self):
res = []
for tr in self:
if tr.consumption > 1.8:
res.append({"speed": tr.speed_average, "consumption_km": tr.consumption_km, "date": tr.start_at,
"consumption": tr.consumption, "consumption_by_temp": tr.get_temperature()})
for trip in self:
if trip.consumption > 1.8:
res.append({"speed": trip.speed_average, "consumption_km": trip.consumption_km, "date": trip.start_at,
"consumption": trip.consumption, "consumption_by_temp": trip.get_temperature()})
return res
def check_and_append(self, tr: Trip):
if tr.consumption_km <= tr.car.max_elec_consumption and tr.consumption_fuel_km <= tr.car.max_fuel_consumption:
self.append(tr)
def check_and_append(self, trip: Trip):
if trip.consumption_km <= trip.car.max_elec_consumption and \
trip.consumption_fuel_km <= trip.car.max_fuel_consumption:
self.append(trip)
return True
logger.debugv("trip discarded")
return False
# flake8: noqa: C901
@staticmethod
def get_trips(vehicles_list: Cars) -> Dict[str, Trips]:
# pylint: disable=too-many-locals,too-many-statements,too-many-nested-blocks
conn = get_db()
vehicles = conn.execute(
"SELECT DISTINCT vin FROM position;").fetchall()
@@ -134,7 +139,7 @@ class Trips(list):
trip_parser = TripParser(car)
start = res[0]
end = res[1]
tr = Trip()
trip = Trip()
# for debugging use this line res = list(map(dict,res))
for x in range(0, len(res) - 2):
logger.debugv("%s mileage:%.1f level:%s level_fuel:%s",
@@ -154,7 +159,7 @@ class Trips(list):
logger.debugv("low speed detected")
if restart_trip:
start = end
tr = Trip()
trip = Trip()
logger.debugv("restart trip at %s mileage:%.1f level:%s level_fuel:%s",
start['Timestamp'], start['mileage'], start['level'], start['level_fuel'])
else:
@@ -183,33 +188,33 @@ class Trips(list):
if end_trip:
logger.debugv("stop trip at %s mileage:%.1f level:%s level_fuel:%s",
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"]
tr.end_at = end["Timestamp"]
tr.add_points(end["longitude"], end["latitude"])
trip.distance = end["mileage"] - start["mileage"] # km
if trip.distance > 0:
trip.start_at = start["Timestamp"]
trip.end_at = end["Timestamp"]
trip.add_points(end["longitude"], end["latitude"])
if end["temperature"] is not None and start["temperature"] is not None:
tr.add_temperature(end["temperature"])
tr.duration = (end["Timestamp"] - start["Timestamp"]).total_seconds() / 3600
tr.speed_average = tr.distance / tr.duration
trip.add_temperature(end["temperature"])
trip.duration = (end["Timestamp"] - start["Timestamp"]).total_seconds() / 3600
trip.speed_average = trip.distance / trip.duration
diff_level, diff_level_fuel = trip_parser.get_level_consumption(start, end)
tr.car = car
trip.car = car
if diff_level != 0:
tr.set_consumption(diff_level) # kw
trip.set_consumption(diff_level) # kw
if diff_level_fuel != 0:
tr.set_fuel_consumption(diff_level_fuel)
tr.mileage = end["mileage"]
trip.set_fuel_consumption(diff_level_fuel)
trip.mileage = end["mileage"]
logger.debugv("Trip: %s -> %s %.1fkm %.2fh %.0fkm/h %.2fkWh %.2fkWh/100km %.2fL "
"%.2fL/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)
trip.start_at, trip.end_at, trip.distance, trip.duration,
trip.speed_average, trip.consumption, trip.consumption_km,
trip.consumption_fuel, trip.consumption_fuel_km, trip.mileage)
# filter bad value
trips.check_and_append(tr)
trips.check_and_append(trip)
start = next_el
tr = Trip()
trip = Trip()
else:
tr.add_points(end["longitude"], end["latitude"])
trip.add_points(end["longitude"], end["latitude"])
end = next_el
trips_by_vin[vin] = trips
return trips_by_vin
trip_parser.py
+4 -3
View File
@@ -1,6 +1,6 @@
from collections.abc import Callable
from Car import Car
from MyLogger import logger
from libs.car import Car
from mylogger import logger
LEVEL = "level"
@@ -49,6 +49,7 @@ class TripParser:
return True
return False
# pylint: disable=unused-argument
def __is_refuel(self, start, end, distance):
fuel_consumption = self.get_level_consumption(start, end)[1]
if fuel_consumption < 0:
@@ -65,4 +66,4 @@ class TripParser:
# A margin of two is set because battery level can increase with regeneration system or temperature change.
# If distance is bigger than 0 but charge bigger than five there is probably missing point and we assume that
# regeneration/temperature can't increase by 5 percent the battery level
return decharge < -2 and (distance == 0 or decharge < -5)
return decharge < -2 and (distance == 0 or decharge < -5)
utils.py
+6 -6
View File
@@ -1,11 +1,11 @@
import traceback
from functools import wraps
from threading import Semaphore, Timer
import requests
import socket
from MyLogger import logger
import requests
from mylogger import logger
def get_temp(latitude: str, longitude: str, api_key: str) -> float:
@@ -27,14 +27,14 @@ def get_temp(latitude: str, longitude: str, api_key: str) -> float:
def rate_limit(limit, every):
def limit_decorator(fn):
def limit_decorator(func):
semaphore = Semaphore(limit)
@wraps(fn)
@wraps(func)
def wrapper(*args, **kwargs):
semaphore.acquire()
try:
return fn(*args, **kwargs)
return func(*args, **kwargs)
finally: # don't catch but ensure semaphore release
timer = Timer(every, semaphore.release)
timer.setDaemon(True) # allows the timer to be canceled on exit
web/abrp.py
+2 -2
View File
@@ -4,8 +4,8 @@ from datetime import datetime
import requests
from Car import Car
from MyLogger import logger
from libs.car import Car
from mylogger import logger
class Abrp:
web/app.py
+11 -7
View File
@@ -1,9 +1,9 @@
import threading
import locale
import dash
import dash_bootstrap_components as dbc
from flask import Flask
import locale
from werkzeug import run_simple
@@ -12,23 +12,26 @@ try:
except ImportError:
from werkzeug import DispatcherMiddleware
from ChargeControl import ChargeControls
from MyLogger import logger
from MyPSACC import MyPSACC
from charge_control import ChargeControls
from mylogger import logger
from my_psacc import MyPSACC
# pylint: disable=invalid-name
app = None
dash_app = None
dispatcher = None
# noinspection PyTypeChecker
myp:MyPSACC = None
myp: MyPSACC = None
# noinspection PyTypeChecker
chc: ChargeControls = None
def start_app(title, base_path, debug: bool, host, port):
global app, dash_app, dispatcher
try:
lang = locale.getlocale()[0].split("_")[0]
locale.setlocale(locale.LC_TIME, ".".join(locale.getlocale())) #make sure LC_TIME is set
locale.setlocale(locale.LC_TIME, ".".join(locale.getlocale())) # make sure LC_TIME is set
locale_url = [f"https://cdn.plot.ly/plotly-locale-{lang}-latest.js"]
except (IndexError, locale.Error):
locale_url = None
@@ -44,9 +47,10 @@ def start_app(title, base_path, debug: bool, host, port):
dash_app = dash.Dash(external_stylesheets=[dbc.themes.BOOTSTRAP], external_scripts=locale_url, title=title,
server=app, requests_pathname_prefix=requests_pathname_prefix)
# keep this line
import web.views
import web.views # pylint: disable=unused-import,import-outside-toplevel
return run_simple(host, port, application, use_reloader=False, use_debugger=debug)
def save_config(my_peugeot: MyPSACC, name):
my_peugeot.save_config(name)
threading.Timer(30, save_config, args=[my_peugeot, name]).start()
web/db.py
+8 -6
View File
@@ -1,13 +1,14 @@
import sqlite3
from datetime import datetime
import pytz
from typing import Callable
import pytz
from MyLogger import logger
from mylogger import logger
callback_fct: Callable[[], None] = lambda: None
default_db_file = 'info.db'
DEFAULT_DB_FILE = 'info.db'
# pylint: disable=invalid-name
db_initialized = False
@@ -37,7 +38,8 @@ def backup(conn):
def init_db(conn):
global db_initialized
conn.execute("CREATE TABLE IF NOT EXISTS position (Timestamp DATETIME PRIMARY KEY, VIN TEXT, longitude REAL, "
"latitude REAL, mileage REAL, level INTEGER, level_fuel INTEGER, moving BOOLEAN, temperature INTEGER);")
"latitude REAL, mileage REAL, level INTEGER, level_fuel INTEGER, moving BOOLEAN,"
" temperature INTEGER);")
make_backup = False
try:
conn.execute("ALTER TABLE position ADD level_fuel INTEGER;")
@@ -61,7 +63,7 @@ def init_db(conn):
db_initialized = True
def get_db(db_file=default_db_file):
def get_db(db_file=DEFAULT_DB_FILE):
sqlite3.register_converter("DATETIME", convert_datetime_from_bytes)
conn = sqlite3.connect(db_file, detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES)
conn.row_factory = sqlite3.Row
@@ -104,4 +106,4 @@ def set_chargings_price(conn, start_at, price):
conn.commit()
if not update:
logger.error("Can't find line to update in the database")
return update
return update
web/figures.py
+16 -13
View File
@@ -6,18 +6,18 @@ import numpy as np
from dash_core_components import Graph
from dash_table.Format import Format, Scheme, Symbol
from dateutil.relativedelta import relativedelta
from pandas import DataFrame
import plotly.express as px
import plotly.graph_objects as go
from Trip import Trips
from pandas import DataFrame
from pandas import options as pandas_options
import dash_html_components as html
from trip import Trips
from libs.elec_price import ElecPrice
def unix_time_millis(dt):
return int(dt.timestamp())
def unix_time_millis(date):
return int(date.timestamp())
def get_marks_from_start_end(start, end):
@@ -44,6 +44,7 @@ def get_marks_from_start_end(start, end):
return None
# pylint: disable=invalid-name
consumption_fig = None
consumption_df = None
trips_map = None
@@ -56,6 +57,7 @@ battery_info = dbc.Alert("No data to show", color="danger")
battery_table = None
# pylint: disable=too-many-locals
def get_figures(trips: Trips, charging: list[dict]):
global consumption_fig, consumption_df, trips_map, consumption_fig_by_speed, table_fig, info, battery_info, \
battery_table, consumption_graph_by_temp
@@ -73,7 +75,7 @@ def get_figures(trips: Trips, charging: list[dict]):
trips_map = px.line_mapbox(lat=lats, lon=lons, hover_name=names,
mapbox_style="stamen-terrain", zoom=12)
# table
nb_format = Format(precision=2, scheme=Scheme.fixed, symbol=Symbol.yes)
nb_format = Format(precision=2, scheme=Scheme.fixed, symbol=Symbol.yes) # pylint: disable=no-member
table_fig = dash_table.DataTable(
id='trips-table',
sort_action='native',
@@ -132,12 +134,13 @@ def get_figures(trips: Trips, charging: list[dict]):
data=[{"name": "Average emission:", "value": "{:.1f} g/km".format(co2_per_km)},
{"name": " ", "value:": "{:.1f} g/kWh".format(co2_per_kw)},
{"name": "Average charge speed:", "value": "{:.3f} kW".format(charge_speed)}])
battery_info = html.Div(children=[html.Tr(
[
html.Td('Average emission:', rowSpan=2),
html.Td("{:.1f} g/km".format(co2_per_km)),
]
),
battery_info = html.Div(children=[
html.Tr(
[
html.Td('Average emission:', rowSpan=2),
html.Td("{:.1f} g/km".format(co2_per_km)),
]
),
html.Tr(
[
"{:.1f} g/kWh".format(co2_per_kw),
@@ -152,7 +155,7 @@ def get_figures(trips: Trips, charging: list[dict]):
html.Tr(
[
html.Td('Average Price:', rowSpan=2),
html.Td("{:.2f} {}/100km".format(price_kw*kw_per_km, ElecPrice.currency)),
html.Td("{:.2f} {}/100km".format(price_kw * kw_per_km, ElecPrice.currency)),
]
),
html.Tr(
@@ -175,7 +178,7 @@ def get_figures(trips: Trips, charging: list[dict]):
{'id': 'kw', 'name': 'consumption', 'type': 'numeric',
'format': deepcopy(nb_format).symbol_suffix(" kWh").precision(2)},
{'id': 'price', 'name': 'price', 'type': 'numeric',
'format': deepcopy(nb_format).symbol_suffix(" "+ElecPrice.currency).precision(2)}],
'format': deepcopy(nb_format).symbol_suffix(" " + ElecPrice.currency).precision(2)}],
data=charging,
editable=True
)
web/views.py
+33 -29
View File
@@ -3,16 +3,16 @@ import traceback
from datetime import datetime, timezone
import dash_bootstrap_components as dbc
from dash.dependencies import Output, Input, MATCH, State
from dash.exceptions import PreventUpdate
import dash_core_components as dcc
import dash_html_components as html
import dash_daq as daq
import pandas as pd
from dash.exceptions import PreventUpdate
from MyLogger import logger
from flask import jsonify, request, Response as FlaskResponse
from Trip import Trips
from mylogger import logger
from trip import Trips
from libs.charging import Charging
from web import figures
@@ -20,6 +20,8 @@ from web import figures
from web.app import app, dash_app, myp, chc
from web.db import set_chargings_price, get_db, set_db_callback
# pylint: disable=invalid-name
RESPONSE = "-response"
EMPTY_DIV = "empty-div"
ABRP_SWITCH = 'abrp-switch'
@@ -38,7 +40,7 @@ def diff_dashtable(data, data_previous, row_id_name="row_id"):
mask = df.ne(df_previous)
df_diff = df[mask].dropna(how="all", axis="columns").dropna(how="all", axis="rows")
changes = []
for idx, row in df_diff.iterrows():
for row in df_diff.items():
row_id = row.name
row.dropna(inplace=True)
for change in row.iteritems():
@@ -88,13 +90,13 @@ def display_value(value):
State("battery-table", "data_previous"),
],
)
def capture_diffs(ts, data, data_previous):
if ts is None:
def capture_diffs(timestamp, data, data_previous):
if timestamp is None:
raise PreventUpdate
diff_data = diff_dashtable(data, data_previous, "start_at")
for el in diff_data:
if el['column_name'] == 'price':
if not set_chargings_price(get_db(), el['start_at'], el['current_value']):
for changed_line in diff_data:
if changed_line['column_name'] == 'price':
if not set_chargings_price(get_db(), changed_line['start_at'], changed_line['current_value']):
logger.error("Can't find line to update in the database")
return ""
@@ -171,7 +173,7 @@ def get_position(vin):
# Set a battery threshold and schedule an hour to stop the charge
@app.route('/charge_control')
def charge_control():
def get_charge_control():
logger.info(request)
vin = request.args['vin']
charge_control = chc.get(vin)
@@ -245,6 +247,7 @@ def __get_control_tabs():
label = car.vin
else:
label = car.label
# pylint: disable=not-callable
tabs.append(dbc.Tab(label=label, id="tab-" + car.vin, children=[
daq.ToggleSwitch(
id={'role': ABRP_SWITCH, 'vin': car.vin},
@@ -262,23 +265,25 @@ def serve_layout():
logger.debug("Create new layout")
try:
figures.get_figures(trips, chargings)
data_div = html.Div([dcc.RangeSlider(
id='date-slider',
min=min_millis,
max=max_millis,
step=step,
marks=marks,
value=[min_millis, max_millis],
),
data_div = html.Div([
dcc.RangeSlider(
id='date-slider',
min=min_millis,
max=max_millis,
step=step,
marks=marks,
value=[min_millis, max_millis],
),
html.Div([
dbc.Tabs([
dbc.Tab(label="Summary", tab_id="summary", children=[
html.H2(id="consumption",
children=figures.info),
dcc.Graph(figure=figures.consumption_fig, id="consumption_fig"),
dcc.Graph(figure=figures.consumption_fig_by_speed, id="consumption_fig_by_speed"),
figures.consumption_graph_by_temp
]),
dbc.Tab(label="Summary", tab_id="summary",
children=[
html.H2(id="consumption",
children=figures.info),
dcc.Graph(figure=figures.consumption_fig, id="consumption_fig"),
dcc.Graph(figure=figures.consumption_fig_by_speed, id="consumption_fig_by_speed"),
figures.consumption_graph_by_temp
]),
dbc.Tab(label="Trips", tab_id="trips", id="tab_trips", children=[figures.table_fig]),
dbc.Tab(label="Battery", tab_id="battery", id="tab_battery", children=[figures.battery_info]),
dbc.Tab(label="Charge", tab_id="charge", id="tab_charge", children=[figures.battery_table]),
@@ -287,9 +292,8 @@ def serve_layout():
dbc.Tab(label="Control", tab_id="control", children=dbc.Tabs(id="control-tabs",
children=__get_control_tabs()))
],
id="tabs",
active_tab="summary",
),
id="tabs",
active_tab="summary"),
html.Div(id=EMPTY_DIV),
])])
except (IndexError, TypeError, NameError):