diff --git a/codecarbon/core/electricitymaps_api.py b/codecarbon/core/electricitymaps_api.py index cb22c0f79..68b9f89fe 100644 --- a/codecarbon/core/electricitymaps_api.py +++ b/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.""" diff --git a/codecarbon/core/emissions.py b/codecarbon/core/emissions.py index 3b2f10fad..9886cf935 100644 --- a/codecarbon/core/emissions.py +++ b/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: " diff --git a/docs/how-to/configuration.md b/docs/how-to/configuration.md index 9f6766aa1..08e78e5af 100644 --- a/docs/how-to/configuration.md +++ b/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). diff --git a/tests/test_electricitymaps_api.py b/tests/test_electricitymaps_api.py index ce81c0e85..4f60f22ff 100644 --- a/tests/test_electricitymaps_api.py +++ b/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="502 Bad Gateway", + 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) diff --git a/tests/test_electricitymaps_cache.py b/tests/test_electricitymaps_cache.py new file mode 100644 index 000000000..88a900d6b --- /dev/null +++ b/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 diff --git a/tests/test_emissions_tracker.py b/tests/test_emissions_tracker.py index 8ab12e5d8..7f1b4a169 100644 --- a/tests/test_emissions_tracker.py +++ b/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,7 +1095,9 @@ 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 @@ -1100,6 +1105,7 @@ def test_cumulative_emissions_with_varying_intensity( 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