Skip to content

Emissions tracker plugin

Emissions Tracker Plugin Module.

This module defines a plugin for tracking carbon emissions using the codecarbon library.

For usage examples, refer to the docstring of the EmissionsTrackerPlugin class.

class EmissionsTrackerPlugin(*args, **kwargs) #

Bases: Plugin

A plugin that tracks carbon emissions using codecarbon library.

Usage Example:

from concurrent.futures import ThreadPoolExecutor
from amltk.scheduling import Scheduler
from amltk.scheduling.plugins.emissions_tracker_plugin import EmissionsTrackerPlugin

def some_function(x: int) -> int:
    return x * 2

executor = ThreadPoolExecutor(max_workers=1)

# Create a Scheduler instance with the executor
scheduler = Scheduler(executor=executor)

# Create a task with the emissions tracker plugin
task = scheduler.task(some_function, plugins=[
    # Pass any codecarbon parameters as args here
    EmissionsTrackerPlugin(log_level="info", save_to_file=False)
])

@scheduler.on_start
def on_start():
    task.submit(5)  # Submit any args here

@task.on_submitted
def on_submitted(future, *args, **kwargs):
    print(f"Task was submitted", future, args, kwargs)

@task.on_done
def on_done(future):
    # Result is the return value of the function
    print("Task done: ", future.result())

scheduler.run()
PARAMETER DESCRIPTION
*args

Additional arguments to pass to codecarbon library.

TYPE: Any DEFAULT: ()

**kwargs

Additional keyword arguments to pass to codecarbon library.

TYPE: Any DEFAULT: {}

You can pass any codecarbon parameters as args to EmissionsTrackerPlugin. Please refer to the official codecarbon documentation for more details: mlco2.github.io/codecarbon/parameters.html

Source code in src/amltk/scheduling/plugins/emissions_tracker_plugin.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def __init__(self, *args: Any, **kwargs: Any):
    """Initialize the EmissionsTrackerPlugin.

    Args:
        *args: Additional arguments to pass to codecarbon library.
        **kwargs: Additional keyword arguments to pass to codecarbon library.

    You can pass any codecarbon parameters as args to EmissionsTrackerPlugin.
    Please refer to the official codecarbon documentation for more details:
    https://mlco2.github.io/codecarbon/parameters.html
    """
    super().__init__()
    self.task: Task | None = None
    self.codecarbon_args = args
    self.codecarbon_kwargs = kwargs

name: ClassVar
classvar attr #

The name of the plugin.

def attach_task(task) #

Attach the plugin to a task.

Source code in src/amltk/scheduling/plugins/emissions_tracker_plugin.py 
112
113
114
def attach_task(self, task: Task) -> None:
    """Attach the plugin to a task."""
    self.task = task

def pre_submit(fn, *args, **kwargs) #

Pre-submit hook.

Source code in src/amltk/scheduling/plugins/emissions_tracker_plugin.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def pre_submit(
    self,
    fn: Callable[P, R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> tuple[Callable[P, R], tuple, dict]:
    """Pre-submit hook."""
    wrapped_f = _EmissionsTrackerWrapper(
        fn,
        self.task,
        *self.codecarbon_args,
        **self.codecarbon_kwargs,
    )
    return wrapped_f, args, kwargs

def copy() #

Return a copy of the plugin.

Source code in src/amltk/scheduling/plugins/emissions_tracker_plugin.py 
131
132
133
def copy(self) -> Self:
    """Return a copy of the plugin."""
    return self.__class__(*self.codecarbon_args, **self.codecarbon_kwargs)

def __rich__() #

Return a rich panel.

Source code in src/amltk/scheduling/plugins/emissions_tracker_plugin.py 
135
136
137
138
139
140
141
142
143
def __rich__(self) -> Panel:
    """Return a rich panel."""
    from rich.panel import Panel

    return Panel(
        f"codecarbon_args: {self.codecarbon_args} "
        f"codecarbon_kwargs: {self.codecarbon_kwargs}",
        title=f"Plugin {self.name}",
    )