Measurement API

Energy measurement decorator using pyRAPL (Linux/Intel) or estimation.

This module provides a decorator to measure energy consumption of Python functions. On Linux systems with Intel x86_64 processors, it uses pyRAPL for hardware-based energy measurement. On Windows and macOS, it uses estimation methods based on CPU time, TDP, and utilization models.

pgsi_analyzer.measurement.energy.measure_energy_to_csv(n: int, csv_filename: str, folder_name: str | Path = 'energy_benchmark')[source]

Decorator to measure energy usage, store system info in a JSON file, and store energy results in a CSV file.

On Linux/Intel systems, uses pyRAPL for hardware-based energy measurement. On Windows and macOS, uses estimation methods based on CPU time and TDP.

Parameters:

  • n – Number of times to run the function and measure energy

  • csv_filename – Base name for the CSV output file (without .csv extension)

  • folder_name – Directory name or Path where results will be stored

Returns:

Decorator function

Examples

>>> @measure_energy_to_csv(n=10, csv_filename="test_energy")
... def my_function():
...     return sum(range(1000))
>>> result = my_function()

Time measurement decorator for execution time profiling.

This module provides a decorator to measure execution time of Python functions and store the results in CSV format along with system information.

pgsi_analyzer.measurement.time.measure_time_to_csv(n: int, csv_filename: str, folder_name: str | Path = 'time_benchmark')[source]

Decorator to measure execution time, store system info in a JSON file, and store execution time results in a CSV file.

Parameters:

  • n – Number of times to run the function and measure execution time

  • csv_filename – Base name for the CSV output file (without .csv extension)

  • folder_name – Directory name or Path where results will be stored

Returns:

Decorator function

Examples

>>> @measure_time_to_csv(n=10, csv_filename="test_time")
... def my_function():
...     return sum(range(1000))
>>> result = my_function()

Energy estimation utilities for platforms without hardware counters.

This module provides energy estimation functions for Windows and macOS systems where Intel RAPL hardware counters are not available. Estimation is based on CPU time, TDP (Thermal Design Power), and utilization models.

pgsi_analyzer.measurement.estimators.get_cpu_tdp(cpu_model: str) → float[source]

Get CPU TDP (Thermal Design Power) for a given CPU model.

Parameters:

cpu_model – CPU model string (case-insensitive)

Returns:

TDP in Watts. Returns default TDP if model not found.

Examples

>>> get_cpu_tdp("Intel Core i7")
65.0
>>> get_cpu_tdp("AMD Ryzen 5")
65.0
>>> get_cpu_tdp("Unknown CPU")
65.0  # Default
pgsi_analyzer.measurement.estimators.resolve_cpu_power_provenance(cpu_model: str) → Dict[str, str][source]

Return audit metadata for CPU power resolution.

pgsi_analyzer.measurement.estimators.estimate_energy_cpu_time(cpu_time_seconds: float, cpu_info: Dict[str, Any] | None = None) → Tuple[float, str, str][source]

Estimate energy consumption based on CPU time and CPU model.

Uses CPU TDP and a utilization model to estimate energy consumption. The model assumes: - Base power consumption: 20% of TDP (idle power) - Active power consumption: TDP (at full load) - Average utilization during execution: 80% (typical for CPU-bound tasks)

Parameters:

  • cpu_time_seconds – CPU time spent executing (seconds)

  • cpu_info – Optional CPU info dictionary. If None, will fetch automatically.

Returns:

Tuple of (energy in microjoules, estimation model name, methodology tag)

Examples

>>> energy, model = estimate_energy_cpu_time(1.0)
>>> energy > 0
True
pgsi_analyzer.measurement.estimators.estimate_energy_from_psutil(duration_seconds: float, cpu_info: Dict[str, Any] | None = None) → Tuple[float, str, str][source]

Estimate energy using psutil to monitor CPU utilization.

When psutil is not available (e.g. on PyPy), falls back to CPU-time-based estimation.

