calculadorapnav 1.0.2

Una librería para cálculos de CO2

calculadorapnav

calculadorapnav is a Python library library based on the codecarbon to track and calculate energy consumption and carbon dioxide (CO₂e) emissions produced by computational processes. It is designed for scientific computing and machine learning environments, helping measure the environmental impact of executed models and programs.

---

Features

• Measures power consumption by CPU, GPU, and RAM.
• Calculates CO₂e emissions based on local and cloud energy intensity.
• Tracking modes: full machine or specific process.
• Data persistence in CSV files, logs, or HTTP endpoints.
• Decorators to automatically measure emissions in functions.
• Concurrency control to avoid simultaneous multiple executions.
• Flexible configuration via parameters or external files.
• Integration of geographic and cloud metadata for accuracy.

---

Installation

Install the library via pip (from PyPI or public repository if available):

pip install calculadorapnav

If not published, clone the repository and install locally:

git clone <repository_URL>
cd calculadorapnav
pip install -e .

Decorator Usage

Use the @track_emissions decorator to automatically measure energy consumption and CO₂e emissions during the execution of a function.

All the training or inference logic must be packaged in a function to which the already built model must be passed as a parameter in order to extract all the data related to the model.

To calculate energy, time, and emissions metrics by epoch and batch, "n_epoch" and "n_batch" must be passed as parameters in the decorator.

e.g.

from calculadorapnav import track_emissions
@track_emissions(
        type_process="training",
        n_epoch=10,
        n_batch=100,
        token="your_api_token",
        model_id="your_model_id",
        endpoint_url="https://fake_endpoint",
        measure_power_secs=0.5,
        output_dir="./results",
        output_file="results.csv",
        save_to_file=True,
        save_to_logger=True,
        logging_logger=None,
        offline=False,
        gpu_ids=None,
        log_level="INFO",
        on_csv_write=None,
        logger_preamble="Training process",
        default_cpu_power=50,
        pue=1.5,
        country_iso_code="ESP",
        allow_multiple_runs=False,
        region=None,
        cloud_provider=None,
        cloud_region=None
)
def train_model(model, n_epoch=10, n_batch=100):
        # muestra un entremiento por epocas de ejemplo
        for epoch in range(n_epoch):
                print(f"Training epoch {epoch + 1}/{n_epoch} with {n_batch} batches.")
                # Simulate training logic here
        pass

Parameters of the @track_emissions decorator

Parameters include:

• token (str, optional): API token for online communication.
• model_id (str, optional): Identifier for the model being tracked.
• endpoint_url (str, optional): URL for the API endpoint to send results.
• type_process (str): Type of process ("training" or "inference").
• type_ram (str, optional): Specific RAM type.
• n_batch (int, optional): Number of batches processed.
• n_epoch (int, optional): Number of epochs for training.
• measure_power_secs (float, optional): Power measurement interval in seconds (default 0.5).
• output_dir (str, optional): Directory path for storing results.
• output_file (str, optional): Filename to save data.
• save_to_file (bool, optional): Whether to save results in a CSV file.
• save_to_logger (bool, optional): Whether to save information to logs.
• output_handlers (List, optional): List of custom output handlers.
• logging_logger (LoggerOutput, optional): Logger instance for output.
• offline (bool, optional): Run in offline mode (no remote communication).
• gpu_ids (List, optional): List of GPU IDs to track.
• log_level (int or str, optional): Logging level.
• on_csv_write (Callable, optional): Callback or hook for CSV writing events.
• logger_preamble (str, optional): Preamble text for logger output.
• default_cpu_power (int, optional): Default CPU power value (in watts).
• pue (float, optional): Power Usage Effectiveness value.
• country_iso_code (str, optional): ISO code for the country (for energy intensity).
• allow_multiple_runs (bool, optional): Allow multiple instances running concurrently.
• region (str, optional): Geographic region for energy intensity.
• cloud_provider (str, optional): Cloud provider for energy intensity.
• cloud_region (str, optional): Cloud region for energy intensity.

---

Example usage of the decorator with parameters

The parameter type_process is required and determines the type of process to be measured; allowed values are "training" and "inference".

If the offline parameter is not entered as True by default the process will be online mode

Offline mode

