Compute emissions based on ongoing carbon intensity

codecarbon/emissions_tracker.py

@@ -337,6 +337,8 @@ def __init__(
         self._start_time: Optional[float] = None
         self._last_measured_time: float = time.perf_counter()
         self._total_energy: Energy = Energy.from_energy(kWh=0)
+        self._total_emissions: float = 0.0
+        self._last_energy_covered: Energy = Energy.from_energy(kWh=0)
         self._total_water: Water = Water.from_litres(litres=0)
         # CPU and RAM utilization tracking
         self._cpu_utilization_history: List[float] = []
@@ -757,28 +759,43 @@ def _persist_data(
             if len(task_emissions_data) > 0:
                 handler.task_out(task_emissions_data, experiment_name)
 
+    def _update_emissions(self) -> None:
+        """
+        Compute emissions for the energy consumed since the last update
+        and add them to the total emissions.
+        """
+        delta_energy = self._total_energy - self._last_energy_covered
+        if delta_energy.kWh > 0:
+            cloud: CloudMetadata = self._get_cloud_metadata()
+            if cloud.is_on_private_infra:
+                delta_emissions = self._emissions.get_private_infra_emissions(
+                    delta_energy, self._geo
+                )
+            else:
+                delta_emissions = self._emissions.get_cloud_emissions(
+                    delta_energy, cloud, self._geo
+                )
+            self._total_emissions += delta_emissions
+            self._last_energy_covered = self._total_energy
+
     def _prepare_emissions_data(self) -> EmissionsData:
         """
         Prepare the emissions data to be sent to the API or written to a file.
         :return: EmissionsData object with the total emissions data.
         """
+        self._update_emissions()
         cloud: CloudMetadata = self._get_cloud_metadata()
         duration: Time = Time.from_seconds(time.perf_counter() - self._start_time)
 
+        emissions = self._total_emissions
         if cloud.is_on_private_infra:
-            emissions = self._emissions.get_private_infra_emissions(
-                self._total_energy, self._geo
-            )  # float: kg co2_eq
             country_name = self._geo.country_name
             country_iso_code = self._geo.country_iso_code
             region = self._geo.region
             on_cloud = "N"
             cloud_provider = ""
             cloud_region = ""
         else:
-            emissions = self._emissions.get_cloud_emissions(
-                self._total_energy, cloud, self._geo
-            )
             # Try to get cloud region metadata, fall back to geo metadata if not found
             try:
                 country_name = self._emissions.get_cloud_country_name(cloud)

tests/test_emissions_tracker.py

@@ -656,3 +656,98 @@ def test_get_detected_hardware(
         self.assertIn("gpu_count", hardware_info)
         self.assertIn("gpu_model", hardware_info)
         self.assertIn("gpu_ids", hardware_info)
+
+    @mock.patch("codecarbon.emissions_tracker.EmissionsTracker._get_geo_metadata")
+    @mock.patch("codecarbon.emissions_tracker.EmissionsTracker._get_cloud_metadata")
+    @mock.patch("codecarbon.core.electricitymaps_api.requests.get")
+    @mock.patch("codecarbon.emissions_tracker.ResourceTracker")
+    @mock.patch(
+        "codecarbon.emissions_tracker.BaseEmissionsTracker.get_detected_hardware"
+    )
+    @mock.patch("codecarbon.emissions_tracker.PeriodicScheduler")
+    def test_cumulative_emissions_with_varying_intensity(
+        self,
+        mock_scheduler,
+        mock_get_hw,
+        mock_resource_tracker,
+        mock_get,
+        mock_cloud,
+        mock_geo,
+        mock_cli_setup,
+        mock_log_values,
+        mocked_get_cloud_metadata_class,
+        mocked_get_gpu_details,
+        mocked_is_gpu_details_available,
+    ):
+        # Setup mocks
+        mock_geo.return_value = mock.MagicMock(
+            latitude=1.0,
+            longitude=1.0,
+            country_iso_code="USA",
+            country_2letter_iso_code="US",
+        )
+        mock_cloud.return_value = mock.MagicMock(
+            is_on_private_infra=True, provider="azure", region="eastus"
+        )
+        mock_get_hw.return_value = {
+            "ram_total_size": 16.0,
+            "cpu_count": 8,
+            "cpu_physical_count": 4,
+            "cpu_model": "Mock CPU",
+            "gpu_count": 0,
+            "gpu_model": "None",
+            "gpu_ids": None,
+        }
+
+        # Mock Electricity Maps API responses with different intensities
+        # 1st call: 100 g/kWh, 2nd call: 200 g/kWh, 3rd call: 300 g/kWh
+        responses = [
+            mock.MagicMock(status_code=200, json=lambda: {"carbonIntensity": 100}),
+            mock.MagicMock(status_code=200, json=lambda: {"carbonIntensity": 200}),
+            mock.MagicMock(status_code=200, json=lambda: {"carbonIntensity": 300}),
+        ]
+        mock_get.side_effect = responses
+
+        tracker = EmissionsTracker(
+            electricitymaps_api_token="test-token",
+            save_to_file=False,
+            measure_power_secs=1,
+            allow_multiple_runs=True,
+        )
+
+        # Manually inject a mock hardware component
+        mock_cpu = mock.MagicMock()
+        from codecarbon.external.hardware import CPU
+
+        mock_cpu.__class__ = CPU
+        # Mock measure_power_and_energy: return 1kWh delta each time
+        mock_cpu.measure_power_and_energy.return_value = (
+            Power.from_watts(100),
+            Energy.from_energy(kWh=1.0),
+        )
+        tracker._hardware = [mock_cpu]
+
+        # Start tracking
+        tracker.start()
+
+        tracker._measure_power_and_energy()
+        # total_energy = 1.0, intensity = 100 => emissions = 0.1 kg
+        data1 = tracker._prepare_emissions_data()
+        self.assertAlmostEqual(data1.emissions, 0.1)
+
+        # Step 2
+        tracker._measure_power_and_energy()
+        # total_energy = 2.0, delta_energy = 1.0, intensity = 200 => delta_emissions = 0.2 kg
+        # total_emissions = 0.3 kg
+        data2 = tracker._prepare_emissions_data()
+        self.assertAlmostEqual(data2.emissions, 0.3)
+
+        # Step 3
+        tracker._measure_power_and_energy()
+        # total_energy = 3.0, delta_energy = 1.0, intensity = 300 => delta_emissions = 0.3 kg
+        # total_emissions = 0.6 kg
+        data3 = tracker._prepare_emissions_data()
+        self.assertAlmostEqual(data3.emissions, 0.6)
+
+        # Verification: If it wasn't cumulative, it would be 3.0 kWh * 300 g/kWh = 0.9 kg
+        self.assertLess(data3.emissions, 0.8)