From 045d5a7b03a49dcc949b3d0c95f51e9df2ebac89 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Wed, 2 Jul 2025 08:45:27 +0000
Subject: [PATCH 1/8] Add force_mode_constant parameter to bypass psutil
 overhead

- Make psutil imports graceful in cpu.py and util.py
- Add force_mode_constant parameter to EmissionsTracker
- Update CPU tracking logic to prioritize force_mode_constant
- Add comprehensive test suite for new functionality
- Update documentation with new parameter details

Co-authored-by: benoit-cty <6603048+benoit-cty@users.noreply.github.com>
---
 codecarbon/core/cpu.py              |  11 ++-
 codecarbon/core/resource_tracker.py |  25 +++++
 codecarbon/core/util.py             |  14 ++-
 codecarbon/emissions_tracker.py     |   4 +
 tests/test_force_constant_mode.py   | 147 ++++++++++++++++++++++++++++
 5 files changed, 199 insertions(+), 2 deletions(-)
 create mode 100644 tests/test_force_constant_mode.py

diff --git a/codecarbon/core/cpu.py b/codecarbon/core/cpu.py
index 74549a14f..1c230294d 100644
--- a/codecarbon/core/cpu.py
+++ b/codecarbon/core/cpu.py
@@ -12,9 +12,15 @@
 from typing import Dict, Optional, Tuple
 
 import pandas as pd
-import psutil
 from rapidfuzz import fuzz, process, utils
 
+try:
+    import psutil
+    PSUTIL_AVAILABLE = True
+except ImportError:
+    PSUTIL_AVAILABLE = False
+    psutil = None
+
 from codecarbon.core.rapl import RAPLFile
 from codecarbon.core.units import Time
 from codecarbon.core.util import count_cpus, detect_cpu_model
@@ -207,6 +213,9 @@ def is_rapl_available(rapl_dir: Optional[str] = None) -> bool:
 
 
 def is_psutil_available():
+    if not PSUTIL_AVAILABLE:
+        logger.debug("psutil module is not available.")
+        return False
     try:
         cpu_times = psutil.cpu_times()
 
diff --git a/codecarbon/core/resource_tracker.py b/codecarbon/core/resource_tracker.py
index 67786189d..9a2088cbc 100644
--- a/codecarbon/core/resource_tracker.py
+++ b/codecarbon/core/resource_tracker.py
@@ -137,6 +137,31 @@ def _setup_fallback_tracking(self, tdp, max_power):
         self.tracker._conf["cpu_model"] = model
 
         if tdp:
