feat(electricity): add ElectricityMaps integration (part 1, recent data)

.github/workflows/test.yml

@@ -44,6 +44,8 @@ jobs:
 
   test-e2e:
     runs-on: ubuntu-latest
+    env:
+      BOAVIZTA_ELECTRICITY_MAPS_API_KEY: ${{ secrets.SIMON_ELECTRICITYMAPS_SANDBOX_KEY }}
     steps:
     - uses: actions/checkout@v4
     - uses: actions/setup-python@v5

boaviztapi/service/electricitymaps.py

@@ -0,0 +1,51 @@
+import time
+
+from electricitymaps import create_client
+from openapi.exceptions import ApiException
+
+from boaviztapi.utils.config import config
+
+_cache: dict[str, tuple[float, dict]] = {}
+
+
+def fetch_carbon_intensity(api_key: str, zone: str) -> dict:
+    """Fetch real-time carbon intensity from the Electricity Maps API.
+
+    Returns a dict matching the factor format used by factors.yml::
+
+        {
+            "unit": "kg CO2eq/kWh",
+            "source": "Electricity Maps API (lifecycle)",
+            "value": <carbonIntensity / 1000>,
+            "min": <carbonIntensity / 1000>,
+            "max": <carbonIntensity / 1000>,
+        }
+
+    Raises ConnectionError on request failure.
+    """
+    cached = _cache.get(zone)
+    if cached is not None:
+        ts, result = cached
+        if time.monotonic() - ts < config.electricity_maps_cache_expiry_seconds:
+            return result
+
+    client = create_client(api_key=api_key)
+
+    try:
+        response = client.carbon_intensity.latest(zone_key=zone)
+    except ApiException as exc:
+        raise ConnectionError(f"Electricity Maps API request failed: {exc}") from exc
+
+    value = response.carbon_intensity / 1000
+
+    result = {
+        "unit": "kg CO2eq/kWh",
+        "source": "Electricity Maps API (lifecycle)",
+        "value": value,
+        "min": value,
+        "max": value,
+    }
+
+    _cache[zone] = (time.monotonic(), result)
+
+    return result

boaviztapi/service/factor_provider.py

@@ -3,6 +3,9 @@
 
 import yaml
 from boaviztapi import data_dir
+from boaviztapi.service.electricitymaps import fetch_carbon_intensity
+from boaviztapi.utils.config import config
+from boaviztapi.utils.country import iso3_to_iso2, is_iso3
 
 config_file = os.path.join(data_dir, "factors.yml")
 impact_factors = yaml.load(Path(config_file).read_text(), Loader=yaml.CSafeLoader)
@@ -26,6 +29,16 @@ def get_gpu_impact_factor(component, phase, impact_type) -> dict:
 
 
 def get_electrical_impact_factor(usage_location, impact_type) -> dict:
+    # Use Electricity Maps if we have an API key, are looking up GWP, and the location is a valid in ISO 3166-1 alpha-3 country.
+    # Note that calling code may pass non-country locations like WOR or EEE, which cannot be used with Electricity Maps.
+    if (
+        config.electricity_maps_api_key
+        and impact_type == "gwp"
+        and is_iso3(usage_location)
+    ):
+        zone = iso3_to_iso2(usage_location)
+        return fetch_carbon_intensity(config.electricity_maps_api_key, zone)
+
     if impact_factors["electricity"].get(usage_location):
         if impact_factors["electricity"].get(usage_location).get(impact_type):
             return impact_factors["electricity"].get(usage_location).get(impact_type)
@@ -88,48 +101,3 @@ def get_iot_impact_factor(functional_block, hsl, impact_type):
                     .get(hsl)["eol"][impact_type]
                 )
     raise NotImplementedError
