diff --git a/codecarbon/core/emissions.py b/codecarbon/core/emissions.py index 953ca47b3..53597ed2d 100644 --- a/codecarbon/core/emissions.py +++ b/codecarbon/core/emissions.py @@ -10,7 +10,7 @@ import pandas as pd -from codecarbon.core import electricitymaps_api +from codecarbon.core import electricitymaps_api, national_grid_eso_api from codecarbon.core.units import EmissionsPerKWh, Energy from codecarbon.external.geography import CloudMetadata, GeoMetadata from codecarbon.external.logger import logger @@ -175,6 +175,21 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float + " >>> Using CodeCarbon's data." ) + if national_grid_eso_api.is_supported(geo): + try: + emissions = national_grid_eso_api.get_emissions(energy, geo) + logger.debug( + "national_grid_eso_api.get_emissions: " + + f"Retrieved emissions for {geo.country_name} using National Grid ESO API: {emissions * 1000} g CO2eq" + ) + return emissions + except Exception as e: + logger.error( + "national_grid_eso_api.get_emissions: " + + str(e) + + " >>> Using CodeCarbon's data." + ) + country_iso_code = ( geo.country_iso_code.upper() if geo.country_iso_code is not None else None ) diff --git a/codecarbon/core/national_grid_eso_api.py b/codecarbon/core/national_grid_eso_api.py new file mode 100644 index 000000000..539a08351 --- /dev/null +++ b/codecarbon/core/national_grid_eso_api.py @@ -0,0 +1,56 @@ +import requests + +from codecarbon.core.units import EmissionsPerKWh, Energy +from codecarbon.external.geography import GeoMetadata + +URL: str = "https://api.carbonintensity.org.uk/intensity" +NATIONAL_GRID_ESO_API_TIMEOUT: int = 30 + + +def get_emissions(energy: Energy, geo: GeoMetadata) -> float: + """ + Calculate the CO2 emissions using the UK National Grid ESO Carbon Intensity API. + + Args: + energy: Energy consumption in kWh. + geo: Geographic metadata; must have country_iso_code == "GBR". + + Returns: + CO2 emissions in kilograms. + + Raises: + NationalGridESOAPIError: If the API request fails or returns unusable data. + """ + resp = requests.get(URL, timeout=NATIONAL_GRID_ESO_API_TIMEOUT) + if resp.status_code != 200: + raise NationalGridESOAPIError( + f"National Grid ESO API returned status {resp.status_code}" + ) + + try: + intensity = resp.json()["data"][0]["intensity"] + except (KeyError, IndexError, TypeError) as e: + raise NationalGridESOAPIError(f"Unexpected response structure: {e}") from e + + # Prefer actual; fall back to forecast when actual has not been published yet. + actual = intensity.get("actual") + carbon_intensity_g_per_kWh = actual if actual is not None else intensity.get("forecast") + + if carbon_intensity_g_per_kWh is None: + raise NationalGridESOAPIError( + "No actual or forecast carbon intensity in response" + ) + + emissions_per_kWh: EmissionsPerKWh = EmissionsPerKWh.from_g_per_kWh( + carbon_intensity_g_per_kWh + ) + return emissions_per_kWh.kgs_per_kWh * energy.kWh + + +def is_supported(geo: GeoMetadata) -> bool: + """Return True when the geo location is within Great Britain (GBR).""" + return geo.country_iso_code == "GBR" + + +class NationalGridESOAPIError(Exception): + pass diff --git a/tests/test_national_grid_eso_api.py b/tests/test_national_grid_eso_api.py new file mode 100644 index 000000000..752b41f61 --- /dev/null +++ b/tests/test_national_grid_eso_api.py @@ -0,0 +1,131 @@ +import unittest + +import pytest +import responses + +from codecarbon.core import national_grid_eso_api +from codecarbon.core.national_grid_eso_api import NationalGridESOAPIError +from codecarbon.core.units import Energy +from codecarbon.external.geography import GeoMetadata + + +class TestNationalGridESOAPI(unittest.TestCase): + def setUp(self) -> None: + self._energy = Energy.from_energy(kWh=10) + self._geo = GeoMetadata( + country_iso_code="GBR", + country_name="United Kingdom", + region=None, + country_2letter_iso_code="GB", + ) + + # ------------------------------------------------------------------ + # is_supported + # ------------------------------------------------------------------ + + def test_is_supported_gbr(self): + assert national_grid_eso_api.is_supported(self._geo) is True + + def test_is_supported_other_country(self): + geo = GeoMetadata( + country_iso_code="FRA", + country_name="France", + region=None, + country_2letter_iso_code="FR", + ) + assert national_grid_eso_api.is_supported(geo) is False + + # ------------------------------------------------------------------ + # get_emissions – happy paths + # ------------------------------------------------------------------ + + @responses.activate + def test_get_emissions_uses_actual(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"data": [{"intensity": {"forecast": 266, "actual": 263, "index": "moderate"}}]}, + status=200, + ) + result = national_grid_eso_api.get_emissions(self._energy, self._geo) + # 263 g/kWh * 0.001 kg/g * 10 kWh = 2.63 kg + assert round(result, 5) == 2.63 + + @responses.activate + def test_get_emissions_uses_actual_zero(self): + # actual=0 is a valid value and must not fall through to forecast + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"data": [{"intensity": {"forecast": 266, "actual": 0, "index": "very low"}}]}, + status=200, + ) + result = national_grid_eso_api.get_emissions(self._energy, self._geo) + # 0 g/kWh * 10 kWh = 0.0 kg + assert result == 0.0 + + @responses.activate + def test_get_emissions_falls_back_to_forecast_when_actual_is_none(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"data": [{"intensity": {"forecast": 266, "actual": None, "index": "moderate"}}]}, + status=200, + ) + result = national_grid_eso_api.get_emissions(self._energy, self._geo) + # 266 g/kWh * 0.001 * 10 kWh = 2.66 kg + assert round(result, 5) == 2.66 + + @responses.activate + def test_get_emissions_falls_back_to_forecast_when_actual_absent(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"data": [{"intensity": {"forecast": 200, "index": "low"}}]}, + status=200, + ) + result = national_grid_eso_api.get_emissions(self._energy, self._geo) + assert round(result, 5) == 2.0 + + # ------------------------------------------------------------------ + # get_emissions – error paths + # ------------------------------------------------------------------ + + @responses.activate + def test_get_emissions_raises_on_http_error(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"error": "Service unavailable"}, + status=503, + ) + with self.assertRaises(NationalGridESOAPIError): + national_grid_eso_api.get_emissions(self._energy, self._geo) + + @responses.activate + def test_get_emissions_raises_when_both_actual_and_forecast_are_none(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"data": [{"intensity": {"forecast": None, "actual": None, "index": "unknown"}}]}, + status=200, + ) + with self.assertRaises(NationalGridESOAPIError): + national_grid_eso_api.get_emissions(self._energy, self._geo) + + @responses.activate + def test_get_emissions_raises_on_malformed_response(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={"unexpected": "shape"}, + status=200, + ) + with self.assertRaises(NationalGridESOAPIError): + national_grid_eso_api.get_emissions(self._energy, self._geo) + + @pytest.mark.integ_test + @unittest.skip("Skip real API call in regular test runs") + def test_get_emissions_real_api(self): + result = national_grid_eso_api.get_emissions(self._energy, self._geo) + assert result > 0