wattameter.readers package

wattameter.readers.base module

class wattameter.readers.base.BaseReader(quantities: Iterable[type[Quantity]]) → None

Bases: ABC

Base class for all readers.

Parameters:: quantities (Iterable[type[Quantity]]) – List of quantities to read.

compute_derived(time_series, data_series, time_unit: Second = 's') → list

Compute the derived quantities from the time series and data series.

The data series is expected to be a sequence generated by successive calls to the read() method.

Parameters:

time_series – One-dimensional array of time readings.
data_series – One- or Two-dimensional array of data readings. The first dimension is expected to be the time dimension.
time_unit (Second) – Unit of the time readings. Defaults to seconds. (default: 's')

Return type:

list

property derived_quantities: list[type[Quantity]]: Return a list of quantities derived from the reading streams.

property derived_tags: list[str]

Return a list of tags with quentities derived from the reading streams.

For example, if one of the quantities is energy, the derived quantity could be power.

abstractmethod get_unit(quantity: type[Quantity]) → Unit

Get the unit for a given quantity.

Return type:: Unit

abstractmethod read() → Iterable

Read the quantities of interest.

Return type:: Iterable

abstract property tags: list[str]: Return a list of tags for each reading stream.

wattameter.readers.nvml module

class wattameter.readers.nvml.DataThroughput(x=0, /)

Bases: Quantity

Data throughput quantity (e.g., NVLink throughput).

static units() → list[type[Unit]]

Return a list of possible units for this quantity.

Return type:: list[type[Unit]]

class wattameter.readers.nvml.NVMLReader(quantities=(<class 'wattameter.readers.utils.Power'>, )) → None

Bases: BaseReader

Reader for NVIDIA Management Library (NVML) to monitor GPU

devices: List of NVML device handles for available GPUs.

UNITS = {<class 'wattameter.readers.nvml.DataThroughput'>: 'KiB', <class 'wattameter.readers.utils.Energy'>: 'mJ', <class 'wattameter.readers.utils.Power'>: 'mW', <class 'wattameter.readers.utils.Temperature'>: 'C'}: Dictionary of measurement units for physical quantities.

get_unit(quantity: type[Quantity]) → Unit

Get the unit for a given quantity.

Return type:: Unit

read() → list[int]

Read the specified quantities for all devices.

Return type:: list[int]

read_energy() → list[int]

Read the current power usage for all devices.

Return type:: list[int]

read_energy_on_device(i: int) → int

Read the energy counter for the i-th device.

Return type:: int

read_nvlink_throughput() → list[tuple[int, int]]

Read the current NVLink throughput for all devices.

Return type:: list[tuple[int, int]]

read_nvlink_throughput_on_device(i: int) → tuple[int, int]

Read the current NVLink throughput for the i-th device.

Return type:: tuple[int, int]

read_power() → list[int]

Read the current power usage for all devices.

Return type:: list[int]

read_power_on_device(i: int) → int

Read the current power usage for the i-th device.

Return type:: int

read_temperature() → list[int]

Read the current temperature for all devices.

Return type:: list[int]

read_temperature_on_device(i: int) → int

Read the temperature for the i-th device.

Return type:: int

read_utilization() → list[tuple[int, int]]

Read the current utilization for all devices.

Return type:: list[tuple[int, int]]

read_utilization_on_device(i: int) → tuple[int, int]

Read the current utilization for the i-th device.

Return type:: tuple[int, int]

property tags: list[str]: Return a list of tags for each reading stream.

wattameter.readers.rapl module

class wattameter.readers.rapl.RAPLDevice(rapl_device_path: str) → None

Bases: BaseReader

Reader for RAPL (Running Average Power Limit) devices.

Parameters:: rapl_device_path (str) – Path to the RAPL device directory.

path: Path to the RAPL device directory, typically under /sys/class/powercap/intel-rapl.

name: Name of the RAPL device, read from the ‘name’ file.

max_energy_range: Maximum energy range, read from the ‘max_energy_range_uj’ file.

compute_derived(time_series, data_series, time_unit: Second = 's')

Compute the derived quantities from the time series and data series.

The data series is expected to be a sequence generated by successive calls to the read() method.

Parameters:

time_series – One-dimensional array of time readings.
data_series – One- or Two-dimensional array of data readings. The first dimension is expected to be the time dimension.
time_unit (Second) – Unit of the time readings. Defaults to seconds. (default: 's')

property derived_quantities: list[type[Quantity]]: Return a list of quantities derived from the reading streams.

property derived_tags: list[str]

Return a list of tags with quentities derived from the reading streams.

For example, if one of the quantities is energy, the derived quantity could be power.

get_unit(quantity: type[Quantity]) → Unit

Get the unit for a given quantity.

Return type:: Unit

