Banc de test energetic pe Raspberry Pi 5 și Pi 3

#!/usr/bin/env python3
"""Citeste starea procesorului (Raspberry Pi 5 sau 3) din sysfs si din vcgencmd."""

import pathlib
import subprocess

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


def citeste(cale):
    """Citeste un fisier din sysfs; intoarce None daca nu exista."""
    try:
        return pathlib.Path(cale).read_text().strip()
    except OSError:
        return None


def frecventa(nucleu=0):
    """Frecventa curenta a unui nucleu, in MHz."""
    khz = citeste(CPUFREQ.format(nucleu, "scaling_cur_freq"))
    return int(khz) / 1000 if khz else None


def guvernator(nucleu=0):
    return citeste(CPUFREQ.format(nucleu, "scaling_governor"))


def trepte_disponibile():
    """Lista treptelor de frecventa, in MHz."""
    val = citeste(CPUFREQ.format(0, "scaling_available_frequencies"))
    if not val:
        # Unele nuclee expun doar limitele, nu si lista completa.
        jos = citeste(CPUFREQ.format(0, "cpuinfo_min_freq"))
        sus = citeste(CPUFREQ.format(0, "cpuinfo_max_freq"))
        return [int(jos) / 1000, int(sus) / 1000] if jos and sus else []
    return [int(v) / 1000 for v in val.split()]


def temperatura():
    """Temperatura procesorului, in grade Celsius."""
    val = citeste("/sys/class/thermal/thermal_zone0/temp")
    return int(val) / 1000 if val else None


# Fiecare bit din vcgencmd get_throttled semnaleaza o problema de alimentare
# sau de temperatura. Bitii 0-3 = acum, bitii 16-19 = s-a intamplat de la pornire.
STEAGURI = {
    0: "subtensiune ACUM",
    1: "frecventa limitata ACUM",
    2: "limitare termica ACUM",
    3: "limita soft de temperatura ACUM",
    16: "a existat subtensiune",
    17: "a existat limitare de frecventa",
    18: "a existat limitare termica",
    19: "s-a atins limita soft de temperatura",
}


def limitari():
    """Intoarce lista problemelor semnalate de firmware."""
    try:
        iesire = subprocess.run(["vcgencmd", "get_throttled"],
                                capture_output=True, text=True, check=True).stdout
    except (OSError, subprocess.CalledProcessError):
        return ["vcgencmd indisponibil"]
    valoare = int(iesire.strip().split("=")[1], 16)
    return [text for bit, text in STEAGURI.items() if valoare & (1 << bit)]


if __name__ == "__main__":
    print("guvernator activ :", guvernator())
    print("trepte (MHz)     :", trepte_disponibile())
    print("temperatura      : %.1f gradeC" % temperatura())
    print()
    for n in range(NUCLEE):
        print("  nucleul %d: %7.1f MHz" % (n, frecventa(n)))
    print()
    probleme = limitari()
    print("limitari         :", ", ".join(probleme) if probleme else "niciuna")

#!/usr/bin/env python3
"""Citeste puterea consumata direct din circuitul de alimentare al placii Pi 5."""

import re
import subprocess

# Exemplu de linie:  VDD_CORE_A current(7)=2.61856000A
LINIE = re.compile(r"^\s*(?P<linie>\S+?)_(?P<tip>[AV])\s+\w+\(\d+\)=(?P<val>[0-9.]+)[AV]\s*$")


def masoara():
    """Intoarce (putere_totala_W, dictionar cu puterea pe fiecare linie)."""
    try:
        iesire = subprocess.run(["vcgencmd", "pmic_read_adc"],
                                capture_output=True, text=True, check=True).stdout
    except (OSError, subprocess.CalledProcessError):
        return None, {}

    brut = {}
    for rand in iesire.splitlines():
        m = LINIE.match(rand)
        if m:
            brut.setdefault(m.group("linie"), {})[m.group("tip")] = float(m.group("val"))

    # Numai liniile care raporteaza si curent, si tensiune pot da o putere.
    # EXT5V si BATT raporteaza doar tensiunea, deci raman pe dinafara.
    detaliu = {nume: v["A"] * v["V"] for nume, v in brut.items() if "A" in v and "V" in v}
    if not detaliu:
        return None, {}      # placa nu are PMIC cu masurare (de exemplu Pi 3)
    return sum(detaliu.values()), detaliu


if __name__ == "__main__":
    total, detaliu = masoara()
    if total is None:
        raise SystemExit("PMIC indisponibil - exista doar pe Raspberry Pi 5 (pe Pi 3 folositi putere_ina219.py)")

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

#!/usr/bin/env python3
"""Citeste tensiunea, curentul si puterea de la un modul INA219."""

import time

import board
from adafruit_ina219 import ADCResolution, BusVoltageRange, INA219

