Merge pull request #1039 from Hinetziedacted/add-nordic-emission-factors

Add Nordic region emission factors and update emissions logic

Author: benoit-cty

CONTRIBUTING.md

@@ -258,7 +258,7 @@ flake8...................................................................Passed
 
 If any of the linters/formatters fail, check the difference with `git diff`, add the differences if there is no behavior changes (isort and black might have change some coding style or import order, this is expected it is their job) with `git add` and finally try to commit again `git commit ...`.
 
-You can also run `pre-commit` with `uv run pre-commit run --all-file` to check all file.
+You can also run `pre-commit` with `uv run pre-commit run --all-files` if you have some changes staged but you are not ready yet to commit.
 
 
 <!-- TOC --><a name="dependencies-management"></a>

codecarbon/core/emissions.py

@@ -156,7 +156,7 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
                 )
 
         compute_with_regional_data: bool = (geo.region is not None) and (
-            geo.country_iso_code.upper() in ["USA", "CAN"]
+            geo.country_iso_code.upper() in ["USA", "CAN", "SWE", "NOR", "FIN"]
         )
 
         if compute_with_regional_data:
@@ -170,16 +170,72 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
                 )
         return self.get_country_emissions(energy, geo)
 
+    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:
+            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()
+            region_data = nordic_data["data"].get(region_upper)
+            if region_data:
+                emission_factor_g = region_data["emission_factor"]
+                emission_factor_kg = emission_factor_g / 1000
+                emissions = emission_factor_kg * energy.kWh
+                logger.debug(
+                    f"Nordic region {geo.region}: Retrieved emissions using static factor "
+                    + f"{emission_factor_g} gCO2eq/kWh: {emissions * 1000} g CO2eq"
+                )
+                return emissions
+        except Exception as e:
+            logger.warning(
+                f"Error loading Nordic emissions data for {geo.region}: {e}. "
+                + "Falling back to default emission calculation."
+            )
+        return None
+
     def get_region_emissions(self, energy: Energy, geo: GeoMetadata) -> float:
         """
         Computes emissions for a region on private infra.
         Given an quantity of power consumed, use regional data
          on emissions per unit power consumed or the mix of energy sources.
         https://github.com/responsibleproblemsolving/energy-usage#calculating-co2-emissions
+
+        get_private_infra_emissions
+        ├─ Electricity Maps API (si token)
+        ├─ get_region_emissions (USA/CAN/SWE/NOR/FIN)
+        │   └─ _try_get_nordic_region_emissions (pour SWE/NOR/FIN)
+        │   └─ country_emissions_data (pour USA)
+        │   └─ country_energy_mix_data (pour CAN)
+        └─ get_country_emissions (fallback)
+
         :param energy: Mean power consumption of the process (kWh)
         :param geo: Country and region metadata.
         :return: CO2 emissions in kg
         """
+        # Handle Nordic regions (Sweden, Norway, Finland electricity bidding zones)
+        nordic_emissions = self._try_get_nordic_region_emissions(energy, geo)
+        if nordic_emissions is not None:
+            return nordic_emissions
+
+        # Handle USA and Canada regional data
         try:
             country_emissions_data = self._data_source.get_country_emissions_data(
                 geo.country_iso_code.lower()

codecarbon/data/private_infra/nordic_emissions.json

@@ -0,0 +1,69 @@
+{
+  "data": {
+    "SE1": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Sweden Bidding Zone 1 (Northern Sweden)",
+      "year": 2024
+    },
+    "SE2": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Sweden Bidding Zone 2 (Central Sweden)",
+      "year": 2024
+    },
+    "SE3": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Sweden Bidding Zone 3 (Southern Sweden)",
+      "year": 2024
+    },
+    "SE4": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Sweden Bidding Zone 4 (Stockholm region)",
+      "year": 2024
+    },
+    "NO1": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Norway Bidding Zone 1 (Oslo)",
+      "year": 2024
+    },
+    "NO2": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Norway Bidding Zone 2 (Southern Norway)",
+      "year": 2024
+    },
+    "NO3": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Norway Bidding Zone 3 (Central Norway)",
+      "year": 2024
+    },
+    "NO4": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Norway Bidding Zone 4 (Northern Norway)",
+      "year": 2024
+    },
+    "NO5": {
+      "emission_factor": 18.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Norway Bidding Zone 5 (Western Norway)",
+      "year": 2024
+    },
+    "FI": {
+      "emission_factor": 72.0,
+      "unit": "gCO2eq/kWh",
+      "description": "Finland",
+      "year": 2025
+    }
+  },
+  "metadata": {
+    "source": "Based on historical averages from ENTSO-E data",
+    "last_updated": "2026-01-24",
+    "notes": "Static emission factors for Nordic regions. Sweden and Norway have very low carbon intensity due to high renewable energy (primarily hydro and nuclear). Finland has higher emissions due to greater fossil fuel dependency."
+  }
+}

codecarbon/input.py

@@ -53,6 +53,11 @@ def _load_static_data() -> None:
     path = _get_resource_path("data/hardware/cpu_power.csv")
     _CACHE["cpu_power"] = pd.read_csv(path)
 
