Energy Test Bench on a Raspberry Pi 5 or Pi 3

#!/usr/bin/env python3
"""Reads the processor state (Raspberry Pi 5 or 3) from sysfs and vcgencmd."""

import pathlib
import subprocess

NUM_CORES = 4
CPUFREQ = "/sys/devices/system/cpu/cpu{}/cpufreq/{}"


def read_value(path):
    """Reads a sysfs file; returns None if it does not exist."""
    try:
        return pathlib.Path(path).read_text().strip()
    except OSError:
        return None


def frequency(core=0):
    """Current frequency of a core, in MHz."""
    khz = read_value(CPUFREQ.format(core, "scaling_cur_freq"))
    return int(khz) / 1000 if khz else None


def governor(core=0):
    return read_value(CPUFREQ.format(core, "scaling_governor"))


def available_steps():
    """List of the frequency steps, in MHz."""
    val = read_value(CPUFREQ.format(0, "scaling_available_frequencies"))
    if not val:
        # Some cores only expose the limits, not the full list.
        low = read_value(CPUFREQ.format(0, "cpuinfo_min_freq"))
        high = read_value(CPUFREQ.format(0, "cpuinfo_max_freq"))
        return [int(low) / 1000, int(high) / 1000] if low and high else []
    return [int(v) / 1000 for v in val.split()]


def temperature():
    """Processor temperature, in degrees Celsius."""
    val = read_value("/sys/class/thermal/thermal_zone0/temp")
    return int(val) / 1000 if val else None


# Each bit of vcgencmd get_throttled flags a power or temperature issue.
# Bits 0-3 = happening now, bits 16-19 = has happened since boot.
FLAGS = {
    0: "undervoltage NOW",
    1: "frequency throttled NOW",
    2: "thermal throttling NOW",
    3: "soft temperature limit NOW",
    16: "undervoltage has occurred",
    17: "frequency throttling has occurred",
    18: "thermal throttling has occurred",
    19: "soft temperature limit has been reached",
}


def throttle_flags():
    """Returns the list of problems reported by the firmware."""
    try:
        output = subprocess.run(["vcgencmd", "get_throttled"],
                                capture_output=True, text=True, check=True).stdout
    except (OSError, subprocess.CalledProcessError):
        return ["vcgencmd unavailable"]
    value = int(output.strip().split("=")[1], 16)
    return [text for bit, text in FLAGS.items() if value & (1 << bit)]


if __name__ == "__main__":
    print("active governor  :", governor())
    print("steps (MHz)      :", available_steps())
    print("temperature      : %.1f degC" % temperature())
    print()
    for n in range(NUM_CORES):
        print("  core %d: %7.1f MHz" % (n, frequency(n)))
    print()
    issues = throttle_flags()
    print("throttling       :", ", ".join(issues) if issues else "none")

#!/usr/bin/env python3
"""Reads the power consumed directly from the Pi 5 board's power management circuit."""

import re
import subprocess

# Example line:  VDD_CORE_A current(7)=2.61856000A
LINE = re.compile(r"^\s*(?P<rail>\S+?)_(?P<kind>[AV])\s+\w+\(\d+\)=(?P<val>[0-9.]+)[AV]\s*$")


def measure():
    """Returns (total_power_W, dict with the power on each rail)."""
    try:
        output = subprocess.run(["vcgencmd", "pmic_read_adc"],
                                capture_output=True, text=True, check=True).stdout
    except (OSError, subprocess.CalledProcessError):
        return None, {}

    raw = {}
    for line in output.splitlines():
        m = LINE.match(line)
        if m:
            raw.setdefault(m.group("rail"), {})[m.group("kind")] = float(m.group("val"))

    # Only rails that report both current and voltage can yield a power.
    # EXT5V and BATT only report the voltage, so they are left out.
    detail = {name: v["A"] * v["V"] for name, v in raw.items() if "A" in v and "V" in v}
    if not detail:
        return None, {}      # the board has no measuring PMIC (e.g. a Pi 3)
    return sum(detail.values()), detail


if __name__ == "__main__":
    total, detail = measure()
    if total is None:
        raise SystemExit("PMIC unavailable - it only exists on a Raspberry Pi 5 (on a Pi 3 use power_ina219.py)")

    for name, p in sorted(detail.items(), key=lambda x: -x[1]):
        percent = 100 * p / total
        bar = "#" * int(percent / 2)
        print("%-12s %6.3f W  %5.1f%%  %s" % (name, p, percent, bar))
    print("-" * 46)
    print("%-12s %6.3f W" % ("TOTAL", total))