-
-
-"""
-_electricity_emission_factors_df = pd.read_csv(
-    os.path.join(data_dir, 'electricity/electricity_impact_factors.csv'))
-class ElecFactorProvider:
-    def get(self, criteria, location, date):
-        pass
-    def get_range(self, criteria, location, date1, date2):
-        pass
-class BoaviztaFactors(ElecFactorProvider):
-    def get(self, criteria, usage_location, date=None):
-        sub = _electricity_emission_factors_df
-        sub = sub[sub['code'] == usage_location]
-        return float(sub[f"{criteria}_emission_factor"]), sub[f"{criteria}_emission_source"].iloc[0], 0, ["The impact factor is averaged over the year"]
-    def get_range(self, criteria, usage_location, date1=None, date2=None):
-        return self.get(criteria,usage_location)
-
-class ElectricityMap(ElecFactorProvider):
-    auth_token = "6QGqlsF7ZcdUN6TMB7jX9DMsYKeGHbVl"
-    url = "https://api-access.electricitymaps.com/2w97h07rvxvuaa1g"
-    now = datetime.now()
-    def get(self, criteria, location, date):
-        zone = self._location_to_em_zone(location)
-        if self.now - timedelta(hours=1) < date < self.now + timedelta(hours=1):
-            return self._get_current(zone)
-        elif date < self.now:
-            return self._get_history(zone, date)
-        else:
-            return NotImplementedError
-    def get_range(self, criteria, zone, date1, date2):
-        pass
-    def _get_current(self, zone):
-        reponse = requests.get(f"{self.url}/carbon-intensity/latest?zone={zone}",
-                headers={"X-BLOBR-KEY": self.auth_token}).json()
-
-        return reponse["carbonIntensity"]/1000, f"electricity map response : {reponse}", 0, []
-    def _get_history(self, zone, date):
-        reponse = requests.get(f"{self.url}/carbon-intensity/history?zone={zone}&datetime={date}",
-                               headers={"X-BLOBR-KEY": self.auth_token }).json()
-
-        return reponse
-    def _location_to_em_zone(self, location):
-        return location
-"""

boaviztapi/utils/config.py

@@ -13,6 +13,10 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
+    # Electricity Maps API key (if set, real-time GWP factors are fetched)
+    electricity_maps_api_key: Optional[str] = None
+    electricity_maps_cache_expiry_seconds: int = 600
+
     # Location and usage defaults
     default_location: str = "EEE"
     default_usage: str = "DEFAULT"

boaviztapi/utils/country.py

@@ -0,0 +1,15 @@
+import pycountry
+
+
+def is_iso3(iso3_code: str) -> bool:
+    """Checks whether country code is present in ISO 3166-1 alpha-3 country list."""
+    return pycountry.countries.get(alpha_3=iso3_code) is not None
+
+
+def iso3_to_iso2(iso3_code: str) -> str:
+    """Convert an ISO 3166-1 alpha-3 country code to alpha-2."""
+    country = pycountry.countries.get(alpha_3=iso3_code)
+    if country is None:
+        raise ValueError(f"Unknown ISO3 country code: '{iso3_code}'")
+
+    return country.alpha_2

docs/docs/Explanations/usage/elec_factors.md

@@ -13,41 +13,52 @@ Users can give their own impact factors.
 
 ## Boavizta's impact factors
 
-Users can use the average impact factors per country available in BoaviztAPI. 
+Users can use the average impact factors per country available in BoaviztAPI.
 
 !!!info
     Impact factors will depend on the `usage_location` defined by the user in usage object. By default, the average european mix is used.
 
-`usage_location` are given in a trigram format, according to the [list of the available countries](countries.md). 
+`usage_location` are given in a trigram format, according to the [list of the available countries](countries.md).
 
 !!!info
     Available countries can be retrieve using the API endpoint `/v1/utils/country_code`.
 
 You can find bellow the data source and methodology used for each impact criteria.
 
