feat(emissions): add UK National Grid ESO live carbon intensity source

codecarbon/core/emissions.py

@@ -8,7 +8,7 @@
 
 from typing import TYPE_CHECKING, Dict, Optional
 
-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
@@ -33,6 +33,7 @@ def __init__(
             str
         ] = None,  # Deprecated, for backward compatibility
         force_carbon_intensity_g_co2e_kwh: Optional[float] = None,
+        use_national_grid_eso_api: bool = False,
     ):
         self._data_source = data_source
 
@@ -47,6 +48,10 @@ def __init__(
 
         self._electricitymaps_api_token = electricitymaps_api_token
         self._force_carbon_intensity_g_co2e_kwh = force_carbon_intensity_g_co2e_kwh
+        # Opt-in only: this makes a live network call on every GBR run if left
+        # on. Defaults to False, mirroring how electricitymaps_api_token being
+        # unset keeps the Electricity Maps call off.
+        self._use_national_grid_eso_api = use_national_grid_eso_api
 
     def get_cloud_emissions(
         self, energy: Energy, cloud: CloudMetadata, geo: GeoMetadata = None
@@ -176,6 +181,21 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
                     + " >>> Using CodeCarbon's data."
                 )
 
+        if self._use_national_grid_eso_api and 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
         )

codecarbon/core/national_grid_eso_api.py

@@ -0,0 +1,106 @@
+import time
+
+import requests
+
+from codecarbon.core.units import EmissionsPerKWh, Energy
+from codecarbon.external.geography import GeoMetadata
+
+# National Grid ESO Carbon Intensity API for Great Britain.
+# Data source: https://carbonintensity.org.uk/ — published by National Grid ESO
+# under the Open Government Licence v3.0. The API reports a single,
+# national-level (Great-Britain-wide) carbon intensity figure; it does not
+# provide regional or sub-national granularity.
+URL: str = "https://api.carbonintensity.org.uk/intensity"
+
+# Kept short deliberately: this call sits in the emissions fallback chain and
+# must not stall a measurement if the API is slow or unreachable.
+NATIONAL_GRID_ESO_API_TIMEOUT: int = 4
+
+# Per-process cache of the last fetched carbon intensity, so a single
+# long-running CodeCarbon session does not re-hit the API on every
+# measurement. Keyed by nothing (single value) since this API only ever
+# describes Great Britain as a whole; re-fetched once the entry expires.
+_CACHE_TTL_SECONDS: int = 300
+_cache: tuple[float, float] | None = None  # (carbon_intensity_g_per_kWh, fetched_at)
+
+
+def _fetch_carbon_intensity_g_per_kWh() -> float:
+    """Call the live API and return the current carbon intensity in gCO2/kWh."""
+    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"
+        )
+
+    return carbon_intensity_g_per_kWh
+
+
+def _get_cached_carbon_intensity_g_per_kWh() -> float:
+    """Return the cached intensity if still fresh, otherwise fetch and cache it."""
+    global _cache
+    now = time.monotonic()
+    if _cache is not None:
+        cached_value, fetched_at = _cache
+        if (now - fetched_at) < _CACHE_TTL_SECONDS:
+            return cached_value
+
+    value = _fetch_carbon_intensity_g_per_kWh()
+    _cache = (value, now)
+    return value
+
+
+def clear_cache() -> None:
+    """Reset the cached carbon intensity value (mainly useful for tests)."""
+    global _cache
+    _cache = None
+
+
+def get_emissions(energy: Energy, geo: GeoMetadata) -> float:
+    """
+    Calculate the CO2 emissions using the UK National Grid ESO Carbon Intensity API.
+
+    This is an opt-in data source: callers must check ``is_supported`` and the
+    caller-side opt-in flag before calling this. The value returned describes
+    Great Britain as a whole; the API does not provide regional figures.
+
+    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.
+    """
+    carbon_intensity_g_per_kWh = _get_cached_carbon_intensity_g_per_kWh()
+
+    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

codecarbon/emissions_tracker.py

@@ -373,6 +373,7 @@ def _ensure_emissions_engine(self) -> None:
             self._data_source,
             self._electricitymaps_api_token,
             force_carbon_intensity_g_co2e_kwh=self.force_carbon_intensity_g_co2e_kwh,
