feat(codecarbon) add raspberry support

Author: prmths128

codecarbon/core/cpu.py

@@ -58,6 +58,13 @@ def is_rapl_available() -> bool:
         return False
 
 
+def is_raspberry() -> bool:
+    """
+    Check if raspberry power util is present
+    """
+    return os.path.exists("/usr/bin/vcgencmd")
+
+
 class IntelPowerGadget:
     """
     A class to interface with Intel Power Gadget for monitoring CPU power consumption on Windows and (non-Apple Silicon) macOS.

codecarbon/core/resource_tracker.py

@@ -4,7 +4,7 @@
 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, RAM, AppleSiliconChip
+from codecarbon.external.hardware import CPU, GPU, RAM, AppleSiliconChip, Raspberry
 from codecarbon.external.logger import logger
 
 
@@ -19,7 +19,9 @@ def set_RAM_tracking(self):
         self.ram_tracker = "3 Watts for 8 GB ratio constant"
         ram = RAM(tracking_mode=self.tracker._tracking_mode)
         self.tracker._conf["ram_total_size"] = ram.machine_memory_GB
-        self.tracker._hardware: List[Union[RAM, CPU, GPU, AppleSiliconChip]] = [ram]
+        self.tracker._hardware: List[
+            Union[RAM, CPU, GPU, AppleSiliconChip, Raspberry]
+        ] = [ram]
 
     def set_CPU_tracking(self):
         logger.info("[setup] CPU Tracking...")
@@ -35,6 +37,18 @@ def set_CPU_tracking(self):
             hardware = CPU.from_utils(self.tracker._output_dir, "intel_rapl")
             self.tracker._hardware.append(hardware)
             self.tracker._conf["cpu_model"] = hardware.get_model()
+        elif cpu.is_raspberry():
+            logger.info("Tracking CPU via raspberry utils")
+            self.cpu_tracker = "raspberry"
+            hardware_cpu = Raspberry.from_utils(
+                self.tracker._output_dir, chip_part="CPU"
+            )
+            self.tracker._hardware.append(hardware_cpu)
+            hardware_ram = Raspberry.from_utils(
+                self.tracker._output_dir, chip_part="RAM"
+            )
+            self.tracker._hardware.append(hardware_ram)
+            self.tracker._conf["cpu_model"] = hardware_cpu.get_model()
         # change code to check if powermetrics needs to be installed or just sudo setup
         elif (
             powermetrics.is_powermetrics_available()

codecarbon/emissions_tracker.py

@@ -20,7 +20,7 @@
 from codecarbon.core.units import Energy, Power, Time
 from codecarbon.core.util import count_cpus, suppress
 from codecarbon.external.geography import CloudMetadata, GeoMetadata
-from codecarbon.external.hardware import CPU, GPU, RAM, AppleSiliconChip
+from codecarbon.external.hardware import CPU, GPU, RAM, AppleSiliconChip, Raspberry
 from codecarbon.external.logger import logger, set_logger_format, set_logger_level
 from codecarbon.external.scheduler import PeriodicScheduler
 from codecarbon.external.task import Task
@@ -705,6 +705,21 @@ def _do_measurements(self) -> None:
                     f"Energy consumed for RAM : {self._total_ram_energy.kWh:.6f} kWh"
                     + f". RAM Power : {self._ram_power.W} W"
                 )
+            elif isinstance(hardware, Raspberry):
+                if hardware.chip_part == "CPU":
+                    self._total_cpu_energy += energy
+                    self._cpu_power = power
+                    logger.info(
+                        f"Energy consumed for all CPUs : {self._total_cpu_energy.kWh:.6f} kWh"
+                        + f". Total CPU Power : {self._cpu_power.W} W"
+                    )
+                elif hardware.chip_part == "RAM":
+                    self._total_ram_energy += energy
+                    self._ram_power = power
+                    logger.info(
+                        f"Energy consumed for RAMs : {self._total_ram_energy.kWh:.6f} kWh"
+                        + f". Total RAM Power : {self._ram_power.W} W"
+                    )
             elif isinstance(hardware, AppleSiliconChip):
                 if hardware.chip_part == "CPU":
                     self._total_cpu_energy += energy

codecarbon/external/hardware.py

@@ -6,6 +6,7 @@
 import subprocess
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from subprocess import run
 from typing import Dict, Iterable, List, Optional, Tuple
 
 import psutil
@@ -487,3 +488,99 @@ def from_utils(
                 logger.warning("Could not read AppleSiliconChip model.")
 
         return cls(output_dir=output_dir, model=model, chip_part=chip_part)
+
+
+@dataclass
+class Raspberry(BaseHardware):
+    def __init__(
+        self,
+        output_dir: str,
+        model: str,
+        chip_part: str = "CPU",
+    ):
+        if chip_part == "CPU":
+            self.WANTED_COMPONENTS = (
+                "3V7_WL_SW",
+                "3V3_SYS",
+                "1V8_SYS",
+                "1V1_SYS",
+                "0V8_SW",
+                "VDD_CORE",
+                "3V3_DAC",
+                "3V3_ADC",
+                "0V8_AON",
+            )
+        elif chip_part == "RAM":
+            self.WANTED_COMPONENTS = ("DDR_VDD2", "DDR_VDD2", "DDR_VDDQ")
+        else:
+            raise Exception("Unknown chip part", chip_part)
+
+        self._output_dir = output_dir
+        self._model = model
+        self.chip_part = chip_part
+
+    def __repr__(self) -> str:
+        return f"Raspberry ({self._model} > {self.chip_part})"
+
+    def _get_power(self) -> Power:
+        """ """
+        measure: Dict = self.get_measure()
+        return Power.from_watts(measure["power"])
+
+    def _get_energy(self, delay: Time) -> Energy:
+        """
+        Get Chip part energy deltas
+        Args:
+            chip_part (str): Chip part to get power from (Processor, GPU, etc.)
+        :return: energy in kWh
+        """
+        energy = Energy.from_power_and_time(
+            power=self._get_power(), time=Time.from_seconds(delay)
+        )
+        return energy
+
+    def total_power(self) -> Power:
+        return self._get_power()
+
+    def start(self):
+        # ...
+        ...
+
+    def get_model(self):
+        return self._model
+
+    def get_measure(self):
+        components = {}
+        res = run(["vcgencmd", "pmic_read_adc"], capture_output=True)
+        lines = res.stdout.decode("utf-8").splitlines()
+        for line in lines:
+            res = re.search(
+                "([A-Z_0-9]+)_[VA] (current|volt)\(([0-9]+)\)=([0-9.]+)",  # noqa: W605
+                line,
+            )
+            component_name, measure_type, idx, value = res.groups()
+            component = components[component_name] = components.get(component_name, {})
+            component[measure_type] = float(value)
+        pi_power = 0
+
+        for component_name, component in components.items():
+            try:
+                component["power"] = component["volt"] * component["current"]
+                if component_name in self.WANTED_COMPONENTS:
+                    pi_power += component["power"]
+            except Exception:
+                ...
+        return {
+            "power": pi_power,
+        }
+
+    @classmethod
+    def from_utils(
+        cls, output_dir: str, model: Optional[str] = None, chip_part: str = "CPU"
+    ) -> "Raspberry":
+        if model is None:
+            model = detect_cpu_model()
+            if model is None:
+                logger.warning("Could not read Raspberry model.")
+
+        return cls(output_dir=output_dir, model=model, chip_part=chip_part)