+    # Nordic country energy mix - used for emissions calculations
+    path = _get_resource_path("data/private_infra/nordic_emissions.json")
+    with open(path) as f:
+        _CACHE["nordic_country_energy_mix"] = json.load(f)
+
 
 # Load static data at module import
 _load_static_data()
@@ -182,6 +187,13 @@ def get_cpu_power_data(self) -> pd.DataFrame:
         """
         return _CACHE["cpu_power"]
 
+    def get_nordic_country_energy_mix_data(self) -> Dict:
+        """
+        Returns Nordic Country Energy Mix Data.
+        Data is cached on first access per country.
+        """
+        return _CACHE["nordic_country_energy_mix"]
+
 
 class DataSourceException(Exception):
     pass

tests/test_emissions.py

@@ -1,4 +1,5 @@
 import unittest
+from unittest.mock import patch
 
 from codecarbon.core.emissions import Emissions
 from codecarbon.core.units import Energy
@@ -172,3 +173,105 @@ 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_NORDIC_REGION(self):
+        # WHEN
+        # Test Nordic region (Sweden SE2)
+
+        emissions = self._emissions.get_private_infra_emissions(
+            Energy.from_energy(kWh=1.0),
+            GeoMetadata(country_iso_code="SWE", country_name="Sweden", region="SE2"),
+        )
+
+        # THEN
+        # Nordic regions use static emission factors from the JSON file
+        # SE2 has an emission factor specified in nordic_country_energy_mix.json
+        assert isinstance(emissions, float)
+        self.assertAlmostEqual(emissions, 0.018, places=6)
+
+    def test_get_emissions_PRIVATE_INFRA_NORDIC_FINLAND(self):
+        # WHEN
+        # Test Nordic region (Finland)
+
+        emissions = self._emissions.get_private_infra_emissions(
+            Energy.from_energy(kWh=2.5),
+            GeoMetadata(country_iso_code="FIN", country_name="Finland", region="FI"),
+        )
+
+        # THEN
+        # Finland (FI) should use Nordic static emission factors
+        assert isinstance(emissions, float)
+        expected_emissions = 0.072 * 2.5
+        self.assertAlmostEqual(emissions, expected_emissions, places=6)
+
+    def test_get_emissions_PRIVATE_INFRA_NORDIC_REGION_uses_static_factor_without_token(
+        self,
+    ):
+        # GIVEN
+        energy = Energy.from_energy(kWh=1.0)
+        geo = GeoMetadata(country_iso_code="SWE", country_name="Sweden", region="SE2")
+
+        # WHEN
+        emissions = self._emissions.get_private_infra_emissions(energy, geo)
+
+        # THEN
+        expected_country = self._emissions.get_country_emissions(energy, geo)
+        nordic_data = self._data_source.get_nordic_country_energy_mix_data()
+        emission_factor_g = nordic_data["data"]["SE2"]["emission_factor"]
+        expected_nordic = (emission_factor_g / 1000) * energy.kWh
+        self.assertAlmostEqual(emissions, expected_nordic, places=6)
+        self.assertNotAlmostEqual(emissions, expected_country, places=4)
+
+    def test_try_get_nordic_region_emissions_returns_none_without_region(self):
+        # GIVEN
+        energy = Energy.from_energy(kWh=1.0)
+        geo = GeoMetadata(country_iso_code="SWE", country_name="Sweden", region=None)
+
+        # WHEN
+        emissions = self._emissions._try_get_nordic_region_emissions(energy, geo)
+
+        # THEN
+        self.assertIsNone(emissions)
+
+    def test_try_get_nordic_region_emissions_returns_none_for_non_nordic_region(self):
+        # GIVEN
+        energy = Energy.from_energy(kWh=1.0)
+        geo = GeoMetadata(country_iso_code="SWE", country_name="Sweden", region="XYZ")
+
+        # WHEN
+        emissions = self._emissions._try_get_nordic_region_emissions(energy, geo)
+
+        # THEN
+        self.assertIsNone(emissions)
+
+    def test_try_get_nordic_region_emissions_returns_none_if_region_data_missing(self):
+        # GIVEN
+        energy = Energy.from_energy(kWh=1.0)
+        geo = GeoMetadata(country_iso_code="SWE", country_name="Sweden", region="SE2")
+
+        # WHEN
+        with patch.object(
+            self._data_source,
+            "get_nordic_country_energy_mix_data",
+            return_value={"data": {}},
+        ):
+            emissions = self._emissions._try_get_nordic_region_emissions(energy, geo)
+
+        # THEN
+        self.assertIsNone(emissions)
+
+    def test_try_get_nordic_region_emissions_returns_none_on_data_loading_error(self):
+        # GIVEN
+        energy = Energy.from_energy(kWh=1.0)
+        geo = GeoMetadata(country_iso_code="SWE", country_name="Sweden", region="SE2")
+
+        # WHEN
+        with patch.object(
+            self._data_source,
+            "get_nordic_country_energy_mix_data",
+            side_effect=Exception("boom"),
+        ):
+            emissions = self._emissions._try_get_nordic_region_emissions(energy, geo)
+
+        # THEN
+        self.assertIsNone(emissions)