#!/usr/bin/env python3
"""Reads voltage, current and power from an INA219 module."""

import time

import board
from adafruit_ina219 import ADCResolution, BusVoltageRange, INA219

i2c = board.I2C()             # automatically uses the I2C-1 bus
sensor = INA219(i2c)          # default address 0x40

# Configuration for quiet measurements: maximum resolution and averaging
# over 128 samples. Reduces noise, at the cost of a slower reading.
sensor.bus_adc_resolution = ADCResolution.ADCRES_12BIT_128S
sensor.shunt_adc_resolution = ADCResolution.ADCRES_12BIT_128S
sensor.bus_voltage_range = BusVoltageRange.RANGE_16V


def measure():
    """Returns (voltage_V, current_mA, power_W)."""
    # The real voltage on the load = the bus voltage + the drop across the shunt.
    voltage = sensor.bus_voltage + sensor.shunt_voltage
    current = sensor.current            # mA
    return voltage, current, voltage * current / 1000.0


if __name__ == "__main__":
    print("%8s %10s %10s" % ("V", "mA", "W"))
    try:
        while True:
            u, i, p = measure()
            print("%8.3f %10.2f %10.4f" % (u, i, p))
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("\nstopped")

#!/usr/bin/env python3
"""Automatically picks where we read the power from:
Pi 5 -> the internal PMIC (power_pmic.py),
Pi 3 -> the INA219 module on the board's supply (power_ina219.py)."""

import power_pmic

SOURCE = None


def measure():
    """No source available: returns the same format, but empty."""
    return None, {}


if power_pmic.measure()[0] is not None:
    SOURCE = "PMIC"
    measure = power_pmic.measure
else:
    try:
        import power_ina219           # initializes the sensor; fails if it is missing
        power_ina219.measure()        # a test reading
    except Exception as e:
        print("No PMIC, and the INA219 does not answer (%s)" % e)
    else:
        SOURCE = "INA219"

        def measure():
            """Same shape as power_pmic.measure(): (total_W, detail)."""
            _, _, p = power_ina219.measure()
            return p, {"BOARD_5V": p}


if __name__ == "__main__":
    total, _ = measure()
    if total is None:
        raise SystemExit("nothing to measure the power with - see sections 8 and 9")
    print("source: %s   power: %.3f W" % (SOURCE, total))

#!/usr/bin/env python3
"""16x2 LCD display for the energy test bench."""

from RPLCD.i2c import CharLCD

# If i2cdetect shows 3f instead of 27, change the address below.
ADDRESS = 0x27


class Panel:
    """Wrapper around the LCD that does not complain if the screen is missing."""

    def __init__(self, address=ADDRESS):
        self.lcd = None
        try:
            self.lcd = CharLCD("PCF8574", address, cols=16, rows=2, auto_linebreaks=False)
            self.lcd.clear()
            # Custom degree symbol, in the screen's character memory.
            self.lcd.create_char(0, [0b01100, 0b10010, 0b10010, 0b01100,
                                     0b00000, 0b00000, 0b00000, 0b00000])
        except Exception as e:
            print("LCD unavailable (%s) - continuing in the console only" % e)

    def write(self, top, bottom):
        """Writes two lines of at most 16 characters each."""
        top, bottom = top[:16].ljust(16), bottom[:16].ljust(16)
        if self.lcd is None:
            print("\r[%s|%s]" % (top, bottom), end="", flush=True)
            return
        # We write at fixed positions, not with clear(): otherwise the screen flickers.
        self.lcd.cursor_pos = (0, 0)
        self.lcd.write_string(top)
        self.lcd.cursor_pos = (1, 0)
        self.lcd.write_string(bottom)

    def close(self):
        if self.lcd is not None:
            self.lcd.clear()
            self.lcd.close(clear=True)


if __name__ == "__main__":
    import time
    from cpu_state import frequency, temperature
    from power import measure

    panel = Panel()
    try:
        while True:
            p, _ = measure()
            panel.write("%4.0fMHz  %4.1fW" % (frequency(), p or 0),
                        "temp %4.1f degC" % temperature())
            time.sleep(0.5)
    except KeyboardInterrupt:
        panel.close()
        print("\nstopped")

#!/usr/bin/env python3
"""Energy test bench: runs the same task under every governor
and compares the time, average power and energy consumed."""

import csv
import subprocess
import sys
import time

from display import Panel
from power import SOURCE, measure
from cpu_state import frequency, governor, throttle_flags, temperature

