feat: cache and back off Electricity Maps carbon intensity

codecarbon/core/electricitymaps_api.py

@@ -1,13 +1,106 @@
-from typing import Any, Dict
+import threading
+import time
+from typing import Any, Dict, Tuple
 
 import requests
 
 from codecarbon.core.units import EmissionsPerKWh, Energy
 from codecarbon.external.geography import GeoMetadata
+from codecarbon.external.logger import logger
+
+_Key = Tuple[Tuple[Tuple[str, Any], ...], str]
 
 URL: str = "https://api.electricitymaps.com/v3/carbon-intensity/latest"
 ELECTRICITYMAPS_API_TIMEOUT: int = 30
 
+# Grid carbon intensity is published hourly at best, while emissions are computed
+# on every measurement tick, so the value is cached instead of refetched.
+ELECTRICITYMAPS_CACHE_TTL: int = 60
+# After a failure (bad token, network down), wait before retrying instead of
+# issuing one doomed request per measurement tick.
+ELECTRICITYMAPS_COOLDOWN: int = 60
+
+# {(sorted query params, token): (monotonic fetch time, intensity in gCO2e/kWh)}
+_cache: Dict[_Key, Tuple[float, float]] = {}
+# {cache key: monotonic time until which requests are skipped}
+# Keyed like the cache: one tracker's bad token must not block another's good one.
+_cooldown: Dict[_Key, float] = {}
+# The state above is read-modify-written from the measurement thread.
+_lock = threading.Lock()
+
+
+def reset_cache() -> None:
+    """Drop the cached carbon intensities and any pending failure cooldown."""
+    with _lock:
+        _cache.clear()
+        _cooldown.clear()
+
+
+def _start_cooldown(key: _Key) -> None:
+    with _lock:
+        _cooldown[key] = time.monotonic() + ELECTRICITYMAPS_COOLDOWN
+
+
+def get_carbon_intensity(
+    geo: GeoMetadata, electricitymaps_api_token: str = ""
+) -> float:
+    """
+    Retrieve the carbon intensity of the grid, in gCO2e/kWh, from the Electricity
+    Maps API (formerly CO2 Signal) for the given geographic location.
+
+    Raises:
+        ElectricityMapsAPIError: the request failed, returned an error, or was
+            skipped because a previous one failed (``...CooldownError``).
+    """
+    if geo.latitude:
+        params: Dict[str, Any] = {"lat": geo.latitude, "lon": geo.longitude}
+    else:
+        params = {"countryCode": geo.country_2letter_iso_code}
+    key = (tuple(sorted(params.items())), electricitymaps_api_token)
+    with _lock:
+        cached = _cache.get(key)
+        cooldown_until = _cooldown.get(key, 0.0)
+    if cached and time.monotonic() - cached[0] <= ELECTRICITYMAPS_CACHE_TTL:
+        logger.debug(
+            f"electricitymaps_api: using cached carbon intensity {cached[1]} gCO2e/kWh"
+        )
+        return cached[1]
+
+    remaining = cooldown_until - time.monotonic()
+    if remaining > 0:
+        raise ElectricityMapsAPICooldownError(
+            "Electricity Maps API is in cooldown after a previous failure, "
+            f"retrying in {remaining:.0f} seconds"
+        )
+
+    try:
+        resp = requests.get(
+            URL,
+            params=params,
+            headers={"auth-token": electricitymaps_api_token},
+            timeout=ELECTRICITYMAPS_API_TIMEOUT,
+        )
+        if resp.status_code != 200:
+            try:
+                body = resp.json()
+            except ValueError:
+                body = {}
+            raise ElectricityMapsAPIError(
+                body.get("error") or body.get("message") or resp.text
+            )
+        # API v3 response structure: carbonIntensity is at the root level
+        carbon_intensity_g_per_kWh = resp.json().get("carbonIntensity")
+        if carbon_intensity_g_per_kWh is None:
+            raise ElectricityMapsAPIError("No carbonIntensity data in response")
+    except Exception:
+        _start_cooldown(key)
+        raise
+
+    with _lock:
+        _cache[key] = (time.monotonic(), carbon_intensity_g_per_kWh)
+        _cooldown.pop(key, None)
+    return carbon_intensity_g_per_kWh
+
 
 def get_emissions(
     energy: Energy, geo: GeoMetadata, electricitymaps_api_token: str = ""
@@ -37,28 +130,7 @@ def get_emissions(
         ElectricityMapsAPIError:
             If the Electricity Maps API request fails or returns an error.
     """
-    params: Dict[str, Any]
-    if geo.latitude:
-        params = {"lat": geo.latitude, "lon": geo.longitude}
-    else:
-        params = {"countryCode": geo.country_2letter_iso_code}
-    resp = requests.get(
-        URL,
-        params=params,
-        headers={"auth-token": electricitymaps_api_token},
-        timeout=ELECTRICITYMAPS_API_TIMEOUT,
-    )
-    if resp.status_code != 200:
-        message = resp.json().get("error") or resp.json().get("message")
-        raise ElectricityMapsAPIError(message)
-
-    # API v3 response structure: carbonIntensity is at the root level
-    response_data = resp.json()
-    carbon_intensity_g_per_kWh = response_data.get("carbonIntensity")
-
-    if carbon_intensity_g_per_kWh is None:
-        raise ElectricityMapsAPIError("No carbonIntensity data in response")
-
+    carbon_intensity_g_per_kWh = get_carbon_intensity(geo, electricitymaps_api_token)
     emissions_per_kWh: EmissionsPerKWh = EmissionsPerKWh.from_g_per_kWh(
         carbon_intensity_g_per_kWh
     )
@@ -67,3 +139,7 @@ def get_emissions(
 
 class ElectricityMapsAPIError(Exception):
     pass
+
+
+class ElectricityMapsAPICooldownError(ElectricityMapsAPIError):
+    """Raised when a request is skipped because a previous one failed."""

codecarbon/core/emissions.py

@@ -169,6 +169,13 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
                     + f"Retrieved emissions for {geo.country_name} using Electricity Maps API :{emissions * 1000} g CO2eq"
                 )
                 return emissions
+            except electricitymaps_api.ElectricityMapsAPICooldownError as e:
+                # The failure that started the cooldown was already logged.
+                logger.debug(
+                    "electricitymaps_api.get_emissions: "
+                    + str(e)
+                    + " >>> Using CodeCarbon's data."
+                )
             except Exception as e:
                 logger.error(
                     "electricitymaps_api.get_emissions: "

docs/how-to/configuration.md

@@ -100,6 +100,17 @@ carbon intensity of your grid. The query runs at the end of each tracking run,
 and also periodically during long runs (every
 `api_call_interval × measure_power_secs` seconds; default: every ~2 minutes).
 
+!!! warning "Carbon intensity is cached for 60 seconds"
+
+    A fetched carbon intensity is reused for 60 seconds before the API is
+    queried again, so measurements taken within that window share the same
+    intensity value. Electricity Maps publishes hourly at best, and the default
+    `measure_power_secs` is 15 seconds, so the cache removes roughly three
+    requests out of four without changing the reported figures. After a failure
+    (invalid token, network down), requests are skipped for an exponentially
+    growing cooldown (30 s up to 1 hour), per location and token, and CodeCarbon
+    falls back to its own country data.
+
 The Electricity Maps API offers a free tier. You can sign up and get a token at
 [electricitymaps.com](https://app.electricitymaps.com/sign-up).
 

tests/test_electricitymaps_api.py

@@ -11,6 +11,7 @@
 class TestElectricityMapsAPI(unittest.TestCase):
     def setUp(self) -> None:
         # GIVEN
+        electricitymaps_api.reset_cache()
         self._energy = Energy.from_energy(kWh=10)
         self._geo = GeoMetadata(
             country_iso_code="FRA",
@@ -43,3 +44,15 @@ def test_get_emissions_with_api_key(self):
         result = electricitymaps_api.get_emissions(self._energy, self._geo, api_key)
         # Should return a positive emissions value
         assert result > 0
+
+    @responses.activate
+    def test_non_json_error_body_falls_back_to_text(self):
+        responses.add(
+            responses.GET,
+            electricitymaps_api.URL,
+            body="<html>502 Bad Gateway</html>",
+            status=502,
+        )
+        with pytest.raises(electricitymaps_api.ElectricityMapsAPIError) as error:
+            electricitymaps_api.get_carbon_intensity(self._geo)
+        assert "502 Bad Gateway" in str(error.value)

tests/test_electricitymaps_cache.py

@@ -0,0 +1,160 @@
+import unittest
+from unittest import mock
+
+import responses
+
+from codecarbon.core import electricitymaps_api
+from codecarbon.core.emissions import Emissions
+from codecarbon.core.units import Energy
+from codecarbon.external.geography import GeoMetadata
+from codecarbon.input import DataSource
+
+
+class TestElectricityMapsCache(unittest.TestCase):
+    def setUp(self) -> None:
+        # GIVEN
+        electricitymaps_api.reset_cache()
+        self._geo = GeoMetadata(
+            country_iso_code="FRA",
+            country_name="France",
+            region=None,
+            country_2letter_iso_code="FR",
+        )
+        self._other_geo = GeoMetadata(
+            country_iso_code="DEU",
+            country_name="Germany",
+            region=None,
+            country_2letter_iso_code="DE",
+        )
+
+    def tearDown(self) -> None:
+        electricitymaps_api.reset_cache()
+
+    def _add_success_response(self, carbon_intensity: float = 58.7) -> None:
+        responses.add(
+            responses.GET,
+            electricitymaps_api.URL,
+            json={"zone": "FR", "carbonIntensity": carbon_intensity},
+            status=200,
+        )
+
+    @responses.activate
+    def test_second_call_within_ttl_does_not_hit_the_api(self):
+        self._add_success_response()
+
+        first = electricitymaps_api.get_carbon_intensity(self._geo)
+        second = electricitymaps_api.get_carbon_intensity(self._geo)
+
+        assert first == second == 58.7
+        assert len(responses.calls) == 1
+
+    @responses.activate
+    def test_a_long_run_issues_a_bounded_number_of_requests(self):
+        self._add_success_response()
+
+        for _ in range(1000):
+            electricitymaps_api.get_carbon_intensity(self._geo)
+
+        assert len(responses.calls) == 1
+
+    @responses.activate
+    def test_expired_cache_entry_is_refetched(self):
+        self._add_success_response()
+
+        with mock.patch.object(electricitymaps_api, "ELECTRICITYMAPS_CACHE_TTL", 0):
+            electricitymaps_api.get_carbon_intensity(self._geo)
+            electricitymaps_api.get_carbon_intensity(self._geo)
+
+        assert len(responses.calls) == 2
+
+    @responses.activate
+    def test_cache_is_keyed_by_location(self):
+        self._add_success_response()
+
+        electricitymaps_api.get_carbon_intensity(self._geo)
+        electricitymaps_api.get_carbon_intensity(self._other_geo)
+
+        assert len(responses.calls) == 2
+
+    @responses.activate
+    def test_failure_puts_the_api_in_cooldown(self):
+        responses.add(
+            responses.GET,
+            electricitymaps_api.URL,
+            json={"error": "invalid token"},
+            status=401,
+        )
+
+        with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError):
+            electricitymaps_api.get_carbon_intensity(self._geo)
+        for _ in range(100):
+            with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError):
+                electricitymaps_api.get_carbon_intensity(self._geo)
+
+        assert len(responses.calls) == 1
+
+    @responses.activate
+    def test_cooldown_is_reset_after_a_successful_call(self):
+        responses.add(
+            responses.GET,
+            electricitymaps_api.URL,
+            json={"error": "invalid token"},
+            status=401,
+        )
+        with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError):
+            electricitymaps_api.get_carbon_intensity(self._geo)
+
+        responses.reset()
+        self._add_success_response()
+        key = next(iter(electricitymaps_api._cooldown))
+        electricitymaps_api._cooldown[key] = 0.0
+        electricitymaps_api.get_carbon_intensity(self._geo)
+
+        assert electricitymaps_api._cooldown == {}
+
+    @responses.activate
+    def test_cooldown_is_not_shared_between_tokens(self):
+        responses.add(
+            responses.GET,
+            electricitymaps_api.URL,
+            json={"error": "invalid token"},
+            status=401,
+        )
+        with self.assertRaises(electricitymaps_api.ElectricityMapsAPIError):
+            electricitymaps_api.get_carbon_intensity(self._geo, "bad-token")
+
+        responses.reset()
+        self._add_success_response()
+        # THEN a tracker with a working token is not blocked by the other's
+        # failure cooldown.
+        assert electricitymaps_api.get_carbon_intensity(self._geo, "good") == 58.7
+
+    @responses.activate
+    def test_cache_is_not_shared_between_tokens(self):
+        self._add_success_response(carbon_intensity=58.7)
+        assert electricitymaps_api.get_carbon_intensity(self._geo, "token-a") == 58.7
+
+        responses.reset()
+        self._add_success_response(carbon_intensity=412.0)
+        # WHEN another tracker in the same process uses a different token, it
+        # must not be served the value cached for the first one.
+        assert electricitymaps_api.get_carbon_intensity(self._geo, "token-b") == 412.0
+
+    @responses.activate
+    def test_cooldown_does_not_log_one_error_per_call(self):
+        responses.add(
+            responses.GET,
+            electricitymaps_api.URL,
+            json={"error": "invalid token"},
+            status=401,
+        )
+        emissions = Emissions(DataSource(), electricitymaps_api_token="bad-token")
+        energy = Energy.from_energy(kWh=1.0)
+
+        with mock.patch("codecarbon.core.emissions.logger") as mock_logger:
+            for _ in range(3):
+                emissions.get_private_infra_emissions(energy, self._geo)
+
+        # THEN only the first, real failure is an error; the calls skipped
+        # during the cooldown stay at debug level.
+        assert mock_logger.error.call_count == 1

tests/test_emissions_tracker.py

@@ -11,6 +11,7 @@
 import requests
 import responses
 
+from codecarbon.core import electricitymaps_api
 from codecarbon.core.units import Energy, Power
 from codecarbon.emissions_tracker import (
     EmissionsTracker,
@@ -1038,6 +1039,8 @@ def test_cumulative_emissions_with_varying_intensity(
         mocked_is_nvidia_system,
     ):
         # Setup mocks
+        electricitymaps_api.reset_cache()
+        self.addCleanup(electricitymaps_api.reset_cache)
         mock_geo.return_value = mock.MagicMock(
             latitude=1.0,
             longitude=1.0,
@@ -1092,14 +1095,17 @@ def test_cumulative_emissions_with_varying_intensity(
         data1 = tracker._prepare_emissions_data()
         self.assertAlmostEqual(data1.emissions, 0.1)
 
-        # Step 2
+        # Step 2: drop the cache so this tick sees a fresh intensity, as it
+        # would once the TTL expires. This test is about cumulating deltas.
+        electricitymaps_api.reset_cache()
         tracker._measure_power_and_energy()
         # total_energy = 2.0, delta_energy = 1.0, intensity = 200 => delta_emissions = 0.2 kg
         # total_emissions = 0.3 kg
         data2 = tracker._prepare_emissions_data()
         self.assertAlmostEqual(data2.emissions, 0.3)
 
         # Step 3
+        electricitymaps_api.reset_cache()
         tracker._measure_power_and_energy()
         # total_energy = 3.0, delta_energy = 1.0, intensity = 300 => delta_emissions = 0.3 kg
         # total_emissions = 0.6 kg