i2c = board.I2C()             # foloseste automat magistrala I2C-1
senzor = INA219(i2c)          # adresa implicita 0x40

# Configurare pentru masuratori linistite: rezolutie maxima si mediere pe
# 128 de esantioane. Reduce zgomotul, cu pretul unei citiri mai lente.
senzor.bus_adc_resolution = ADCResolution.ADCRES_12BIT_128S
senzor.shunt_adc_resolution = ADCResolution.ADCRES_12BIT_128S
senzor.bus_voltage_range = BusVoltageRange.RANGE_16V


def masoara():
    """Intoarce (tensiune_V, curent_mA, putere_W)."""
    # Tensiunea reala pe sarcina = tensiunea pe magistrala + caderea pe sunt.
    tensiune = senzor.bus_voltage + senzor.shunt_voltage
    curent = senzor.current            # mA
    return tensiune, curent, tensiune * curent / 1000.0


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

#!/usr/bin/env python3
"""Alege automat de unde citim puterea:
Pi 5 -> circuitul PMIC intern (putere_pmic.py),
Pi 3 -> modulul INA219 de pe alimentarea placii (putere_ina219.py)."""

import putere_pmic

SURSA = None


def masoara():
    """Nicio sursa disponibila: intoarce acelasi format, dar gol."""
    return None, {}


if putere_pmic.masoara()[0] is not None:
    SURSA = "PMIC"
    masoara = putere_pmic.masoara
else:
    try:
        import putere_ina219          # initializeaza senzorul; esueaza daca lipseste
        putere_ina219.masoara()       # o citire de proba
    except Exception as e:
        print("Fara PMIC, iar INA219 nu raspunde (%s)" % e)
    else:
        SURSA = "INA219"

        def masoara():
            """Aceeasi forma ca putere_pmic.masoara(): (total_W, detaliu)."""
            _, _, p = putere_ina219.masoara()
            return p, {"PLACA_5V": p}


if __name__ == "__main__":
    total, _ = masoara()
    if total is None:
        raise SystemExit("nu am cu ce masura puterea - vezi sectiunile 8 si 9")
    print("sursa: %s   putere: %.3f W" % (SURSA, total))

#!/usr/bin/env python3
"""Afisaj LCD 16x2 pentru bancul de test energetic."""

from RPLCD.i2c import CharLCD

# Daca i2cdetect arata 3f in loc de 27, schimbati adresa mai jos.
ADRESA = 0x27


class Panou:
    """Invelis peste LCD care nu se plange daca ecranul lipseste."""

    def __init__(self, adresa=ADRESA):
        self.lcd = None
        try:
            self.lcd = CharLCD("PCF8574", adresa, cols=16, rows=2, auto_linebreaks=False)
            self.lcd.clear()
            # Simbol propriu pentru grade, in memoria de caractere a ecranului.
            self.lcd.create_char(0, [0b01100, 0b10010, 0b10010, 0b01100,
                                     0b00000, 0b00000, 0b00000, 0b00000])
        except Exception as e:
            print("LCD indisponibil (%s) - continui doar in consola" % e)

    def scrie(self, sus, jos):
        """Scrie doua randuri de maximum 16 caractere fiecare."""
        sus, jos = sus[:16].ljust(16), jos[:16].ljust(16)
        if self.lcd is None:
            print("\r[%s|%s]" % (sus, jos), end="", flush=True)
            return
        # Scriem pe pozitii fixe, nu cu clear(): altfel ecranul palpaie.
        self.lcd.cursor_pos = (0, 0)
        self.lcd.write_string(sus)
        self.lcd.cursor_pos = (1, 0)
        self.lcd.write_string(jos)

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


if __name__ == "__main__":
    import time
    from stare_cpu import frecventa, temperatura
    from putere import masoara

    panou = Panou()
    try:
        while True:
            p, _ = masoara()
            panou.scrie("%4.0fMHz  %4.1fW" % (frecventa(), p or 0),
                        "temp %4.1f gradeC" % temperatura())
            time.sleep(0.5)
    except KeyboardInterrupt:
        panou.inchide()
        print("\noprit")

#!/usr/bin/env python3
"""Banc de test energetic: ruleaza aceeasi sarcina sub fiecare guvernator
si compara timpul, puterea medie si energia consumata."""

import csv
import subprocess
import sys
import time

from afisaj import Panou
from putere import SURSA, masoara
from stare_cpu import frecventa, guvernator, limitari, temperatura

GUVERNATOARE = ["powersave", "ondemand", "schedutil", "performance"]
CICLI = 400_000_000          # ajustati ca sa dea aproximativ 10 secunde (pe Pi 3: mai putin)
PERIOADA = 0.2               # intervalul de esantionare a puterii, in secunde
NUCLEE = 4


