Add option for simulated CPU/GPU

Author: PedramBakh

README.md

@@ -22,7 +22,7 @@ Kindly cite our work if you use **carbontracker** in a scientific publication:
   note={arXiv:2007.03051},
   year={2020}}
 ```
-
+_
 ## Installation
 ### PyPi
 ```
@@ -66,6 +66,18 @@ Wrap any of your scripts (python, bash, etc.):
   Sets the level of verbosity.
 - `decimal_precision` (default=6):
   Desired decimal precision of reported values.
+- `sim_cpu` (default=None):
+  Name of the simulated CPU. If set, will use simulated CPU power measurements.
+- `sim_cpu_tdp` (default=None):
+  Thermal Design Power (TDP) in Watts for the simulated CPU. Required if `sim_cpu` is set.
+- `sim_cpu_util` (default=None):
+  CPU utilization factor between 0 and 1. If not set, defaults to 0.5 (50% utilization).
+- `sim_gpu` (default=None):
+  Name of the simulated GPU. If set, will use simulated GPU power measurements.
+- `sim_gpu_watts` (default=None):
+  Power consumption in Watts for the simulated GPU. Required if `sim_gpu` is set.
+- `sim_gpu_util` (default=None):
+  GPU utilization factor between 0 and 1. If not set, defaults to 0.5 (50% utilization).
 
 #### Example usage
 

carbontracker/cli.py

@@ -31,6 +31,12 @@ def main():
         --parse (path, optional): Directory containing the log files to parse.
         --report (path, optional): Generate a PDF report from a log file.
         --output (path, optional): Output path for the generated report. Defaults to 'carbon_report.pdf'
+        --sim-cpu (str, optional): Simulated CPU name (overrides detection)
+        --sim-cpu-tdp (float, optional): Simulated CPU TDP in Watts
+        --sim-cpu-util (float, optional): Simulated CPU utilization (0.0 to 1.0)
+        --sim-gpu (str, optional): Simulated GPU name (overrides detection)
+        --sim-gpu-watts (float, optional): Simulated GPU power consumption in Watts
+        --sim-gpu-util (float, optional): Simulated GPU utilization (0.0 to 1.0)
 
     Example:
         Tracking the carbon intensity of `script.py`.
@@ -41,6 +47,10 @@ def main():
 
             $ carbontracker --log_dir='./logs' --api_keys='{"electricitymaps": "API_KEY_EXAMPLE"}' python script.py
 
+        Using simulated hardware:
+
+            $ carbontracker --sim-cpu "Intel Xeon" --sim-cpu-tdp 150 --sim-gpu "NVIDIA A100" --sim-gpu-watts 400 python script.py
+
         Parsing logs:
 
             $ carbontracker --parse ./internal_logs
@@ -64,6 +74,26 @@ def main():
     cli_parser.add_argument("--report", type=str, help="Generate a PDF report from a log file.")
     cli_parser.add_argument("--output", type=str, default="carbon_report.pdf", 
                           help="Output path for the generated report.")
+    
+    # Add simulated hardware arguments
+    cli_parser.add_argument("--sim-cpu", type=str, 
+                          help="Simulated CPU name (overrides detection). REQUIRED with --sim-cpu-tdp", 
+                          default=None)
+    cli_parser.add_argument("--sim-cpu-tdp", type=float, 
+                          help="Simulated CPU TDP in Watts. REQUIRED when --sim-cpu is specified", 
+                          default=None)
+    cli_parser.add_argument("--sim-cpu-util", type=float, 
+                          help="Simulated CPU utilization (0.0 to 1.0). Defaults to 0.5 if not specified", 
+                          default=None)
+    cli_parser.add_argument("--sim-gpu", type=str, 
+                          help="Simulated GPU name (overrides detection). REQUIRED with --sim-gpu-watts", 
+                          default=None)
+    cli_parser.add_argument("--sim-gpu-watts", type=float, 
+                          help="Simulated GPU power consumption in Watts. REQUIRED when --sim-gpu is specified", 
+                          default=None)
+    cli_parser.add_argument("--sim-gpu-util", type=float, 
+                          help="Simulated GPU utilization (0.0 to 1.0). Defaults to 0.5 if not specified", 
+                          default=None)
 
     # Parse known arguments only
     known_args, remaining_args = cli_parser.parse_known_args()
@@ -82,7 +112,16 @@ def main():
     api_keys = ast.literal_eval(known_args.api_keys) if known_args.api_keys else None
 
     tracker = CarbonTracker(
-        epochs=1, log_dir=known_args.log_dir, epochs_before_pred=0, api_keys=api_keys
+        epochs=1, 
+        log_dir=known_args.log_dir, 
+        epochs_before_pred=0, 
+        api_keys=api_keys,
+        sim_cpu=known_args.sim_cpu,
+        sim_cpu_tdp=known_args.sim_cpu_tdp,
+        sim_cpu_util=known_args.sim_cpu_util,
+        sim_gpu=known_args.sim_gpu,
+        sim_gpu_watts=known_args.sim_gpu_watts,
+        sim_gpu_util=known_args.sim_gpu_util
     )
     tracker.epoch_start()
 

