feat: cache and back off Electricity Maps carbon intensity

codecarbon/core/electricitymaps_api.py

@@ -1,13 +1,152 @@
-from typing import Any, Dict
+import threading
+import time
+from typing import Any, Dict, Optional, Tuple
 
 import requests
 
 from codecarbon.core.units import EmissionsPerKWh, Energy
 from codecarbon.external.geography import GeoMetadata
+from codecarbon.external.logger import logger
 
 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 = 300
+# After a failure (bad token, network down), retry with an exponential cooldown
+# instead of issuing one doomed request per measurement tick.
+ELECTRICITYMAPS_COOLDOWN_MIN: int = 30
+ELECTRICITYMAPS_COOLDOWN_MAX: int = 3600
+
+# {cache key: (monotonic fetch time, carbon intensity in gCO2e/kWh)}
+_cache: Dict[str, Tuple[float, float]] = {}
+_cooldown_until: float = 0.0
+_cooldown_duration: float = 0.0
+# Emissions are computed from a background measurement thread, so every
+# read-modify-write of the state above is serialised. The lock is never held
+# across the HTTP request.
+_lock = threading.Lock()
+
+
+def reset_cache() -> None:
+    """Drop the cached carbon intensities and any pending failure cooldown."""
+    global _cooldown_until, _cooldown_duration
+    with _lock:
+        _cache.clear()
+        _cooldown_until = 0.0
+        _cooldown_duration = 0.0
+
+
+def _cache_key(params: Dict[str, Any], electricitymaps_api_token: str) -> str:
+    # The token is part of the key: two trackers in one process may use
+    # different tokens, and must not share a cached value. Only an opaque,
+    # process-local marker is kept, so the raw secret is never held in the
+    # cache nor rendered in logs. builtin hash() is randomly seeded per
+    # process and is not a password digest: it is used to tell tokens apart,
+    # never to protect one.
+    joined = ",".join(f"{key}={params[key]}" for key in sorted(params))
+    return f"{joined},token={hash(electricitymaps_api_token):x}"
+
+
+def _get_cached_carbon_intensity(key: str) -> Optional[float]:
+    with _lock:
+        cached = _cache.get(key)
+    if cached is None:
+        return None
+    fetched_at, carbon_intensity_g_per_kWh = cached
+    if time.monotonic() - fetched_at > ELECTRICITYMAPS_CACHE_TTL:
+        return None
+    return carbon_intensity_g_per_kWh
+
+
+def _start_cooldown() -> None:
+    global _cooldown_until, _cooldown_duration
+    with _lock:
+        _cooldown_duration = min(
+            ELECTRICITYMAPS_COOLDOWN_MAX,
+            max(ELECTRICITYMAPS_COOLDOWN_MIN, _cooldown_duration * 2),
+        )
+        _cooldown_until = time.monotonic() + _cooldown_duration
+
+
+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.
+
+    Values are cached for ``ELECTRICITYMAPS_CACHE_TTL`` seconds, and failures put
+    the API in an exponential cooldown during which no request is issued.
+
+    Args:
+        geo (GeoMetadata):
+            Geographic metadata, including either latitude/longitude
+            or a country code.
+        electricitymaps_api_token (str, optional):
+            The API token for authenticating with the Electricity Maps API
+            (default is an empty string).
+
+    Returns:
+        float:
+            The carbon intensity of the grid, in grams of CO2eq per kWh.
+
+    Raises:
+        ElectricityMapsAPIError:
+            If the Electricity Maps API request fails, returns an error, or is
+            currently in a failure cooldown.
+    """
+    global _cooldown_duration
+    params: Dict[str, Any]
+    if geo.latitude:
+        params = {"lat": geo.latitude, "lon": geo.longitude}
+    else:
+        params = {"countryCode": geo.country_2letter_iso_code}
+
+    key = _cache_key(params, electricitymaps_api_token)
+    cached_carbon_intensity = _get_cached_carbon_intensity(key)
+    if cached_carbon_intensity is not None:
+        logger.debug(
+            "electricitymaps_api: using cached carbon intensity "
+            f"{cached_carbon_intensity} gCO2e/kWh for {key}"
+        )
+        return cached_carbon_intensity
+
+    with _lock:
+        cooldown_until = _cooldown_until
+    if time.monotonic() < cooldown_until:
+        raise ElectricityMapsAPICooldownError(
+            "Electricity Maps API is in cooldown after a previous failure, "
+            f"retrying in {cooldown_until - time.monotonic():.0f} seconds"
+        )
+
+    try:
+        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")
+    except Exception:
+        _start_cooldown()
+        raise
+
+    with _lock:
+        _cooldown_duration = 0.0
+        _cache[key] = (time.monotonic(), carbon_intensity_g_per_kWh)
+    return carbon_intensity_g_per_kWh
+
 
 def get_emissions(
     energy: Energy, geo: GeoMetadata, electricitymaps_api_token: str = ""
@@ -37,28 +176,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 +185,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,14 @@ 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 as an
+                # error: skipped requests must not log one line per tick.
+                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,16 @@ 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 5 minutes"
+
+    A fetched carbon intensity is reused for 5 minutes before the API is
+    queried again, so measurements taken within that window share the same
+    intensity value. Electricity Maps publishes hourly at best, but this does
+    mean a run shorter than 5 minutes converts all of its energy with a single
+    intensity reading rather than one per tick. After a failure (invalid token,
+    network down), requests are skipped for an exponentially growing cooldown
+    (30 s up to 1 hour) 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",