When running offline, use:

• offline=True
• country_iso_code, e.g. "ESP"
• type_process (process type: "training" or "inference")
• When running on cloud services you can specify:
  • cloud_provider
  • cloud_region

For executions in private infrastructure:

from calculadorapnav import track_emissions

@track_emissions(
        offline=True,
        country_iso_code="ESP",
        type_process="training",
        # other parameters...
)
def train_model(model=model, n_epoch=10, n_batch=100):
        # Your model training code
        pass

train_model()

For cloud executions:

from calculadorapnav import track_emissions

@track_emissions(
        offline=True,
        cloud_provider="aws",
        cloud_region="eu-west-1",
        type_process="training",
        # other parameters...
)
def train_model(model=model, n_epoch=10, n_batch=100):
        # Your model training code
        pass

train_model()

---

Online mode

When running online, it requires at least:

Geospatial location will be obtained automatically. If it cannot be determined, "Madrid, Spain" will be used by default.

If the training or inference time is too long, it is recommended to enter the token when prompted by the console, not as a parameter in the decorator.

Example:

from calculadorapnav import track_emissions

@track_emissions(
    token="your_api_token",
    type_process="training",
    model_id="your_model_id",
    endpoint_url="https://fake_endpoint",
    # other parameters...
)
def train_model(model=model, n_epoch=10, n_batch=100):
    # Your model training code
    pass

train_model()

---

Extraction of general data from the model

To extract general model data and calculate energy consumption, time, and CO₂ emissions associated with the output, the following apply:

• "n_epoch" and "n_batch" must be passed as decorator inputs to estimate time and energy consumption.
• Model-related data like "model_size", "n_parameters", "n_features", or "size_weights" must be set outside the training/inference function and passed as arguments to the decorated function.

Example of custom model initialization:

from calculadorapnav import track_emissions
import tensorflow as tf
from tensorflow.keras import layers, losses, optimizers

# Define the model
class MiModelo(tf.keras.Model):
    def __init__(self):
        super(MiModelo, self).__init__()
        self.fc1 = layers.Dense(5, activation='relu')
        self.fc2 = layers.Dense(1)

    def call(self, x):
        x = self.fc1(x)
        x = self.fc2(x)
        return x

def build_model():
    modelo = MiModelo()
    criterio = losses.MeanSquaredError()
    optimizador = optimizers.SGD(learning_rate=0.01, momentum=0.9)
    return modelo, criterio, optimizador

@track_emissions(
    type_process="training",
    token="token",
    measure_power_secs=10,
    type_ram="DDR5",
    n_epoch=10,
    n_batch=100,
    endpoint_url="https://fake_endpoint",
)
def train(modelo, criterio, optimizador, inputs, targets):
    for epoch in range(10):
        with tf.GradientTape() as tape:
            outputs = modelo(inputs)
            loss = criterio(targets, outputs)

        grads = tape.gradient(loss, modelo.trainable_variables)
        optimizador.apply_gradients(zip(grads, modelo.trainable_variables))

        print(f'Epoch {epoch + 1}/10, Loss: {loss.numpy():.4f}')

def orquestador():
    inputs = tf.random.normal([100, 10])
    targets = tf.random.normal([100, 1])
    modelo, criterio, optimizador = build_model()
    train(modelo, criterio, optimizador, inputs, targets)

orquestador()

---

Output Handlers

The outputs available include:

save_to_file

Stores measurement results in CSV files, one for each process type ("training" and "inference"). The writing mode ("append" or "update") can be controlled with the on_csv_write parameter.

save_endpoint

Sends results to the calculadorapnav API for output calculations and result storage in a database. Available only in online mode. Requires token and model_id. These will be requested if not provided.

---

Advanced Usage

• Metadata capture functions.
• Locking mechanisms to control concurrency.
• Persistence and output customization.
• Online mode uses up-to-date emission data from the calculadorapnav API.
• Offline mode uses local stored data for emission estimates.
• Easy instrumentation of existing code with @track_emissions.
• Results include hardware energy consumption, CO₂e emission estimates and general data of the model.

---

License

This project is licensed under the MIT License. See the LICENSE file for details.

---

Contact

https://algoritmosverdes.gob.es/es