carbontracker/components/component.py

@@ -11,6 +11,8 @@
 from typing import Iterable, List, Union, Type, Sized
 from carbontracker.loggerutil import Logger
 import os
+from carbontracker.components.cpu.sim_cpu import SimulatedCPUHandler
+from carbontracker.components.gpu.sim_gpu import SimulatedGPUHandler
 
 COMPONENTS = [
     {
@@ -37,25 +39,64 @@ def error_by_name(name) -> Exception:
     raise exceptions.ComponentNameError()
 
 
-def handlers_by_name(name) -> List[Type[Handler]]:
+def handlers_by_name(
+    name, 
+    sim_cpu=None, 
+    sim_cpu_tdp=None, 
+    sim_cpu_util=None,
+    sim_gpu=None, 
+    sim_gpu_watts=None,
+    sim_gpu_util=None
+):
     for comp in COMPONENTS:
         if comp["name"] == name:
+            if name == "cpu" and sim_cpu and sim_cpu_tdp:
+                return [lambda pids, devices_by_pid: SimulatedCPUHandler(
+                    sim_cpu, 
+                    float(sim_cpu_tdp),
+                    float(sim_cpu_util) if sim_cpu_util is not None else 0.5
+                )]
+            elif name == "gpu" and sim_gpu and sim_gpu_watts:
+                return [lambda pids, devices_by_pid: SimulatedGPUHandler(
+                    sim_gpu, 
+                    float(sim_gpu_watts),
+                    float(sim_gpu_util) if sim_gpu_util is not None else 0.5
+                )]
             return comp["handlers"]
     raise exceptions.ComponentNameError()
 
 
 class Component:
-    def __init__(self, name: str, pids: Iterable[int], devices_by_pid: bool, logger: Logger):
+    def __init__(
+        self, 
+        name: str, 
+        pids: Iterable[int], 
+        devices_by_pid: bool, 
+        logger: Logger, 
+        sim_cpu=None,
+        sim_cpu_tdp=None,
+        sim_cpu_util=None,
+        sim_gpu=None,
+        sim_gpu_watts=None,
+        sim_gpu_util=None
+    ):
         self.name = name
         if name not in component_names():
             raise exceptions.ComponentNameError(
                 f"No component found with name '{self.name}'."
             )
         self._handler = self._determine_handler(
-            pids=pids, devices_by_pid=devices_by_pid
+            pids=pids, 
+            devices_by_pid=devices_by_pid, 
+            sim_cpu=sim_cpu,
+            sim_cpu_tdp=sim_cpu_tdp,
+            sim_cpu_util=sim_cpu_util,
+            sim_gpu=sim_gpu,
+            sim_gpu_watts=sim_gpu_watts,
+            sim_gpu_util=sim_gpu_util
         )
         self.power_usages: List[List[float]] = []
-        self.cur_epoch: int = -1  # Sentry
+        self.cur_epoch: int = -1
         self.logger = logger
 
     @property
@@ -65,11 +106,27 @@ def handler(self) -> Handler:
         return self._handler
 
     def _determine_handler(
-        self, pids: Iterable[int], devices_by_pid: bool
+        self, 
+        pids: Iterable[int], 
+        devices_by_pid: bool, 
+        sim_cpu=None,
+        sim_cpu_tdp=None,
+        sim_cpu_util=None,
+        sim_gpu=None,
+        sim_gpu_watts=None,
+        sim_gpu_util=None
     ) -> Union[Handler, None]:
-        handlers = handlers_by_name(self.name)
+        handlers = handlers_by_name(
+            self.name, 
+            sim_cpu=sim_cpu,
+            sim_cpu_tdp=sim_cpu_tdp,
+            sim_cpu_util=sim_cpu_util,
+            sim_gpu=sim_gpu,
+            sim_gpu_watts=sim_gpu_watts,
+            sim_gpu_util=sim_gpu_util
+        )
         for h in handlers:
-            handler = h(pids=pids, devices_by_pid=devices_by_pid)
+            handler = h(pids=pids, devices_by_pid=devices_by_pid) if callable(h) else h(pids=pids, devices_by_pid=devices_by_pid)
             if handler.available():
                 return handler
         return None
@@ -158,16 +215,47 @@ def shutdown(self):
 
 
 def create_components(
-    components: str, pids: Iterable[int], devices_by_pid: bool, logger: Logger
+    components: str, 
+    pids: Iterable[int], 
+    devices_by_pid: bool, 
+    logger: Logger, 
+    sim_cpu=None,
+    sim_cpu_tdp=None,
+    sim_cpu_util=None,
+    sim_gpu=None,
+    sim_gpu_watts=None,
+    sim_gpu_util=None
 ) -> List[Component]:
     components = components.strip().replace(" ", "").lower()
     if components == "all":
         return [
-            Component(name=comp_name, pids=pids, devices_by_pid=devices_by_pid, logger=logger)
+            Component(
+                name=comp_name, 
+                pids=pids, 
+                devices_by_pid=devices_by_pid, 
+                logger=logger, 
+                sim_cpu=sim_cpu,
+                sim_cpu_tdp=sim_cpu_tdp,
+                sim_cpu_util=sim_cpu_util,
+                sim_gpu=sim_gpu,
+                sim_gpu_watts=sim_gpu_watts,
+                sim_gpu_util=sim_gpu_util
+            )
             for comp_name in component_names()
         ]
     else:
         return [
-            Component(name=comp_name, pids=pids, devices_by_pid=devices_by_pid, logger=logger)
+            Component(
+                name=comp_name, 
+                pids=pids, 
+                devices_by_pid=devices_by_pid, 
+                logger=logger, 
+                sim_cpu=sim_cpu,
+                sim_cpu_tdp=sim_cpu_tdp,
+                sim_cpu_util=sim_cpu_util,
+                sim_gpu=sim_gpu,
+                sim_gpu_watts=sim_gpu_watts,
+                sim_gpu_util=sim_gpu_util
+            )
             for comp_name in components.split(",")
         ]

carbontracker/components/cpu/generic.py

@@ -68,10 +68,10 @@ def init(self):
 
         if self.tdp is None:
             self.tdp = self.average_tdp
-            logger.err_warn(f"No matching TDP found for CPU: {self.cpu_brand}. Using average TDP of {self.tdp:.2f}W as fallback.")
+            logger.err_warn(f"No matching TDP found for CPU: {self.cpu_brand}. Using average TDP of {self.tdp:.2f}W at 50% utilization as fallback.")
         else:
             self.tdp = self.tdp / 2  # 50% utilization
-            logger.err_info(f"Using TDP of {self.tdp:.2f}W for {self.cpu_brand}")
+            logger.err_info(f"Using TDP of {self.tdp:.2f}W for {self.cpu_brand} at 50% utilization")
 
     def find_matching_tdp(self) -> Optional[float]:
         # Try direct match

carbontracker/components/cpu/sim_cpu.py

@@ -0,0 +1,30 @@
+from carbontracker.components.handler import Handler
+from typing import List
+
+class SimulatedCPUHandler(Handler):
+    def __init__(self, name: str, tdp: float, utilization: float = 0.5):
+        super().__init__(pids=[], devices_by_pid=False)
+        if not isinstance(name, str) or not name.strip():
+            raise ValueError("CPU name must be a non-empty string.")
+        if tdp is None or not isinstance(tdp, (int, float)) or tdp <= 0:
+            raise ValueError("CPU TDP must be a positive number.")
+        if not isinstance(utilization, (int, float)) or not (0.0 <= utilization <= 1.0):
+            raise ValueError("CPU utilization must be between 0.0 and 1.0.")
+        self.cpu_brand = name
+        self.utilization = utilization
+        self.tdp = tdp * utilization
+
+    def devices(self) -> List[str]:
+        return [self.cpu_brand]
+
+    def available(self) -> bool:
+        return True
+
+    def power_usage(self) -> List[float]:
+        return [self.tdp]
+
+    def init(self):
+        print(f"Using simulated CPU: {self.cpu_brand} with TDP: {self.tdp:.2f}W (at {self.utilization*100:.0f}% utilization)")
+
+    def shutdown(self):
+        pass

carbontracker/components/gpu/sim_gpu.py

@@ -0,0 +1,30 @@
+from carbontracker.components.handler import Handler
+from typing import List
+
+class SimulatedGPUHandler(Handler):
+    def __init__(self, name: str, watts: float, utilization: float = 0.5):
+        super().__init__(pids=[], devices_by_pid=False)
+        if not isinstance(name, str) or not name.strip():
+            raise ValueError("GPU name must be a non-empty string.")
+        if watts is None or not isinstance(watts, (int, float)) or watts <= 0:
+            raise ValueError("GPU watts must be a positive number.")
+        if not isinstance(utilization, (int, float)) or not (0.0 <= utilization <= 1.0):
+            raise ValueError("GPU utilization must be between 0.0 and 1.0.")
+        self.gpu_brand = name
+        self.utilization = utilization
+        self.watts = watts * utilization
+
+    def devices(self) -> List[str]:
+        return [self.gpu_brand]
+
+    def available(self) -> bool:
+        return True
+
+    def power_usage(self) -> List[float]:
+        return [self.watts]
+
+    def init(self):
+        print(f"Using simulated GPU: {self.gpu_brand} with power consumption: {self.watts:.2f}W (at {self.utilization*100:.0f}% utilization)")
+
+    def shutdown(self):
+        pass