mirror of
https://github.com/flobz/psa_car_controller.git
synced 2026-08-26 10:17:18 +00:00
Florian BEZANNIER
+125
@@ -0,0 +1,125 @@
|
||||
import json
|
||||
from copy import copy
|
||||
|
||||
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):
|
||||
self.vin = vin
|
||||
model = None
|
||||
if label is not None:
|
||||
model = CarModel.find_model_by_name(label)
|
||||
if model is None:
|
||||
model = CarModel.find_model_by_vin(self.vin)
|
||||
label = model.name
|
||||
self.vehicle_id = vehicle_id
|
||||
self.label = label
|
||||
self.brand = brand
|
||||
self._status = None
|
||||
self.abrp_name = abrp_name or model.abrp_name
|
||||
self.battery_power = battery_power or model.battery_power
|
||||
self.fuel_capacity = fuel_capacity or model.fuel_capacity
|
||||
self.max_elec_consumption = max_elec_consumption or model.max_elec_consumption # kwh/100Km
|
||||
self.max_fuel_consumption = max_fuel_consumption or model.max_fuel_consumption # L/100Km
|
||||
|
||||
def set_model_name(self, name):
|
||||
self.label = name
|
||||
|
||||
def is_electric(self) -> bool:
|
||||
return self.fuel_capacity == 0 and self.battery_power > 0
|
||||
|
||||
def is_thermal(self) -> bool:
|
||||
return self.fuel_capacity > 0 and self.battery_power == 0
|
||||
|
||||
def is_hybrid(self) -> bool:
|
||||
return self.fuel_capacity > 0 and self.battery_power > 0
|
||||
|
||||
def get_status(self):
|
||||
if self.status is not None:
|
||||
return self.status
|
||||
logger.error("status of %s is None", self.vin)
|
||||
raise ValueError("status of %s is None")
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, data: dict):
|
||||
return cls(**data)
|
||||
|
||||
def to_dict(self):
|
||||
car_dict = copy(self.__dict__)
|
||||
car_dict.pop("_status")
|
||||
return car_dict
|
||||
|
||||
def __str__(self):
|
||||
return str(self.to_dict())
|
||||
|
||||
def get_abrp_name(self):
|
||||
if self.abrp_name is not None:
|
||||
return self.abrp_name
|
||||
raise ValueError("ABRP model is not set")
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
return self._status
|
||||
|
||||
@status.setter
|
||||
def status(self, value: CarStatus):
|
||||
self._status = value
|
||||
if self._status is not None and self.status.__class__ != CarStatus:
|
||||
self._status.__class__ = CarStatus
|
||||
self._status.correct()
|
||||
|
||||
|
||||
class Cars(list):
|
||||
def __init__(self, *args):
|
||||
list.__init__(self, *args)
|
||||
self.config_filename = "../cars.json"
|
||||
|
||||
def get_car_by_vin(self, vin) -> Car:
|
||||
for car in self:
|
||||
if car.vin == vin:
|
||||
return car
|
||||
return None
|
||||
|
||||
def get_car_by_id(self, vehicle_id) -> Car:
|
||||
for car in self:
|
||||
if car.vehicle_id == vehicle_id:
|
||||
return car
|
||||
return None
|
||||
|
||||
def add(self, car: Car):
|
||||
if self.get_car_by_id(car.vehicle_id) is None:
|
||||
self.append(car)
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, data: list):
|
||||
cars = list(map(Car.from_json, data))
|
||||
return cls(cars)
|
||||
|
||||
def __str__(self):
|
||||
return str(list(map(str, self)))
|
||||
|
||||
def save_cars(self, name=None):
|
||||
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 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 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 ex:
|
||||
logger.debug(ex)
|
||||
return Cars()
|
||||
+3
-1
@@ -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)
|
||||
|
||||
+2
-1
@@ -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,
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
from typing import List
|
||||
|
||||
from libs.elec_price import ElecPrice
|
||||
from web.db import get_db, set_chargings_price, clean_battery
|
||||
|
||||
elec_price = ElecPrice.read_config()
|
||||
|
||||
|
||||
class Charging:
|
||||
@staticmethod
|
||||
def get_chargings(mini=None, maxi=None) -> List[dict]:
|
||||
conn = get_db()
|
||||
if mini is not None:
|
||||
if maxi is not None:
|
||||
res = conn.execute("select * from battery WHERE start_at>=? and start_at<=?", (mini, maxi)).fetchall()
|
||||
else:
|
||||
res = conn.execute("select * from battery WHERE start_at>=?", (mini,)).fetchall()
|
||||
elif maxi is not None:
|
||||
res = conn.execute("select * from battery WHERE start_at<=?", (maxi,)).fetchall()
|
||||
else:
|
||||
res = conn.execute("select * from battery").fetchall()
|
||||
conn.close()
|
||||
return list(map(dict, res))
|
||||
|
||||
@staticmethod
|
||||
def set_default_price():
|
||||
if elec_price.is_enable():
|
||||
conn = get_db()
|
||||
charge_list = list(map(dict, conn.execute("SELECT * FROM battery WHERE price IS NULL").fetchall()))
|
||||
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, 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, consumption_kw, price, start_at, vin))
|
||||
clean_battery(conn)
|
||||
@@ -0,0 +1,95 @@
|
||||
from datetime import datetime, timezone, timedelta
|
||||
import configparser
|
||||
from statistics import mean
|
||||
|
||||
CONFIG_FILENAME = "config.ini"
|
||||
|
||||
|
||||
def set_number(value):
|
||||
try:
|
||||
return float(value)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def utc_to_local(utc_dt):
|
||||
return utc_dt.replace(tzinfo=timezone.utc).astimezone(tz=None)
|
||||
|
||||
|
||||
class ElecPrice:
|
||||
currency = ""
|
||||
|
||||
def __init__(self, day_price, night_price=None, nights_hours=None):
|
||||
self.day_price = set_number(day_price)
|
||||
self.night_price = set_number(night_price)
|
||||
self.nights_hour = None
|
||||
self.set_night_hour(nights_hours)
|
||||
self.config_filename = CONFIG_FILENAME
|
||||
|
||||
def set_night_hour(self, value):
|
||||
if value is not None and isinstance(value, list):
|
||||
self.nights_hour = []
|
||||
for hours in value:
|
||||
self.nights_hour.append(list(map(int, hours)))
|
||||
|
||||
@staticmethod
|
||||
def compare_hour(date: datetime, hour, minute):
|
||||
if date.hour < hour:
|
||||
return False
|
||||
if date.hour == hour and date.minute < minute:
|
||||
return False
|
||||
return True
|
||||
|
||||
def get_instant_price(self, date):
|
||||
local_date = utc_to_local(date)
|
||||
if self.night_price is None:
|
||||
return self.day_price
|
||||
if self.compare_hour(local_date, self.nights_hour[0][0], self.nights_hour[0][1]) or \
|
||||
not self.compare_hour(local_date, self.nights_hour[1][0], self.nights_hour[1][1]):
|
||||
return self.night_price
|
||||
return self.day_price
|
||||
|
||||
def get_price(self, start, end, consumption):
|
||||
prices = []
|
||||
date = start
|
||||
while date < end:
|
||||
prices.append(self.get_instant_price(date))
|
||||
date = date + timedelta(minutes=30)
|
||||
return round(consumption * mean(prices), 2)
|
||||
|
||||
def is_enable(self):
|
||||
return self.day_price is not None
|
||||
|
||||
@staticmethod
|
||||
def read_config(name=CONFIG_FILENAME):
|
||||
config = configparser.ConfigParser()
|
||||
if len(config.read(name)) == 0:
|
||||
ElecPrice.write_default_config(name)
|
||||
config.read(name)
|
||||
elec_config = config["Electricity config"]
|
||||
if len(elec_config["night price"]) > 0:
|
||||
night_hours = []
|
||||
night_price = elec_config["night price"]
|
||||
for hour in [elec_config["night hour start"], elec_config["night hour end"]]:
|
||||
night_hours.append(hour.split("h"))
|
||||
else:
|
||||
night_hours = None
|
||||
night_price = None
|
||||
ElecPrice.currency = config["General"]["currency"]
|
||||
return ElecPrice(elec_config["day price"], night_price, night_hours)
|
||||
|
||||
@staticmethod
|
||||
def write_default_config(name=CONFIG_FILENAME):
|
||||
config = configparser.ConfigParser()
|
||||
config["General"] = {
|
||||
"currency": "€"
|
||||
}
|
||||
config["Electricity config"] = {
|
||||
"day price": "",
|
||||
"night price": "",
|
||||
"night hour start": "",
|
||||
"night hour end": ""
|
||||
}
|
||||
|
||||
with open(name, "w") as f:
|
||||
config.write(f)
|
||||
@@ -0,0 +1,78 @@
|
||||
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
|
||||
return None
|
||||
Reference in New Issue
Block a user