add co2signal api

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

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)

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