mirror of
https://github.com/flobz/psa_car_controller.git
synced 2026-08-22 17:36:15 +00:00
apply autopep8
This commit is contained in:
@@ -35,4 +35,4 @@ BRAND = {"com.psa.mym.myopel": {"realm": "clientsB2COpel", "brand_code": "OP", "
|
||||
"com.psa.mym.mycitroen": {"realm": "clientsB2CCitroen", "brand_code": "AC", "app_name": "MyCitroen"},
|
||||
"com.psa.mym.myds": {"realm": "clientsB2CDS", "brand_code": "DS", "app_name": "MyDS"},
|
||||
"com.psa.mym.myvauxhall": {"realm": "clientsB2CVauxhall", "brand_code": "VX", "app_name": "MyVauxhall"}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ DEFAULT_VERSION = "529"
|
||||
def filter_load(string: str):
|
||||
return string.replace("&", "&")
|
||||
|
||||
|
||||
class IWData:
|
||||
# pylint: disable=invalid-name,too-many-branches,too-many-statements
|
||||
def __init__(self, IW):
|
||||
|
||||
@@ -256,7 +256,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
|
||||
|
||||
@@ -305,7 +305,7 @@ def save_otp(obj, filename="otp.bin"):
|
||||
|
||||
class RenameUnpickler(pickle.Unpickler):
|
||||
def find_class(self, module, name):
|
||||
renamed_module = "psa."+module.lower()
|
||||
renamed_module = "psa." + module.lower()
|
||||
return super().find_class(renamed_module, name)
|
||||
|
||||
|
||||
|
||||
@@ -32,7 +32,7 @@ class ApkParser:
|
||||
self.client_secret = resources.get_string(package_name, "PSA_API_CLIENT_SECRET_PROD")[1]
|
||||
self.host_brandid_prod = resources.get_string(package_name, "HOST_BRANDID_PROD")[1]
|
||||
self.culture = self.__get_cultures_code(a.get_file("res/raw/cultures.json"), self.country_code)
|
||||
## Get Customer id
|
||||
# Get Customer id
|
||||
self.site_code = BRAND[package_name]["brand_code"] + "_" + self.country_code + "_ESP"
|
||||
pfx_cert = a.get_file("assets/MWPMYMA1.pfx")
|
||||
save_key_to_pem(pfx_cert, b"y5Y2my5B")
|
||||
@@ -51,4 +51,4 @@ def save_key_to_pem(pfx_data, pfx_password):
|
||||
with open("certs/private.pem", "wb") as f:
|
||||
f.write(private_key.private_bytes(encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption()))
|
||||
encryption_algorithm=serialization.NoEncryption()))
|
||||
|
||||
@@ -51,7 +51,7 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code,
|
||||
f"sitecode: {apk_parser.site_code}"
|
||||
try:
|
||||
msg += res.text
|
||||
except: # pylint: disable=bare-except
|
||||
except BaseException:
|
||||
pass
|
||||
logger.error(msg)
|
||||
raise Exception(msg) from ex
|
||||
@@ -81,7 +81,7 @@ def firstLaunchConfig(package_name, client_email, client_password, country_code,
|
||||
msg = traceback.format_exc()
|
||||
try:
|
||||
msg += res2.text
|
||||
except: # pylint: disable=bare-except
|
||||
except BaseException:
|
||||
pass
|
||||
logger.error(msg)
|
||||
raise Exception(msg) from ex
|
||||
|
||||
@@ -35,9 +35,9 @@ def urlretrieve_from_github(user, repo, directory, filename, branch="main"):
|
||||
r = requests.get("https://github.com/{}/{}/raw/{}/{}{}".format(user, repo, branch, directory, filename),
|
||||
headers={
|
||||
"Accept": "application/vnd.github.VERSION.raw"
|
||||
},
|
||||
stream=True
|
||||
)
|
||||
},
|
||||
stream=True
|
||||
)
|
||||
|
||||
r.raise_for_status()
|
||||
for chunk in r.iter_content(1024):
|
||||
|
||||
@@ -9,6 +9,7 @@ from psacc.model.car import Car
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Abrp:
|
||||
api_key = "1e28ad14-df16-49f0-97da-364c9154b44a"
|
||||
url = "https://api.iternio.com/1/tlm/send"
|
||||
|
||||
@@ -92,7 +92,7 @@ class ChargeControl:
|
||||
if next_in_second < self.psacc.info_refresh_rate:
|
||||
periodicity = next_in_second
|
||||
thread = threading.Timer(periodicity, self.process)
|
||||
thread.setDaemon(True) # pylint: disable=deprecated-method
|
||||
thread.setDaemon(True)
|
||||
thread.start()
|
||||
else:
|
||||
if self._next_stop_hour is not None and self._next_stop_hour < now:
|
||||
@@ -100,7 +100,7 @@ class ChargeControl:
|
||||
self.retry_count = 0
|
||||
except (AttributeError, ValueError):
|
||||
logger.exception("Probably can't retrieve all information from API:")
|
||||
except: # pylint: disable=bare-except
|
||||
except BaseException:
|
||||
logger.exception("Charge control:")
|
||||
|
||||
def get_dict(self):
|
||||
|
||||
@@ -53,10 +53,8 @@ class Charging:
|
||||
Database.update_charge(charge)
|
||||
Database.clean_battery(conn)
|
||||
|
||||
|
||||
|
||||
@staticmethod
|
||||
def is_charge_ended(charge:'Charge'):
|
||||
def is_charge_ended(charge: 'Charge'):
|
||||
return not charge or charge.stop_at
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -15,6 +15,7 @@ CO2_SIGNAL_URL = "https://api.co2signal.com"
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Ecomix:
|
||||
_cache = {}
|
||||
co2_signal_key = None
|
||||
|
||||
@@ -122,13 +122,13 @@ class PSAClient:
|
||||
self.get_vehicle_info(car.vin)
|
||||
for callback in self.info_callback:
|
||||
callback()
|
||||
except: # pylint: disable=bare-except
|
||||
except BaseException:
|
||||
logger.exception("refresh_vehicle_info: ")
|
||||
sleep(self.info_refresh_rate)
|
||||
|
||||
def start_refresh_thread(self):
|
||||
if self.refresh_thread is None:
|
||||
self.refresh_thread = threading.Thread(target=self.__refresh_vehicle_info,daemon=True)
|
||||
self.refresh_thread = threading.Thread(target=self.__refresh_vehicle_info, daemon=True)
|
||||
self.refresh_thread.start()
|
||||
|
||||
def get_vehicles(self):
|
||||
|
||||
@@ -8,6 +8,7 @@ LEVEL_FUEL = 8
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TripParser:
|
||||
def __init__(self, car: Car):
|
||||
self.car = car
|
||||
|
||||
@@ -8,6 +8,7 @@ from .car import Car
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Points:
|
||||
def __init__(self, latitude, longitude):
|
||||
self.latitude = latitude
|
||||
|
||||
@@ -15,20 +15,20 @@ logger = logging.getLogger(__name__)
|
||||
DEFAULT_CONFIG = """[General]
|
||||
currency = €
|
||||
# minimum trip length in km so it's added to stats and map in website
|
||||
minimum trip length =
|
||||
minimum trip length =
|
||||
|
||||
[Electricity config]
|
||||
# price by kw/h
|
||||
day price =
|
||||
day price =
|
||||
night price =
|
||||
# ex: 22h30
|
||||
night hour start =
|
||||
# ex: 6h00
|
||||
night hour end =
|
||||
dc charge price =
|
||||
high speed dc charge price =
|
||||
dc charge price =
|
||||
high speed dc charge price =
|
||||
# minimum power in kW that should be delivered during a charge so it can be considered as a high speed charger
|
||||
high speed dc charge threshold =
|
||||
high speed dc charge threshold =
|
||||
"""
|
||||
|
||||
|
||||
@@ -150,7 +150,7 @@ class ConfigRepository(BaseModel):
|
||||
def _read_file(name):
|
||||
if name is None:
|
||||
name = CONFIG_FILENAME
|
||||
with open(name, "r") as f:
|
||||
with open(name, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
@staticmethod
|
||||
@@ -186,7 +186,7 @@ class ConfigRepository(BaseModel):
|
||||
|
||||
@staticmethod
|
||||
def _write(name, config):
|
||||
with open(name, "w") as f:
|
||||
with open(name, "w", encoding="utf-8") as f:
|
||||
config.write(f)
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -187,7 +187,7 @@ class Database:
|
||||
@staticmethod
|
||||
def get_battery_curve(conn, start_at, stop_at, vin):
|
||||
battery_curves = []
|
||||
res = conn.execute("""SELECT date, level, rate, autonomy
|
||||
res = conn.execute("""SELECT date, level, rate, autonomy
|
||||
FROM battery_curve
|
||||
WHERE start_at=? and date<=? and VIN=?;""",
|
||||
(start_at, stop_at, vin)).fetchall()
|
||||
|
||||
@@ -9,7 +9,6 @@ from dash.dependencies import Output, Input
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
|
||||
class Graph:
|
||||
def __init__(self, graph_id, x, y: [], figure):
|
||||
self.graph_id = graph_id
|
||||
@@ -97,7 +96,7 @@ class FigureFilter:
|
||||
res = "{"
|
||||
i = ord("a")
|
||||
for table in self.tables:
|
||||
res+= f'"{table.table_id}": {chr(i)},'
|
||||
res += f'"{table.table_id}": {chr(i)},'
|
||||
i += 1
|
||||
res = res[:-1] + "}"
|
||||
return res
|
||||
|
||||
@@ -109,7 +109,7 @@ def log_layout():
|
||||
"white-space": "pre-line"},
|
||||
children=log_text,
|
||||
className="m-3 bg-light h5"),
|
||||
html.Div(id="empty-div")])
|
||||
html.Div(id="empty-div")])
|
||||
|
||||
|
||||
def config_layout(activeTabs="log"):
|
||||
|
||||
@@ -357,8 +357,8 @@ def serve_layout():
|
||||
],
|
||||
id="tab_trips_popup",
|
||||
size="xl",
|
||||
)
|
||||
]),
|
||||
)
|
||||
]),
|
||||
dbc.Tab(label="Charge", tab_id="charge", id="tab_charge",
|
||||
children=[figures.battery_table,
|
||||
dbc.Modal(
|
||||
|
||||
Reference in New Issue
Block a user