diff --git a/MyPSACC.py b/MyPSACC.py index 6b7deae..f429672 100644 --- a/MyPSACC.py +++ b/MyPSACC.py @@ -36,7 +36,8 @@ realm_info = { "clientsB2CCitroen": {"oauth_url": "https://idpcvs.citroen.com/am/oauth2/access_token", "app_name": "MyCitroen"}, "clientsB2CDS": {"oauth_url": "https://idpcvs.driveds.com/am/oauth2/access_token", "app_name": "MyDS"}, "clientsB2COpel": {"oauth_url": "https://idpcvs.opel.com/am/oauth2/access_token", "app_name": "MyOpel"}, - "clientsB2CVauxhall": {"oauth_url": "https://idpcvs.vauxhall.co.uk/am/oauth2/access_token", "app_name": "MyVauxhall"} + "clientsB2CVauxhall": {"oauth_url": "https://idpcvs.vauxhall.co.uk/am/oauth2/access_token", + "app_name": "MyVauxhall"} } authorize_service = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize" @@ -51,8 +52,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', @@ -124,7 +123,7 @@ class MyPSACC: self.manager.init_with_user_credentials(user, password, self.realm) def __init__(self, refresh_token, client_id, client_secret, remote_refresh_token, customer_id, realm, country_code, - proxies=None, weather_api=None, abrp=None): + proxies=None, weather_api=None, abrp=None, co2_signal_api=None): self.realm = realm self.service_information = ServiceInformation(authorize_service, realm_info[self.realm]['oauth_url'], @@ -164,6 +163,7 @@ class MyPSACC: self.abrp: Abrp = Abrp(**abrp) self.set_proxies(proxies) self.config_file = DEFAULT_CONFIG_FILENAME + self.co2_signal_api = co2_signal_api def get_app_name(self): return realm_info[self.realm]['app_name'] @@ -488,8 +488,9 @@ class MyPSACC: config = dict(**json.loads(config_str)) if "country_code" not in config: config["country_code"] = input("What is your country code ? (ex: FR, GB, DE, ES...)\n") - if "abrp" not in config: - config["abrp"] = None + for new_el in ["abrp", "co2_signal_api"]: + if new_el not in config: + config[new_el] = None psacc = MyPSACC(**config) psacc.config_file = name return psacc @@ -561,6 +562,7 @@ class MyPSACC: if not in_progress: conn.execute("INSERT INTO battery(start_at,start_level,VIN) VALUES(?,?,?)", (charge_date, level, vin)) conn.commit() + Ecomix.get_data_from_co2_signal(latitude, longitude, self.co2_signal_api) else: try: start_at, stop_at, start_level = conn.execute( @@ -568,7 +570,8 @@ class MyPSACC: "DESC limit 1", (vin,)).fetchone() in_progress = stop_at is None if in_progress: - co2_per_kw = Ecomix.get_co2_per_kw(start_at, charge_date, latitude, longitude) + 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 conn.execute( "UPDATE battery set stop_at=?, end_level=?, co2=?, kw=? WHERE start_at=? and VIN=?", @@ -613,11 +616,13 @@ class MyPSACC: for key, value in self.__dict__.items(): yield key, value + 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"]: + "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] return mpd diff --git a/README.md b/README.md index 7fd42df..cdc8f52 100644 --- a/README.md +++ b/README.md @@ -99,7 +99,8 @@ We will retrieve these informations: ![Screenshot_20210128_104519](https://user-images.githubusercontent.com/48728684/106119895-01c98d80-6156-11eb-8969-9e8bc24f3677.png) - You have to add an api key from https://home.openweathermap.org/ in your config file, to be able to see your consumption vs exterior temperature. +- You have to add an api key from https://home.openweathermap.org/ in your config file, to be able to see your consumption vs exterior temperature. +- You have to add an api key from https://co2signal.com/ to have your C02 emission by KM (in France the key isn't needed). ## Connect your home automation system: - [Domoticz](docs/domoticz/Domoticz.md) - [HomeAssistant](https://github.com/Flodu31/HomeAssistant-PeugeotIntegration) diff --git a/ecomix.py b/ecomix.py index 94da2e0..49e703b 100644 --- a/ecomix.py +++ b/ecomix.py @@ -1,13 +1,19 @@ -from datetime import datetime +from datetime import datetime, timedelta from statistics import mean, StatisticsError import xml.etree.cElementTree as ElT import requests import reverse_geocode +import traceback +import numbers from MyLogger import logger +CO2_SIGNAL_REQ_INTERVAL = 600 + class Ecomix: + _cache = {} + @staticmethod def get_data_france(start, end): start_str = start.strftime("%d/%m/%Y") @@ -39,18 +45,65 @@ class Ecomix: return None @staticmethod - def get_co2_per_kw(start: datetime, end: datetime, latitude, longitude): + def get_data_from_co2_signal(latitude, longitude, co2_signal_key): + if co2_signal_key is not None: + try: + country_code = Ecomix.get_country(latitude, longitude) + assert country_code is not None + if country_code not in Ecomix._cache: + Ecomix._cache[country_code] = [] + elif len(Ecomix._cache[country_code]) > 0 and \ + (datetime.now()-Ecomix._cache[country_code][-1][0]).total_seconds() < CO2_SIGNAL_REQ_INTERVAL: + return False + res = requests.get("https://api.co2signal.com/v1/latest", + headers={"auth-token": co2_signal_key}, + params={"countryCode": country_code}) + data = res.json() + value = data["data"]["carbonIntensity"] + assert isinstance(value, numbers.Number) + Ecomix._cache[country_code].append([datetime.now(), value]) + return data["status"] == "ok" + except (AssertionError, NameError, KeyError): + logger.debug(traceback.format_exc()) + return False + else: + return False + + @staticmethod + def clean_cache(): + max_date = datetime.now() - timedelta(days=1) + for country in Ecomix._cache.keys(): + 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]) + if len(co2_per_kw) == 0: + return None + return mean(co2_per_kw) + + @staticmethod + def get_country(latitude, longitude): try: location = reverse_geocode.search([(latitude, longitude)])[0] country_code = location["country_code"] - except UnicodeDecodeError: + return country_code + except (UnicodeDecodeError, IndexError): logger.error("Can't find country for %s %s", latitude, longitude) - country_code = None - except IndexError: - country_code = None - # todo implement other countries - if country_code == 'FR': + return None + + @staticmethod + def get_co2_per_kw(start: datetime, end: datetime, latitude, longitude, from_cache=False): + co2_per_kw = None + country_code = Ecomix.get_country(latitude, longitude) + if country_code is None: + return None + if from_cache: + co2_per_kw = Ecomix.get_co2_from_signal_cache(start, end, country_code) + elif country_code == 'FR': co2_per_kw = Ecomix.get_data_france(start, end) - else: - co2_per_kw = None return co2_per_kw