Updated Emissions, EmissionsTracker, OfflineEmissionsTracker, and @track_emissions.

Author: benoit-cty

codecarbon/core/emissions.py

@@ -25,7 +25,7 @@ def __init__(
         co2_signal_api_token: Optional[
             str
         ] = None,  # Deprecated, for backward compatibility
-        custom_carbon_intensity_g_co2e_kwh: Optional[float] = None,
+        force_carbon_intensity_g_co2e_kwh: Optional[float] = None,
     ):
         self._data_source = data_source
 
@@ -39,7 +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
+        self._force_carbon_intensity_g_co2e_kwh = force_carbon_intensity_g_co2e_kwh
 
     def get_cloud_emissions(
         self, energy: Energy, cloud: CloudMetadata, geo: GeoMetadata = None
@@ -52,11 +52,11 @@ def get_cloud_emissions(
         :return: CO2 emissions in kg
         """
 
-        if self._custom_carbon_intensity_g_co2e_kwh is not None:
+        if self._force_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"
+                f"Using forced carbon intensity for cloud emissions: {self._force_carbon_intensity_g_co2e_kwh} gCO2e/kWh"
             )
-            return energy.kWh * (self._custom_carbon_intensity_g_co2e_kwh / 1000.0)
+            return energy.kWh * (self._force_carbon_intensity_g_co2e_kwh / 1000.0)
 
         df: pd.DataFrame = self._data_source.get_cloud_emissions_data()
         try:
@@ -146,11 +146,11 @@ 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:
+        if self._force_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"
+                f"Using forced carbon intensity for private infrastructure emissions: {self._force_carbon_intensity_g_co2e_kwh} gCO2e/kWh"
             )
-            return energy.kWh * (self._custom_carbon_intensity_g_co2e_kwh / 1000.0)
+            return energy.kWh * (self._force_carbon_intensity_g_co2e_kwh / 1000.0)
 
         if self._electricitymaps_api_token:
             try:

codecarbon/emissions_tracker.py

@@ -195,6 +195,7 @@ def __init__(
         force_ram_power: Optional[int] = _sentinel,
         pue: Optional[float] = _sentinel,
         wue: Optional[float] = _sentinel,
+        force_carbon_intensity_g_co2e_kwh: Optional[float] = _sentinel,
         force_mode_cpu_load: Optional[bool] = _sentinel,
         allow_multiple_runs: Optional[bool] = _sentinel,
         rapl_include_dram: Optional[bool] = _sentinel,
@@ -259,11 +260,13 @@ def __init__(
                                 then RAM power (W) = Number of RAM Slots × 5 Watts.
         :param pue: PUE (Power Usage Effectiveness) of the data center where the
                     experiment is being run.
+        :param wue: WUE (Water Usage Effectiveness) of the data center. Units of L/kWh:
+                    litres of water consumed per kilowatt-hour of electricity consumed.
+        :param force_carbon_intensity_g_co2e_kwh: Override grid carbon intensity
+                                                  in gCO2e/kWh for emissions calculations.
         :param force_mode_cpu_load: Force the addition of a CPU in MODE_CPU_LOAD
         :param allow_multiple_runs: Allow multiple CodeCarbon instances on the same machine.
                                     Defaults to True since v3 (was False in v2).
-        :param wue: WUE (Water Usage Effectiveness) of the data center. Units of L/kWh:
-                    litres of water consumed per kilowatt-hour of electricity consumed.
         :param rapl_include_dram: Include DRAM (memory) power in RAPL measurements on Linux,
                                   defaults to False. When True, measures CPU package + DRAM.
                                   Only affects systems where RAPL exposes separate DRAM domains.
@@ -276,33 +279,31 @@ 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"
+        self._set_from_conf(
+            force_carbon_intensity_g_co2e_kwh,
+            "force_carbon_intensity_g_co2e_kwh",
+            None,
+            float,
         )
         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:
+        if self._force_carbon_intensity_g_co2e_kwh is not None:
+            try:
+                value = float(self._force_carbon_intensity_g_co2e_kwh)
+                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 numeric value. Using default calculation methods."
+                        f"CODECARBON : Invalid value for force_carbon_intensity_g_co2e_kwh: '{self._force_carbon_intensity_g_co2e_kwh}'. "
+                        "It must be a non-negative number. Using default calculation methods."
                     )
-        self.custom_carbon_intensity_g_co2e_kwh = parsed_intensity
+            except (ValueError, TypeError):
+                logger.warning(
+                    f"CODECARBON : Invalid value for force_carbon_intensity_g_co2e_kwh: '{self._force_carbon_intensity_g_co2e_kwh}'. "
+                    "It must be a numeric value. Using default calculation methods."
+                )
+        self._force_carbon_intensity_g_co2e_kwh = parsed_intensity
+        self._conf["force_carbon_intensity_g_co2e_kwh"] = parsed_intensity
+        self.force_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(
@@ -380,9 +381,9 @@ def __init__(
             experiment_id, "experiment_id", "5b0fa12a-3dd7-45bb-9766-cc326314d9f1"
         )
 
-        if self.custom_carbon_intensity_g_co2e_kwh is not None:
+        if self.force_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."
+                f"CODECARBON : Using forced carbon intensity: {self.force_carbon_intensity_g_co2e_kwh} gCO2e/kWh."
             )
 
         assert self._tracking_mode in ["machine", "process"]
@@ -480,7 +481,7 @@ def __init__(
         self._emissions: Emissions = Emissions(
             self._data_source,
             self._electricitymaps_api_token,
-            custom_carbon_intensity_g_co2e_kwh=self.custom_carbon_intensity_g_co2e_kwh,
+            force_carbon_intensity_g_co2e_kwh=self.force_carbon_intensity_g_co2e_kwh,
         )
         self._init_output_methods(api_key=self._api_key)
 
@@ -1344,6 +1345,7 @@ def track_emissions(
     force_ram_power: Optional[int] = _sentinel,
     pue: Optional[float] = _sentinel,
     wue: Optional[float] = _sentinel,
+    force_carbon_intensity_g_co2e_kwh: Optional[float] = _sentinel,
     allow_multiple_runs: Optional[bool] = _sentinel,
     rapl_include_dram: Optional[bool] = _sentinel,
     rapl_prefer_psys: Optional[bool] = _sentinel,
@@ -1426,6 +1428,8 @@ def track_emissions(
     :param pue: PUE (Power Usage Effectiveness) of the data center.
     :param wue: WUE (Water Usage Effectiveness) of the data center. Units of L/kWh:
                 litres of water consumed per kilowatt-hour of electricity consumed.
+    :param force_carbon_intensity_g_co2e_kwh: Override grid carbon intensity
+                         in gCO2e/kWh for emissions calculations.
     :param rapl_include_dram: Include DRAM in RAPL measurements on Linux (default: False).
                               When True, measures CPU package + DRAM.
     :param rapl_prefer_psys: Prefer psys over package domains for RAPL on Linux
@@ -1481,6 +1485,7 @@ def wrapped_fn(*args, **kwargs):
                     force_ram_power=force_ram_power,
                     pue=pue,
                     wue=wue,
+                    force_carbon_intensity_g_co2e_kwh=force_carbon_intensity_g_co2e_kwh,
                     allow_multiple_runs=allow_multiple_runs,
                     rapl_include_dram=rapl_include_dram,
                     rapl_prefer_psys=rapl_prefer_psys,
@@ -1515,6 +1520,7 @@ def wrapped_fn(*args, **kwargs):
                     force_ram_power=force_ram_power,
                     pue=pue,
                     wue=wue,
+                    force_carbon_intensity_g_co2e_kwh=force_carbon_intensity_g_co2e_kwh,
                     allow_multiple_runs=allow_multiple_runs,
                     rapl_include_dram=rapl_include_dram,
                     rapl_prefer_psys=rapl_prefer_psys,

tests/test_config.py

@@ -11,7 +11,11 @@
     parse_env_config,
     parse_gpu_ids,
 )
-from codecarbon.emissions_tracker import EmissionsTracker
+from codecarbon.emissions_tracker import (
+    EmissionsTracker,
+    OfflineEmissionsTracker,
+    track_emissions,
+)
 from codecarbon.external.hardware import GPU
 from tests.testutils import get_custom_mock_open
 
@@ -199,7 +203,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
+            force_carbon_intensity_g_co2e_kwh=123.4
             """
         )
         local_conf = dedent(
@@ -226,10 +230,59 @@ 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.force_carbon_intensity_g_co2e_kwh, 123.4)
             self.assertEqual(tracker._project_name, "test-project")
             self.assertTrue(tracker._save_to_file)
 
+    def test_force_carbon_intensity_constructor_overrides_config(self):
+        global_conf = dedent(
+            """\
+            [codecarbon]
+            force_carbon_intensity_g_co2e_kwh=123.4
+            """
+        )
+
+        with patch("builtins.open", new_callable=get_custom_mock_open(global_conf, "")):
+            with patch("os.path.exists", return_value=True):
+                tracker = EmissionsTracker(
+                    force_carbon_intensity_g_co2e_kwh=456.7,
+                    save_to_file=False,
+                    allow_multiple_runs=True,
+                )
+
+        self.assertEqual(tracker.force_carbon_intensity_g_co2e_kwh, 456.7)
+        self.assertEqual(tracker._conf["force_carbon_intensity_g_co2e_kwh"], 456.7)
+
+    def test_offline_tracker_accepts_force_carbon_intensity_parameter(self):
+        with patch("builtins.open", new_callable=get_custom_mock_open("", "")):
+            with patch("os.path.exists", return_value=True):
+                tracker = OfflineEmissionsTracker(
+                    country_iso_code="FRA",
+                    force_carbon_intensity_g_co2e_kwh=0,
+                    save_to_file=False,
+                    allow_multiple_runs=True,
+                )
+
+        self.assertEqual(tracker.force_carbon_intensity_g_co2e_kwh, 0.0)
+
+    def test_track_emissions_forwards_force_carbon_intensity_parameter(self):
+        with patch("codecarbon.emissions_tracker.EmissionsTracker") as tracker_class:
+
+            @track_emissions(
+                force_carbon_intensity_g_co2e_kwh=321.0,
+                save_to_file=False,
+            )
+            def tracked_function():
+                return "success"
+
+            self.assertEqual(tracked_function(), "success")
+
+        tracker_class.assert_called_once()
+        self.assertEqual(
+            tracker_class.call_args.kwargs["force_carbon_intensity_g_co2e_kwh"],
+            321.0,
+        )
+
     @mock.patch.dict(
         os.environ,
         {

tests/test_emissions.py

@@ -175,9 +175,9 @@ def test_get_emissions_PRIVATE_INFRA_unknown_country(self):
         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):
+    def test_private_infra_uses_forced_intensity_when_set(self, mocked_get_emissions):
         emissions_calculator = Emissions(
-            self._data_source, custom_carbon_intensity_g_co2e_kwh=50.0
+            self._data_source, force_carbon_intensity_g_co2e_kwh=50.0
         )
 
         emissions = emissions_calculator.get_private_infra_emissions(
@@ -188,9 +188,9 @@ def test_private_infra_uses_custom_intensity_when_set(self, mocked_get_emissions
         self.assertAlmostEqual(emissions, 0.1, places=6)
         mocked_get_emissions.assert_not_called()
 
-    def test_cloud_uses_custom_intensity_when_set(self):
+    def test_cloud_uses_forced_intensity_when_set(self):
         emissions_calculator = Emissions(
-            self._data_source, custom_carbon_intensity_g_co2e_kwh=100.0
+            self._data_source, force_carbon_intensity_g_co2e_kwh=100.0
         )
 
         emissions = emissions_calculator.get_cloud_emissions(