Skip to content

energy_rapl

energy_rapl

CPU RAPL energy monitor — reads Intel/AMD RAPL counters from sysfs.

Classes

RaplDomain

RaplDomain(path: Path)

A single RAPL power domain (e.g., intel-rapl:0, intel-rapl:0:0).

Source code in src/openjarvis/telemetry/energy_rapl.py
def __init__(self, path: Path) -> None:
    self.path = path
    self.name = self._read_name()
    self.max_energy_uj = self._read_max_energy()
Functions
read_energy_uj
read_energy_uj() -> int

Read the current energy counter value in microjoules.

Source code in src/openjarvis/telemetry/energy_rapl.py
def read_energy_uj(self) -> int:
    """Read the current energy counter value in microjoules."""
    energy_file = self.path / "energy_uj"
    try:
        return int(energy_file.read_text().strip())
    except (OSError, PermissionError, ValueError) as exc:
        logger.debug("Failed to read RAPL energy: %s", exc)
        return 0

RaplEnergyMonitor

RaplEnergyMonitor(poll_interval_ms: int = 50, rapl_base: Path = _RAPL_BASE)

Bases: EnergyMonitor

CPU energy monitor reading Intel RAPL counters from sysfs.

No external dependencies — reads directly from /sys/class/powercap/intel-rapl/. Handles counter wrap-around using max_energy_range_uj.

Source code in src/openjarvis/telemetry/energy_rapl.py
def __init__(
    self,
    poll_interval_ms: int = 50,
    rapl_base: Path = _RAPL_BASE,
) -> None:
    self._poll_interval_ms = poll_interval_ms
    self._rapl_base = rapl_base
    self._domains: List[RaplDomain] = []
    self._initialized = False

    if platform.system() == "Linux":
        try:
            self._domains = _discover_domains(rapl_base)
            self._initialized = len(self._domains) > 0
        except Exception as exc:
            logger.debug("RAPL monitor initialization failed: %s", exc)
            self._initialized = False