feat: add scheduler job metadata to output

codecarbon/core/api_client.py

@@ -22,6 +22,11 @@
 from codecarbon.external.logger import logger
 
 
+def _round_or_none(value: float | None) -> float | None:
+    """Round a coordinate, keeping None when it is unknown."""
+    return None if value is None else round(value, 1)
+
+
 def get_datetime_with_timezone():
     import arrow
 
@@ -242,8 +247,8 @@ def _create_run(self, experiment_id: str):
                 gpu_count=self.conf.get("gpu_count"),
                 gpu_model=self.conf.get("gpu_model"),
                 # Reduce precision for Privacy
-                longitude=round(self.conf.get("longitude", 0), 1),
-                latitude=round(self.conf.get("latitude", 0), 1),
+                longitude=_round_or_none(self.conf.get("longitude")),
+                latitude=_round_or_none(self.conf.get("latitude")),
                 region=self.conf.get("region"),
                 provider=self.conf.get("provider"),
                 ram_total_size=self.conf.get("ram_total_size"),

codecarbon/core/schedulers.py

@@ -0,0 +1,27 @@
+"""
+Detection of the HPC batch scheduler job identity.
+
+Schedulers export the identity of the running job into the environment of every
+job step, so no scheduler library is needed: reading ``os.environ`` is enough.
+
+SLURM is detected automatically. Any other scheduler is supported through the
+generic ``CODECARBON_SCHEDULER_JOB_ID`` environment variable, which also takes
+precedence over the auto-detected value, so a site can map its own scheduler in
+one line of shell.
+"""
+
+import os
+
+
+def detect_scheduler_job_id() -> str:
+    """
+    Return the batch scheduler job ID of the current process, or "" outside of
+    a batch job.
+
+    Only the job ID is collected: it is the join key, and everything else the
+    scheduler knows about the job (name, account, partition, node) is one
+    ``sacct -j <id>`` away.
+    """
+    return os.environ.get("CODECARBON_SCHEDULER_JOB_ID") or os.environ.get(
+        "SLURM_JOB_ID", ""
+    )

codecarbon/emissions_tracker.py

@@ -21,6 +21,7 @@
 
 from codecarbon._version import __version__
 from codecarbon.core.config import get_hierarchical_config, normalize_gpu_ids
+from codecarbon.core.schedulers import detect_scheduler_job_id
 from codecarbon.core.units import Energy, Power, Time, Water
 from codecarbon.core.util import count_cpus, count_physical_cpus, suppress
 from codecarbon.external.hardware import CPU, GPU, AppleSiliconChip
@@ -600,6 +601,9 @@ def __init__(
         assert self._tracking_mode in ["machine", "process"]
         set_logger_level(self._log_level)
         set_logger_format(self._logger_preamble)
+        # The job identity cannot change during the process' life, so it is read
+        # once here rather than on every flush.
+        self._scheduler_job_id = detect_scheduler_job_id()
         self._initialize_runtime_state()
         self._initialize_scheduler_state()
         self._initialize_emissions_context()
@@ -1098,6 +1102,7 @@ def _prepare_emissions_data(self) -> EmissionsData:
             tracking_mode=self._conf.get("tracking_mode"),
             pue=self._pue,
             wue=self._wue,
+            scheduler_job_id=self._scheduler_job_id,
         )
         logger.debug(total_emissions)
         return total_emissions

codecarbon/output_methods/emissions_data.py

@@ -47,6 +47,8 @@ class EmissionsData:
     on_cloud: str = "N"
     pue: float = 1
     wue: float = 0
+    # Batch scheduler job ID, empty outside of an HPC job.
+    scheduler_job_id: str = ""
 
     @property
     def values(self) -> OrderedDict:

codecarbon/output_methods/file.py

@@ -108,22 +108,19 @@ def out(self, total: EmissionsData, _):
         else:
             df = pd.read_csv(self.save_file_path)
             df_run = df.loc[df.run_id == total.run_id]
