feat(codecarbon) add raspberry support

codecarbon/core/cpu.py

@@ -248,6 +248,24 @@ def is_psutil_available():
         return False
 
 
+def is_raspberry() -> bool:
+    """
+    Check if the Raspberry Pi PMIC power readings are available.
+
+    `vcgencmd` ships with every Raspberry Pi OS, but only the Pi 5 answers
+    `pmic_read_adc`: older boards fall back to the TDP of their CPU.
+    """
+    if not os.path.exists("/usr/bin/vcgencmd"):
+        return False
+    try:
+        res = subprocess.run(
+            ["vcgencmd", "pmic_read_adc"], capture_output=True, timeout=5
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+    return res.returncode == 0 and b"volt(" in res.stdout
+
+
 class IntelPowerGadget:
     """
     A class to interface with Intel Power Gadget for monitoring CPU power consumption on Windows and (non-Apple Silicon) macOS.

codecarbon/core/hardware_cache.py

@@ -36,6 +36,7 @@ class HardwareKind(str, Enum):
     CPU = "cpu"
     APPLE_CHIP = "apple_chip"
     GPU = "gpu"
+    RASPBERRY = "raspberry"
 
 
 _cache_lock = threading.Lock()
@@ -108,6 +109,8 @@ def _hardware_kind(hw) -> HardwareKind:
         return HardwareKind.APPLE_CHIP
     if name == "GPU":
         return HardwareKind.GPU
+    if name == "Raspberry":
+        return HardwareKind.RASPBERRY
     raise TypeError(f"Unsupported hardware type for cache: {type(hw)}")
 
 
@@ -148,11 +151,17 @@ def _spec_from_hardware(hw) -> Dict[str, Any]:
     if kind == HardwareKind.GPU:
         gpu_ids = _canonical_gpu_ids(hw.gpu_ids)
         return {"kind": kind.value, "gpu_ids": list(gpu_ids) if gpu_ids else None}
+    if kind == HardwareKind.RASPBERRY:
+        return {
+            "kind": kind.value,
+            "model": hw._model,
+            "chip_part": hw.chip_part,
+        }
     raise TypeError(f"Unsupported hardware type for cache: {type(hw)}")
 
 
 def _hardware_from_spec(spec: Dict[str, Any], output_dir: str):
-    from codecarbon.external.hardware import CPU, GPU, AppleSiliconChip
+    from codecarbon.external.hardware import CPU, GPU, AppleSiliconChip, Raspberry
     from codecarbon.external.ram import RAM
 
     try:
@@ -185,6 +194,12 @@ def _hardware_from_spec(spec: Dict[str, Any], output_dir: str):
     if kind == HardwareKind.GPU:
         gpu_ids = _canonical_gpu_ids(spec.get("gpu_ids"))
         return GPU.from_utils(gpu_ids=list(gpu_ids) if gpu_ids else None)
+    if kind == HardwareKind.RASPBERRY:
+        return Raspberry(
+            output_dir=output_dir,
+            model=spec["model"],
+            chip_part=spec["chip_part"],
+        )
     raise ValueError(f"Unknown hardware spec kind: {kind}")
 
 

codecarbon/core/resource_tracker.py

@@ -12,7 +12,13 @@
     is_mac_os,
     is_windows_os,
 )
-from codecarbon.external.hardware import CPU, GPU, MODE_CPU_LOAD, AppleSiliconChip
+from codecarbon.external.hardware import (
+    CPU,
+    GPU,
+    MODE_CPU_LOAD,
+    AppleSiliconChip,
+    Raspberry,
+)
 from codecarbon.external.logger import logger
 from codecarbon.external.ram import RAM
 
@@ -39,7 +45,14 @@ def set_RAM_tracking(self):
             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]
+        self.tracker._hardware: List[
+            Union[RAM, CPU, GPU, AppleSiliconChip, Raspberry]
+        ] = [ram]
+        if cpu.is_raspberry():
+            # The PMIC measures the DRAM rails, which beats any RAM estimate
+            self.tracker._hardware = [
+                Raspberry.from_utils(self.tracker._output_dir, chip_part="RAM")
+            ]
 
     def _setup_cpu_load_mode(self, tdp, max_power):
         """Set up CPU tracking in load mode using psutil."""
@@ -102,6 +115,15 @@ def _setup_rapl(self):
         self.tracker._conf["cpu_model"] = hardware_cpu.get_model()
         return True
 
+    def _setup_raspberry(self):
+        """Set up CPU tracking using the Raspberry Pi power interface."""
+        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)
+        self.tracker._conf["cpu_model"] = hardware_cpu.get_model()
+        return True
+
     def _setup_powermetrics(self):
         """Set up CPU and GPU tracking using PowerMetrics (Apple Silicon)."""
         logger.info("Tracking Apple CPU and GPU via PowerMetrics")
@@ -223,6 +245,9 @@ def _try_platform_cpu_backend(self) -> bool:
         if is_linux_os() and cpu.is_rapl_available():
             self._setup_rapl()
             return True
+        if cpu.is_raspberry():
+            self._setup_raspberry()
+            return True
         if is_mac_os():
             cpu_model = detect_cpu_model() or ""
             if is_mac_arm(cpu_model):

codecarbon/emissions_tracker.py

@@ -23,7 +23,7 @@
 from codecarbon.core.config import get_hierarchical_config, normalize_gpu_ids
 from codecarbon.core.units import Energy, Power, Time, Water
 from codecarbon.core.util import count_cpus, count_physical_cpus, suppress
-from codecarbon.external.hardware import CPU, GPU, AppleSiliconChip
+from codecarbon.external.hardware import CPU, GPU, AppleSiliconChip, Raspberry
 from codecarbon.external.logger import logger, set_logger_format, set_logger_level
 from codecarbon.external.ram import RAM
 from codecarbon.external.scheduler import PeriodicScheduler
@@ -1236,6 +1236,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

@@ -7,6 +7,7 @@
 import time
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from subprocess import run
 from typing import Dict, Iterable, List, Optional, Tuple
 
 import psutil
@@ -548,3 +549,100 @@ def from_utils(
                 logger.warning("Could not read AppleSiliconChip model.")
 
         return cls(output_dir=output_dir, model=model, chip_part=chip_part)
+
+
+# One `vcgencmd pmic_read_adc` line, e.g. "VDD_CORE_A current(7)=3.00000000A"
+PMIC_LINE_RE = re.compile(r"([A-Z_0-9]+)_[VA] (current|volt)\(([0-9]+)\)=([0-9.]+)")
+
+
+@dataclass
+class Raspberry(BaseHardware):
+    def __init__(
+        self,
+        output_dir: str,
+        model: str,
+        chip_part: str = "CPU",
+    ):
+        # The "CPU" rails cover the whole SoC and board except the DRAM and
+        # HDMI rails, as the PMIC has no rail for the CPU cores alone.
+        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_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 get_model(self):
+        return self._model
+
+    def get_measure(self):
+        components = {}
+        res = run(["vcgencmd", "pmic_read_adc"], capture_output=True, timeout=5)
+        lines = res.stdout.decode("utf-8").splitlines()
+        for line in lines:
+            match = PMIC_LINE_RE.search(line)
+            if match is None:
+                continue
+            component_name, measure_type, idx, value = match.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)

tests/test_hardware_cache.py

@@ -156,6 +156,40 @@ def test_spec_from_hardware_windows_emi_cpu():
     assert "rapl_dir" not in spec
 
 
+def test_hardware_kind_raspberry():
+    raspberry_hw = type("Raspberry", (), {})()
+    assert hardware_cache._hardware_kind(raspberry_hw) == "raspberry"
+
+
+def test_spec_from_hardware_raspberry():
+    raspberry_hw = type(
+        "Raspberry",
+        (),
+        {"_model": "Raspberry Pi 5", "chip_part": "CPU"},
+    )()
+    assert hardware_cache._spec_from_hardware(raspberry_hw) == {
+        "kind": "raspberry",
+        "model": "Raspberry Pi 5",
+        "chip_part": "CPU",
+    }
+
+
+def test_spec_and_rebuild_roundtrip_for_raspberry():
+    spec = {"kind": "raspberry", "model": "Raspberry Pi 5", "chip_part": "CPU"}
+    fake_pi = SimpleNamespace(_model="Raspberry Pi 5")
+    with patch(
+        "codecarbon.external.hardware.Raspberry",
+        return_value=fake_pi,
+    ) as mock_pi_cls:
+        rebuilt = hardware_cache._hardware_from_spec(spec, "out")
+    mock_pi_cls.assert_called_once_with(
+        output_dir="out",
+        model="Raspberry Pi 5",
+        chip_part="CPU",
+    )
+    assert rebuilt._model == "Raspberry Pi 5"
+
+
 def test_spec_and_rebuild_roundtrip_for_apple_chip():
     spec = {"kind": "apple_chip", "model": "Apple M1", "chip_part": "CPU"}
     fake_chip = SimpleNamespace(_model="Apple M1")