Monitors CPU percent during execution and applies power models based on CPU type and actual utilization.

Parameters:

  • duration_seconds – Duration of execution (seconds)

  • cpu_info – Optional CPU info dictionary. If None, will fetch automatically.

Returns:

Tuple of (energy in microjoules, estimation model name, methodology tag)

Examples

>>> energy, model = estimate_energy_from_psutil(1.0)
>>> energy > 0
True
pgsi_analyzer.measurement.estimators.estimate_windows(cpu_time_seconds: float, cpu_info: Dict[str, Any] | None = None) → Tuple[float, str, str][source]

Windows-specific energy estimation.

Uses CPU time-based estimation optimized for Windows systems.

Parameters:

  • cpu_time_seconds – CPU time spent executing (seconds)

  • cpu_info – Optional CPU info dictionary

Returns:

Tuple of (energy in microjoules, estimation model name, methodology tag)

pgsi_analyzer.measurement.estimators.estimate_macos(cpu_time_seconds: float, cpu_info: Dict[str, Any] | None = None) → Tuple[float, str, str][source]

macOS-specific energy estimation.

Uses CPU time-based estimation optimized for macOS systems. Includes special handling for Apple Silicon (M-series) processors.

Parameters:

  • cpu_time_seconds – CPU time spent executing (seconds)

  • cpu_info – Optional CPU info dictionary

Returns:

Tuple of (energy in microjoules, estimation model name, methodology tag)

pgsi_analyzer.measurement.estimators.estimate_energy(cpu_time_seconds: float, cpu_info: Dict[str, Any] | None = None) → Tuple[float, str, str][source]

Platform-agnostic energy estimation function.

Automatically selects the appropriate estimation method based on platform.

Parameters:

  • cpu_time_seconds – CPU time spent executing (seconds)

  • cpu_info – Optional CPU info dictionary

Returns:

Tuple of (energy in microjoules, estimation model name, methodology tag)

Examples

>>> energy, model, methodology = estimate_energy(1.0)
>>> energy > 0
True
>>> methodology in ("dataset_tdp", "generic_tdp")
True
pgsi_analyzer.measurement.estimators.estimate_energy_from_codecarbon(cpu_time_seconds: float, tracker: Any | None = None, emissions_kg: float | None = None, cpu_info: Dict[str, Any] | None = None) → Tuple[float, str, str][source]

Estimate energy using CodeCarbon tracker output when available.

If tracker metadata is insufficient to recover energy directly, this falls back to the CPU-TDP model to keep behavior deterministic across platforms.

CPU power resolver backed by a packaged dataset.

class pgsi_analyzer.measurement.cpu_power_resolver.CpuPowerResolution(tdp_watts: float, match_type: str, matched_model: str, source: str, normalized_query: str)[source]

Bases: object

Structured result returned by resolve_cpu_power().

Examples

>>> result = CpuPowerResolution(
...     tdp_watts=65.0,
...     match_type="default",
...     matched_model="generic_default",
...     source="generic_tdp_default",
...     normalized_query="unknown",
... )
>>> result.tdp_watts
65.0
tdp_watts: float
match_type: str
matched_model: str
source: str
normalized_query: str
pgsi_analyzer.measurement.cpu_power_resolver.resolve_cpu_power(cpu_name: str) → CpuPowerResolution[source]

Resolve CPU TDP from dataset with exact, regex, fuzzy, then default fallback.

This function is intentionally defensive because benchmark users run on many hosts with inconsistent CPU naming formats. The multi-step strategy maximizes match rate while still producing audit-friendly provenance when falling back to defaults.

Parameters:

cpu_name – CPU model string detected from the executing machine.

Returns:

Structured resolution metadata used by energy estimators.

Return type:

CpuPowerResolution

Examples

>>> result = resolve_cpu_power("Intel Core i7-10750H")
>>> result.match_type in {"exact", "regex", "fuzzy", "default"}
True