-            if len(df_run) < 1:
-                df = pd.concat([df, new_df])
-            elif len(df_run) > 1:
+            if len(df_run) > 1:
                 logger.warning(
                     f"CSV contains more than 1 ({len(df_run)})"
                     + f" rows with current run ID ({total.run_id})."
                     + "Appending instead of updating."
                 )
                 df = pd.concat([df, new_df])
             else:
-                update_values = {}
-                for col, val in dict(total.values).items():
-                    update_values[col] = df[col].dtype.type(val)
-                df.loc[df.run_id == total.run_id, update_values.keys()] = (
-                    update_values.values()
-                )
+                # Drop the previous row for this run (if any) and re-append it.
+                # Assigning column by column would coerce values to the dtype
+                # pandas inferred for the existing column, which breaks for
+                # columns that are empty in every row (read back as float64).
+                df = pd.concat([df.loc[df.run_id != total.run_id], new_df])
             df.to_csv(self.save_file_path, index=False)
 
     def task_out(self, data: List[TaskEmissionsData], experiment_name: str):

docs/how-to/agent-instructions.md

@@ -60,8 +60,7 @@ Here's what you need to know to navigate and contribute effectively.
    # Run specific test
    uv run pytest tests/test_emissions_tracker.py
 
-   # Lint and format
-   uv run task lint
+   # Lint and format (runs the pre-commit hooks, which rewrite files in place)
    uv run task format
    ```
 
@@ -112,7 +111,7 @@ Here's what you need to know to navigate and contribute effectively.
 1. **Check existing tests** in `tests/` for similar functionality
 2. **Add unit tests** first (test-driven development)
 3. **Update documentation** if public interface changes
-4. **Follow coding style**: Use `uv run task format` and `uv run task lint`
+4. **Follow coding style**: Use `uv run task format`
 
 ### API Development
 1. **Follow FastAPI patterns** - see routers in `carbonserver/carbonserver/api/routers/`
@@ -134,8 +133,7 @@ uv run task -l
 
 # Main tasks:
 # - test-package: Core package testing
-# - lint: Code linting and style checks
-# - format: Code formatting
+# - format: Lint and format, by running the pre-commit hooks
 # - test-api-unit: API unit tests
 # - test-api-integ: API integration tests
 # - dashboard: Run API locally

docs/how-to/slurm.md

@@ -160,6 +160,43 @@ tail -f logs/<job_id>.out
 sinfo
 ```
 
+## The job ID in the output
+
+When CodeCarbon runs inside a SLURM job step it reads `SLURM_JOB_ID` from the
+environment SLURM already provides and stores it on every emissions record, in the
+`scheduler_job_id` column. There is nothing to enable and no code to change. Outside
+of a job the column is empty, so nothing changes for non-HPC users.
+
+This makes `emissions.csv` directly joinable against SLURM accounting, which knows
+everything else about the job already:
+
+```bash
+sacct -j 1234567 --format=JobID,JobName,Account,Partition,Elapsed,AllocTRES --parsable2
+```
+
+!!! tip "You no longer need `CODECARBON_PROJECT_NAME=$SLURM_JOB_ID`"
+    Overloading the project name with the job ID used to be the only way to tell runs
+    apart. Keep `project_name` for your project and use `scheduler_job_id` for the job.
+
+### Other schedulers
+
+Set `CODECARBON_SCHEDULER_JOB_ID` and the column is filled the same way. This is how
+PBS, LSF or OAR sites get it without CodeCarbon needing to know about their scheduler:
+
+```bash
+export CODECARBON_SCHEDULER_JOB_ID=$PBS_JOBID
+```
+
+It takes precedence over the auto-detected SLURM value, so it also works for
+correcting the field on a site whose SLURM configuration is unusual.
+
+!!! warning "One tracker per node"
+    Power is a property of the node, not of a rank. If you launch CodeCarbon on every
+    rank of a multi-node job in `machine` tracking mode, each one measures the whole
+    node and your total is multiplied by the number of ranks. Start the tracker on one
+    rank per node (for example when `SLURM_LOCALID` is `0`), or use `process` tracking
+    mode.
+
 ## Troubleshooting
 
 ### Error: AMD GPU detected but amdsmi is not properly configured