+## Electricity Maps Integration
+
+You can use [Electricity Maps](https://app.electricitymaps.com/) to load live electricity impact factors by setting the `BOAVIZTA_ELECTRICITY_MAPS_API_KEY` environment variable to a valid API key for the Electricity Maps API.
+
+Response data for each zone is cached in memory for 10 minutes, but this duration can be configured via the `BOAVIZTA_ELECTRICITY_MAPS_CACHE_EXPIRY_SECONDS` environment variable.
+
+!!!info
+    Electricity Maps currently only supports GWP as an impact factor. As a result, only the `gwp` factor of the _usage_ part of the impact will be based on Electricity Maps data.
+
+!!!info
+    Electricity Maps data is only available for given zones, and it does not provide any kind of global/regional averages. Therefore, the Boavizta location codes of `WOR` and `EEE` will also default to the default hard-coded factors.
+
 ### GWP - Global warming potential factor
 
-_Source_ : 
+_Source_ :
 
 * For Europe (2019): [Quantification of the carbon intensity of electricity produced and used in Europe](https://www.sciencedirect.com/science/article/pii/S0306261921012149)
-* For the rest of the world: [Ember Climate](https://ember-climate.org/data/data-explorer) 
-
+* For the rest of the world: [Ember Climate](https://ember-climate.org/data/data-explorer)
 
 ### PE - Primary energy factor
 
-_Source_ : 
+_Source_ :
 
-PE impact factor are not available in open access. 
+PE impact factor are not available in open access.
 We use the consumption of fossil resources per kwh (APDf/kwh) per country and extrapolate this consumption to renewable energy :
 
 ```PE/kwh = ADPf/kwh / (1-%RenewableEnergyInMix)```
 
 * `%RenewableEnergyInMix` (2016): [List of countries by renewable electricity production](https://en.wikipedia.org/wiki/List_of_countries_by_renewable_electricity_production) from IRENA
-* `ADPf` (2011): [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/) 
+* `ADPf` (2011): [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/)
 
 ### Other impact factors
 
-| Criteria | Implemented | Source                                                   | 
+| Criteria | Implemented | Source                                                   |
 |----------|-------------|----------------------------------------------------------|
 | adp      | yes         | [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/)    |
 | gwppb    | yes         | [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/)    |
@@ -69,7 +80,3 @@ We use the consumption of fossil resources per kwh (APDf/kwh) per country and ex
 | epf      | yes         | [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/)    |
 | epm      | yes         | [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/)    |
 | ept      | yes         | [Base IMPACTS® ADEME](https://base-impacts.ademe.fr/)    |
-
-## Electricity map integration
-
-Coming soon...

docs/docs/config.md

@@ -102,3 +102,7 @@ cpu_name_fuzzymatch_threshold: 62
 ```
 
 This can be overridden with the `BOAVIZTA_CPU_NAME_FUZZYMATCH_THRESHOLD` environment variable.
+
+## Electricity Maps integration
+
+The Electricity Maps integration can be activated by setting the `electricity_maps_api_key` parameter. This can be overridden with the `BOAVIZTA_ELECTRICITY_MAPS_API_KEY` environment variable.

pyproject.toml

@@ -21,6 +21,9 @@ mangum = "^0.20.0"
 importlib-metadata = "^8.7.1"
 pyyaml = "^6.0.3"
 toml = "^0.10.2"
+requests = "^2.32.3"
+pycountry = "^24.6.1"
+electricitymaps = "^0.0.1"
 
 # Security updates
 aiohttp = "^3.13.3"

tests/api/test_cloud_e2e.py

@@ -29,11 +29,14 @@ def _generate_cloud_provider_urls():
                         for instances_row in instances_reader:
                             instance_id = instances_row.get("id")
                             request = CloudInstanceRequest(
-                                provider_name, instance_id, use_url_params=True
+                                provider_name,
+                                instance_id,
+                                use_url_params=True,
+                                usage={
+                                    "usage_location": "FRA",
+                                },
                             )
 
-                            print(f"{provider_name} - {instance_id}")
-
                             urls.append((request.to_url()))
 
                 except FileNotFoundError:

tests/api/test_electricity_maps_e2e.py

@@ -0,0 +1,40 @@
+import pytest
+
+from boaviztapi import config
+from boaviztapi.service.factor_provider import get_electrical_impact_factor
+
+pytestmark = [
+    pytest.mark.e2e,
+    pytest.mark.skipif(
+        not config.electricity_maps_api_key,
+        reason="Electricity Maps API key not set",
+    ),
+]
+
+
+class TestGetElectricalImpactFactor:
+    def test_returns_gwp_factor_for_known_country(self):
+        result = get_electrical_impact_factor("FRA", "gwp")
+
+        assert result["unit"] == "kg CO2eq/kWh"
+        assert result["source"] == "Electricity Maps API (lifecycle)"
+        assert isinstance(result["value"], float)
+        assert result["value"] > 0
+        assert result["min"] == result["value"]
+        assert result["max"] == result["value"]
+
+    def test_non_gwp_falls_back_to_hardcoded(self):
+        result = get_electrical_impact_factor("FRA", "pe")
+
+        assert result["value"] == 11.289
+        assert result["source"] == "ADPf / (1-%renewable_energy)"
+
+    def test_world_location_falls_back_to_hardcoded(self):
+        result = get_electrical_impact_factor("WOR", "gwp")
+
+        assert result["value"] == 0.39
+        assert result["source"] == "Average of all country in the csv"
+
+    def test_unknown_country_falls_back_to_not_implemented(self):
+        with pytest.raises(NotImplementedError):
+            get_electrical_impact_factor("ZZZ", "gwp")

tests/conftest.py

@@ -12,6 +12,9 @@
     bv_config.cpu_name_fuzzymatch_threshold = 60
     bv_config.default_case = "DEFAULT"
     bv_config.default_server = "DEFAULT"
+    bv_config.electricity_maps_api_key = (
+        ""  # Deactivate Electricity Maps integration for unit tests
+    )
 
 
 def pytest_configure(config):