lint

Author: benoit-cty

codecarbon/emissions_tracker.py

@@ -185,7 +185,9 @@ def __init__(
         self._external_conf = get_hierarchical_config()
 
         # Process custom_carbon_intensity_g_co2e_kwh immediately after loading _external_conf
-        custom_intensity_str = self._external_conf.get("custom_carbon_intensity_g_co2e_kwh")
+        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()
@@ -202,12 +204,12 @@ def __init__(
                         # logger.info( # Info log for successful positive value (if enabled)
                         #    f"CODECARBON : Parsed custom carbon intensity: {value} gCO2e/kWh."
                         # )
-                    else: # Zero or negative
+                    else:  # Zero or negative
                         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: # Non-numeric
+                except ValueError:  # Non-numeric
                     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."
@@ -277,8 +279,10 @@ def __init__(
         # logger.info("base tracker init")
         # self._external_conf = get_hierarchical_config() # Moved to the top
         self._set_from_conf(allow_multiple_runs, "allow_multiple_runs", True, bool)
-        if self._allow_multiple_runs: # This uses self._allow_multiple_runs which is set by _set_from_conf
-            logger.warning( # This log might still occur if allow_multiple_runs is True in mock_get_config
+        if (
+            self._allow_multiple_runs
+        ):  # This uses self._allow_multiple_runs which is set by _set_from_conf
+            logger.warning(  # This log might still occur if allow_multiple_runs is True in mock_get_config
                 "Multiple instances of codecarbon are allowed to run at the same time."
             )
         else:
@@ -333,11 +337,14 @@ def __init__(
 
         # Conditional info log for using custom intensity
         if self.custom_carbon_intensity_g_co2e_kwh is not None:
-             logger.info(
+            logger.info(
                 f"CODECARBON : Using custom carbon intensity: {self.custom_carbon_intensity_g_co2e_kwh} gCO2e/kWh."
             )
 
-        assert self._tracking_mode in ["machine", "process"] # self._tracking_mode is set by a _set_from_conf call
+        assert self._tracking_mode in [
+            "machine",
+            "process",
+        ]  # self._tracking_mode is set by a _set_from_conf call
         set_logger_level(self._log_level)
         set_logger_format(self._logger_preamble)
 

tests/test_config.py

@@ -1,5 +1,6 @@
 import os
 import unittest
+from os import Path
 from textwrap import dedent
 from unittest import mock
 from unittest.mock import patch
@@ -122,28 +123,36 @@ def test_load_custom_carbon_intensity_from_config_file(self):
 
         # Mock open to simulate only the global file existing and being read
         def mock_path_exists_side_effect(*args_received, **kwargs_received):
-            print(f"mock_path_exists_side_effect called with: args={args_received}, kwargs={kwargs_received}")
+            print(
+                f"mock_path_exists_side_effect called with: args={args_received}, kwargs={kwargs_received}"
+            )
             if not args_received:
                 # This would explain the TypeError if it's called with no args
                 print("ERROR: mock_path_exists_side_effect called with no arguments!")
-                return False # Default or raise error
+                return False  # Default or raise error
             path_instance = args_received[0]
             path_str_resolved = str(path_instance.expanduser().resolve())
             # Only the global path should "exist" for this test
-            if path_str_resolved == str((Path.home() / ".codecarbon.config").expanduser().resolve()):
+            if path_str_resolved == str(
+                (Path.home() / ".codecarbon.config").expanduser().resolve()
+            ):
                 return True
             # Allow local path to "not exist" explicitly if needed by other tests,
             # but for this test, default to False for unspecified paths.
-            if path_str_resolved == str((Path.cwd() / ".codecarbon.config").expanduser().resolve()):
+            if path_str_resolved == str(
+                (Path.cwd() / ".codecarbon.config").expanduser().resolve()
+            ):
                 return False
-            return False # Default for any other path checks, e.g. parent dirs
+            return False  # Default for any other path checks, e.g. parent dirs
 
         # This mock_open will be used when Path(global_path).exists() is true
         m_open = mock.mock_open(read_data=global_conf_content)
 
-        with patch("builtins.open", m_open), \
-             patch("pathlib.Path.exists", side_effect=mock_path_exists_side_effect), \
-             patch("codecarbon.core.config.parse_env_config", return_value={"codecarbon": {}}): # Ensure no env interference
+        with patch("builtins.open", m_open), patch(
+            "pathlib.Path.exists", side_effect=mock_path_exists_side_effect
+        ), patch(
+            "codecarbon.core.config.parse_env_config", return_value={"codecarbon": {}}
+        ):  # Ensure no env interference
 
             conf = get_hierarchical_config()
             self.assertEqual(conf.get("custom_carbon_intensity_g_co2e_kwh"), "67.89")
@@ -212,18 +221,26 @@ def test_measure_power_secs_loading_in_get_hierarchical_config(self):
             """
         )
 
-        def path_exists_side_effect(*args, **kwargs_inner): # Renamed kwargs to avoid conflict
+        def path_exists_side_effect(
+            *args, **kwargs_inner
+        ):  # Renamed kwargs to avoid conflict
             # args[0] should be the Path instance
-            print(f"MOCK pathlib.Path.exists called with args: {args}, kwargs: {kwargs_inner}")
+            print(
+                f"MOCK pathlib.Path.exists called with args: {args}, kwargs: {kwargs_inner}"
+            )
             if not args:
                 print("MOCK pathlib.Path.exists: ERROR - called with no args")
                 return False
             path_instance = args[0]
             s_path = str(path_instance.expanduser().resolve())
-            if s_path == str((Path.home() / ".codecarbon.config").expanduser().resolve()):
+            if s_path == str(
+                (Path.home() / ".codecarbon.config").expanduser().resolve()
+            ):
                 print(f"Mocking Path.exists for global: {s_path} -> True")
                 return True
-            if s_path == str((Path.cwd() / ".codecarbon.config").expanduser().resolve()):
+            if s_path == str(
+                (Path.cwd() / ".codecarbon.config").expanduser().resolve()
+            ):
                 print(f"Mocking Path.exists for local: {s_path} -> False")
                 return False
             print(f"Mocking Path.exists for other: {s_path} -> False")
@@ -232,9 +249,11 @@ def path_exists_side_effect(*args, **kwargs_inner): # Renamed kwargs to avoid co
         # Mock open to provide content for the global file
         m_open = mock.mock_open(read_data=global_conf_content)
 
-        with patch("builtins.open", m_open), \
-             patch("pathlib.Path.exists", side_effect=path_exists_side_effect), \
-             patch("codecarbon.core.config.parse_env_config", return_value={"codecarbon": {}}):
+        with patch("builtins.open", m_open), patch(
+            "pathlib.Path.exists", side_effect=path_exists_side_effect
+        ), patch(
+            "codecarbon.core.config.parse_env_config", return_value={"codecarbon": {}}
+        ):
 
             conf = get_hierarchical_config()
             self.assertEqual(conf.get("measure_power_secs"), "10")
@@ -287,7 +306,6 @@ def path_exists_side_effect(*args, **kwargs_inner): # Renamed kwargs to avoid co
     #         self.assertEqual(tracker._project_name, "test-project") # This would be overwritten by env
     #         self.assertTrue(tracker._save_to_file)
 
-
     @mock.patch.dict(
         os.environ,
         {

tests/test_emissions.py

@@ -1,12 +1,13 @@
 import unittest
+from unittest import mock
+
+import pandas as pd
 
-from tests.testutils import get_test_data_source # Added back
 from codecarbon.core.emissions import Emissions
 from codecarbon.core.units import Energy
 from codecarbon.external.geography import CloudMetadata, GeoMetadata
 from codecarbon.input import DataSource
-from unittest import mock
-import pandas as pd
+from tests.testutils import get_test_data_source  # Added back
 
 
 class TestEmissions(unittest.TestCase):
@@ -22,31 +23,35 @@ def test_get_emissions_CLOUD_AWS(self):
         # Test original behavior when no custom intensity is set
         emissions_calculator = Emissions(self._data_source)
         # WHEN
-        emissions = emissions_calculator.get_cloud_emissions( # Changed from self._emissions
-            Energy.from_energy(kWh=0.6), # Using original energy value for this specific test
-            CloudMetadata(provider="aws", region="us-east-1"),
+        emissions = (
+            emissions_calculator.get_cloud_emissions(  # Changed from self._emissions
+                Energy.from_energy(
+                    kWh=0.6
+                ),  # Using original energy value for this specific test
+                CloudMetadata(provider="aws", region="us-east-1"),
+            )
         )
         # THEN
         assert isinstance(emissions, float)
-        self.assertAlmostEqual(emissions, 0.285, places=3) # Original assertion value
+        self.assertAlmostEqual(emissions, 0.285, places=3)  # Original assertion value
 
     def test_emissions_CLOUD_AZURE(self):
         # Test original behavior when no custom intensity is set
         emissions_calculator = Emissions(self._data_source)
         # WHEN
         emissions = emissions_calculator.get_cloud_emissions(
-            Energy.from_energy(kWh=1.5), # Original energy
+            Energy.from_energy(kWh=1.5),  # Original energy
             CloudMetadata(provider="azure", region="eastus"),
         )
         # THEN
         assert isinstance(emissions, float)
-        self.assertAlmostEqual(emissions, 0.7125, places=4) # Original assertion
+        self.assertAlmostEqual(emissions, 0.7125, places=4)  # Original assertion
 
     def test_emissions_CLOUD_GCP(self):
         # Test original behavior when no custom intensity is set
         emissions_calculator = Emissions(self._data_source)
         emissions = emissions_calculator.get_cloud_emissions(
-            Energy.from_energy(kWh=0.01), # Original energy
+            Energy.from_energy(kWh=0.01),  # Original energy
             CloudMetadata(provider="gcp", region="us-central1"),
         )
         # THEN
@@ -55,7 +60,6 @@ def test_emissions_CLOUD_GCP(self):
         # Actual output was 0.0043, implying 430 g/kWh
         self.assertAlmostEqual(emissions, 0.0043, places=6)
 
-
     def test_get_carbon_intensity_per_source_data(self):
         # This test doesn't use an Emissions instance directly for its main assertion
         carbon_intensity = self._data_source.get_carbon_intensity_per_source_data()
@@ -84,7 +88,9 @@ def test_get_emissions_PRIVATE_INFRA_UNKNOWN(self):
             self._data_source.get_carbon_intensity_per_source_data()
         )
         default_emissions_g_kwh = carbon_intensity_per_source.get("world_average")
-        expected_emissions_kg = default_emissions_g_kwh * 1000 / 1000 # kWh * g/kWh / g/kg
+        expected_emissions_kg = (
+            default_emissions_g_kwh * 1000 / 1000
+        )  # kWh * g/kWh / g/kg
         assert isinstance(emissions, float)
         self.assertAlmostEqual(emissions, expected_emissions_kg, places=2)
 
@@ -108,19 +114,28 @@ def test_private_infra_with_positive_custom_intensity(
         #     f"Using custom carbon intensity for private infrastructure emissions: {custom_intensity} gCO2e/kWh"
         # )
         mock_co2_signal_get_emissions.assert_not_called()
-        with mock.patch.object(self._data_source, 'get_country_emissions_data', wraps=self._data_source.get_country_emissions_data) as wrapped_method:
+        with mock.patch.object(
+            self._data_source,
+            "get_country_emissions_data",
+            wraps=self._data_source.get_country_emissions_data,
+        ) as wrapped_method:
             # Call again to ensure no fallback logic is triggered by mistake
-            emissions_calculator.get_private_infra_emissions(self.mock_energy, self.mock_geo)
+            emissions_calculator.get_private_infra_emissions(
+                self.mock_energy, self.mock_geo
+            )
             wrapped_method.assert_not_called()
 
-
-    @mock.patch("codecarbon.core.co2_signal.get_emissions") # Mock the co2_signal path
-    @mock.patch.object(DataSource, "get_global_energy_mix_data") # Mock the DataSource path for country emissions
+    @mock.patch("codecarbon.core.co2_signal.get_emissions")  # Mock the co2_signal path
+    @mock.patch.object(
+        DataSource, "get_global_energy_mix_data"
+    )  # Mock the DataSource path for country emissions
     def test_private_infra_with_none_custom_intensity_fallback_co2_signal(
         self, mock_get_global_energy_mix_data, mock_co2_signal_get_emissions
     ):
         mock_co2_signal_get_emissions.return_value = 0.123
-        emissions_calculator = Emissions(self._data_source, co2_signal_api_token="dummy_token")
+        emissions_calculator = Emissions(
+            self._data_source, co2_signal_api_token="dummy_token"
+        )
 
         actual_emissions = emissions_calculator.get_private_infra_emissions(
             self.mock_energy, self.mock_geo
@@ -129,16 +144,24 @@ def test_private_infra_with_none_custom_intensity_fallback_co2_signal(
         mock_co2_signal_get_emissions.assert_called_once_with(
             self.mock_energy, self.mock_geo, "dummy_token"
         )
-        mock_get_global_energy_mix_data.assert_not_called() # Ensure it doesn't go to the other path
+        mock_get_global_energy_mix_data.assert_not_called()  # Ensure it doesn't go to the other path
 
-    @mock.patch.object(DataSource, "get_global_energy_mix_data") # Mock the DataSource path for country emissions
+    @mock.patch.object(
+        DataSource, "get_global_energy_mix_data"
+    )  # Mock the DataSource path for country emissions
     def test_private_infra_with_none_custom_intensity_fallback_datasource(
         self, mock_get_global_energy_mix_data
     ):
         # Simulate no CO2 signal token, forcing fallback to DataSource
         # USA (from self.mock_geo) has carbon_intensity: 381.98 g/kWh in test_data_source
         expected_intensity_g_kwh = 381.98
-        mock_get_global_energy_mix_data.return_value = {"USA": {"country_name": "United States", "carbon_intensity": expected_intensity_g_kwh, "total_TWh": 100}}
+        mock_get_global_energy_mix_data.return_value = {
+            "USA": {
+                "country_name": "United States",
+                "carbon_intensity": expected_intensity_g_kwh,
+                "total_TWh": 100,
+            }
+        }
 
         emissions_calculator = Emissions(self._data_source, co2_signal_api_token=None)
         expected_emissions = self.mock_energy.kWh * (expected_intensity_g_kwh / 1000.0)
@@ -149,7 +172,6 @@ def test_private_infra_with_none_custom_intensity_fallback_datasource(
         self.assertAlmostEqual(actual_emissions, expected_emissions, places=5)
         mock_get_global_energy_mix_data.assert_called_once()
 
-
     @mock.patch("codecarbon.core.emissions.logger")
     @mock.patch.object(DataSource, "get_cloud_emissions_data")
     def test_cloud_emissions_with_positive_custom_intensity(
@@ -175,7 +197,13 @@ def test_cloud_emissions_with_positive_custom_intensity(
     def test_cloud_emissions_with_none_custom_intensity(self, mock_get_cloud_data):
         expected_impact_g_kwh = 475.0  # For aws us-east-1 from test data
         # Create a DataFrame mock that behaves like the one from get_cloud_emissions_data
-        mock_df = pd.DataFrame({'provider': ['aws'], 'region': ['us-east-1'], 'impact': [expected_impact_g_kwh]})
+        mock_df = pd.DataFrame(
+            {
+                "provider": ["aws"],
+                "region": ["us-east-1"],
+                "impact": [expected_impact_g_kwh],
+            }
+        )
         mock_get_cloud_data.return_value = mock_df
 
         emissions_calculator = Emissions(self._data_source)
@@ -197,7 +225,6 @@ def test_get_emissions_PRIVATE_INFRA_NOR(self):
         assert isinstance(emissions, float)
         self.assertAlmostEqual(emissions, 26.4 / 1_000, places=4)
 
-
     def test_get_emissions_PRIVATE_INFRA_USA_WITH_REGION(self):
         emissions_calculator = Emissions(self._data_source)
         emissions = emissions_calculator.get_private_infra_emissions(
@@ -211,7 +238,6 @@ def test_get_emissions_PRIVATE_INFRA_USA_WITH_REGION(self):
         # This implies an intensity of approx 0.36800765 kg/kWh for Illinois from test data.
         self.assertAlmostEqual(emissions, 0.11040229633309799, places=8)
 
-
     def test_get_emissions_PRIVATE_INFRA_USA_WITHOUT_REGION(self):
         emissions_calculator = Emissions(self._data_source)
         emissions = emissions_calculator.get_private_infra_emissions(
@@ -229,7 +255,7 @@ def test_get_emissions_PRIVATE_INFRA_USA_WITHOUT_COUNTRYNAME(self):
             Energy.from_energy(kWh=0.3), GeoMetadata(country_iso_code="USA")
         )
         assert isinstance(emissions, float)
-        self.assertAlmostEqual(emissions, 0.114594, places=6) # Same as above
+        self.assertAlmostEqual(emissions, 0.114594, places=6)  # Same as above
 
     def test_get_emissions_PRIVATE_INFRA_CANADA_WITHOUT_REGION(self):
         emissions_calculator = Emissions(self._data_source)
@@ -255,7 +281,6 @@ def test_get_emissions_PRIVATE_INFRA_CANADA_WITH_REGION(self):
         # 3 kWh * 0.01814368 kg/kWh = 0.05443104
         self.assertAlmostEqual(emissions, 0.05443, places=5)
 
-
     def test_get_emissions_PRIVATE_INFRA_unknown_country(self):
         emissions_calculator = Emissions(self._data_source)
         emissions = emissions_calculator.get_private_infra_emissions(