Added support to CodeCarbon.

Author: fmind

poetry.lock

@@ -33,6 +33,7 @@ pydantic = "^2.5.3"
 pydantic-settings = "^2.1.0"
 scikit-learn = "^1.4.0"
 mlflow-skinny = "^2.10.2"
+codecarbon = "^2.3.4"
 
 [tool.poetry.group.dev.dependencies]
 invoke = "^2.2.0"
@@ -50,6 +51,11 @@ pytest-xdist = "^3.5.0"
 pandera = { extras = ["mypy"], version = "^0.18.0" }
 ruff = "^0.2.2"
 
+[tool.poetry.group.carbons.dependencies]
+dash = "^2.15.0"
+dash-bootstrap-components = "^1.5.0"
+fire = "^0.5.0"
+
 [tool.poetry.group.notebooks.dependencies]
 ipykernel = "^6.29.0"
 nbformat = "^5.9.2"
@@ -59,12 +65,16 @@ nbformat = "^5.9.2"
 [tool.coverage.run]
 branch = true
 source = ["src"]
+omit = ["__main__.py"]
 
 [tool.mypy]
 check_untyped_defs = true
 ignore_missing_imports = true
 plugins = ["pandera.mypy", "pydantic.mypy"]
 
+[tool.pytest.ini_options]
+addopts = "--numprocesses='auto'"
+
 [tool.ruff]
 fix = true
 line-length = 100

pyproject.toml

@@ -33,6 +33,7 @@ class Job(abc.ABC, pdt.BaseModel, strict=True):
     KIND: str
 
     logger_service: services.LoggerService = services.LoggerService()
+    carbon_service: services.CarbonService = services.CarbonService()
     mlflow_service: services.MLflowService = services.MLflowService()
 
     def __enter__(self) -> T.Self:
@@ -41,7 +42,10 @@ def __enter__(self) -> T.Self:
         Returns:
             T.Self: return the current object.
         """
-        self.logger_service.start()
+        self.logger_service.start()  # start then log
+        logger.debug("[START] Logger service: {}", self.logger_service)
+        logger.debug("[START] Carbon service: {}", self.carbon_service)
+        self.carbon_service.start()
         logger.debug("[START] MLflow service: {}", self.mlflow_service)
         self.mlflow_service.start()
         return self
@@ -59,6 +63,9 @@ def __exit__(self, exc_type, exc_value, traceback) -> T.Literal[False]:
         """
         logger.debug("[STOP] MLflow service: {}", self.mlflow_service)
         self.mlflow_service.stop()
+        logger.debug("[STOP] Carbon service: {}", self.carbon_service)
+        self.carbon_service.stop()
+        logger.debug("[STOP] Logger service: {}", self.carbon_service)
         self.logger_service.stop()
         return False
 

src/bikes/jobs.py

@@ -3,9 +3,11 @@
 # %% IMPORTS
 
 import abc
+import os
 import sys
 import typing as T
 
+import codecarbon as cc
 import mlflow
 import pydantic as pdt
 from loguru import logger
@@ -78,6 +80,36 @@ def start(self) -> None:
         logger.add(**config)
 
 