read() → list[int]

Read the quantities of interest.

Return type:: list[int]

read_energy() → int

Read the energy counter for the i-th device.

Return type:: int

property tags: list[str]: Return a list of tags for each reading stream.

class wattameter.readers.rapl.RAPLReader(rapl_dir='/sys/class/powercap/intel-rapl/subsystem') → None

Bases: BaseReader

Reader for RAPL devices in the system.

Parameters:: rapl_dir – Directory where RAPL devices are located, typically /sys/class/powercap/intel-rapl/subsystem. (default: '/sys/class/powercap/intel-rapl/subsystem')

rapl_dir: Directory where RAPL devices are located.

devices: List of RAPLDevice instances for available RAPL devices.

compute_derived(time_series: Iterable, data_series: Iterable, time_unit: Second = 's')

Compute the derived quantities from the time series and data series.

The data series is expected to be a sequence generated by successive calls to the read() method.

Parameters:

time_series (Iterable) – One-dimensional array of time readings.
data_series (Iterable) – One- or Two-dimensional array of data readings. The first dimension is expected to be the time dimension.
time_unit (Second) – Unit of the time readings. Defaults to seconds. (default: 's')

property derived_quantities: list[type[Quantity]]: Return a list of quantities derived from the reading streams.

property derived_tags: list[str]

Return a list of tags with quentities derived from the reading streams.

For example, if one of the quantities is energy, the derived quantity could be power.

get_unit(quantity: type[Quantity]) → Unit

Get the unit for a given quantity.

Return type:: Unit

read() → list[int]

Read the current energy counter for all RAPL devices.

Return type:: list[int]

read_energy() → list[int]

Read the current energy counter for all RAPL devices.

Return type:: list[int]

read_energy_on_device(i: int) → int

Read the energy counter of the i-th device.

Return type:: int

property tags: list[str]: Return a list of tags for each reading stream.

wattameter.readers.utils module

class wattameter.readers.utils.Byte(si_prefix_str: str = '') → None

Bases: Unit

Byte unit.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float

class wattameter.readers.utils.Celsius(si_prefix_str: str = '') → None

Bases: Unit

Celsius unit.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float

class wattameter.readers.utils.Energy(x=0, /)

Bases: Quantity

A physical quantity representing energy.

static units() → list[type[Unit]]

Return a list of possible units for this quantity.

Return type:: list[type[Unit]]

class wattameter.readers.utils.Joule(si_prefix_str: str = '') → None

Bases: Unit

Joule unit.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float

class wattameter.readers.utils.Power(x=0, /)

Bases: Quantity

A physical quantity representing power.

static units() → list[type[Unit]]

Return a list of possible units for this quantity.

Return type:: list[type[Unit]]

class wattameter.readers.utils.Quantity(x=0, /)

Bases: float

A base class for physical quantities.

abstractmethod static units() → list[type[Unit]]

Return a list of possible units for this quantity.

Return type:: list[type[Unit]]

class wattameter.readers.utils.SIPrefix(*values)

Bases: Enum

An enumeration of SI prefixes.

GIBI = 1073741824

GIGA = 1000000000.0

KIBI = 1024

KILO = 1000.0

MEGA = 1000000.0

MIBI = 1048576

MICRO = 1e-06

MILLI = 0.001

NANO = 1e-09

NONE = 1.0

PEBI = 1125899906842624

TEBI = 1099511627776

TERA = 1000000000000.0

classmethod from_string(prefix: str) → SIPrefix

Get the SIPrefix enum member from a string.

Parameters:: prefix (str) – The SI prefix as a string (e.g., “m” for milli, “k” for kilo).
Raises:: ValueError – If the prefix is not recognized.
Return type:: SIPrefix

class wattameter.readers.utils.Second(si_prefix_str: str = '') → None

Bases: Unit

Second unit.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float

class wattameter.readers.utils.Temperature(x=0, /)

Bases: Quantity

A physical quantity representing temperature.

static units() → list[type[Unit]]

Return a list of possible units for this quantity.

Return type:: list[type[Unit]]

class wattameter.readers.utils.Unit(si_prefix_str: str = '') → None

Bases: str

Default unit class.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float

class wattameter.readers.utils.Utilization(x=0, /)

Bases: Quantity

Percentage utilization quantity.

static units() → list[type[Unit]]

Return a list of possible units for this quantity.

Return type:: list[type[Unit]]

class wattameter.readers.utils.Watt(si_prefix_str: str = '') → None

Bases: Unit

Watt unit.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float

class wattameter.readers.utils.WattHour(si_prefix_str: str = '') → None

Bases: Unit

Watt-hour unit.

static symbol() → str

Return the symbol of the unit.

Return type:: str

to_si() → float

Convert 1 unit to the SI unit.

Return type:: float