+            use_national_grid_eso_api=self._use_national_grid_eso_api,
         )
 
     def _ensure_geo_metadata(self) -> None:
@@ -424,6 +425,7 @@ def __init__(
         allow_multiple_runs: Optional[bool] = _sentinel,
         rapl_include_dram: Optional[bool] = _sentinel,
         rapl_prefer_psys: Optional[bool] = _sentinel,
+        use_national_grid_eso_api: Optional[bool] = _sentinel,
     ):
         """
         :param project_name: Project name for current experiment run, default name
@@ -519,6 +521,14 @@ def __init__(
                                  (CPU + chipset + PCIe). When False, uses package domains which
                                  are more reliable. Note: psys can report higher values than
                                  CPU TDP and may be unreliable on older systems.
+        :param use_national_grid_eso_api: Opt-in: when True and the detected location is
+                                 Great Britain, fetch live carbon intensity from the UK
+                                 National Grid ESO Carbon Intensity API
+                                 (https://carbonintensity.org.uk/, Open Government Licence
+                                 v3.0) instead of CodeCarbon's static country-average data.
+                                 This makes a network call (short timeout, cached for the
+                                 duration of the run) and reports a Great-Britain-wide
+                                 figure, not a regional one. Defaults to False.
         """
 
         # logger.info("base tracker init")
@@ -583,6 +593,9 @@ def __init__(
         self._set_from_conf(logger_preamble, "logger_preamble", "")
         self._set_from_conf(force_cpu_power, "force_cpu_power", None, float)
         self._set_from_conf(force_ram_power, "force_ram_power", None, float)
+        self._set_from_conf(
+            use_national_grid_eso_api, "use_national_grid_eso_api", False, bool
+        )
         self._set_from_conf(pue, "pue", 1.0, float)
         self._set_from_conf(wue, "wue", 0, float)
         self._set_from_conf(force_mode_cpu_load, "force_mode_cpu_load", False, bool)
@@ -1512,6 +1525,7 @@ def track_emissions(
     allow_multiple_runs: Optional[bool] = _sentinel,
     rapl_include_dram: Optional[bool] = _sentinel,
     rapl_prefer_psys: Optional[bool] = _sentinel,
+    use_national_grid_eso_api: Optional[bool] = _sentinel,
 ):
     """
     Decorator that supports both `EmissionsTracker` and `OfflineEmissionsTracker`
@@ -1594,6 +1608,11 @@ def track_emissions(
                               When True, measures CPU package + DRAM.
     :param rapl_prefer_psys: Prefer psys over package domains for RAPL on Linux
                              (default: False). When True, uses total platform power.
+    :param use_national_grid_eso_api: Opt-in: when True and the detected location
+                             is Great Britain, fetch live carbon intensity from the
+                             UK National Grid ESO Carbon Intensity API instead of
+                             CodeCarbon's static country-average data. Defaults to
+                             False.
 
     :return: The decorated function
     """
@@ -1650,6 +1669,7 @@ def wrapped_fn(*args, **kwargs):
                     allow_multiple_runs=allow_multiple_runs,
                     rapl_include_dram=rapl_include_dram,
                     rapl_prefer_psys=rapl_prefer_psys,
+                    use_national_grid_eso_api=use_national_grid_eso_api,
                 )
             else:
                 tracker = EmissionsTracker(
@@ -1686,6 +1706,7 @@ def wrapped_fn(*args, **kwargs):
                     allow_multiple_runs=allow_multiple_runs,
                     rapl_include_dram=rapl_include_dram,
                     rapl_prefer_psys=rapl_prefer_psys,
+                    use_national_grid_eso_api=use_national_grid_eso_api,
                 )
             tracker.start()
             try:

docs/how-to/configuration.md

@@ -122,6 +122,39 @@ EmissionsTracker(electricitymaps_api_token="your-token-here")
     compatibility but is deprecated and will be removed in a future version.
     Use `electricitymaps_api_token` instead.
 
