From f8f0e283521e996797b59bd1f13b5c17136e7bd7 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 17:51:45 +0200 Subject: [PATCH 1/3] feat: add SCI report output method Emit an ISO/IEC 21031 Software Carbon Intensity report alongside the existing output methods. E and I come from the measured run; R and M are user declarations and are reported as undeclared rather than guessed. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/emissions_tracker.py | 12 +- codecarbon/output_methods/base_output.py | 3 +- codecarbon/output_methods/sci.py | 265 +++++++++++++++++++++++ docs/how-to/examples.md | 1 + docs/reference/output.md | 90 +++++++- examples/sci_output.py | 39 ++++ tests/test_sci_output.py | 253 ++++++++++++++++++++++ 7 files changed, 660 insertions(+), 3 deletions(-) create mode 100644 codecarbon/output_methods/sci.py create mode 100644 examples/sci_output.py create mode 100644 tests/test_sci_output.py diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 96ed00c91..29d7d9ef4 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -446,7 +446,7 @@ def __init__( EmissionsTracker(output_methods=[OutputMethod.CSV, OutputMethod.API]) Available values: ``CSV``, ``API``, ``LOGGER``, - ``PROMETHEUS``, ``LOGFIRE``, ``BOAMPS``. + ``PROMETHEUS``, ``LOGFIRE``, ``BOAMPS``, ``SCI``. When provided, the individual ``save_to_*`` flags are ignored. Defaults to ``[OutputMethod.CSV]``. Can also be set in config as a comma-separated string: @@ -620,6 +620,7 @@ def _init_output_methods(self, *, api_key: str = None): from codecarbon.output_methods.http import CodeCarbonAPIOutput, HTTPOutput from codecarbon.output_methods.metrics.logfire import LogfireOutput from codecarbon.output_methods.metrics.prometheus import PrometheusOutput + from codecarbon.output_methods.sci import SCIOutput methods = set(self._output_methods) if self._output_methods else set() @@ -668,6 +669,15 @@ def _init_output_methods(self, *, api_key: str = None): if OutputMethod.BOAMPS in methods: self._output_handlers.append(BoAmpsOutput(output_dir=self._output_dir)) + if OutputMethod.SCI in methods: + context_file = self._external_conf.get("sci_context_file") + if context_file: + self._output_handlers.append( + SCIOutput.from_file(context_file, output_dir=self._output_dir) + ) + else: + self._output_handlers.append(SCIOutput(output_dir=self._output_dir)) + def get_detected_hardware(self) -> Dict[str, Any]: """ Get the detected hardware. diff --git a/codecarbon/output_methods/base_output.py b/codecarbon/output_methods/base_output.py index 373d23edd..6bc9ee205 100644 --- a/codecarbon/output_methods/base_output.py +++ b/codecarbon/output_methods/base_output.py @@ -15,7 +15,7 @@ class OutputMethod(str, Enum): ) Available values: ``CSV``, ``API``, ``LOGGER``, ``PROMETHEUS``, - ``LOGFIRE``, ``BOAMPS``. + ``LOGFIRE``, ``BOAMPS``, ``SCI``. .. note:: HTTP output is not configured here; it is enabled by setting the @@ -28,6 +28,7 @@ class OutputMethod(str, Enum): PROMETHEUS = "prometheus" LOGFIRE = "logfire" BOAMPS = "boamps" + SCI = "sci" class BaseOutput: diff --git a/codecarbon/output_methods/sci.py b/codecarbon/output_methods/sci.py new file mode 100644 index 000000000..33b7df34d --- /dev/null +++ b/codecarbon/output_methods/sci.py @@ -0,0 +1,265 @@ +""" +SCI (Software Carbon Intensity) output handler for CodeCarbon. + +Writes a report in the shape defined by the Green Software Foundation's +Software Carbon Intensity specification, standardized as ISO/IEC 21031:2024:: + + SCI = (E * I + M) / R + +CodeCarbon measures ``E`` (energy consumed) and applies ``I`` (carbon +intensity). ``R``, the functional unit, and ``M``, the embodied emissions, +are declarations only the user can make, so they are never guessed: an +undeclared term is reported as undeclared. +""" + +import json +import os +from dataclasses import dataclass +from typing import List, Optional + +from codecarbon.external.logger import logger +from codecarbon.output_methods.base_output import BaseOutput +from codecarbon.output_methods.emissions_data import EmissionsData, TaskEmissionsData + +SCI_SPEC_VERSION = "ISO/IEC 21031:2024" + + +@dataclass +class FunctionalUnit: + """The ``R`` term: what one unit of the software is, and how many happened.""" + + name: str + count: float = 0.0 + + @classmethod + def from_dict(cls, data: dict) -> "FunctionalUnit": + return cls(name=data.get("name", ""), count=data.get("count", 0.0)) + + +@dataclass +class EmbodiedDeclaration: + """The ``M`` term: user-declared embodied emissions, in gCO2eq.""" + + gco2e: float = 0.0 + source: str = "" + + @classmethod + def from_dict(cls, data: dict) -> "EmbodiedDeclaration": + return cls(gco2e=data.get("gCO2e", 0.0), source=data.get("source", "")) + + +def map_emissions_to_sci( + emissions: EmissionsData, + functional_unit: Optional[FunctionalUnit] = None, + embodied: Optional[EmbodiedDeclaration] = None, + reporter: Optional[dict] = None, + boundary: str = "", +) -> dict: + """ + Build an SCI report dictionary from CodeCarbon emissions data. + + ``I`` is derived from the measured values as ``emissions * 1000 / + energy_consumed`` rather than recomputed, so the report is consistent by + construction with the CSV output, and already accounts for cloud region, + PUE and the country/region fallbacks. + + Args: + emissions: CodeCarbon emissions data from a completed run. + functional_unit: The user-declared ``R`` term. + embodied: The user-declared ``M`` term. + reporter: Free-form reporter identification (organization, contact). + boundary: Free-form description of the measurement boundary. + + Returns: + A JSON-serializable dict. ``sci`` is ``None``, with a ``status`` + explaining why, when ``R`` was not declared. + """ + energy_kwh = emissions.energy_consumed + intensity = emissions.emissions * 1000 / energy_kwh if energy_kwh else 0.0 + embodied = embodied or EmbodiedDeclaration() + unit_count = functional_unit.count if functional_unit else 0.0 + + report = { + "specVersion": SCI_SPEC_VERSION, + "generatedBy": f"codecarbon {emissions.codecarbon_version}", + "runId": str(emissions.run_id), + "projectName": emissions.project_name, + "timestamp": emissions.timestamp, + "sci": None, + "unit": None, + "terms": { + "E_kWh": energy_kwh, + "I_gCO2e_per_kWh": intensity, + "M_gCO2e": embodied.gco2e, + "R": unit_count, + "R_name": functional_unit.name if functional_unit else None, + }, + "provenance": { + "I_source": ( + "codecarbon regional intensity, " + f"country_iso_code={emissions.country_iso_code}" + ), + "M_source": embodied.source or "not declared", + "measurementBoundary": boundary + or "Application only; excludes client devices and network.", + "durationSeconds": emissions.duration, + "hardware": { + "cpu": emissions.cpu_model, + "gpu": emissions.gpu_model, + "ramTotalGB": emissions.ram_total_size, + }, + "pue": getattr(emissions, "pue", 1), + }, + } + if reporter: + report["reporter"] = reporter + + if unit_count > 0: + report["sci"] = (energy_kwh * intensity + embodied.gco2e) / unit_count + report["unit"] = f"gCO2eq per {functional_unit.name or 'functional unit'}" + else: + report["status"] = ( + "No functional unit (R) declared, so SCI could not be computed. " + "Declare one via SCIOutput(functional_unit=...), " + "set_functional_unit_count() or a context file." + ) + return report + + +class SCIOutput(BaseOutput): + """ + Output handler that writes SCI-formatted JSON reports. + + Usage: + # Programmatic + handler = SCIOutput( + functional_unit=FunctionalUnit(name="inference request", count=10_000), + ) + tracker = EmissionsTracker(output_handlers=[handler]) + + # From context file + handler = SCIOutput.from_file("sci_context.json") + tracker = EmissionsTracker(output_handlers=[handler]) + """ + + def __init__( + self, + output_dir: str = ".", + functional_unit: Optional[FunctionalUnit] = None, + embodied: Optional[EmbodiedDeclaration] = None, + reporter: Optional[dict] = None, + boundary: str = "", + ): + os.makedirs(output_dir, exist_ok=True) + self._output_dir = output_dir + self._functional_unit = functional_unit + self._embodied = embodied + self._reporter = reporter + self._boundary = boundary + + @classmethod + def from_file(cls, context_file_path: str, output_dir: str = ".") -> "SCIOutput": + """ + Load the user-declared SCI terms from a JSON file. + + The context file holds what CodeCarbon cannot know:: + + { + "functionalUnit": {"name": "inference request", "count": 10000}, + "embodied": {"gCO2e": 42.5, "source": "manufacturer LCA, 4y"}, + "reporter": {"organization": "Acme"}, + "boundary": "Application only." + } + + Args: + context_file_path: Path to the SCI context JSON file. + output_dir: Directory to write output reports to. + + Returns: + A configured SCIOutput instance. + + Raises: + FileNotFoundError: If the context file does not exist. + json.JSONDecodeError: If the context file contains invalid JSON. + """ + try: + with open(context_file_path) as f: + context = json.load(f) + except FileNotFoundError: + raise FileNotFoundError(f"SCI context file not found: {context_file_path}") + + functional_unit = None + embodied = None + + if "functionalUnit" in context: + functional_unit = FunctionalUnit.from_dict(context["functionalUnit"]) + + if "embodied" in context: + embodied = EmbodiedDeclaration.from_dict(context["embodied"]) + + return cls( + output_dir=output_dir, + functional_unit=functional_unit, + embodied=embodied, + reporter=context.get("reporter"), + boundary=context.get("boundary", ""), + ) + + def set_functional_unit_count(self, count: float, name: Optional[str] = None): + """Declare how many functional units the run covered, once it is known.""" + if self._functional_unit is None: + self._functional_unit = FunctionalUnit(name=name or "", count=count) + else: + self._functional_unit.count = count + if name is not None: + self._functional_unit.name = name + + def _write(self, report: dict, file_name: str): + file_path = os.path.join(self._output_dir, file_name) + with open(file_path, "w") as f: + json.dump(report, f, indent=2) + logger.info(f"SCI report saved to {os.path.abspath(file_path)}") + + def out(self, total: EmissionsData, delta: EmissionsData): + """Write the final SCI report as a JSON file.""" + try: + report = map_emissions_to_sci( + total, + functional_unit=self._functional_unit, + embodied=self._embodied, + reporter=self._reporter, + boundary=self._boundary, + ) + self._write(report, f"sci_report_{total.run_id}.json") + except Exception as e: + logger.error(f"Failed to write SCI report: {e}", exc_info=True) + + def task_out(self, data: List[TaskEmissionsData], experiment_name: str): + """Write one SCI report per task, sharing the run-level declarations.""" + try: + reports = [] + for task in data: + report = map_emissions_to_sci( + task, + functional_unit=self._functional_unit, + embodied=self._embodied, + reporter=self._reporter, + boundary=self._boundary, + ) + report["taskName"] = task.task_name + reports.append(report) + if not reports: + return + self._write( + { + "specVersion": SCI_SPEC_VERSION, + "experimentName": experiment_name, + "tasks": reports, + }, + f"sci_report_tasks_{data[0].run_id}.json", + ) + except Exception as e: + logger.error(f"Failed to write SCI task report: {e}", exc_info=True) + + def live_out(self, total: EmissionsData, delta: EmissionsData): + """No-op: SCI reports are final, not incremental.""" diff --git a/docs/how-to/examples.md b/docs/how-to/examples.md index 06f084aca..50e415aac 100644 --- a/docs/how-to/examples.md +++ b/docs/how-to/examples.md @@ -62,6 +62,7 @@ The directory [examples/](https://github.com/mlco2/codecarbon/tree/master/exampl | Example | Type | Description | |---------|------|-------------| | [boamps_output.py](https://github.com/mlco2/codecarbon/blob/master/examples/boamps_output.py) | Python Script | Write the output in [BoAmps](https://github.com/Boavizta/BoAmps) format. | +| [sci_output.py](https://github.com/mlco2/codecarbon/blob/master/examples/sci_output.py) | Python Script | Write a [Software Carbon Intensity](https://sci.greensoftware.foundation/) (ISO/IEC 21031) report. | | [logging_to_file.py](https://github.com/mlco2/codecarbon/blob/master/examples/logging_to_file.py) | Python Script | Save emissions data to a local CSV file | | [logging_to_file_exclusive_run.py](https://github.com/mlco2/codecarbon/blob/master/examples/logging_to_file_exclusive_run.py) | Python Script | Long-running process with exclusive file logging | | [logging_to_google_cloud.py](https://github.com/mlco2/codecarbon/blob/master/examples/logging_to_google_cloud.py) | Python Script | Send emissions data to Google Cloud Logging | diff --git a/docs/reference/output.md b/docs/reference/output.md index 720e0a883..7e4168c8e 100644 --- a/docs/reference/output.md +++ b/docs/reference/output.md @@ -13,7 +13,7 @@ tracker = EmissionsTracker( ) ``` -Available values: `CSV`, `API`, `LOGGER`, `PROMETHEUS`, `LOGFIRE`, `BOAMPS`. +Available values: `CSV`, `API`, `LOGGER`, `PROMETHEUS`, `LOGFIRE`, `BOAMPS`, `SCI`. It can also be set in the config file as a comma-separated string, e.g. `output_methods=csv,api`. HTTP output is enabled separately via the `emissions_endpoint` parameter. @@ -211,3 +211,91 @@ You can send all your data to the CodeCarbon API so you have your historical dat ## Logger Output See [Collecting emissions to a logger](../how-to/logging.md). + +## SCI + +The [Software Carbon Intensity](https://sci.greensoftware.foundation/) specification, standardised as ISO/IEC 21031:2024, expresses a workload's footprint as a rate: + +``` +SCI = (E * I + M) / R +``` + +CodeCarbon measures `E` (energy consumed, kWh) and applies `I` (carbon intensity, gCO2eq/kWh). The two remaining terms are declarations that only you can make: + +- `R`, the functional unit — one request, one training run, 1k tokens. There is no sensible default, so if you do not declare one the report is still written, with `sci: null` and a `status` field saying why. +- `M`, the embodied emissions attributable to the run. CodeCarbon has no hardware manufacturing data and will not invent any; when you do not declare a figure the report says `"M_source": "not declared"` so a reader can see the report is a partial one. + +### How to use it + +```python-skip +from codecarbon import EmissionsTracker +from codecarbon.output_methods.sci import EmbodiedDeclaration, FunctionalUnit, SCIOutput + +sci = SCIOutput( + functional_unit=FunctionalUnit(name="inference request", count=10_000), + embodied=EmbodiedDeclaration(gCO2e=42.5, source="vendor LCA, 4y amortization"), + output_dir="reports", +) +tracker = EmissionsTracker(output_handlers=[sci]) +``` + +When the count is only known at the end of the run, declare it before stopping the tracker: + +```python-skip +sci.set_functional_unit_count(len(results)) +``` + +The declarations can also live in a JSON context file, which is what the `sci` output method reads when you enable it through configuration: + +```json +{ + "functionalUnit": {"name": "inference request", "count": 10000}, + "embodied": {"gCO2e": 42.5, "source": "manufacturer LCA, 4y amortization"}, + "reporter": {"organization": "Acme", "contact": "green@acme.example"}, + "boundary": "Application only; excludes client devices and network." +} +``` + +```ini +[codecarbon] +output_methods = csv,sci +sci_context_file = ./sci_context.json +``` + +```python-skip +from codecarbon.output_methods.sci import SCIOutput + +sci = SCIOutput.from_file("sci_context.json") +``` + +CodeCarbon writes a final report named `sci_report_.json` in `output_dir`, plus `sci_report_tasks_.json` when tasks are used. `I` is derived as `emissions * 1000 / energy_consumed` rather than recomputed, so the report is consistent with the CSV by construction; the PUE that was applied is recorded separately in the provenance block. + +Sample output: +```json +{ + "specVersion": "ISO/IEC 21031:2024", + "generatedBy": "codecarbon 3.2.6", + "runId": "79e4408f-ec31-476f-a2c5-8ca7f53e6cc7", + "projectName": "my_project", + "timestamp": "2025-01-15T10:30:00", + "sci": 0.0000127, + "unit": "gCO2eq per inference request", + "terms": { + "E_kWh": 0.1007, + "I_gCO2e_per_kWh": 417.08, + "M_gCO2e": 42.5, + "R": 10000, + "R_name": "inference request" + }, + "provenance": { + "I_source": "codecarbon regional intensity, country_iso_code=FRA", + "M_source": "manufacturer LCA, 4y amortization", + "measurementBoundary": "Application only; excludes client devices and network.", + "durationSeconds": 3600.0, + "hardware": {"cpu": "Intel Xeon", "gpu": "NVIDIA A100", "ramTotalGB": 64.0}, + "pue": 1 + } +} +``` + +See [examples/sci_output.py](https://github.com/mlco2/codecarbon/blob/master/examples/sci_output.py) for a runnable example. diff --git a/examples/sci_output.py b/examples/sci_output.py new file mode 100644 index 000000000..5b63e3f62 --- /dev/null +++ b/examples/sci_output.py @@ -0,0 +1,39 @@ +""" +Write a Software Carbon Intensity (ISO/IEC 21031) report. + +The functional unit `R` is a declaration: only you know what one unit of your +software is. Here it is one "inference request", counted as the workload runs. +""" + +from codecarbon import EmissionsTracker +from codecarbon.output_methods.sci import FunctionalUnit, SCIOutput + +REQUESTS = 100 + + +def handle_request(number): + a = 0 + for i in range(int(1e6)): + a = a + i**number + return a + + +sci = SCIOutput( + functional_unit=FunctionalUnit(name="inference request"), + # embodied=EmbodiedDeclaration(gCO2e=42.5, source="vendor LCA, 4y amortization"), +) +tracker = EmissionsTracker( + measure_power_secs=10, + output_methods=["csv"], + output_handlers=[sci], +) +try: + tracker.start() + for i in range(REQUESTS): + handle_request(i % 3) +finally: + sci.set_functional_unit_count(REQUESTS) + emissions = tracker.stop() + +print(f"Emissions: {emissions} kg") +print(f"SCI report written to ./sci_report_{tracker.run_id}.json") diff --git a/tests/test_sci_output.py b/tests/test_sci_output.py new file mode 100644 index 000000000..79b4e48af --- /dev/null +++ b/tests/test_sci_output.py @@ -0,0 +1,253 @@ +""" +Test suite for the SCI (ISO/IEC 21031) output method. + +Sections: +A. Mapping and arithmetic +B. Undeclared terms (R and M) +C. Output handler +D. Context file loading +""" + +import json +import os +import shutil +import tempfile +import unittest + +from codecarbon.output_methods.base_output import OutputMethod +from codecarbon.output_methods.emissions_data import EmissionsData, TaskEmissionsData +from codecarbon.output_methods.sci import ( + SCI_SPEC_VERSION, + EmbodiedDeclaration, + FunctionalUnit, + SCIOutput, + map_emissions_to_sci, +) + + +def _make_emissions_data(**overrides) -> EmissionsData: + """Create a realistic EmissionsData instance for testing.""" + defaults = dict( + timestamp="2025-01-15T10:30:00", + project_name="test_project", + run_id="550e8400-e29b-41d4-a716-446655440000", + experiment_id="exp-001", + duration=3600.0, + emissions=0.042, + emissions_rate=1.17e-05, + cpu_power=12.5, + gpu_power=85.0, + ram_power=3.2, + cpu_energy=0.0125, + gpu_energy=0.085, + ram_energy=0.0032, + energy_consumed=0.1007, + water_consumed=0.0, + country_name="France", + country_iso_code="FRA", + region="ile-de-france", + cloud_provider="", + cloud_region="", + os="Linux", + python_version="3.11.0", + codecarbon_version="3.0.0", + cpu_count=8, + cpu_model="Intel Xeon", + gpu_count=1, + gpu_model="NVIDIA A100", + longitude=2.35, + latitude=48.85, + ram_total_size=64.0, + tracking_mode="machine", + ) + defaults.update(overrides) + return EmissionsData(**defaults) + + +def _make_task_data(task_name: str, **overrides) -> TaskEmissionsData: + data = _make_emissions_data(**overrides).values + data.pop("experiment_id") + data.pop("pue") + data.pop("wue") + return TaskEmissionsData(task_name=task_name, **data) + + +class TestSCIMapping(unittest.TestCase): + """A. Mapping and arithmetic.""" + + def test_sci_formula(self): + data = _make_emissions_data() + report = map_emissions_to_sci( + data, + functional_unit=FunctionalUnit(name="inference request", count=10_000), + embodied=EmbodiedDeclaration(gco2e=42.5, source="vendor LCA"), + ) + expected_i = data.emissions * 1000 / data.energy_consumed + expected_sci = (data.energy_consumed * expected_i + 42.5) / 10_000 + + self.assertAlmostEqual(report["sci"], expected_sci) + self.assertAlmostEqual(report["terms"]["I_gCO2e_per_kWh"], expected_i) + self.assertEqual(report["terms"]["E_kWh"], data.energy_consumed) + self.assertEqual(report["terms"]["M_gCO2e"], 42.5) + self.assertEqual(report["terms"]["R"], 10_000) + self.assertEqual(report["unit"], "gCO2eq per inference request") + self.assertEqual(report["specVersion"], SCI_SPEC_VERSION) + self.assertEqual(report["provenance"]["M_source"], "vendor LCA") + self.assertIn("FRA", report["provenance"]["I_source"]) + + def test_provenance_fields(self): + report = map_emissions_to_sci( + _make_emissions_data(), + functional_unit=FunctionalUnit(name="request", count=1), + reporter={"organization": "Acme"}, + boundary="Application only.", + ) + provenance = report["provenance"] + self.assertEqual(provenance["hardware"]["cpu"], "Intel Xeon") + self.assertEqual(provenance["hardware"]["gpu"], "NVIDIA A100") + self.assertEqual(provenance["hardware"]["ramTotalGB"], 64.0) + self.assertEqual(provenance["durationSeconds"], 3600.0) + self.assertEqual(provenance["measurementBoundary"], "Application only.") + self.assertEqual(report["reporter"], {"organization": "Acme"}) + + +class TestUndeclaredTerms(unittest.TestCase): + """B. Undeclared terms.""" + + def test_no_functional_unit(self): + report = map_emissions_to_sci(_make_emissions_data()) + self.assertIsNone(report["sci"]) + self.assertIn("status", report) + self.assertIn("functional unit", report["status"]) + + def test_zero_functional_unit_count(self): + report = map_emissions_to_sci( + _make_emissions_data(), functional_unit=FunctionalUnit(name="req", count=0) + ) + self.assertIsNone(report["sci"]) + self.assertIn("status", report) + + def test_embodied_not_declared(self): + report = map_emissions_to_sci( + _make_emissions_data(), + functional_unit=FunctionalUnit(name="req", count=10), + ) + self.assertEqual(report["terms"]["M_gCO2e"], 0.0) + self.assertEqual(report["provenance"]["M_source"], "not declared") + + def test_zero_energy_does_not_raise(self): + report = map_emissions_to_sci( + _make_emissions_data(energy_consumed=0.0, emissions=0.0), + functional_unit=FunctionalUnit(name="req", count=10), + ) + self.assertEqual(report["terms"]["I_gCO2e_per_kWh"], 0.0) + self.assertEqual(report["sci"], 0.0) + + +class TestSCIOutput(unittest.TestCase): + """C. Output handler.""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_out_writes_report(self): + data = _make_emissions_data() + handler = SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=100), + ) + handler.out(data, data) + + file_path = os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json") + self.assertTrue(os.path.exists(file_path)) + with open(file_path) as f: + report = json.load(f) + self.assertIsNotNone(report["sci"]) + + def test_set_functional_unit_count(self): + data = _make_emissions_data() + handler = SCIOutput(output_dir=self.temp_dir) + handler.set_functional_unit_count(500, name="request") + handler.out(data, data) + + with open(os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json")) as f: + report = json.load(f) + self.assertEqual(report["terms"]["R"], 500) + self.assertEqual(report["terms"]["R_name"], "request") + + def test_task_out_writes_one_entry_per_task(self): + tasks = [_make_task_data("train"), _make_task_data("infer")] + handler = SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=100), + ) + handler.task_out(tasks, "experiment") + + file_path = os.path.join( + self.temp_dir, f"sci_report_tasks_{tasks[0].run_id}.json" + ) + with open(file_path) as f: + report = json.load(f) + self.assertEqual([t["taskName"] for t in report["tasks"]], ["train", "infer"]) + + def test_live_out_is_noop(self): + data = _make_emissions_data() + SCIOutput(output_dir=self.temp_dir).live_out(data, data) + self.assertEqual(os.listdir(self.temp_dir), []) + + def test_output_method_enum(self): + self.assertEqual(OutputMethod.SCI.value, "sci") + + +class TestContextFile(unittest.TestCase): + """D. Context file loading.""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def _write_context(self, content: str) -> str: + path = os.path.join(self.temp_dir, "sci_context.json") + with open(path, "w") as f: + f.write(content) + return path + + def test_from_file(self): + path = self._write_context( + json.dumps( + { + "functionalUnit": {"name": "inference request", "count": 10000}, + "embodied": {"gCO2e": 42.5, "source": "manufacturer LCA"}, + "reporter": {"organization": "Acme"}, + "boundary": "Application only.", + } + ) + ) + handler = SCIOutput.from_file(path, output_dir=self.temp_dir) + data = _make_emissions_data() + handler.out(data, data) + + with open(os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json")) as f: + report = json.load(f) + self.assertEqual(report["terms"]["R"], 10000) + self.assertEqual(report["terms"]["M_gCO2e"], 42.5) + self.assertEqual(report["provenance"]["M_source"], "manufacturer LCA") + self.assertEqual(report["reporter"], {"organization": "Acme"}) + + def test_from_file_missing(self): + with self.assertRaises(FileNotFoundError): + SCIOutput.from_file(os.path.join(self.temp_dir, "nope.json")) + + def test_from_file_malformed(self): + path = self._write_context("{not json") + with self.assertRaises(json.JSONDecodeError): + SCIOutput.from_file(path, output_dir=self.temp_dir) + + +if __name__ == "__main__": + unittest.main() From 51b6dadce266bb60a6ec6b0de76ff40e151a9430 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:12:15 +0200 Subject: [PATCH 2/3] test: cover SCI output branches and tracker Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_sci_output.py | 158 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/tests/test_sci_output.py b/tests/test_sci_output.py index 79b4e48af..c1424ca17 100644 --- a/tests/test_sci_output.py +++ b/tests/test_sci_output.py @@ -6,6 +6,7 @@ B. Undeclared terms (R and M) C. Output handler D. Context file loading +E. Tracker registration """ import json @@ -14,6 +15,7 @@ import tempfile import unittest +from codecarbon.emissions_tracker import BaseEmissionsTracker from codecarbon.output_methods.base_output import OutputMethod from codecarbon.output_methods.emissions_data import EmissionsData, TaskEmissionsData from codecarbon.output_methods.sci import ( @@ -193,6 +195,77 @@ def test_task_out_writes_one_entry_per_task(self): report = json.load(f) self.assertEqual([t["taskName"] for t in report["tasks"]], ["train", "infer"]) + def test_set_functional_unit_count_updates_existing_unit(self): + data = _make_emissions_data() + handler = SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=1), + ) + handler.set_functional_unit_count(750) + handler.out(data, data) + + with open(os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json")) as f: + report = json.load(f) + self.assertEqual(report["terms"]["R"], 750) + # name is kept when not given again + self.assertEqual(report["terms"]["R_name"], "request") + self.assertAlmostEqual(report["sci"], data.emissions * 1000 / 750, places=10) + + def test_set_functional_unit_count_renames_existing_unit(self): + handler = SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=1), + ) + handler.set_functional_unit_count(10, name="image") + data = _make_emissions_data() + handler.out(data, data) + + with open(os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json")) as f: + report = json.load(f) + self.assertEqual(report["terms"]["R_name"], "image") + self.assertEqual(report["unit"], "gCO2eq per image") + + def test_out_logs_and_swallows_errors(self): + handler = SCIOutput(output_dir=self.temp_dir) + with self.assertLogs("codecarbon", level="ERROR") as logs: + handler.out(object(), object()) + self.assertTrue(any("Failed to write SCI report" in m for m in logs.output)) + self.assertEqual(os.listdir(self.temp_dir), []) + + def test_task_out_without_tasks_writes_nothing(self): + SCIOutput(output_dir=self.temp_dir).task_out([], "experiment") + self.assertEqual(os.listdir(self.temp_dir), []) + + def test_task_out_logs_and_swallows_errors(self): + handler = SCIOutput(output_dir=self.temp_dir) + with self.assertLogs("codecarbon", level="ERROR") as logs: + handler.task_out([object()], "experiment") + self.assertTrue( + any("Failed to write SCI task report" in m for m in logs.output) + ) + self.assertEqual(os.listdir(self.temp_dir), []) + + def test_task_out_computes_sci_per_task(self): + tasks = [ + _make_task_data("train", emissions=0.2, energy_consumed=0.5), + _make_task_data("infer", emissions=0.1, energy_consumed=0.5), + ] + handler = SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=100), + embodied=EmbodiedDeclaration(gco2e=10.0, source="vendor LCA"), + ) + handler.task_out(tasks, "experiment") + + with open( + os.path.join(self.temp_dir, f"sci_report_tasks_{tasks[0].run_id}.json") + ) as f: + report = json.load(f) + self.assertEqual(report["experimentName"], "experiment") + self.assertAlmostEqual(report["tasks"][0]["sci"], (0.2 * 1000 + 10.0) / 100) + self.assertAlmostEqual(report["tasks"][1]["sci"], (0.1 * 1000 + 10.0) / 100) + self.assertEqual(report["tasks"][0]["provenance"]["M_source"], "vendor LCA") + def test_live_out_is_noop(self): data = _make_emissions_data() SCIOutput(output_dir=self.temp_dir).live_out(data, data) @@ -239,6 +312,35 @@ def test_from_file(self): self.assertEqual(report["provenance"]["M_source"], "manufacturer LCA") self.assertEqual(report["reporter"], {"organization": "Acme"}) + def test_from_file_empty_context_declares_nothing(self): + path = self._write_context("{}") + handler = SCIOutput.from_file(path, output_dir=self.temp_dir) + data = _make_emissions_data() + handler.out(data, data) + + with open(os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json")) as f: + report = json.load(f) + self.assertIsNone(report["sci"]) + self.assertIsNone(report["terms"]["R_name"]) + self.assertEqual(report["terms"]["M_gCO2e"], 0.0) + self.assertNotIn("reporter", report) + + def test_from_file_partial_declarations_use_defaults(self): + path = self._write_context( + json.dumps({"functionalUnit": {"count": 4}, "embodied": {"gCO2e": 8.0}}) + ) + handler = SCIOutput.from_file(path, output_dir=self.temp_dir) + data = _make_emissions_data() + handler.out(data, data) + + with open(os.path.join(self.temp_dir, f"sci_report_{data.run_id}.json")) as f: + report = json.load(f) + self.assertEqual(report["terms"]["R"], 4) + self.assertEqual(report["terms"]["R_name"], "") + self.assertEqual(report["unit"], "gCO2eq per functional unit") + self.assertEqual(report["provenance"]["M_source"], "not declared") + self.assertAlmostEqual(report["sci"], (data.emissions * 1000 + 8.0) / 4) + def test_from_file_missing(self): with self.assertRaises(FileNotFoundError): SCIOutput.from_file(os.path.join(self.temp_dir, "nope.json")) @@ -249,5 +351,61 @@ def test_from_file_malformed(self): SCIOutput.from_file(path, output_dir=self.temp_dir) +class _TrackerStub: + """Minimal stand-in exercising ``_init_output_methods`` without hardware.""" + + def __init__(self, output_dir: str, external_conf: dict): + self._output_methods = [OutputMethod.SCI] + self._emissions_endpoint = None + self._external_conf = external_conf + self._output_dir = output_dir + self._output_handlers = [] + + def init(self): + BaseEmissionsTracker._init_output_methods(self, api_key=None) + return self._output_handlers + + +class TestTrackerRegistration(unittest.TestCase): + """E. Tracker registration of OutputMethod.SCI.""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_sci_method_adds_handler_without_declarations(self): + handlers = _TrackerStub(self.temp_dir, {}).init() + sci_handlers = [h for h in handlers if isinstance(h, SCIOutput)] + self.assertEqual(len(sci_handlers), 1) + self.assertIsNone(sci_handlers[0]._functional_unit) + self.assertEqual(sci_handlers[0]._output_dir, self.temp_dir) + + def test_sci_context_file_from_config_is_loaded(self): + context_path = os.path.join(self.temp_dir, "sci_context.json") + with open(context_path, "w") as f: + json.dump( + { + "functionalUnit": {"name": "request", "count": 250}, + "embodied": {"gCO2e": 12.0, "source": "vendor LCA"}, + }, + f, + ) + handlers = _TrackerStub( + self.temp_dir, {"sci_context_file": context_path} + ).init() + handler = next(h for h in handlers if isinstance(h, SCIOutput)) + self.assertEqual(handler._functional_unit, FunctionalUnit("request", 250)) + self.assertEqual(handler._embodied, EmbodiedDeclaration(12.0, "vendor LCA")) + + def test_sci_context_file_missing_raises(self): + with self.assertRaises(FileNotFoundError): + _TrackerStub( + self.temp_dir, + {"sci_context_file": os.path.join(self.temp_dir, "nope.json")}, + ).init() + + if __name__ == "__main__": unittest.main() From e06fdc34251ca27e2632bff9152329c5f57435ce Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:57:27 +0200 Subject: [PATCH 3/3] fix: apportion embodied carbon (M) across tasks in the SCI report task_out() handed the same declared M to every task, so a run with 5 tasks reported the device's full embodied carbon 5 times and each per-task sci included 100% of it. Split M by each task's share of the run duration. Chosen over dropping M from task reports because the split is exhaustive: the per-task figures now sum back to the run-level report, and M_source records the share applied. Also document that `output_methods = csv,sci` alone can never produce a non-null sci unless the context file hardcodes functionalUnit.count. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/output_methods/sci.py | 26 ++++++++++++- docs/reference/output.md | 5 ++- tests/test_sci_output.py | 63 ++++++++++++++++++++++++++++++-- 3 files changed, 88 insertions(+), 6 deletions(-) diff --git a/codecarbon/output_methods/sci.py b/codecarbon/output_methods/sci.py index 33b7df34d..6437cd0cd 100644 --- a/codecarbon/output_methods/sci.py +++ b/codecarbon/output_methods/sci.py @@ -234,15 +234,37 @@ def out(self, total: EmissionsData, delta: EmissionsData): except Exception as e: logger.error(f"Failed to write SCI report: {e}", exc_info=True) + def _embodied_share(self, share: float) -> Optional[EmbodiedDeclaration]: + """The slice of the declared ``M`` that belongs to one task.""" + if self._embodied is None: + return None + return EmbodiedDeclaration( + gco2e=self._embodied.gco2e * share, + source=( + f"{self._embodied.source or 'declared'}; " + f"apportioned by duration ({share:.1%} of the run)" + ), + ) + def task_out(self, data: List[TaskEmissionsData], experiment_name: str): - """Write one SCI report per task, sharing the run-level declarations.""" + """ + Write one SCI report per task, sharing the run-level declarations. + + ``M`` is embodied carbon of the whole device over the whole run, so each + task gets the slice matching its share of the run's duration. Giving + every task the full ``M`` would report the device's embodied carbon once + per task; apportioning it keeps the per-task figures summing back to the + run-level report. + """ try: + total_duration = sum(task.duration for task in data) reports = [] for task in data: + share = task.duration / total_duration if total_duration else 0.0 report = map_emissions_to_sci( task, functional_unit=self._functional_unit, - embodied=self._embodied, + embodied=self._embodied_share(share), reporter=self._reporter, boundary=self._boundary, ) diff --git a/docs/reference/output.md b/docs/reference/output.md index 7e4168c8e..27090a097 100644 --- a/docs/reference/output.md +++ b/docs/reference/output.md @@ -268,7 +268,10 @@ from codecarbon.output_methods.sci import SCIOutput sci = SCIOutput.from_file("sci_context.json") ``` -CodeCarbon writes a final report named `sci_report_.json` in `output_dir`, plus `sci_report_tasks_.json` when tasks are used. `I` is derived as `emissions * 1000 / energy_consumed` rather than recomputed, so the report is consistent with the CSV by construction; the PUE that was applied is recorded separately in the provenance block. +!!! warning "The configuration-only path needs a count in the context file" + `set_functional_unit_count()` needs a reference to the handler, and the handler created by `output_methods = csv,sci` is owned by the tracker. So with configuration alone, `sci` is `null` in every report unless `functionalUnit.count` is hardcoded in `sci_context_file`. If the count is only known at the end of the run, construct `SCIOutput` yourself and pass it via `output_handlers=[...]`. + +CodeCarbon writes a final report named `sci_report_.json` in `output_dir`, plus `sci_report_tasks_.json` when tasks are used. In the task report, `M` is apportioned across tasks by their share of the run's duration: the declared `M` is the embodied carbon of the device for the whole run, so giving each task the full figure would count it once per task. `I` is derived as `emissions * 1000 / energy_consumed` rather than recomputed, so the report is consistent with the CSV by construction; the PUE that was applied is recorded separately in the provenance block. Sample output: ```json diff --git a/tests/test_sci_output.py b/tests/test_sci_output.py index c1424ca17..16864c48c 100644 --- a/tests/test_sci_output.py +++ b/tests/test_sci_output.py @@ -262,9 +262,66 @@ def test_task_out_computes_sci_per_task(self): ) as f: report = json.load(f) self.assertEqual(report["experimentName"], "experiment") - self.assertAlmostEqual(report["tasks"][0]["sci"], (0.2 * 1000 + 10.0) / 100) - self.assertAlmostEqual(report["tasks"][1]["sci"], (0.1 * 1000 + 10.0) / 100) - self.assertEqual(report["tasks"][0]["provenance"]["M_source"], "vendor LCA") + # equal durations, so each task carries half of the declared M + self.assertAlmostEqual(report["tasks"][0]["sci"], (0.2 * 1000 + 5.0) / 100) + self.assertAlmostEqual(report["tasks"][1]["sci"], (0.1 * 1000 + 5.0) / 100) + self.assertIn("vendor LCA", report["tasks"][0]["provenance"]["M_source"]) + self.assertIn( + "apportioned by duration", report["tasks"][0]["provenance"]["M_source"] + ) + + def test_task_out_apportions_embodied_by_duration(self): + # The whole point: M is the device's embodied carbon for the whole run. + # Handing every task the full M would report it once per task. + tasks = [ + _make_task_data("short", duration=100.0), + _make_task_data("long", duration=300.0), + ] + handler = SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=100), + embodied=EmbodiedDeclaration(gco2e=40.0, source="vendor LCA"), + ) + handler.task_out(tasks, "experiment") + + with open( + os.path.join(self.temp_dir, f"sci_report_tasks_{tasks[0].run_id}.json") + ) as f: + report = json.load(f) + shares = [t["terms"]["M_gCO2e"] for t in report["tasks"]] + self.assertAlmostEqual(shares[0], 10.0) + self.assertAlmostEqual(shares[1], 30.0) + # the split is exhaustive: the run's M is counted exactly once + self.assertAlmostEqual(sum(shares), 40.0) + self.assertIn("25.0% of the run", report["tasks"][0]["provenance"]["M_source"]) + + def test_task_out_without_embodied_stays_undeclared(self): + tasks = [_make_task_data("train")] + SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=100), + ).task_out(tasks, "experiment") + + with open( + os.path.join(self.temp_dir, f"sci_report_tasks_{tasks[0].run_id}.json") + ) as f: + report = json.load(f) + self.assertEqual(report["tasks"][0]["terms"]["M_gCO2e"], 0.0) + self.assertEqual(report["tasks"][0]["provenance"]["M_source"], "not declared") + + def test_task_out_with_zero_duration_does_not_divide_by_zero(self): + tasks = [_make_task_data("instant", duration=0.0)] + SCIOutput( + output_dir=self.temp_dir, + functional_unit=FunctionalUnit(name="request", count=100), + embodied=EmbodiedDeclaration(gco2e=40.0, source="vendor LCA"), + ).task_out(tasks, "experiment") + + with open( + os.path.join(self.temp_dir, f"sci_report_tasks_{tasks[0].run_id}.json") + ) as f: + report = json.load(f) + self.assertEqual(report["tasks"][0]["terms"]["M_gCO2e"], 0.0) def test_live_out_is_noop(self): data = _make_emissions_data()