Merge pull request #1224 from kbarkhad/issue_swedish_region

Fix: Look up failure for nordic regions using online emissions tracker

Author: benoit-cty

codecarbon/core/emissions.py

@@ -16,6 +16,12 @@
 from codecarbon.external.logger import logger
 from codecarbon.input import DataSource, DataSourceException
 
+_NORDIC_REGIONS_BY_COUNTRY = {
+    "SWE": {"SE1", "SE2", "SE3", "SE4"},
+    "NOR": {"NO1", "NO2", "NO3", "NO4", "NO5"},
+    "FIN": {"FI"},
+}
+
 
 class Emissions:
     def __init__(
@@ -169,8 +175,11 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
                     + " >>> Using CodeCarbon's data."
                 )
 
+        country_iso_code = (
+            geo.country_iso_code.upper() if geo.country_iso_code is not None else None
+        )
         compute_with_regional_data: bool = (geo.region is not None) and (
-            geo.country_iso_code.upper() in ["USA", "CAN", "SWE", "NOR", "FIN"]
+            country_iso_code in ["USA", "CAN"] or self._is_supported_nordic_region(geo)
         )
 
         if compute_with_regional_data:
@@ -187,24 +196,10 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
     def _try_get_nordic_region_emissions(
         self, energy: Energy, geo: GeoMetadata
     ) -> Optional[float]:
-        nordic_regions = {
-            "SE1",
-            "SE2",
-            "SE3",
-            "SE4",
-            "NO1",
-            "NO2",
-            "NO3",
-            "NO4",
-            "NO5",
-            "FI",
-        }
-        if geo.region is None:
+        if not self._is_supported_nordic_region(geo):
             return None
 
         region_upper = geo.region.upper()
-        if region_upper not in nordic_regions:
-            return None
 
         try:
             nordic_data = self._data_source.get_nordic_country_energy_mix_data()
@@ -225,6 +220,16 @@ def _try_get_nordic_region_emissions(
             )
         return None
 
+    @staticmethod
+    def _is_supported_nordic_region(geo: GeoMetadata) -> bool:
+        if geo.country_iso_code is None or geo.region is None:
+            return False
+
+        country_regions = _NORDIC_REGIONS_BY_COUNTRY.get(
+            geo.country_iso_code.upper(), set()
+        )
+        return geo.region.upper() in country_regions
+
     def get_region_emissions(self, energy: Energy, geo: GeoMetadata) -> float:
         """
         Computes emissions for a region on private infra.
@@ -248,6 +253,11 @@ def get_region_emissions(self, energy: Energy, geo: GeoMetadata) -> float:
         nordic_emissions = self._try_get_nordic_region_emissions(energy, geo)
         if nordic_emissions is not None:
             return nordic_emissions
+        if (
+            geo.country_iso_code is not None
+            and geo.country_iso_code.upper() in _NORDIC_REGIONS_BY_COUNTRY
+        ):
+            return self.get_country_emissions(energy, geo)
 
         # Handle USA and Canada regional data
         try:

tests/test_emissions.py

@@ -174,6 +174,15 @@ def test_get_emissions_PRIVATE_INFRA_unknown_country(self):
         assert isinstance(emissions, float)
         self.assertAlmostEqual(emissions, 0.475, places=2)
 
+    def test_get_emissions_PRIVATE_INFRA_without_country_metadata(self):
+        emissions = self._emissions.get_private_infra_emissions(
+            Energy.from_energy(kWh=1),
+            GeoMetadata(country_iso_code=None, country_name=None, region=None),
+        )
+
+        assert isinstance(emissions, float)
+        self.assertAlmostEqual(emissions, 0.475, places=2)
+
     @patch("codecarbon.core.electricitymaps_api.get_emissions")
     def test_private_infra_uses_forced_intensity_when_set(self, mocked_get_emissions):
         emissions_calculator = Emissions(
@@ -270,6 +279,69 @@ def test_try_get_nordic_region_emissions_returns_none_for_non_nordic_region(self
         # THEN
         self.assertIsNone(emissions)
 
+    def test_nordic_admin_regions_fall_back_to_country_without_regional_lookup(self):
+        energy = Energy.from_energy(kWh=1.0)
+        admin_regions = [
+            ("SWE", "Sweden", "Stockholm County"),
+            ("NOR", "Norway", "Oslo"),
+            ("FIN", "Finland", "Uusimaa"),
+        ]
+
+        for country_iso_code, country_name, region in admin_regions:
+            with self.subTest(country_iso_code=country_iso_code, region=region):
+                geo = GeoMetadata(
+                    country_iso_code=country_iso_code,
+                    country_name=country_name,
+                    region=region,
+                )
+                expected_emissions = self._emissions.get_country_emissions(energy, geo)
+
+                with (
+                    patch.object(
+                        self._data_source, "get_nordic_country_energy_mix_data"
+                    ) as get_nordic_data,
+                    patch.object(
+                        self._data_source, "get_country_emissions_data"
+                    ) as get_regional_emissions_data,
+                    patch.object(
+                        self._data_source, "get_country_energy_mix_data"
+                    ) as get_regional_energy_mix_data,
+                    patch("codecarbon.core.emissions.logger.error") as log_error,
+                    patch("codecarbon.core.emissions.logger.warning") as log_warning,
+                ):
+                    emissions = self._emissions.get_private_infra_emissions(energy, geo)
+
+                self.assertAlmostEqual(emissions, expected_emissions, places=6)
+                get_nordic_data.assert_not_called()
+                get_regional_emissions_data.assert_not_called()
+                get_regional_energy_mix_data.assert_not_called()
+                log_error.assert_not_called()
+                log_warning.assert_not_called()
+
+    def test_nordic_region_missing_static_data_falls_back_to_country(self):
+        energy = Energy.from_energy(kWh=1.0)
+        geo = GeoMetadata(country_iso_code="SWE", country_name="Sweden", region="SE2")
+        expected_emissions = self._emissions.get_country_emissions(energy, geo)
+
+        with (
+            patch.object(
+                self._data_source,
+                "get_nordic_country_energy_mix_data",
+                return_value={"data": {}},
+            ),
+            patch.object(
+                self._data_source, "get_country_emissions_data"
+            ) as get_regional_emissions_data,
+            patch.object(
+                self._data_source, "get_country_energy_mix_data"
+            ) as get_regional_energy_mix_data,
+        ):
+            emissions = self._emissions.get_private_infra_emissions(energy, geo)
+
+        self.assertAlmostEqual(emissions, expected_emissions, places=6)
+        get_regional_emissions_data.assert_not_called()
+        get_regional_energy_mix_data.assert_not_called()
+
     def test_try_get_nordic_region_emissions_returns_none_if_region_data_missing(self):
         # GIVEN
         energy = Energy.from_energy(kWh=1.0)