GOVERNORS = ["powersave", "ondemand", "schedutil", "performance"]
CYCLES = 400_000_000          # adjust to get roughly 10 seconds (less on a Pi 3)
PERIOD = 0.2                  # power sampling interval, in seconds
NUM_CORES = 4


def set_governor(name):
    """Writes the same governor to every core. Requires root privileges."""
    for n in range(NUM_CORES):
        path = "/sys/devices/system/cpu/cpu%d/cpufreq/scaling_governor" % n
        subprocess.run(["sudo", "tee", path], input=name.encode(),
                       stdout=subprocess.DEVNULL, check=True)
    time.sleep(2)            # let the governor settle


def measure_idle(seconds=5):
    """Average power with the system idle - the reference for the useful energy."""
    samples = []
    end = time.time() + seconds
    while time.time() < end:
        p, _ = measure()
        if p:
            samples.append(p)
        time.sleep(PERIOD)
    return sum(samples) / len(samples) if samples else 0.0


def run_benchmark(panel, governor_name, p_idle):
    """Runs the task and samples the power while it is working."""
    set_governor(governor_name)

    process = subprocess.Popen(["./bench", str(CYCLES)], stdout=subprocess.DEVNULL)
    samples, frequencies = [], []
    t0 = time.time()

    while process.poll() is None:
        p, _ = measure()
        if p:
            samples.append(p)
        f = frequency()
        if f:
            frequencies.append(f)
        panel.write("%-11s%4.1fW" % (governor_name[:11], p or 0),
                    "%4.0fMHz  %4.1fC" % (f or 0, temperature()))
        time.sleep(PERIOD)

    duration = time.time() - t0
    p_avg = sum(samples) / len(samples) if samples else 0.0
    f_avg = sum(frequencies) / len(frequencies) if frequencies else 0.0

    return {
        "governor": governor_name,
        "duration_s": round(duration, 3),
        "avg_frequency_MHz": round(f_avg, 1),
        "avg_power_W": round(p_avg, 3),
        # Total energy consumed over the duration of the task.
        "energy_J": round(p_avg * duration, 2),
        # Energy above the idle consumption - the part actually "paid" for the computation.
        "useful_energy_J": round((p_avg - p_idle) * duration, 2),
        # How much computation we get for every joule spent.
        "Mcycles_per_J": round(CYCLES / 1e6 / max(p_avg * duration, 1e-9), 2),
        "temperature_C": round(temperature(), 1),
    }


def main():
    if SOURCE is None:
        raise SystemExit("Nothing to measure the power with - see sections 8 and 9.")
    print("Power measurement source: %s" % SOURCE)

    # Keep only the governors the kernel actually has.
    with open("/sys/devices/system/cpu/cpu0/cpufreq/scaling_available_governors") as f:
        available = f.read().split()
    to_run = [g for g in GOVERNORS if g in available]

    initial = governor()
    panel = Panel()

    issues = throttle_flags()
    if issues:
        print("WARNING - power supply or temperature:", ", ".join(issues))
        print("The results will not be comparable. Continuing anyway in 5 seconds...")
        time.sleep(5)

    print("Measuring idle consumption...")
    p_idle = measure_idle()
    print("Idle: %.3f W\n" % p_idle)

    results = []
    try:
        for g in to_run:
            print("Running with governor %s..." % g)
            r = run_benchmark(panel, g, p_idle)
            results.append(r)
            print("   %5.2f s   %5.2f W   %7.1f J   %6.0f MHz"
                  % (r["duration_s"], r["avg_power_W"], r["energy_J"],
                     r["avg_frequency_MHz"]))
    finally:
        set_governor(initial)     # leave the system as we found it
        panel.close()

    with open("results.csv", "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=list(results[0].keys()))
        writer.writeheader()
        writer.writerows(results)

    print("\n%-12s %8s %8s %9s %10s" % ("governor", "duration", "power", "energy", "Mcycles/J"))
    for r in results:
        print("%-12s %7.2fs %7.2fW %8.1fJ %9.2f"
              % (r["governor"], r["duration_s"], r["avg_power_W"],
                 r["energy_J"], r["Mcycles_per_J"]))

    fastest = min(results, key=lambda r: r["duration_s"])
    most_efficient = min(results, key=lambda r: r["energy_J"])
    print("\nfastest        : %s" % fastest["governor"])
    print("most efficient : %s" % most_efficient["governor"])
    if fastest["governor"] == most_efficient["governor"]:
        print("-> Same winner: 'race to idle' works on this platform.")
    else:
        print("-> Different winners: there is a trade-off between speed and energy.")


if __name__ == "__main__":
    sys.exit(main())