Spaces:
Sleeping
Sleeping
| from collections import Counter | |
| from typing import List, Union | |
| from codecarbon.core import cpu, gpu, powermetrics | |
| from codecarbon.core.config import parse_gpu_ids | |
| from codecarbon.core.util import detect_cpu_model, is_linux_os, is_mac_os, is_windows_os | |
| from codecarbon.external.hardware import CPU, GPU, MODE_CPU_LOAD, AppleSiliconChip | |
| from codecarbon.external.logger import logger | |
| from codecarbon.external.ram import RAM | |
| class ResourceTracker: | |
| cpu_tracker = gpu_tracker = ram_tracker = "Unspecified" | |
| def __init__(self, tracker): | |
| self.tracker = tracker | |
| def set_RAM_tracking(self): | |
| logger.info("[setup] RAM Tracking...") | |
| if self.tracker._force_ram_power is not None: | |
| self.ram_tracker = ( | |
| f"User specified constant: {self.tracker._force_ram_power} Watts" | |
| ) | |
| logger.info( | |
| f"Using user-provided RAM power: {self.tracker._force_ram_power} Watts" | |
| ) | |
| else: | |
| self.ram_tracker = "RAM power estimation model" | |
| ram = RAM( | |
| tracking_mode=self.tracker._tracking_mode, | |
| force_ram_power=self.tracker._force_ram_power, | |
| ) | |
| self.tracker._conf["ram_total_size"] = ram.machine_memory_GB | |
| self.tracker._hardware: List[Union[RAM, CPU, GPU, AppleSiliconChip]] = [ram] | |
| def set_CPU_tracking(self): | |
| logger.info("[setup] CPU Tracking...") | |
| cpu_number = self.tracker._conf.get("cpu_physical_count") | |
| tdp = cpu.TDP() | |
| if self.tracker._force_cpu_power is not None: | |
| logger.info( | |
| f"Using user-provided CPU power: {self.tracker._force_cpu_power} Watts" | |
| ) | |
| self.cpu_tracker = "User Input TDP constant" | |
| max_power = self.tracker._force_cpu_power | |
| else: | |
| max_power = tdp.tdp * cpu_number if tdp.tdp is not None else None | |
| 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(): | |
| # Register a CPU with MODE_CPU_LOAD | |
| model = tdp.model | |
| hardware_cpu = CPU.from_utils( | |
| self.tracker._output_dir, | |
| MODE_CPU_LOAD, | |
| model, | |
| max_power, | |
| tracking_mode=self.tracker._tracking_mode, | |
| ) | |
| self.cpu_tracker = MODE_CPU_LOAD | |
| self.tracker._conf["cpu_model"] = hardware_cpu.get_model() | |
| self.tracker._hardware.append(hardware_cpu) | |
| return | |
| else: | |
| logger.warning( | |
| "Force CPU load mode requested but psutil is not available." | |
| ) | |
| if cpu.is_powergadget_available() and self.tracker._force_cpu_power is None: | |
| logger.info("Tracking Intel CPU via Power Gadget") | |
| self.cpu_tracker = "Power Gadget" | |
| hardware_cpu = CPU.from_utils( | |
| self.tracker._output_dir, "intel_power_gadget" | |
| ) | |
| self.tracker._hardware.append(hardware_cpu) | |
| self.tracker._conf["cpu_model"] = hardware_cpu.get_model() | |
| elif cpu.is_rapl_available(): | |
| logger.info("Tracking Intel CPU via RAPL interface") | |
| self.cpu_tracker = "RAPL" | |
| hardware_cpu = CPU.from_utils( | |
| output_dir=self.tracker._output_dir, mode="intel_rapl" | |
| ) | |
| self.tracker._hardware.append(hardware_cpu) | |
| self.tracker._conf["cpu_model"] = hardware_cpu.get_model() | |
| if "AMD Ryzen Threadripper" in self.tracker._conf["cpu_model"]: | |
| logger.warning( | |
| "The RAPL energy and power reported is divided by 2 for all 'AMD Ryzen Threadripper' as it seems to give better results." | |
| ) | |
| # change code to check if powermetrics needs to be installed or just sudo setup | |
| elif ( | |
| powermetrics.is_powermetrics_available() | |
| and self.tracker._force_cpu_power is None | |
| ): | |
| logger.info("Tracking Apple CPU and GPU via PowerMetrics") | |
| self.gpu_tracker = "PowerMetrics" | |
| self.cpu_tracker = "PowerMetrics" | |
| hardware_cpu = AppleSiliconChip.from_utils( | |
| self.tracker._output_dir, chip_part="CPU" | |
| ) | |
| self.tracker._hardware.append(hardware_cpu) | |
| self.tracker._conf["cpu_model"] = hardware_cpu.get_model() | |
| hardware_gpu = AppleSiliconChip.from_utils( | |
| self.tracker._output_dir, chip_part="GPU" | |
| ) | |
| self.tracker._hardware.append(hardware_gpu) | |
| self.tracker._conf["gpu_model"] = hardware_gpu.get_model() | |
| self.tracker._conf["gpu_count"] = 1 | |
| else: | |
| # Explain what to install to increase accuracy | |
| cpu_tracking_install_instructions = "" | |
| if is_mac_os(): | |
| if ( | |
| "M1" in detect_cpu_model() | |
| or "M2" in detect_cpu_model() | |
| or "M3" in detect_cpu_model() | |
| ): | |
| cpu_tracking_install_instructions = "" | |
| cpu_tracking_install_instructions = "Mac OS and ARM processor detected: Please enable PowerMetrics sudo to measure CPU" | |
| else: | |
| cpu_tracking_install_instructions = "Mac OS detected: Please install Intel Power Gadget or enable PowerMetrics sudo to measure CPU" | |
| elif is_windows_os(): | |
| cpu_tracking_install_instructions = "Windows OS detected: Please install Intel Power Gadget to measure CPU" | |
| elif is_linux_os(): | |
| cpu_tracking_install_instructions = "Linux OS detected: Please ensure RAPL files exist at /sys/class/powercap/intel-rapl/subsystem to measure CPU" | |
| logger.warning( | |
| f"No CPU tracking mode found. Falling back on estimation based on TDP for CPU. \n {cpu_tracking_install_instructions}\n" | |
| ) | |
| self.cpu_tracker = "TDP constant" | |
| model = tdp.model | |
| if (max_power is None) and self.tracker._force_cpu_power: | |
| # We haven't been able to calculate CPU power but user has input a default one. We use it | |
| 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 | |
| logger.info(f"CPU Model on constant consumption mode: {model}") | |
| self.tracker._conf["cpu_model"] = model | |
| if tdp: | |
| if cpu.is_psutil_available(): | |
| logger.warning( | |
| "No CPU tracking mode found. Falling back on CPU load mode." | |
| ) | |
| hardware_cpu = CPU.from_utils( | |
| self.tracker._output_dir, | |
| MODE_CPU_LOAD, | |
| model, | |
| max_power, | |
| tracking_mode=self.tracker._tracking_mode, | |
| ) | |
| self.cpu_tracker = MODE_CPU_LOAD | |
| 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.cpu_tracker = "global constant" | |
| self.tracker._hardware.append(hardware_cpu) | |
| else: | |
| if cpu.is_psutil_available(): | |
| logger.warning( | |
| "Failed to match CPU TDP constant. Falling back on CPU load mode." | |
| ) | |
| hardware_cpu = CPU.from_utils( | |
| self.tracker._output_dir, | |
| MODE_CPU_LOAD, | |
| model, | |
| max_power, | |
| tracking_mode=self.tracker._tracking_mode, | |
| ) | |
| 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") | |
| self.tracker._hardware.append(hardware_cpu) | |
| def set_GPU_tracking(self): | |
| logger.info("[setup] GPU Tracking...") | |
| if self.tracker._gpu_ids: | |
| # If _gpu_ids is a string or a list of int, parse it to a list of ints | |
| if isinstance(self.tracker._gpu_ids, str) or ( | |
| isinstance(self.tracker._gpu_ids, list) | |
| and all(isinstance(gpu_id, int) for gpu_id in self.tracker._gpu_ids) | |
| ): | |
| self.tracker._gpu_ids: List = parse_gpu_ids(self.tracker._gpu_ids) | |
| self.tracker._conf["gpu_ids"] = self.tracker._gpu_ids | |
| self.tracker._conf["gpu_count"] = len(self.tracker._gpu_ids) | |
| else: | |
| logger.warning( | |
| "Invalid gpu_ids format. Expected a string or a list of ints." | |
| ) | |
| if gpu.is_gpu_details_available(): | |
| logger.info("Tracking Nvidia GPU via pynvml") | |
| gpu_devices = GPU.from_utils(self.tracker._gpu_ids) | |
| self.tracker._hardware.append(gpu_devices) | |
| gpu_names = [n["name"] for n in gpu_devices.devices.get_gpu_static_info()] | |
| gpu_names_dict = Counter(gpu_names) | |
| self.tracker._conf["gpu_model"] = "".join( | |
| [f"{i} x {name}" for name, i in gpu_names_dict.items()] | |
| ) | |
| if self.tracker._conf.get("gpu_count") is None: | |
| self.tracker._conf["gpu_count"] = len( | |
| gpu_devices.devices.get_gpu_static_info() | |
| ) | |
| self.gpu_tracker = "pynvml" | |
| else: | |
| logger.info("No GPU found.") | |
| def set_CPU_GPU_ram_tracking(self): | |
| """ | |
| Set up CPU, GPU and RAM tracking based on the user's configuration. | |
| param tracker: BaseEmissionsTracker object | |
| """ | |
| self.set_RAM_tracking() | |
| self.set_CPU_tracking() | |
| self.set_GPU_tracking() | |
| logger.info( | |
| f"""The below tracking methods have been set up: | |
| RAM Tracking Method: {self.ram_tracker} | |
| CPU Tracking Method: {self.cpu_tracker} | |
| GPU Tracking Method: {self.gpu_tracker} | |
| """ | |
| ) | |