From b4a7324bbf661af254295e6c948ec126409f3394 Mon Sep 17 00:00:00 2001 From: Dell Date: Mon, 22 Jun 2026 14:24:20 +0100 Subject: [PATCH 1/2] feat(emissions): add UK National Grid ESO live carbon intensity source Adds codecarbon/core/national_grid_eso_api.py, which calls the public National Grid ESO Carbon Intensity API (https://api.carbonintensity.org.uk/intensity) to retrieve half-hourly live carbon intensity for Great Britain. The actual reading is preferred; forecast is used as a fallback when actual has not yet been published. A genuine zero-intensity actual value is handled correctly via an explicit `is not None` check rather than a truthiness test. Integrates the new source into the get_private_infra_emissions fallback chain in emissions.py immediately after ElectricityMaps: when geo.country_iso_code == GBR and no ElectricityMaps token is configured, the live API is tried first and the existing static country-average remains the final fallback on failure. No behaviour changes for any other country. Closes #947 --- codecarbon/core/emissions.py | 17 ++- codecarbon/core/national_grid_eso_api.py | 56 ++++++++++ tests/test_national_grid_eso_api.py | 131 +++++++++++++++++++++++ 3 files changed, 203 insertions(+), 1 deletion(-) create mode 100644 codecarbon/core/national_grid_eso_api.py create mode 100644 tests/test_national_grid_eso_api.py 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 From d78c98176c9b175aa1bad76c80893aaa79c32401 Mon Sep 17 00:00:00 2001 From: Sufiyan Ul Rehman Date: Fri, 9 Oct 2026 11:18:31 +0100 Subject: [PATCH 2/2] Make National Grid ESO API opt-in, add short timeout and per-run cache Address review feedback on the live UK carbon intensity integration: - Gate the live call behind a new use_national_grid_eso_api option (default False), configurable via constructor, decorator, .codecarbon.config or environment variable, mirroring how the Electricity Maps call is only enabled when a token is provided. - Reduce the request timeout from 30s to 4s so an unreachable API cannot stall a measurement. - Cache the fetched intensity (5 minute TTL) so repeated measurements in one run do not re-hit the API. - Document the OGL v3.0 attribution and that the figure is national (Great Britain-wide), not regional. - Add tests for caching and for the opt-in gate in Emissions. --- codecarbon/core/emissions.py | 7 +- codecarbon/core/national_grid_eso_api.py | 78 ++++++++++++++--- codecarbon/emissions_tracker.py | 21 +++++ docs/how-to/configuration.md | 33 +++++++ tests/test_emissions.py | 73 ++++++++++++++++ tests/test_national_grid_eso_api.py | 105 ++++++++++++++++++++++- 6 files changed, 298 insertions(+), 19 deletions(-) diff --git a/codecarbon/core/emissions.py b/codecarbon/core/emissions.py index 76e669790..696643dc3 100644 --- a/codecarbon/core/emissions.py +++ b/codecarbon/core/emissions.py @@ -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,7 +181,7 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float + " >>> Using CodeCarbon's data." ) - if national_grid_eso_api.is_supported(geo): + 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( diff --git a/codecarbon/core/national_grid_eso_api.py b/codecarbon/core/national_grid_eso_api.py index 539a08351..7d6e0909e 100644 --- a/codecarbon/core/national_grid_eso_api.py +++ b/codecarbon/core/national_grid_eso_api.py @@ -1,26 +1,31 @@ +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" -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. +# 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 - Args: - energy: Energy consumption in kWh. - geo: Geographic metadata; must have country_iso_code == "GBR". +# 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) - Returns: - CO2 emissions in kilograms. - Raises: - NationalGridESOAPIError: If the API request fails or returns unusable data. - """ +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( @@ -34,13 +39,58 @@ def get_emissions(energy: Energy, geo: GeoMetadata) -> float: # 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") + 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 ) diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 96ed00c91..239a5927f 100644 --- a/codecarbon/emissions_tracker.py +++ b/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: diff --git a/docs/how-to/configuration.md b/docs/how-to/configuration.md index 9f6766aa1..78245ebec 100644 --- a/docs/how-to/configuration.md +++ b/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: diff --git a/tests/test_emissions.py b/tests/test_emissions.py index 9084ef3ea..524de49b3 100644 --- a/tests/test_emissions.py +++ b/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() diff --git a/tests/test_national_grid_eso_api.py b/tests/test_national_grid_eso_api.py index 752b41f61..43c8fcd02 100644 --- a/tests/test_national_grid_eso_api.py +++ b/tests/test_national_grid_eso_api.py @@ -18,6 +18,12 @@ def setUp(self) -> None: region=None, country_2letter_iso_code="GB", ) + # Each test expects its own mocked HTTP call; the module-level cache + # would otherwise serve a stale value from a previous test. + national_grid_eso_api.clear_cache() + + def tearDown(self) -> None: + national_grid_eso_api.clear_cache() # ------------------------------------------------------------------ # is_supported @@ -44,7 +50,11 @@ 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"}}]}, + json={ + "data": [ + {"intensity": {"forecast": 266, "actual": 263, "index": "moderate"}} + ] + }, status=200, ) result = national_grid_eso_api.get_emissions(self._energy, self._geo) @@ -57,7 +67,11 @@ def test_get_emissions_uses_actual_zero(self): responses.add( responses.GET, national_grid_eso_api.URL, - json={"data": [{"intensity": {"forecast": 266, "actual": 0, "index": "very low"}}]}, + json={ + "data": [ + {"intensity": {"forecast": 266, "actual": 0, "index": "very low"}} + ] + }, status=200, ) result = national_grid_eso_api.get_emissions(self._energy, self._geo) @@ -69,7 +83,17 @@ 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"}}]}, + json={ + "data": [ + { + "intensity": { + "forecast": 266, + "actual": None, + "index": "moderate", + } + } + ] + }, status=200, ) result = national_grid_eso_api.get_emissions(self._energy, self._geo) @@ -107,7 +131,17 @@ 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"}}]}, + json={ + "data": [ + { + "intensity": { + "forecast": None, + "actual": None, + "index": "unknown", + } + } + ] + }, status=200, ) with self.assertRaises(NationalGridESOAPIError): @@ -129,3 +163,66 @@ def test_get_emissions_raises_on_malformed_response(self): def test_get_emissions_real_api(self): result = national_grid_eso_api.get_emissions(self._energy, self._geo) assert result > 0 + + # ------------------------------------------------------------------ + # caching + # ------------------------------------------------------------------ + + @responses.activate + def test_get_emissions_caches_within_ttl(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={ + "data": [ + {"intensity": {"forecast": 266, "actual": 263, "index": "moderate"}} + ] + }, + status=200, + ) + national_grid_eso_api.get_emissions(self._energy, self._geo) + national_grid_eso_api.get_emissions(self._energy, self._geo) + national_grid_eso_api.get_emissions(self._energy, self._geo) + # Only the first call should have hit the network; the rest are + # served from the per-run cache. + assert len(responses.calls) == 1 + + @responses.activate + def test_get_emissions_refetches_after_cache_cleared(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={ + "data": [ + {"intensity": {"forecast": 266, "actual": 263, "index": "moderate"}} + ] + }, + status=200, + ) + national_grid_eso_api.get_emissions(self._energy, self._geo) + national_grid_eso_api.clear_cache() + national_grid_eso_api.get_emissions(self._energy, self._geo) + assert len(responses.calls) == 2 + + @responses.activate + def test_get_emissions_refetches_after_ttl_expires(self): + responses.add( + responses.GET, + national_grid_eso_api.URL, + json={ + "data": [ + {"intensity": {"forecast": 266, "actual": 263, "index": "moderate"}} + ] + }, + status=200, + ) + national_grid_eso_api.get_emissions(self._energy, self._geo) + # Simulate TTL expiry without sleeping in the test. + national_grid_eso_api._cache = ( + national_grid_eso_api._cache[0], + national_grid_eso_api._cache[1] + - national_grid_eso_api._CACHE_TTL_SECONDS + - 1, + ) + national_grid_eso_api.get_emissions(self._energy, self._geo) + assert len(responses.calls) == 2