def seteaza_guvernator(nume):
    """Scrie acelasi guvernator pe toate nucleele. Necesita drepturi de root."""
    for n in range(NUCLEE):
        cale = "/sys/devices/system/cpu/cpu%d/cpufreq/scaling_governor" % n
        subprocess.run(["sudo", "tee", cale], input=nume.encode(),
                       stdout=subprocess.DEVNULL, check=True)
    time.sleep(2)            # lasam guvernatorul sa se aseze


def masoara_repaus(secunde=5):
    """Puterea medie cu sistemul in repaus - referinta pentru energia utila."""
    probe = []
    sfarsit = time.time() + secunde
    while time.time() < sfarsit:
        p, _ = masoara()
        if p:
            probe.append(p)
        time.sleep(PERIOADA)
    return sum(probe) / len(probe) if probe else 0.0


def ruleaza(panou, nume_guvernator, p_repaus):
    """Ruleaza sarcina si esantioneaza puterea cat timp aceasta lucreaza."""
    seteaza_guvernator(nume_guvernator)

    proces = subprocess.Popen(["./bench", str(CICLI)], stdout=subprocess.DEVNULL)
    probe, frecvente = [], []
    t0 = time.time()

    while proces.poll() is None:
        p, _ = masoara()
        if p:
            probe.append(p)
        f = frecventa()
        if f:
            frecvente.append(f)
        panou.scrie("%-11s%4.1fW" % (nume_guvernator[:11], p or 0),
                    "%4.0fMHz  %4.1fC" % (f or 0, temperatura()))
        time.sleep(PERIOADA)

    durata = time.time() - t0
    p_medie = sum(probe) / len(probe) if probe else 0.0
    f_medie = sum(frecvente) / len(frecvente) if frecvente else 0.0

    return {
        "guvernator": nume_guvernator,
        "durata_s": round(durata, 3),
        "frecventa_medie_MHz": round(f_medie, 1),
        "putere_medie_W": round(p_medie, 3),
        # Energia totala consumata pe durata sarcinii.
        "energie_J": round(p_medie * durata, 2),
        # Energia peste consumul de repaus - partea "platita" efectiv pentru calcul.
        "energie_utila_J": round((p_medie - p_repaus) * durata, 2),
        # Cat calcul obtinem pentru fiecare joule cheltuit.
        "Mcicli_pe_J": round(CICLI / 1e6 / max(p_medie * durata, 1e-9), 2),
        "temperatura_C": round(temperatura(), 1),
    }


def main():
    if SURSA is None:
        raise SystemExit("Nu am cu ce masura puterea - vezi sectiunile 8 si 9.")
    print("Sursa de masura a puterii: %s" % SURSA)

    # Pastram doar guvernatoarele pe care nucleul le are efectiv.
    with open("/sys/devices/system/cpu/cpu0/cpufreq/scaling_available_governors") as f:
        disponibile = f.read().split()
    lista = [g for g in GUVERNATOARE if g in disponibile]

    initial = guvernator()
    panou = Panou()

    probleme = limitari()
    if probleme:
        print("ATENTIE - alimentare sau temperatura:", ", ".join(probleme))
        print("Rezultatele nu vor fi comparabile. Continui oricum in 5 secunde...")
        time.sleep(5)

    print("Masor consumul in repaus...")
    p_repaus = masoara_repaus()
    print("Repaus: %.3f W\n" % p_repaus)

    rezultate = []
    try:
        for g in lista:
            print("Rulez cu guvernatorul %s..." % g)
            r = ruleaza(panou, g, p_repaus)
            rezultate.append(r)
            print("   %5.2f s   %5.2f W   %7.1f J   %6.0f MHz"
                  % (r["durata_s"], r["putere_medie_W"], r["energie_J"],
                     r["frecventa_medie_MHz"]))
    finally:
        seteaza_guvernator(initial)     # lasam sistemul cum l-am gasit
        panou.inchide()

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

    print("\n%-12s %8s %8s %9s %10s" % ("guvernator", "durata", "putere", "energie", "Mcicli/J"))
    for r in rezultate:
        print("%-12s %7.2fs %7.2fW %8.1fJ %9.2f"
              % (r["guvernator"], r["durata_s"], r["putere_medie_W"],
                 r["energie_J"], r["Mcicli_pe_J"]))

    cel_mai_rapid = min(rezultate, key=lambda r: r["durata_s"])
    cel_mai_econom = min(rezultate, key=lambda r: r["energie_J"])
    print("\ncel mai rapid  : %s" % cel_mai_rapid["guvernator"])
    print("cel mai econom : %s" % cel_mai_econom["guvernator"])
    if cel_mai_rapid["guvernator"] == cel_mai_econom["guvernator"]:
        print("-> Acelasi castigator: 'race to idle' functioneaza pe aceasta platforma.")
    else:
        print("-> Castigatori diferiti: exista un compromis intre viteza si energie.")


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