+## National Grid ESO API (Great Britain)
+
+For runs detected as being in Great Britain, CodeCarbon can optionally use
+live carbon intensity data from the
+[National Grid ESO Carbon Intensity API](https://carbonintensity.org.uk/)
+instead of the static country-average data, by setting
+`use_national_grid_eso_api=True`. This is **opt-in and off by default**,
+the same way the Electricity Maps integration above is off until you provide
+a token — it is not used automatically.
+
+``` python
+EmissionsTracker(use_national_grid_eso_api=True)
+```
+
+Or in your config file:
+
+``` ini
+[codecarbon]
+use_national_grid_eso_api = true
+```
+
+**Attribution and terms of use:** data is published by National Grid ESO
+under the [Open Government Licence v3.0](https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/).
+See [carbonintensity.org.uk](https://carbonintensity.org.uk/) for full terms.
+
+**Scope:** the API reports a single **Great-Britain-wide (national)** carbon
+intensity figure. It does not provide regional or sub-national data, so this
+is not a substitute for CodeCarbon's USA/Canada/Nordic regional emissions
+support.
+
+If `electricitymaps_api_token` is also set, Electricity Maps takes priority
+and this source is only used as a fallback if that call fails.
+
 ## Tracking Mode
 
 The `tracking_mode` parameter controls how CodeCarbon measures power consumption. It accepts two values:

tests/test_emissions.py

@@ -373,3 +373,76 @@ def test_try_get_nordic_region_emissions_returns_none_on_data_loading_error(self
 
         # THEN
         self.assertIsNone(emissions)
+
+    # ------------------------------------------------------------------
+    # National Grid ESO API (opt-in)
+    # ------------------------------------------------------------------
+
+    @patch("codecarbon.core.national_grid_eso_api.get_emissions")
+    def test_national_grid_eso_not_called_by_default(self, mocked_get_emissions):
+        # GIVEN: use_national_grid_eso_api defaults to False
+        emissions_calculator = Emissions(self._data_source)
+        geo = GeoMetadata(country_iso_code="GBR", country_name="United Kingdom")
+
+        # WHEN
+        emissions_calculator.get_private_infra_emissions(Energy.from_energy(kWh=1), geo)
+
+        # THEN: the live API is never called unless explicitly opted in
+        mocked_get_emissions.assert_not_called()
+
+    @patch("codecarbon.core.national_grid_eso_api.get_emissions")
+    def test_national_grid_eso_called_when_opted_in(self, mocked_get_emissions):
+        # GIVEN
+        mocked_get_emissions.return_value = 0.5
+        emissions_calculator = Emissions(
+            self._data_source, use_national_grid_eso_api=True
+        )
+        geo = GeoMetadata(country_iso_code="GBR", country_name="United Kingdom")
+        energy = Energy.from_energy(kWh=1)
+
+        # WHEN
+        emissions = emissions_calculator.get_private_infra_emissions(energy, geo)
+
+        # THEN
+        mocked_get_emissions.assert_called_once_with(energy, geo)
+        self.assertEqual(emissions, 0.5)
+
+    @patch("codecarbon.core.national_grid_eso_api.get_emissions")
+    def test_national_grid_eso_not_called_for_other_countries_even_when_opted_in(
+        self, mocked_get_emissions
+    ):
+        # GIVEN
+        emissions_calculator = Emissions(
+            self._data_source, use_national_grid_eso_api=True
+        )
+        geo = GeoMetadata(country_iso_code="FRA", country_name="France")
+
+        # WHEN
+        emissions_calculator.get_private_infra_emissions(Energy.from_energy(kWh=1), geo)
+
+        # THEN
+        mocked_get_emissions.assert_not_called()
+
+    @patch("codecarbon.core.national_grid_eso_api.get_emissions")
+    @patch("codecarbon.core.electricitymaps_api.get_emissions")
+    def test_national_grid_eso_not_called_when_electricitymaps_succeeds(
+        self, mocked_em_get_emissions, mocked_ngeso_get_emissions
+    ):
+        # GIVEN: both Electricity Maps and National Grid ESO are opted in,
+        # but Electricity Maps takes priority and succeeds
+        mocked_em_get_emissions.return_value = 0.3
+        emissions_calculator = Emissions(
+            self._data_source,
+            electricitymaps_api_token="token",
+            use_national_grid_eso_api=True,
+        )
+        geo = GeoMetadata(country_iso_code="GBR", country_name="United Kingdom")
+
+        # WHEN
+        emissions = emissions_calculator.get_private_infra_emissions(
+            Energy.from_energy(kWh=1), geo
+        )
+
+        # THEN
+        self.assertEqual(emissions, 0.3)
+        mocked_ngeso_get_emissions.assert_not_called()