+            max_power = self.tracker._force_cpu_power
+        else:
+            max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
+        
+        # Check for forced constant mode first
+        if self.tracker._conf.get("force_mode_constant", False):
+            logger.info("Force constant mode requested - bypassing psutil and using constant CPU power")
+            model = tdp.model
+            if max_power is None and self.tracker._force_cpu_power:
+                max_power = self.tracker._force_cpu_power
+                logger.debug(f"Using user input TDP for constant mode: {max_power} W")
+                self.cpu_tracker = "User Input TDP constant"
+            else:
+                self.cpu_tracker = "TDP constant"
+            logger.info(f"CPU Model on forced constant consumption mode: {model}")
+            self.tracker._conf["cpu_model"] = model
+            hardware_cpu = CPU.from_utils(
+                self.tracker._output_dir, "constant", model, max_power
+            )
+            self.tracker._hardware.append(hardware_cpu)
+            return
+            
+        if self.tracker._conf.get("force_mode_cpu_load", False) and (
+            tdp.tdp is not None or self.tracker._force_cpu_power is not None
+        ):
             if cpu.is_psutil_available():
                 logger.warning(
                     "No CPU tracking mode found. Falling back on CPU load mode."
diff --git a/codecarbon/core/util.py b/codecarbon/core/util.py
index 744b2e3e5..3121a0660 100644
--- a/codecarbon/core/util.py
+++ b/codecarbon/core/util.py
@@ -9,7 +9,13 @@
 from typing import Optional, Union
 
 import cpuinfo
-import psutil
+
+try:
+    import psutil
+    PSUTIL_AVAILABLE = True
+except ImportError:
+    PSUTIL_AVAILABLE = False
+    psutil = None
 
 from codecarbon.external.logger import logger
 
@@ -146,6 +152,12 @@ def _windows_get_physical_sockets():
 
 
 def count_cpus() -> int:
+    if not PSUTIL_AVAILABLE:
+        logger.warning("psutil not available, using fallback CPU count detection")
+        # Fallback to using os.cpu_count() or physical CPU count
+        cpu_count = os.cpu_count()
+        return cpu_count if cpu_count is not None else 1
+    
     if SLURM_JOB_ID is None:
         return psutil.cpu_count()
 
diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py
index ad712c9f0..0336f6bee 100644
--- a/codecarbon/emissions_tracker.py
+++ b/codecarbon/emissions_tracker.py
@@ -197,6 +197,7 @@ def __init__(
         wue: Optional[float] = _sentinel,
         force_carbon_intensity_g_co2e_kwh: Optional[float] = _sentinel,
         force_mode_cpu_load: Optional[bool] = _sentinel,
+        force_mode_constant: Optional[bool] = _sentinel,
         allow_multiple_runs: Optional[bool] = _sentinel,
         rapl_include_dram: Optional[bool] = _sentinel,
         rapl_prefer_psys: Optional[bool] = _sentinel,
@@ -275,6 +276,8 @@ def __init__(
                                  (CPU + chipset + PCIe). When False, uses package domains which
                                  are more reliable. Note: psys can report higher values than
                                  CPU TDP and may be unreliable on older systems.
+        :param force_mode_constant: Force the addition of a CPU in constant mode, bypassing psutil
+        :param allow_multiple_runs: Allow multiple instances of codecarbon running in parallel. Defaults to False.
         """
 
         # logger.info("base tracker init")
@@ -377,6 +380,7 @@ def __init__(
         self._set_from_conf(force_mode_cpu_load, "force_mode_cpu_load", False, bool)
         self._set_from_conf(rapl_include_dram, "rapl_include_dram", False, bool)
         self._set_from_conf(rapl_prefer_psys, "rapl_prefer_psys", False, bool)
+        self._set_from_conf(force_mode_constant, "force_mode_constant", False, bool)
         self._set_from_conf(
             experiment_id, "experiment_id", "5b0fa12a-3dd7-45bb-9766-cc326314d9f1"
         )
diff --git a/tests/test_force_constant_mode.py b/tests/test_force_constant_mode.py
new file mode 100644
index 000000000..ef46fecbd
--- /dev/null
+++ b/tests/test_force_constant_mode.py
@@ -0,0 +1,147 @@
+import os
+import tempfile
+import time
+import unittest
+from unittest import mock
+
+import pandas as pd
+
+from codecarbon.core import cpu
+from codecarbon.emissions_tracker import (
+    EmissionsTracker,
+    OfflineEmissionsTracker,
+)
+
+
+def light_computation(run_time_secs: int = 1):
+    end_time: float = (
+        time.perf_counter() + run_time_secs
+    )  # Run for `run_time_secs` seconds
+    while time.perf_counter() < end_time:
+        pass
+
+
+class TestForceConstantMode(unittest.TestCase):
+    def setUp(self) -> None:
+        self.project_name = "project_TestForceConstantMode"
+        self.emissions_file = "emissions-test-TestForceConstantMode.csv"
+        self.emissions_path = tempfile.gettempdir()
+        self.emissions_file_path = os.path.join(
+            self.emissions_path, self.emissions_file
+        )
+        if os.path.isfile(self.emissions_file_path):
+            os.remove(self.emissions_file_path)
+
+    def tearDown(self) -> None:
+        if os.path.isfile(self.emissions_file_path):
+            os.remove(self.emissions_file_path)
+
+    def test_force_constant_mode_online(self):
+        """Test force_mode_constant parameter with online tracker"""
+        tracker = EmissionsTracker(
+            output_dir=self.emissions_path, 
+            output_file=self.emissions_file,
+            force_mode_constant=True
+        )
+        tracker.start()
+        light_computation(run_time_secs=1)
+        emissions = tracker.stop()
+        
+        # Check that emissions were calculated
+        assert isinstance(emissions, float)
+        self.assertNotEqual(emissions, 0.0)
+        
+        # Verify output file was created
+        self.verify_output_file(self.emissions_file_path)
+        
+        # Check CSV content shows constant mode
+        df = pd.read_csv(self.emissions_file_path)
+        # The cpu_power should be a constant value (not varying like in load mode)
+        self.assertGreater(df["cpu_power"].iloc[0], 0)
+
+    def test_force_constant_mode_offline(self):
+        """Test force_mode_constant parameter with offline tracker"""
+        tracker = OfflineEmissionsTracker(
+            country_iso_code="USA",
+            output_dir=self.emissions_path,
+            output_file=self.emissions_file,
+            force_mode_constant=True
+        )
+        tracker.start()
+        light_computation(run_time_secs=1)
+        emissions = tracker.stop()
+        
+        assert isinstance(emissions, float)
+        self.assertNotEqual(emissions, 0.0)
+        self.verify_output_file(self.emissions_file_path)
+
+    def test_force_constant_mode_with_custom_cpu_power(self):
+        """Test force_mode_constant with custom CPU power"""
+        custom_cpu_power = 200  # 200W
+        tracker = EmissionsTracker(
+            output_dir=self.emissions_path,
+            output_file=self.emissions_file,
+            force_mode_constant=True,
+            force_cpu_power=custom_cpu_power
+        )
+        tracker.start()
+        light_computation(run_time_secs=1)
+        emissions = tracker.stop()
+        
+        assert isinstance(emissions, float)
+        self.assertNotEqual(emissions, 0.0)
+        
+        # Check that the custom CPU power was used
+        df = pd.read_csv(self.emissions_file_path)
+        # CPU power should be 50% of the TDP (constant mode assumption)
+        expected_cpu_power = custom_cpu_power / 2
+        self.assertEqual(df["cpu_power"].iloc[0], expected_cpu_power)
+
+    @mock.patch("codecarbon.core.cpu.PSUTIL_AVAILABLE", False)
+    @mock.patch("codecarbon.core.util.PSUTIL_AVAILABLE", False)
+    def test_force_constant_mode_without_psutil(self):
+        """Test that force_mode_constant works when psutil is not available"""
+        tracker = EmissionsTracker(
+            output_dir=self.emissions_path,
+            output_file=self.emissions_file,
+            force_mode_constant=True
+        )
+        tracker.start()
+        light_computation(run_time_secs=1)
+        emissions = tracker.stop()
+        
+        assert isinstance(emissions, float)
+        self.assertNotEqual(emissions, 0.0)
+        self.verify_output_file(self.emissions_file_path)
+
+    def test_force_constant_mode_takes_precedence_over_cpu_load(self):
+        """Test that force_mode_constant takes precedence over force_mode_cpu_load"""
+        tracker = EmissionsTracker(
+            output_dir=self.emissions_path,
+            output_file=self.emissions_file,
+            force_mode_constant=True,
+            force_mode_cpu_load=True  # This should be ignored
+        )
+        tracker.start()
+        light_computation(run_time_secs=1)
+        emissions = tracker.stop()
+        
+        assert isinstance(emissions, float)
+        self.assertNotEqual(emissions, 0.0)
+        self.verify_output_file(self.emissions_file_path)
+
+    def verify_output_file(self, file_path: str) -> None:
+        """Verify that the output CSV file exists and has expected structure"""
+        with open(file_path, "r") as f:
+            lines = [line.rstrip() for line in f]
+        assert len(lines) == 2  # Header + 1 data row
+        
+        # Check that it's a valid CSV with expected columns
+        df = pd.read_csv(file_path)
+        expected_columns = ["emissions", "cpu_power", "cpu_energy"]
+        for col in expected_columns:
+            self.assertIn(col, df.columns)
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file

From 12d0cd8d7bf0e67c7c5a382191ecbf3a312d418b Mon Sep 17 00:00:00 2001
From: benoit-cty <4-benoit-cty@users.noreply.git.leximpact.dev>
Date: Wed, 24 Sep 2025 19:43:24 +0200
Subject: [PATCH 2/8] Lint

---
 tests/test_force_constant_mode.py | 33 +++++++++++++++----------------
 1 file changed, 16 insertions(+), 17 deletions(-)

diff --git a/tests/test_force_constant_mode.py b/tests/test_force_constant_mode.py
index ef46fecbd..4b1ca8429 100644
--- a/tests/test_force_constant_mode.py
+++ b/tests/test_force_constant_mode.py
@@ -6,7 +6,6 @@
 
 import pandas as pd
 
-from codecarbon.core import cpu
 from codecarbon.emissions_tracker import (
     EmissionsTracker,
     OfflineEmissionsTracker,
@@ -39,21 +38,21 @@ def tearDown(self) -> None:
     def test_force_constant_mode_online(self):
         """Test force_mode_constant parameter with online tracker"""
         tracker = EmissionsTracker(
-            output_dir=self.emissions_path, 
+            output_dir=self.emissions_path,
             output_file=self.emissions_file,
-            force_mode_constant=True
+            force_mode_constant=True,
         )
         tracker.start()
         light_computation(run_time_secs=1)
         emissions = tracker.stop()
-        
+
         # Check that emissions were calculated
         assert isinstance(emissions, float)
         self.assertNotEqual(emissions, 0.0)
-        
+
         # Verify output file was created
         self.verify_output_file(self.emissions_file_path)
-        
+
         # Check CSV content shows constant mode
         df = pd.read_csv(self.emissions_file_path)
         # The cpu_power should be a constant value (not varying like in load mode)
@@ -65,12 +64,12 @@ def test_force_constant_mode_offline(self):
             country_iso_code="USA",
             output_dir=self.emissions_path,
             output_file=self.emissions_file,
-            force_mode_constant=True
+            force_mode_constant=True,
         )
         tracker.start()
         light_computation(run_time_secs=1)
         emissions = tracker.stop()
-        
+
         assert isinstance(emissions, float)
         self.assertNotEqual(emissions, 0.0)
         self.verify_output_file(self.emissions_file_path)
@@ -82,15 +81,15 @@ def test_force_constant_mode_with_custom_cpu_power(self):
             output_dir=self.emissions_path,
             output_file=self.emissions_file,
             force_mode_constant=True,
-            force_cpu_power=custom_cpu_power
+            force_cpu_power=custom_cpu_power,
         )
         tracker.start()
         light_computation(run_time_secs=1)
         emissions = tracker.stop()
-        
+
         assert isinstance(emissions, float)
         self.assertNotEqual(emissions, 0.0)
-        
+
         # Check that the custom CPU power was used
         df = pd.read_csv(self.emissions_file_path)
         # CPU power should be 50% of the TDP (constant mode assumption)
@@ -104,12 +103,12 @@ def test_force_constant_mode_without_psutil(self):
         tracker = EmissionsTracker(
             output_dir=self.emissions_path,
             output_file=self.emissions_file,
-            force_mode_constant=True
+            force_mode_constant=True,
         )
         tracker.start()
         light_computation(run_time_secs=1)
         emissions = tracker.stop()
-        
+
         assert isinstance(emissions, float)
         self.assertNotEqual(emissions, 0.0)
         self.verify_output_file(self.emissions_file_path)
@@ -120,12 +119,12 @@ def test_force_constant_mode_takes_precedence_over_cpu_load(self):
             output_dir=self.emissions_path,
             output_file=self.emissions_file,
             force_mode_constant=True,
-            force_mode_cpu_load=True  # This should be ignored
+            force_mode_cpu_load=True,  # This should be ignored
         )
         tracker.start()
         light_computation(run_time_secs=1)
         emissions = tracker.stop()
-        
+
         assert isinstance(emissions, float)
         self.assertNotEqual(emissions, 0.0)
         self.verify_output_file(self.emissions_file_path)
@@ -135,7 +134,7 @@ def verify_output_file(self, file_path: str) -> None:
         with open(file_path, "r") as f:
             lines = [line.rstrip() for line in f]
         assert len(lines) == 2  # Header + 1 data row
-        
+
         # Check that it's a valid CSV with expected columns
         df = pd.read_csv(file_path)
         expected_columns = ["emissions", "cpu_power", "cpu_energy"]
@@ -144,4 +143,4 @@ def verify_output_file(self, file_path: str) -> None:
 
 
 if __name__ == "__main__":
-    unittest.main()
\ No newline at end of file
+    unittest.main()

From 4573735cf817c544246a9f0ce88910aa29ba70ce Mon Sep 17 00:00:00 2001
From: benoit-cty <4-benoit-cty@users.noreply.git.leximpact.dev>
Date: Wed, 24 Sep 2025 19:45:53 +0200
Subject: [PATCH 3/8] Lint

---
 codecarbon/core/cpu.py              | 1 +
 codecarbon/core/resource_tracker.py | 8 +++++---
 codecarbon/core/util.py             | 3 ++-
 3 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/codecarbon/core/cpu.py b/codecarbon/core/cpu.py
index 1c230294d..ad74e4481 100644
--- a/codecarbon/core/cpu.py
+++ b/codecarbon/core/cpu.py
@@ -16,6 +16,7 @@
 
 try:
     import psutil
+
     PSUTIL_AVAILABLE = True
 except ImportError:
     PSUTIL_AVAILABLE = False
diff --git a/codecarbon/core/resource_tracker.py b/codecarbon/core/resource_tracker.py
index 9a2088cbc..e158c4437 100644
--- a/codecarbon/core/resource_tracker.py
+++ b/codecarbon/core/resource_tracker.py
@@ -140,10 +140,12 @@ def _setup_fallback_tracking(self, tdp, max_power):
             max_power = self.tracker._force_cpu_power
         else:
             max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
-        
+
         # Check for forced constant mode first
         if self.tracker._conf.get("force_mode_constant", False):
-            logger.info("Force constant mode requested - bypassing psutil and using constant CPU power")
+            logger.info(
+                "Force constant mode requested - bypassing psutil and using constant CPU power"
+            )
             model = tdp.model
             if max_power is None and self.tracker._force_cpu_power:
                 max_power = self.tracker._force_cpu_power
@@ -158,7 +160,7 @@ def _setup_fallback_tracking(self, tdp, max_power):
             )
             self.tracker._hardware.append(hardware_cpu)
             return
-            
+
         if self.tracker._conf.get("force_mode_cpu_load", False) and (
             tdp.tdp is not None or self.tracker._force_cpu_power is not None
         ):
diff --git a/codecarbon/core/util.py b/codecarbon/core/util.py
index 3121a0660..afa65e054 100644
--- a/codecarbon/core/util.py
+++ b/codecarbon/core/util.py
@@ -12,6 +12,7 @@
 
 try:
     import psutil
+
     PSUTIL_AVAILABLE = True
 except ImportError:
     PSUTIL_AVAILABLE = False
@@ -157,7 +158,7 @@ def count_cpus() -> int:
         # Fallback to using os.cpu_count() or physical CPU count
         cpu_count = os.cpu_count()
         return cpu_count if cpu_count is not None else 1
-    
+
     if SLURM_JOB_ID is None:
         return psutil.cpu_count()
 

From fb9c3a97ec1cde90ec0234085cd8c006a08da0f4 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Thu, 21 May 2026 21:52:46 +0200
Subject: [PATCH 4/8] Fix rebase code

---
 codecarbon/core/resource_tracker.py | 5 -----
 1 file changed, 5 deletions(-)

diff --git a/codecarbon/core/resource_tracker.py b/codecarbon/core/resource_tracker.py
index e158c4437..42b28d2e8 100644
--- a/codecarbon/core/resource_tracker.py
+++ b/codecarbon/core/resource_tracker.py
@@ -136,11 +136,6 @@ def _setup_fallback_tracking(self, tdp, max_power):
         logger.info(f"CPU Model on constant consumption mode: {model}")
         self.tracker._conf["cpu_model"] = model
 
-        if tdp:
-            max_power = self.tracker._force_cpu_power
-        else:
-            max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
-
         # Check for forced constant mode first
         if self.tracker._conf.get("force_mode_constant", False):
             logger.info(

From 068f08b741f18be5c4dabb6c415c671344a1bec5 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Fri, 22 May 2026 23:32:43 +0200
Subject: [PATCH 5/8] Fix precedence of force_mode_constant

---
 codecarbon/core/resource_tracker.py | 18 +++++++++++
 tests/test_resource_tracker.py      | 49 +++++++++++++++++++++++++++++
 2 files changed, 67 insertions(+)

diff --git a/codecarbon/core/resource_tracker.py b/codecarbon/core/resource_tracker.py
index 42b28d2e8..66229ea47 100644
--- a/codecarbon/core/resource_tracker.py
+++ b/codecarbon/core/resource_tracker.py
@@ -214,6 +214,24 @@ def set_CPU_tracking(self):
             self.cpu_tracker = "User Input TDP constant"
             max_power = self.tracker._force_cpu_power
 
+        # Force constant mode takes precedence over every other CPU tracking backend.
+        if self.tracker._conf.get("force_mode_constant", False):
+            if tdp is None:
+                tdp = cpu.TDP()
+            if max_power is None:
+                max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
+
+            logger.info(
+                "Force constant mode requested - bypassing dynamic CPU power backends"
+            )
+            self.cpu_tracker = "TDP constant"
+            self.tracker._conf["cpu_model"] = tdp.model
+            hardware_cpu = CPU.from_utils(
+                self.tracker._output_dir, "constant", tdp.model, max_power
+            )
+            self.tracker._hardware.append(hardware_cpu)
+            return
+
         # Try force CPU load mode if requested
         if self.tracker._conf.get("force_mode_cpu_load", False):
             if tdp is None:
diff --git a/tests/test_resource_tracker.py b/tests/test_resource_tracker.py
index 632aee464..c79d57744 100644
--- a/tests/test_resource_tracker.py
+++ b/tests/test_resource_tracker.py
@@ -214,6 +214,55 @@ def test_set_cpu_tracking_force_mode_uses_cpu_load_and_returns():
     mock_setup.assert_called_once_with(fake_tdp, 80)
 
 
+def test_set_cpu_tracking_force_mode_constant_takes_precedence_over_all_backends():
+    tracker = make_tracker(
+        _conf={
+            "cpu_physical_count": 4,
+            "force_mode_constant": True,
+            "force_mode_cpu_load": True,
+        }
+    )
+    resource_tracker = ResourceTracker(tracker)
+    fake_tdp = SimpleNamespace(tdp=20, model="CPU")
+    fake_cpu = MagicMock()
+
+    with (
+        patch("codecarbon.core.resource_tracker.cpu.TDP", return_value=fake_tdp),
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=fake_cpu
+        ) as mock_from_utils,
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_powergadget_available",
+            return_value=True,
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_rapl_available", return_value=True
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.powermetrics.is_powermetrics_available",
+            return_value=True,
+        ),
+        patch.object(resource_tracker, "_setup_cpu_load_mode") as mock_setup_cpu_load,
+        patch.object(
+            resource_tracker, "_setup_power_gadget"
+        ) as mock_setup_power_gadget,
+        patch.object(resource_tracker, "_setup_rapl") as mock_setup_rapl,
+        patch.object(
+            resource_tracker, "_setup_powermetrics"
+        ) as mock_setup_powermetrics,
+    ):
+        resource_tracker.set_CPU_tracking()
+
+    mock_from_utils.assert_called_once_with("out", "constant", "CPU", 80)
+    mock_setup_cpu_load.assert_not_called()
+    mock_setup_power_gadget.assert_not_called()
+    mock_setup_rapl.assert_not_called()
+    mock_setup_powermetrics.assert_not_called()
+    assert resource_tracker.cpu_tracker == "TDP constant"
+    assert tracker._conf["cpu_model"] == "CPU"
+    assert tracker._hardware == [fake_cpu]
+
+
 def test_set_cpu_tracking_prefers_power_gadget():
     tracker = make_tracker()
     resource_tracker = ResourceTracker(tracker)

From be53b27925fcf2472f12b0c3fb38133616751e38 Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Sat, 23 May 2026 16:46:48 +0200
Subject: [PATCH 6/8] fix test

---
 codecarbon/core/resource_tracker.py | 26 ++++++++++++++++++--------
 tests/test_resource_tracker.py      | 25 +++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 8 deletions(-)

diff --git a/codecarbon/core/resource_tracker.py b/codecarbon/core/resource_tracker.py
index 66229ea47..326d8536d 100644
--- a/codecarbon/core/resource_tracker.py
+++ b/codecarbon/core/resource_tracker.py
@@ -127,11 +127,12 @@ def _setup_fallback_tracking(self, tdp, max_power):
         self.cpu_tracker = "TDP constant"
         model = tdp.model
 
-        if (max_power is None) and self.tracker._force_cpu_power:
+        if self.tracker._force_cpu_power is not None:
             user_input_power = self.tracker._force_cpu_power
             logger.debug(f"Using user input TDP: {user_input_power} W")
             self.cpu_tracker = "User Input TDP constant"
-            max_power = user_input_power
+            if max_power is None:
+                max_power = user_input_power
 
         logger.info(f"CPU Model on constant consumption mode: {model}")
         self.tracker._conf["cpu_model"] = model
@@ -178,7 +179,8 @@ def _setup_fallback_tracking(self, tdp, max_power):
                 hardware_cpu = CPU.from_utils(
                     self.tracker._output_dir, "constant", model, max_power
                 )
-                self.cpu_tracker = "global constant"
+                if max_power is None:
+                    self.cpu_tracker = "global constant"
             self.tracker._hardware.append(hardware_cpu)
         else:
             if cpu.is_psutil_available():
@@ -194,11 +196,19 @@ def _setup_fallback_tracking(self, tdp, max_power):
                 )
                 self.cpu_tracker = MODE_CPU_LOAD
             else:
-                logger.warning(
-                    "Failed to match CPU TDP constant. Falling back on a global constant."
-                )
-                self.cpu_tracker = "global constant"
-                hardware_cpu = CPU.from_utils(self.tracker._output_dir, "constant")
+                if max_power is None:
+                    logger.warning(
+                        "Failed to match CPU TDP constant. Falling back on a global constant."
+                    )
+                    self.cpu_tracker = "global constant"
+                    hardware_cpu = CPU.from_utils(self.tracker._output_dir, "constant")
+                else:
+                    logger.warning(
+                        "No CPU tracking mode found. Falling back on CPU constant mode."
+                    )
+                    hardware_cpu = CPU.from_utils(
+                        self.tracker._output_dir, "constant", model, max_power
+                    )
             self.tracker._hardware.append(hardware_cpu)
 
     def set_CPU_tracking(self):
diff --git a/tests/test_resource_tracker.py b/tests/test_resource_tracker.py
index c79d57744..60600e694 100644
--- a/tests/test_resource_tracker.py
+++ b/tests/test_resource_tracker.py
@@ -198,6 +198,31 @@ def __bool__(self):
     assert tracker._hardware == [hardware_cpu]
 
 
+def test_setup_fallback_tracking_uses_forced_power_without_psutil():
+    tracker = make_tracker(_force_cpu_power=42)
+    resource_tracker = ResourceTracker(tracker)
+    hardware_cpu = MagicMock()
+    tdp = SimpleNamespace(model="Unknown CPU", tdp=None)
+
+    with (
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_psutil_available",
+            return_value=False,
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=hardware_cpu
+        ) as mock_from_utils,
+        patch.object(
+            resource_tracker, "_get_install_instructions", return_value="instructions"
+        ),
+    ):
+        resource_tracker._setup_fallback_tracking(tdp, 42)
+
+    mock_from_utils.assert_called_once_with("out", "constant", "Unknown CPU", 42)
+    assert resource_tracker.cpu_tracker == "User Input TDP constant"
+    assert tracker._hardware == [hardware_cpu]
+
+
 def test_set_cpu_tracking_force_mode_uses_cpu_load_and_returns():
     tracker = make_tracker(_conf={"cpu_physical_count": 4, "force_mode_cpu_load": True})
     resource_tracker = ResourceTracker(tracker)

From 8e279f6400ffe9c94a75792c7f14f58f6d13c11f Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Wed, 27 May 2026 18:42:35 +0200
Subject: [PATCH 7/8] Add tests

---
 codecarbon/core/resource_tracker.py |  63 +++++++-----
 tests/test_core_util.py             |  29 ++++++
 tests/test_cpu.py                   |  27 +++++
 tests/test_resource_tracker.py      | 149 ++++++++++++++++++++++++++++
 4 files changed, 241 insertions(+), 27 deletions(-)

diff --git a/codecarbon/core/resource_tracker.py b/codecarbon/core/resource_tracker.py
index 326d8536d..b1a605cd1 100644
--- a/codecarbon/core/resource_tracker.py
+++ b/codecarbon/core/resource_tracker.py
@@ -143,9 +143,7 @@ def _setup_fallback_tracking(self, tdp, max_power):
                 "Force constant mode requested - bypassing psutil and using constant CPU power"
             )
             model = tdp.model
-            if max_power is None and self.tracker._force_cpu_power:
-                max_power = self.tracker._force_cpu_power
-                logger.debug(f"Using user input TDP for constant mode: {max_power} W")
+            if self.tracker._force_cpu_power is not None:
                 self.cpu_tracker = "User Input TDP constant"
             else:
                 self.cpu_tracker = "TDP constant"
@@ -211,6 +209,30 @@ def _setup_fallback_tracking(self, tdp, max_power):
                     )
             self.tracker._hardware.append(hardware_cpu)
 
+    def _resolve_tdp_and_max_power(self, cpu_number, tdp=None, max_power=None):
+        """Resolve CPU TDP object and max power estimate."""
+        if tdp is None:
+            tdp = cpu.TDP()
+        if max_power is None:
+            max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
+        return tdp, max_power
+
+    def _setup_preferred_cpu_backend(self):
+        """Set up the best available CPU backend when not forced."""
+        if cpu.is_powergadget_available() and self.tracker._force_cpu_power is None:
+            self._setup_power_gadget()
+            return True
+        if cpu.is_rapl_available() and self.tracker._force_cpu_power is None:
+            self._setup_rapl()
+            return True
+        if (
+            powermetrics.is_powermetrics_available()
+            and self.tracker._force_cpu_power is None
+        ):
+            self._setup_powermetrics()
+            return True
+        return False
+
     def set_CPU_tracking(self):
         logger.info("[setup] CPU Tracking...")
         cpu_number = self.tracker._conf.get("cpu_physical_count")
@@ -226,15 +248,15 @@ def set_CPU_tracking(self):
 
         # Force constant mode takes precedence over every other CPU tracking backend.
         if self.tracker._conf.get("force_mode_constant", False):
-            if tdp is None:
-                tdp = cpu.TDP()
-            if max_power is None:
-                max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
+            tdp, max_power = self._resolve_tdp_and_max_power(cpu_number, tdp, max_power)
 
             logger.info(
                 "Force constant mode requested - bypassing dynamic CPU power backends"
             )
-            self.cpu_tracker = "TDP constant"
+            if self.tracker._force_cpu_power is not None:
+                self.cpu_tracker = "User Input TDP constant"
+            else:
+                self.cpu_tracker = "TDP constant"
             self.tracker._conf["cpu_model"] = tdp.model
             hardware_cpu = CPU.from_utils(
                 self.tracker._output_dir, "constant", tdp.model, max_power
@@ -244,30 +266,17 @@ def set_CPU_tracking(self):
 
         # Try force CPU load mode if requested
         if self.tracker._conf.get("force_mode_cpu_load", False):
-            if tdp is None:
-                tdp = cpu.TDP()
-            if max_power is None:
-                max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
+            tdp, max_power = self._resolve_tdp_and_max_power(cpu_number, tdp, max_power)
             if tdp.tdp is not None or self.tracker._force_cpu_power is not None:
                 if self._setup_cpu_load_mode(tdp, max_power):
                     return
 
         # Try various tracking methods in order of preference
-        if cpu.is_powergadget_available() and self.tracker._force_cpu_power is None:
-            self._setup_power_gadget()
-        elif cpu.is_rapl_available() and self.tracker._force_cpu_power is None:
-            self._setup_rapl()
-        elif (
-            powermetrics.is_powermetrics_available()
-            and self.tracker._force_cpu_power is None
-        ):
-            self._setup_powermetrics()
-        else:
-            if tdp is None:
-                tdp = cpu.TDP()
-            if max_power is None:
-                max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None
-            self._setup_fallback_tracking(tdp, max_power)
+        if self._setup_preferred_cpu_backend():
+            return
+
+        tdp, max_power = self._resolve_tdp_and_max_power(cpu_number, tdp, max_power)
+        self._setup_fallback_tracking(tdp, max_power)
 
     def set_GPU_tracking(self):
         logger.info("[setup] GPU Tracking...")
diff --git a/tests/test_core_util.py b/tests/test_core_util.py
index 4eb1ce2a7..406d176fa 100644
--- a/tests/test_core_util.py
+++ b/tests/test_core_util.py
@@ -1,3 +1,5 @@
+import builtins
+import importlib.util
 import shutil
 import tempfile
 from unittest import mock
@@ -13,6 +15,21 @@
 )
 
 
+def _load_module_without_psutil(module_name, file_path):
+    real_import = builtins.__import__
+
+    def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name == "psutil":
+            raise ImportError("psutil unavailable for test")
+        return real_import(name, globals, locals, fromlist, level)
+
+    spec = importlib.util.spec_from_file_location(module_name, file_path)
+    module = importlib.util.module_from_spec(spec)
+    with mock.patch("builtins.__import__", side_effect=fake_import):
+        spec.loader.exec_module(module)
+    return module
+
+
 def test_detect_cpu_model_caching():
     """Test that detect_cpu_model() results are cached."""
     # Clear cache to ensure clean state
@@ -102,6 +119,18 @@ def test_count_cpus_no_slurm():
             assert count_cpus() == 4
 
 
+def test_util_module_import_without_psutil_uses_cpu_count_fallback():
+    util_module = _load_module_without_psutil(
+        "codecarbon.core.util_no_psutil_test",
+        resolve_path("/home/benoit/CODECARBON/codecarbon/codecarbon/core/util.py"),
+    )
+
+    assert util_module.PSUTIL_AVAILABLE is False
+    assert util_module.psutil is None
+    with mock.patch.object(util_module.os, "cpu_count", return_value=6):
+        assert util_module.count_cpus() == 6
+
+
 def test_count_cpus_slurm():
     with mock.patch("codecarbon.core.util.SLURM_JOB_ID", "12345"):
         with mock.patch(
diff --git a/tests/test_cpu.py b/tests/test_cpu.py
index 1e1308812..7850267d8 100644
--- a/tests/test_cpu.py
+++ b/tests/test_cpu.py
@@ -1,3 +1,5 @@
+import builtins
+import importlib.util
 import os
 import subprocess
 import sys
@@ -30,7 +32,32 @@
 from codecarbon.input import DataSource
 
 
+def _load_module_without_psutil(module_name, file_path):
+    real_import = builtins.__import__
+
+    def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name == "psutil":
+            raise ImportError("psutil unavailable for test")
+        return real_import(name, globals, locals, fromlist, level)
+
+    spec = importlib.util.spec_from_file_location(module_name, file_path)
+    module = importlib.util.module_from_spec(spec)
+    with mock.patch("builtins.__import__", side_effect=fake_import):
+        spec.loader.exec_module(module)
+    return module
+
+
 class TestCPU(unittest.TestCase):
+    def test_cpu_module_import_without_psutil_sets_fallback_flag(self):
+        cpu_module = _load_module_without_psutil(
+            "codecarbon.core.cpu_no_psutil_test",
+            sys.modules["codecarbon.core.cpu"].__file__,
+        )
+
+        self.assertFalse(cpu_module.PSUTIL_AVAILABLE)
+        self.assertIsNone(cpu_module.psutil)
+        self.assertFalse(cpu_module.is_psutil_available())
+
     @mock.patch("codecarbon.core.cpu.IntelPowerGadget", side_effect=Exception("boom"))
     def test_is_powergadget_available_returns_false_on_exception(
         self, mock_powergadget
diff --git a/tests/test_resource_tracker.py b/tests/test_resource_tracker.py
index 60600e694..186a36e42 100644
--- a/tests/test_resource_tracker.py
+++ b/tests/test_resource_tracker.py
@@ -223,6 +223,109 @@ def test_setup_fallback_tracking_uses_forced_power_without_psutil():
     assert tracker._hardware == [hardware_cpu]
 
 
+def test_setup_fallback_tracking_force_mode_constant_uses_constant_cpu_with_user_power():
+    tracker = make_tracker(
+        _conf={"cpu_physical_count": 2, "force_mode_constant": True},
+        _force_cpu_power=42,
+    )
+    resource_tracker = ResourceTracker(tracker)
+    hardware_cpu = MagicMock()
+    tdp = SimpleNamespace(model="Forced CPU", tdp=None)
+
+    with (
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=hardware_cpu
+        ) as mock_from_utils,
+        patch.object(
+            resource_tracker, "_get_install_instructions", return_value="instructions"
+        ),
+    ):
+        resource_tracker._setup_fallback_tracking(tdp, None)
+
+    mock_from_utils.assert_called_once_with("out", "constant", "Forced CPU", 42)
+    assert resource_tracker.cpu_tracker == "User Input TDP constant"
+    assert tracker._conf["cpu_model"] == "Forced CPU"
+    assert tracker._hardware == [hardware_cpu]
+
+
+def test_setup_fallback_tracking_force_mode_constant_preserves_existing_max_power():
+    tracker = make_tracker(_conf={"cpu_physical_count": 2, "force_mode_constant": True})
+    resource_tracker = ResourceTracker(tracker)
+    hardware_cpu = MagicMock()
+    tdp = SimpleNamespace(model="Forced CPU", tdp=75)
+
+    with (
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=hardware_cpu
+        ) as mock_from_utils,
+        patch.object(
+            resource_tracker, "_get_install_instructions", return_value="instructions"
+        ),
+    ):
+        resource_tracker._setup_fallback_tracking(tdp, 150)
+
+    mock_from_utils.assert_called_once_with("out", "constant", "Forced CPU", 150)
+    assert resource_tracker.cpu_tracker == "TDP constant"
+    assert tracker._conf["cpu_model"] == "Forced CPU"
+    assert tracker._hardware == [hardware_cpu]
+
+
+def test_setup_fallback_tracking_force_mode_cpu_load_uses_cpu_load_when_psutil_available():
+    tracker = make_tracker(_conf={"cpu_physical_count": 2, "force_mode_cpu_load": True})
+    resource_tracker = ResourceTracker(tracker)
+    hardware_cpu = MagicMock()
+    tdp = SimpleNamespace(model="Load CPU", tdp=75)
+
+    with (
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_psutil_available",
+            return_value=True,
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=hardware_cpu
+        ) as mock_from_utils,
+        patch.object(
+            resource_tracker, "_get_install_instructions", return_value="instructions"
+        ),
+    ):
+        resource_tracker._setup_fallback_tracking(tdp, 150)
+
+    mock_from_utils.assert_called_once_with(
+        "out",
+        MODE_CPU_LOAD,
+        "Load CPU",
+        150,
+        tracking_mode="machine",
+    )
+    assert resource_tracker.cpu_tracker == MODE_CPU_LOAD
+    assert tracker._hardware == [hardware_cpu]
+
+
+def test_setup_fallback_tracking_force_mode_cpu_load_uses_constant_without_psutil():
+    tracker = make_tracker(_conf={"cpu_physical_count": 2, "force_mode_cpu_load": True})
+    resource_tracker = ResourceTracker(tracker)
+    hardware_cpu = MagicMock()
+    tdp = SimpleNamespace(model="Load CPU", tdp=75)
+
+    with (
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_psutil_available",
+            return_value=False,
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=hardware_cpu
+        ) as mock_from_utils,
+        patch.object(
+            resource_tracker, "_get_install_instructions", return_value="instructions"
+        ),
+    ):
+        resource_tracker._setup_fallback_tracking(tdp, None)
+
+    mock_from_utils.assert_called_once_with("out", "constant", "Load CPU", None)
+    assert resource_tracker.cpu_tracker == "global constant"
+    assert tracker._hardware == [hardware_cpu]
+
+
 def test_set_cpu_tracking_force_mode_uses_cpu_load_and_returns():
     tracker = make_tracker(_conf={"cpu_physical_count": 4, "force_mode_cpu_load": True})
     resource_tracker = ResourceTracker(tracker)
@@ -288,6 +391,29 @@ def test_set_cpu_tracking_force_mode_constant_takes_precedence_over_all_backends
     assert tracker._hardware == [fake_cpu]
 
 
+def test_set_cpu_tracking_force_mode_constant_uses_user_power_without_recomputing():
+    tracker = make_tracker(
+        _conf={"cpu_physical_count": 4, "force_mode_constant": True},
+        _force_cpu_power=55,
+    )
+    resource_tracker = ResourceTracker(tracker)
+    fake_tdp = SimpleNamespace(tdp=None, model="CPU")
+    fake_cpu = MagicMock()
+
+    with (
+        patch("codecarbon.core.resource_tracker.cpu.TDP", return_value=fake_tdp),
+        patch(
+            "codecarbon.core.resource_tracker.CPU.from_utils", return_value=fake_cpu
+        ) as mock_from_utils,
+    ):
+        resource_tracker.set_CPU_tracking()
+
+    mock_from_utils.assert_called_once_with("out", "constant", "CPU", 55)
+    assert resource_tracker.cpu_tracker == "User Input TDP constant"
+    assert tracker._conf["cpu_model"] == "CPU"
+    assert tracker._hardware == [fake_cpu]
+
+
 def test_set_cpu_tracking_prefers_power_gadget():
     tracker = make_tracker()
     resource_tracker = ResourceTracker(tracker)
@@ -334,6 +460,29 @@ def test_set_cpu_tracking_prefers_rapl_before_powermetrics():
     mock_rapl.assert_called_once_with()
 
 
+def test_set_cpu_tracking_prefers_powermetrics_when_other_backends_unavailable():
+    tracker = make_tracker()
+    resource_tracker = ResourceTracker(tracker)
+
+    with (
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_powergadget_available",
+            return_value=False,
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.cpu.is_rapl_available", return_value=False
+        ),
+        patch(
+            "codecarbon.core.resource_tracker.powermetrics.is_powermetrics_available",
+            return_value=True,
+        ),
+        patch.object(resource_tracker, "_setup_powermetrics") as mock_powermetrics,
+    ):
+        resource_tracker.set_CPU_tracking()
+
+    mock_powermetrics.assert_called_once_with()
+
+
 def test_set_cpu_tracking_falls_back_when_forced_power_is_set():
     tracker = make_tracker(_force_cpu_power=42)
     resource_tracker = ResourceTracker(tracker)

From a3fab12dc5479d65eac2d741c8d2fd36bb92532e Mon Sep 17 00:00:00 2001
From: benoit-cty <anne@onyme.fr>
Date: Wed, 27 May 2026 22:07:35 +0200
Subject: [PATCH 8/8] fix path

---
 tests/test_core_util.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_core_util.py b/tests/test_core_util.py
index 406d176fa..b88886ae4 100644
--- a/tests/test_core_util.py
+++ b/tests/test_core_util.py
@@ -122,7 +122,7 @@ def test_count_cpus_no_slurm():
 def test_util_module_import_without_psutil_uses_cpu_count_fallback():
     util_module = _load_module_without_psutil(
         "codecarbon.core.util_no_psutil_test",
-        resolve_path("/home/benoit/CODECARBON/codecarbon/codecarbon/core/util.py"),
+        __import__("codecarbon.core.util", fromlist=["util"]).__file__,
     )
 
     assert util_module.PSUTIL_AVAILABLE is False