docs/reference/output.md

@@ -64,6 +64,17 @@ The package has an in-built logger that logs data into a CSV file named `emissio
 | gpu_utilization_percent | Average GPU utilization during tracking period (%) |
 | ram_utilization_percent | Average RAM utilization during tracking period (%) |
 | ram_used_gb | Average RAM used during tracking period (GB) |
+| scheduler_job_id | Batch scheduler job ID, e.g. the value of `SLURM_JOB_ID`. Empty outside of an HPC job |
+
+`scheduler_job_id` is filled in automatically, see
+[Using CodeCarbon on SLURM](../how-to/slurm.md#the-job-id-in-the-output).
+
+!!! warning "Existing `emissions.csv` files are rotated once"
+
+    This column changes the CSV header. On the first run after upgrading,
+    CodeCarbon backs up an existing `emissions.csv` next to it and starts a new
+    file with the new header. Nothing is lost, but a pipeline reading a fixed
+    path will see a file with only the new rows in it.
 
 !!! note
     Developers can enhance the Output interface by implementing a custom class that extends `BaseOutput` at `codecarbon/output.py`. For example, to log into a database.

pyproject.toml

@@ -84,7 +84,6 @@ dev = [
     "taskipy",
     "bumpver",
     "pre-commit",
-    "ruff",
     "black",
     "mypy",
     "pytest",
@@ -138,10 +137,10 @@ carbonserver-api-requirements = "uv pip compile carbonserver/pyproject.toml --ge
 build-doc = "uv run --only-group doc zensical build -f mkdocs.yml && uv run --only-group doc python scripts/check_docs_links.py site"
 precommit-install = "pre-commit install"
 precommit-update = "pre-commit autoupdate"
-precommit = "c"
 mypy-check = "mypy -m codecarbon --ignore-missing-imports --no-strict-optional --disable-error-code attr-defined --disable-error-code assignment --disable-error-code misc"
-lint = "black --check --diff . && ruff check . && mypy ."
-format = "black . && ruff check --fix --exit-non-zero-on-fix ."
+# No `lint` task: the pre-commit hooks fix what they can, so this rewrites
+# files rather than only reporting. Use `mypy-check` for a read-only check.
+format = "pre-commit run --all-files"
 test-package = "CODECARBON_ALLOW_MULTIPLE_RUNS=True pytest --ignore=tests/test_viz_data.py -vv -m 'not integ_test' tests/"
 test-coverage = "CODECARBON_ALLOW_MULTIPLE_RUNS=True pytest --cov --cov-report=xml --ignore=tests/test_viz_data.py -vv -m 'not integ_test' tests/"
 test-package-integ = "CODECARBON_ALLOW_MULTIPLE_RUNS=True python -m pytest -vv tests/"

tests/output_methods/test_file.py

@@ -169,6 +169,62 @@ def test_file_output_out_update_file_exists_one_matchingrows(self):
         df = pd.read_csv(os.path.join(self.temp_dir, "test.csv"))
         self.assertEqual(df["cpu_power"].iloc[0], 2)
 
+    def test_file_output_out_update_with_always_empty_columns(self):
+        """Regression test: updating a run must not coerce incoming values to the
+        dtype pandas inferred for the existing column.
+
+        An OfflineEmissionsTracker leaves longitude/latitude empty, and
+        gpu_count/gpu_model are empty on CPU-only machines. Such columns are read
+        back from the CSV as float64, so the previous implementation evaluated
+        numpy.float64("") / numpy.float64(None) and raised on the second write.
+        """
+        empty_columns_data = EmissionsData(
+            timestamp="2023-01-01T00:00:00",
+            project_name="test_project",
+            run_id="test_run_id",
+            experiment_id="test_experiment_id",
+            duration=10,
+            emissions=0.5,
+            emissions_rate=0.05,
+            cpu_power=20,
+            gpu_power=0,
+            ram_power=5,
+            cpu_energy=200,
+            gpu_energy=0,
+            ram_energy=50,
+            energy_consumed=250,
+            water_consumed=0.1,
+            country_name="Testland",
+            country_iso_code="TS",
+            region="Test Region",
+            cloud_provider="",
+            cloud_region="",
+            os="TestOS",
+            python_version="3.8",
+            codecarbon_version="2.0",
+            cpu_count=4,
+            cpu_model="Test CPU",
+            gpu_count=None,
+            gpu_model=None,
+            longitude="",
+            latitude="",
+            ram_total_size=16,
+            tracking_mode="machine",
+        )
+
+        file_output = FileOutput("test.csv", self.temp_dir, on_csv_write="update")
+        file_output.out(empty_columns_data, None)
+
+        empty_columns_data.cpu_power = 2
+        # This should not raise.
+        file_output.out(empty_columns_data, None)
+
+        df = pd.read_csv(os.path.join(self.temp_dir, "test.csv"))
+        self.assertEqual(len(df), 1)
+        self.assertEqual(df["cpu_power"].iloc[0], 2)
+        self.assertIn("longitude", df.columns)
+        self.assertIn("gpu_model", df.columns)
+
     # def test_file_output_out_consistent_column_ordering(self):
     #     file_output = FileOutput("test.csv", self.temp_dir, on_csv_write="append")
     #     file_output.out(self.emissions_data, None)

tests/test_api_call.py

@@ -138,6 +138,39 @@ def test_call_api(self):
             assert payload["ram_utilization_percent"] == 56.5
             assert payload["wue"] == 0.8
 
+    def test_create_run_rounds_coordinates(self):
+        with requests_mock.Mocker() as m:
+            m.post("http://test.com/runs", json={"id": "run-1"}, status_code=201)
+            api = ApiClient(
+                endpoint_url="http://test.com",
+                experiment_id="exp-1",
+                conf=conf,
+                create_run_automatically=False,
+            )
+
+            api._create_run("exp-1")
+
+            payload = m.last_request.json()
+            self.assertEqual(payload["longitude"], -7.6)
+            self.assertEqual(payload["latitude"], 33.6)
+
+    def test_create_run_keeps_unknown_coordinates_null(self):
+        offline_conf = dict(conf, longitude=None, latitude=None)
+        with requests_mock.Mocker() as m:
+            m.post("http://test.com/runs", json={"id": "run-1"}, status_code=201)
+            api = ApiClient(
+                endpoint_url="http://test.com",
+                experiment_id="exp-1",
+                conf=offline_conf,
+                create_run_automatically=False,
+            )
+
+            self.assertEqual(api._create_run("exp-1"), "run-1")
+
+            payload = m.last_request.json()
+            self.assertIsNone(payload["longitude"])
+            self.assertIsNone(payload["latitude"])
+
     def test_check_auth_raises_on_error(self):
         with requests_mock.Mocker() as m:
             m.get("http://test.com/auth/check", text="bad", status_code=401)

tests/test_data/emissions_valid_headers.csv

@@ -1,2 +1,2 @@
-timestamp,project_name,run_id,experiment_id,duration,emissions,emissions_rate,cpu_power,gpu_power,ram_power,cpu_energy,gpu_energy,ram_energy,energy_consumed,water_consumed,country_name,country_iso_code,region,cloud_provider,cloud_region,os,python_version,codecarbon_version,cpu_count,cpu_model,gpu_count,gpu_model,longitude,latitude,ram_total_size,tracking_mode,cpu_utilization_percent,gpu_utilization_percent,ram_utilization_percent,ram_used_gb,on_cloud,pue,wue
-2021-09-23T15:04:51,codecarbon,0a578547-1d6b-4e2f-be0c-7ad10f2f7c97,test,161.20380687713623,0.0004490989249167,0.0027859076880178,0.269999999999999,0.0,12.884901888000002,0.0,0,0.00057442898176,0.00057442898176,0.1,Morocco,MAR,casablanca-settat,,,macOS-10.15.7-x86_64-i386-64bit,3.8.0,2.1.3,12,Intel(R) Core(TM) i7-8850H CPU @ 2.60GHz,,,-7.9084,33.5932,,machine,0.0,0.0,0.0,0.0,N,1.0,0.0
+timestamp,project_name,run_id,experiment_id,duration,emissions,emissions_rate,cpu_power,gpu_power,ram_power,cpu_energy,gpu_energy,ram_energy,energy_consumed,water_consumed,country_name,country_iso_code,region,cloud_provider,cloud_region,os,python_version,codecarbon_version,cpu_count,cpu_model,gpu_count,gpu_model,longitude,latitude,ram_total_size,tracking_mode,cpu_utilization_percent,gpu_utilization_percent,ram_utilization_percent,ram_used_gb,on_cloud,pue,wue,scheduler_job_id
+2021-09-23T15:04:51,codecarbon,0a578547-1d6b-4e2f-be0c-7ad10f2f7c97,test,161.20380687713623,0.0004490989249167,0.0027859076880178,0.269999999999999,0.0,12.884901888000002,0.0,0,0.00057442898176,0.00057442898176,0.1,Morocco,MAR,casablanca-settat,,,macOS-10.15.7-x86_64-i386-64bit,3.8.0,2.1.3,12,Intel(R) Core(TM) i7-8850H CPU @ 2.60GHz,,,-7.9084,33.5932,,machine,0.0,0.0,0.0,0.0,N,1.0,0.0,

tests/test_schedulers.py

@@ -0,0 +1,47 @@
+import unittest
+from unittest import mock
+
+from codecarbon.core.schedulers import detect_scheduler_job_id
+
+
+class TestSchedulers(unittest.TestCase):
+    @mock.patch.dict("os.environ", {"SLURM_JOB_ID": "1234567"}, clear=True)
+    def test_slurm_job_id_is_detected(self):
+        self.assertEqual("1234567", detect_scheduler_job_id())
+
+    @mock.patch.dict("os.environ", {}, clear=True)
+    def test_no_scheduler_env_is_inert(self):
+        self.assertEqual("", detect_scheduler_job_id())
+
+    @mock.patch.dict(
+        "os.environ", {"CODECARBON_SCHEDULER_JOB_ID": "99.pbsserver"}, clear=True
+    )
+    def test_generic_env_contract_without_slurm(self):
+        self.assertEqual("99.pbsserver", detect_scheduler_job_id())
+
+    @mock.patch.dict(
+        "os.environ",
+        {"SLURM_JOB_ID": "1234567", "CODECARBON_SCHEDULER_JOB_ID": "override"},
+        clear=True,
+    )
+    def test_generic_env_contract_overrides_slurm(self):
+        self.assertEqual("override", detect_scheduler_job_id())
+
+
+class TestSchedulerJobIdOnEmissionsData(unittest.TestCase):
+    @mock.patch.dict("os.environ", {"SLURM_JOB_ID": "1234567"}, clear=True)
+    def test_job_id_reaches_the_emissions_data(self):
+        from codecarbon.emissions_tracker import OfflineEmissionsTracker
+
+        tracker = OfflineEmissionsTracker(
+            country_iso_code="FRA", output_methods=[], allow_multiple_runs=True
+        )
+        tracker.start()
+        try:
+            data = tracker._prepare_emissions_data()
+        finally:
+            tracker.stop()
+
+        self.assertEqual("1234567", data.scheduler_job_id)
+        # The new field must be part of the CSV columns.
+        self.assertIn("scheduler_job_id", data.values)