electricity: start on EM implementation

Author: Shillaker

.github/workflows/test.yml

@@ -44,6 +44,8 @@ jobs:
 
   test-e2e:
     runs-on: ubuntu-latest
+    env:
+      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,50 @@
+import requests
+
+ELECTRICITY_MAPS_LATEST_URL = (
+    "https://api.electricitymap.org/v3/carbon-intensity/latest"
+)
+
+
+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 and ValueError when the
+    response does not contain the expected data.
+    """
+    params = {"zone": zone, "emissionFactorType": "lifecycle"}
+    headers = {"auth-token": api_key}
+
+    try:
+        response = requests.get(
+            ELECTRICITY_MAPS_LATEST_URL, params=params, headers=headers, timeout=10
+        )
+        response.raise_for_status()
+    except requests.RequestException as exc:
+        raise ConnectionError(f"Electricity Maps API request failed: {exc}") from exc
+
+    data = response.json()
+    carbon_intensity = data.get("carbonIntensity")
+    if carbon_intensity is None:
+        raise ValueError(
+            f"Electricity Maps API response missing 'carbonIntensity': {data}"
+        )
+
+    value = carbon_intensity / 1000
+
+    return {
+        "unit": "kg CO2eq/kWh",
+        "source": "Electricity Maps API (lifecycle)",
+        "value": value,
+        "min": value,
+        "max": value,
+    }

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
 
 config_file = os.path.join(data_dir, "factors.yml")
 impact_factors = yaml.load(Path(config_file).read_text(), Loader=yaml.CSafeLoader)
@@ -26,6 +29,10 @@ def get_gpu_impact_factor(component, phase, impact_type) -> dict:
 
 
 def get_electrical_impact_factor(usage_location, impact_type) -> dict:
+    if config.electricitymaps_api_key and impact_type == "gwp":
+        zone = iso3_to_iso2(usage_location)
+        return fetch_carbon_intensity(config.electricitymaps_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 +95,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,9 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
+    # Electricity Maps API key (if set, real-time GWP factors are fetched)
+    electricitymaps_api_key: Optional[str] = None
+
     # Location and usage defaults
     default_location: str = "EEE"
     default_usage: str = "DEFAULT"

boaviztapi/utils/country.py

@@ -0,0 +1,19 @@
+import pycountry
+
+
+def iso3_to_iso2(iso3_code: str) -> str:
+    """Convert an ISO 3166-1 alpha-3 country code to alpha-2.
+
+    Raises ValueError for codes that cannot be mapped to a valid
+    Electricity Maps zone (e.g. WOR, EEE, or unknown codes).
+    """
+    if iso3_code in ("WOR", "EEE"):
+        raise ValueError(
+            f"Cannot convert '{iso3_code}' to an Electricity Maps zone code"
+        )
+
+    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,46 @@ 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 or historical electricity impact factors by setting the `ELECTRICITY_MAPS_API_KEY` environment variable to a valid API key for the Electricity Maps API.
+
+**Note:** 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.
+
 ### 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 +74,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,13 @@ 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 modifying the following parameter:
+
+```
+electricity_maps_api_key: <your API key>
+```
+
+These can be overridden with the `ELECTRICITY_MAPS_API_KEY` environment variable.

poetry.lock

@@ -21,6 +21,8 @@ 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"
 
 # Security updates
 aiohttp = "^3.13.3"

pyproject.toml

@@ -0,0 +1,55 @@
+import os
+
+import pytest
+
+from boaviztapi.service.factor_provider import get_electrical_impact_factor
+
+ELECTRICITY_MAPS_API_KEY = os.environ.get("ELECTRICITY_MAPS_API_KEY")
+
+pytestmark = [
+    pytest.mark.e2e,
+    pytest.mark.skipif(
+        not ELECTRICITY_MAPS_API_KEY,
+        reason="ELECTRICITY_MAPS_API_KEY environment variable not set",
+    ),
+]
+
+
+@pytest.fixture(autouse=True)
+def _set_api_key(monkeypatch):
+    """Inject the real API key into the application config."""
+    monkeypatch.setattr(
+        "boaviztapi.service.factor_provider.config.electricitymaps_api_key",
+        ELECTRICITY_MAPS_API_KEY,
+    )
+
+
+class TestGetElectricalImpactFactor:
+    def test_returns_gwp_factor_for_known_country(self):
+        """get_electrical_impact_factor returns a valid GWP factor via Electricity Maps."""
+        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):
+        """Non-gwp criteria fall back to hardcoded factors even when API key is set."""
+        result = get_electrical_impact_factor("FRA", "pe")
+
+        assert "value" in result
+        assert "unit" in result
+        assert result.get("source") != "Electricity Maps API (lifecycle)"
+
+    def test_raises_value_error_for_unmappable_location(self):
+        """Locations like WOR cannot be converted to an Electricity Maps zone."""
+        with pytest.raises(ValueError, match="Cannot convert"):
+            get_electrical_impact_factor("WOR", "gwp")
+
+    def test_raises_value_error_for_unknown_country(self):
+        """Unknown ISO3 codes raise ValueError."""
+        with pytest.raises(ValueError, match="Unknown ISO3 country code"):
+            get_electrical_impact_factor("ZZZ", "gwp")