Allow custom carbon intensity configuration

codecarbon/core/emissions.py

@@ -25,6 +25,7 @@ def __init__(
         co2_signal_api_token: Optional[
             str
         ] = None,  # Deprecated, for backward compatibility
+        custom_carbon_intensity_g_co2e_kwh: Optional[float] = None,
     ):
         self._data_source = data_source
 
@@ -38,6 +39,7 @@ def __init__(
                 electricitymaps_api_token = co2_signal_api_token
 
         self._electricitymaps_api_token = electricitymaps_api_token
+        self._custom_carbon_intensity_g_co2e_kwh = custom_carbon_intensity_g_co2e_kwh
 
     def get_cloud_emissions(
         self, energy: Energy, cloud: CloudMetadata, geo: GeoMetadata = None
@@ -50,6 +52,12 @@ def get_cloud_emissions(
         :return: CO2 emissions in kg
         """
 
+        if self._custom_carbon_intensity_g_co2e_kwh is not None:
+            logger.info(
+                f"Using custom carbon intensity for cloud emissions: {self._custom_carbon_intensity_g_co2e_kwh} gCO2e/kWh"
+            )
+            return energy.kWh * (self._custom_carbon_intensity_g_co2e_kwh / 1000.0)
+
         df: pd.DataFrame = self._data_source.get_cloud_emissions_data()
         try:
             emissions_per_kWh: EmissionsPerKWh = EmissionsPerKWh.from_g_per_kWh(
@@ -138,6 +146,12 @@ def get_private_infra_emissions(self, energy: Energy, geo: GeoMetadata) -> float
         :param geo: Country and region metadata
         :return: CO2 emissions in kg
         """
+        if self._custom_carbon_intensity_g_co2e_kwh is not None:
+            logger.info(
+                f"Using custom carbon intensity for private infrastructure emissions: {self._custom_carbon_intensity_g_co2e_kwh} gCO2e/kWh"
+            )
+            return energy.kWh * (self._custom_carbon_intensity_g_co2e_kwh / 1000.0)
+
         if self._electricitymaps_api_token:
             try:
                 emissions = electricitymaps_api.get_emissions(

codecarbon/emissions_tracker.py

@@ -276,6 +276,33 @@ def __init__(
 
         # logger.info("base tracker init")
         self._external_conf = get_hierarchical_config()
+        custom_intensity_str = self._external_conf.get(
+            "custom_carbon_intensity_g_co2e_kwh"
+        )
+        parsed_intensity = None
+        if custom_intensity_str is not None:
+            custom_intensity_str_stripped = custom_intensity_str.strip()
+            if custom_intensity_str_stripped == "":
+                logger.warning(
+                    f"CODECARBON : Invalid value for custom_carbon_intensity_g_co2e_kwh: '{custom_intensity_str}'. "
+                    "It cannot be empty or whitespace. Using default calculation methods."
+                )
+            else:
+                try:
+                    value = float(custom_intensity_str_stripped)
+                    if value > 0:
+                        parsed_intensity = value
+                    else:
+                        logger.warning(
+                            f"CODECARBON : Invalid value for custom_carbon_intensity_g_co2e_kwh: '{custom_intensity_str_stripped}'. "
+                            "It must be a positive number. Using default calculation methods."
+                        )
+                except ValueError:
+                    logger.warning(
+                        f"CODECARBON : Invalid value for custom_carbon_intensity_g_co2e_kwh: '{custom_intensity_str_stripped}'. "
+                        "It must be a numeric value. Using default calculation methods."
+                    )
+        self.custom_carbon_intensity_g_co2e_kwh = parsed_intensity
         self._set_from_conf(allow_multiple_runs, "allow_multiple_runs", True, bool)
         if self._allow_multiple_runs:
             logger.warning(
@@ -353,6 +380,11 @@ def __init__(
             experiment_id, "experiment_id", "5b0fa12a-3dd7-45bb-9766-cc326314d9f1"
         )
 
+        if self.custom_carbon_intensity_g_co2e_kwh is not None:
+            logger.info(
+                f"CODECARBON : Using custom carbon intensity: {self.custom_carbon_intensity_g_co2e_kwh} gCO2e/kWh."
+            )
+
         assert self._tracking_mode in ["machine", "process"]
         set_logger_level(self._log_level)
         set_logger_format(self._logger_preamble)
@@ -446,7 +478,9 @@ def __init__(
             self._conf["provider"] = cloud.provider
 
         self._emissions: Emissions = Emissions(
-            self._data_source, self._electricitymaps_api_token
+            self._data_source,
+            self._electricitymaps_api_token,
+            custom_carbon_intensity_g_co2e_kwh=self.custom_carbon_intensity_g_co2e_kwh,
         )
         self._init_output_methods(api_key=self._api_key)
 

tests/test_config.py

@@ -199,6 +199,7 @@ def test_full_hierarchy(self):
             force_ram_power=50.5
             output_dir=ERROR:not overwritten
             save_to_file=ERROR:not overwritten
+            custom_carbon_intensity_g_co2e_kwh=123.4
             """
         )
         local_conf = dedent(
@@ -225,6 +226,7 @@ def test_full_hierarchy(self):
             self.assertEqual(tracker._emissions_endpoint, "http://testhost:2000")
             self.assertEqual(tracker._gpu_ids, ["0", "1"])
             self.assertEqual(tracker._electricitymaps_api_token, "signal-token")
+            self.assertEqual(tracker.custom_carbon_intensity_g_co2e_kwh, 123.4)
             self.assertEqual(tracker._project_name, "test-project")
             self.assertTrue(tracker._save_to_file)
 

tests/test_emissions.py

@@ -174,6 +174,32 @@ def test_get_emissions_PRIVATE_INFRA_unknown_country(self):
         assert isinstance(emissions, float)
         self.assertAlmostEqual(emissions, 0.475, places=2)
 
+    @patch("codecarbon.core.electricitymaps_api.get_emissions")
+    def test_private_infra_uses_custom_intensity_when_set(self, mocked_get_emissions):
+        emissions_calculator = Emissions(
+            self._data_source, custom_carbon_intensity_g_co2e_kwh=50.0
+        )
+
+        emissions = emissions_calculator.get_private_infra_emissions(
+            Energy.from_energy(kWh=2),
+            GeoMetadata(country_iso_code="CAN", country_name="Canada"),
+        )
+
+        self.assertAlmostEqual(emissions, 0.1, places=6)
+        mocked_get_emissions.assert_not_called()
+
+    def test_cloud_uses_custom_intensity_when_set(self):
+        emissions_calculator = Emissions(
+            self._data_source, custom_carbon_intensity_g_co2e_kwh=100.0
+        )
+
+        emissions = emissions_calculator.get_cloud_emissions(
+            Energy.from_energy(kWh=2),
+            CloudMetadata(provider="aws", region="us-east-1"),
+        )
+
+        self.assertAlmostEqual(emissions, 0.2, places=6)
+
     def test_get_emissions_PRIVATE_INFRA_NORDIC_REGION(self):
         # WHEN
         # Test Nordic region (Sweden SE2)