+class CarbonService(Service):
+    """Service for tracking carbon emissions."""
+
+    # public
+    # - inputs
+    log_level: str = "ERROR"
+    project_name: str = "bikes"
+    measure_power_secs: int = 5
+    # - outputs
+    output_dir: str = "outputs"
+    output_file: str = "emissions.csv"
+    on_csv_write: str = "append"
+    # - offline
+    country_iso_code: str = "LUX"
+    # private
+    _tracker: cc.OfflineEmissionsTracker | None = None
+
+    def start(self):
+        """Start the carbon service."""
+        os.makedirs(self.output_dir, exist_ok=True)  # create output dir
+        self._tracker = cc.OfflineEmissionsTracker(**self.model_dump())
+        self._tracker.start()
+
+    def stop(self):
+        """Stop the carbon service."""
+        assert self._tracker, "Carbon tracker should be started!"
+        self._tracker.flush()
+        self._tracker.stop()
+
+
 class MLflowService(Service):
     """Service for MLflow tracking and registry.
 

src/bikes/services.py

@@ -6,14 +6,15 @@
 
 from invoke import Collection
 
-from . import checks, cleans, containers, dags, docs, formats, installs, packages
+from . import carbons, checks, cleans, containers, dags, docs, formats, installs, packages
 
 # %% NAMESPACES
 
 ns = Collection()
 
 # %% COLLECTIONS
 
+ns.add_collection(carbons)
 ns.add_collection(checks)
 ns.add_collection(cleans)
 ns.add_collection(dags, default=True)

tasks/__init__.py

@@ -0,0 +1,19 @@
+"""Carbons tasks for pyinvoke."""
+
+# %% IMPORTS
+
+from invoke import task
+from invoke.context import Context
+
+# %% TASKS
+
+
+@task
+def board(ctx: Context, filepath: str = "outputs/emissions.csv", port: int = 8050) -> None:
+    """Visualize caron emissions data at file path from the carbon board app."""
+    ctx.run(f"poetry run carbonboard --filepath={filepath} --port={port}")
+
+
+@task(pre=[board], default=True)
+def all(_: Context) -> None:
+    """Run all carbon tasks."""

tasks/carbons.py

@@ -8,7 +8,6 @@
 # %% CONFIGS
 
 COVERAGE_FAIL_UNDER = 80
-PYTEST_N_PROCESSES = "auto"
 
 # %% TASKS
 
@@ -40,16 +39,13 @@ def code(ctx: Context) -> None:
 @task
 def test(ctx: Context) -> None:
     """Check the tests with pytest."""
-    ctx.run("poetry run pytest --numprocesses={PYTEST_N_PROCESSES} tests/")
+    ctx.run("poetry run pytest tests/")
 
 
 @task
 def coverage(ctx: Context) -> None:
     """Check the coverage with coverage."""
-    ctx.run(
-        f"poetry run pytest --numprocesses={PYTEST_N_PROCESSES}"
-        f" --cov=src/ --cov-fail-under={COVERAGE_FAIL_UNDER} tests/"
-    )
+    ctx.run(f"poetry run pytest --cov=src/ --cov-fail-under={COVERAGE_FAIL_UNDER} tests/")
 
 
 @task(pre=[poetry, format, type, code, coverage], default=True)

tasks/checks.py

@@ -8,7 +8,6 @@
 import mlflow
 import omegaconf
 import pytest
-
 from bikes import datasets, metrics, models, registers, schemas, searchers, services, splitters
 
 # %% CONFIGS
@@ -60,6 +59,12 @@ def outputs_path(data_path: str) -> str:
     return os.path.join(data_path, "outputs.parquet")
 
 
+@pytest.fixture(scope="function")
+def tmp_carbon_path(tmp_path: str) -> str:
+    """Return a tmp path of the carbon folder."""
+    return os.path.join(tmp_path, "carbons")
+
+
 @pytest.fixture(scope="function")
 def tmp_outputs_path(tmp_path: str) -> str:
     """Return a tmp path for the outputs dataset."""
@@ -232,13 +237,22 @@ def default_metric() -> metrics.SklearnMetric:
 
 
 @pytest.fixture(scope="session", autouse=True)
-def logger_service():
+def logger_service() -> services.LoggerService:
     """Return and start the logger service."""
     service = services.LoggerService(colorize=False, diagnose=True)
     service.start()  # ready to be used
     return service
 
 
+@pytest.fixture(scope="function")
+def carbon_service(tmp_carbon_path: str) -> T.Generator[services.CarbonService, None, None]:
+    """Return and start the carbon service."""
+    service = services.CarbonService(output_dir=tmp_carbon_path)
+    service.start()  # ready to be used
+    yield service
+    service.stop()
+
+
 @pytest.fixture(scope="function", autouse=True)
 def mlflow_service(tmp_path: str) -> services.MLflowService:
     """Return and start the mlflow service."""

tests/conftest.py

@@ -16,6 +16,7 @@ def test_tuning_job(
     default_searcher: searchers.GridCVSearcher,
     time_series_splitter: splitters.TimeSeriesSplitter,
     logger_service: services.LoggerService,
+    carbon_service: services.CarbonService,
     mlflow_service: services.MLflowService,
 ):
     # given
@@ -30,6 +31,7 @@ def test_tuning_job(
         searcher=default_searcher,
         splitter=time_series_splitter,
         logger_service=logger_service,
+        carbon_service=carbon_service,
         mlflow_service=mlflow_service,
     )
     mlflow_client = mlflow_service.client()
@@ -81,6 +83,7 @@ def test_training_job(
     default_signer: registers.Signer,
     train_test_splitter: splitters.TrainTestSplitter,
     logger_service: services.LoggerService,
+    carbon_service: services.CarbonService,
     mlflow_service: services.MLflowService,
 ):
     # given
@@ -98,6 +101,7 @@ def test_training_job(
         splitter=train_test_splitter,
         registry_alias=registry_alias,
         logger_service=logger_service,
+        carbon_service=carbon_service,
         mlflow_service=mlflow_service,
     )
     mlflow_client = mlflow_service.client()
@@ -187,6 +191,7 @@ def test_inference_job(
     default_alias: str,
     default_mlflow_model_version: registers.Version,
     logger_service: services.LoggerService,
+    carbon_service: services.CarbonService,
     mlflow_service: services.MLflowService,
 ):
     # given
@@ -197,6 +202,7 @@ def test_inference_job(
         registry_alias=default_alias,
         loader=default_loader,
         logger_service=logger_service,
+        carbon_service=carbon_service,
         mlflow_service=mlflow_service,
     )
     # when

tests/test_jobs.py

@@ -1,5 +1,7 @@
 # %% IMPORTS
 
+import os
+
 import mlflow
 from bikes import services
 from loguru import logger
@@ -21,6 +23,20 @@ def test_logger_service(capsys):
     assert "DEBUG" not in capture.out, "Debug should not be logged!"
 
 
+def test_carbon_service(tmp_carbon_path: str):
+    # given
+    output_dir = tmp_carbon_path
+    output_file = "emissions-test.csv"
+    output_path = os.path.join(output_dir, output_file)
+    service = services.CarbonService(output_dir=output_dir, output_file=output_file)
+    # when
+    service.start()
+    service.stop()
+    # then
+    assert os.path.exists(output_path), "Output path should be created!"
+    assert os.stat(output_path).st_size > 0, "Output path should not be empty!"
+
+
 def test_mlflow_service(mlflow_service: services.MLflowService):
     # given
     service = mlflow_service