diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 270723466..23440e46a 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -414,6 +414,7 @@ def __init__( tracking_mode: Optional[str] = _sentinel, log_level: Optional[Union[int, str]] = _sentinel, on_csv_write: Optional[str] = _sentinel, + csv_run_name: Optional[str] = _sentinel, logger_preamble: Optional[str] = _sentinel, force_cpu_power: Optional[int] = _sentinel, force_ram_power: Optional[int] = _sentinel, @@ -492,6 +493,11 @@ def __init__( :param on_csv_write: When calling tracker.flush() manually: "update" to overwrite the existing run_id row, or "append" to add a new row to the CSV file. Defaults to "append". + :param csv_run_name: Optional CSV filename for interval measurement export. + When set, CodeCarbon appends a row on the same cadence as + API/Prometheus (``api_call_interval * measure_power_secs``). + Use ``"auto"`` (or an empty string) to name the file + ``emissions_.csv``. Defaults to None (disabled). :param logger_preamble: String to systematically include in the logger. messages. Defaults to "". :param force_cpu_power: Force the CPU max power consumption in watts. Use this if you @@ -578,6 +584,7 @@ def __init__( self._set_from_conf(output_handlers, "output_handlers", []) self._set_from_conf(tracking_mode, "tracking_mode", "machine") self._set_from_conf(on_csv_write, "on_csv_write", "append") + self._set_from_conf(csv_run_name, "csv_run_name", None) self._set_from_conf(logger_preamble, "logger_preamble", "") self._set_from_conf(force_cpu_power, "force_cpu_power", None, float) self._set_from_conf(force_ram_power, "force_ram_power", None, float) @@ -609,7 +616,7 @@ def _init_output_methods(self, *, api_key: str = None): """ methods = set(self._output_methods) if self._output_methods else set() - if not methods and not self._emissions_endpoint: + if not methods and not self._emissions_endpoint and self._csv_run_name is None: self.run_id = uuid.uuid4() return @@ -620,15 +627,15 @@ def _init_output_methods(self, *, api_key: str = None): from codecarbon.output_methods.metrics.prometheus import PrometheusOutput methods = set(self._output_methods) if self._output_methods else set() + csv_file_handler = None if OutputMethod.CSV in methods: - self._output_handlers.append( - FileOutput( - self._output_file, - self._output_dir, - self._on_csv_write, - ) + csv_file_handler = FileOutput( + self._output_file, + self._output_dir, + self._on_csv_write, ) + self._output_handlers.append(csv_file_handler) if OutputMethod.LOGGER in methods: self._output_handlers.append(self._logging_logger) @@ -666,6 +673,54 @@ def _init_output_methods(self, *, api_key: str = None): if OutputMethod.BOAMPS in methods: self._output_handlers.append(BoAmpsOutput(output_dir=self._output_dir)) + self._init_interval_csv_output(csv_file_handler) + + def _resolve_csv_run_filename(self) -> Optional[str]: + """Return the interval CSV filename, or None when interval export is disabled.""" + if self._csv_run_name is None: + return None + name = self._csv_run_name.strip() if isinstance(self._csv_run_name, str) else "" + if name in ("", "auto"): + return f"emissions_{self.run_id}.csv" + return name + + def _init_interval_csv_output(self, csv_file_handler) -> None: + """ + Opt-in interval CSV export (#467): write measurement rows on the same + cadence as API/Prometheus live outputs. + """ + from codecarbon.output_methods.file import FileOutput + + interval_file_name = self._resolve_csv_run_filename() + if interval_file_name is None: + return + + if ( + csv_file_handler is not None + and csv_file_handler.output_file_name == interval_file_name + ): + csv_file_handler.enable_live_out = True + logger.info( + "Interval CSV export enabled on %s " + "(every api_call_interval * measure_power_secs)", + interval_file_name, + ) + return + + self._output_handlers.append( + FileOutput( + interval_file_name, + self._output_dir, + on_csv_write="append", + enable_live_out=True, + ) + ) + logger.info( + "Interval CSV export enabled: %s " + "(every api_call_interval * measure_power_secs)", + interval_file_name, + ) + def get_detected_hardware(self) -> Dict[str, Any]: """ Get the detected hardware. @@ -1495,6 +1550,7 @@ def track_emissions( tracking_mode: Optional[str] = _sentinel, log_level: Optional[Union[int, str]] = _sentinel, on_csv_write: Optional[str] = _sentinel, + csv_run_name: Optional[str] = _sentinel, logger_preamble: Optional[str] = _sentinel, offline: Optional[bool] = _sentinel, country_iso_code: Optional[str] = _sentinel, @@ -1555,6 +1611,8 @@ def track_emissions( Defaults to "info". :param on_csv_write: When calling tracker.flush() manually: "update" to overwrite the existing run_id row, or "append" to add a new row. Defaults to "append". + :param csv_run_name: Optional CSV filename for interval measurement export. See + EmissionsTracker. Defaults to None (disabled). :param logger_preamble: String to systematically include in the logger. messages. Defaults to "". :param allow_multiple_runs: Allow multiple CodeCarbon instances on the same machine. @@ -1633,6 +1691,7 @@ def wrapped_fn(*args, **kwargs): tracking_mode=tracking_mode, log_level=log_level, on_csv_write=on_csv_write, + csv_run_name=csv_run_name, logger_preamble=logger_preamble, country_iso_code=country_iso_code, region=region, @@ -1674,6 +1733,7 @@ def wrapped_fn(*args, **kwargs): tracking_mode=tracking_mode, log_level=log_level, on_csv_write=on_csv_write, + csv_run_name=csv_run_name, logger_preamble=logger_preamble, force_cpu_power=force_cpu_power, force_ram_power=force_ram_power, diff --git a/codecarbon/output_methods/file.py b/codecarbon/output_methods/file.py index 6a13d5b41..eb59f6d5b 100644 --- a/codecarbon/output_methods/file.py +++ b/codecarbon/output_methods/file.py @@ -22,7 +22,11 @@ class FileOutput(BaseOutput): """ def __init__( - self, output_file_name: str, output_dir: str, on_csv_write: str = "append" + self, + output_file_name: str, + output_dir: str, + on_csv_write: str = "append", + enable_live_out: bool = False, ): """ Initialize the FileOutput object. @@ -31,6 +35,8 @@ def __init__( output_file_name: name of file to write to. output_dir: path to directory to write to. on_csv_write: "append" or "update", whether or not to append or overwrite a file if it exists + enable_live_out: when True, also write on live measurement intervals + (same cadence as API/Prometheus: ``api_call_interval * measure_power_secs``). Raises: ValueError: If the on_csv_write value is invalid. @@ -46,6 +52,7 @@ def __init__( raise OSError(f"Folder '{output_dir}' doesn't exist !") self.output_dir: str = output_dir self.on_csv_write: str = on_csv_write + self.enable_live_out: bool = enable_live_out self.save_file_path = os.path.join(self.output_dir, self.output_file_name) logger.info( f"Emissions data (if any) will be saved to file {os.path.abspath(self.save_file_path)}" @@ -69,6 +76,11 @@ def has_valid_headers(self, data: EmissionsData) -> bool: return True return sorted(headers) == sorted(data.values.keys()) + def live_out(self, total: EmissionsData, delta: EmissionsData): + """Write a measurement row on the live interval when enabled.""" + if self.enable_live_out: + self.out(total, delta) + def out(self, total: EmissionsData, _): """ Save the emissions data from a whole run to a CSV file. diff --git a/docs/reference/output.md b/docs/reference/output.md index 720e0a883..ce0b2de40 100644 --- a/docs/reference/output.md +++ b/docs/reference/output.md @@ -28,6 +28,18 @@ It can also be set in the config file as a comma-separated string, e.g. The package has an in-built logger that logs data into a CSV file named `emissions.csv` in the `output_dir`, provided as an input parameter (defaults to the current directory), for each experiment tracked across projects. +By default that file is written once at the end of the run. To also append a row on the same live cadence as API/Prometheus (`api_call_interval × measure_power_secs`), set `csv_run_name`: + +```python-skip +from codecarbon import EmissionsTracker + +tracker = EmissionsTracker( + csv_run_name="emissions_live.csv", # or "auto" / "" → emissions_.csv +) +``` + +Use the same name as `output_file` to enable live rows on the primary CSV. Leave `csv_run_name` unset to keep the default final-only behavior. + | Field | Description | |-------|-------------| | timestamp | Time of the experiment in `%Y-%m-%dT%H:%M:%S` format | diff --git a/tests/output_methods/test_file.py b/tests/output_methods/test_file.py index e8bccfdf0..9e55f68af 100644 --- a/tests/output_methods/test_file.py +++ b/tests/output_methods/test_file.py @@ -413,3 +413,22 @@ def test_out_append_large_file_fast_path(self): df = pd.read_csv(os.path.join(self.temp_dir, "test.csv")) self.assertEqual(len(df), 2) + + def test_live_out_noop_when_disabled(self): + file_output = FileOutput("test.csv", self.temp_dir, on_csv_write="append") + file_output.live_out(self.emissions_data, self.emissions_data) + self.assertFalse(os.path.isfile(file_output.save_file_path)) + + def test_live_out_appends_when_enabled(self): + file_output = FileOutput( + "interval.csv", + self.temp_dir, + on_csv_write="append", + enable_live_out=True, + ) + file_output.live_out(self.emissions_data, self.emissions_data) + file_output.live_out(self.emissions_data, self.emissions_data) + + df = pd.read_csv(file_output.save_file_path) + self.assertEqual(len(df), 2) + self.assertEqual(df.iloc[0]["run_id"], "test_run_id") diff --git a/tests/test_emissions_tracker.py b/tests/test_emissions_tracker.py index 8ab12e5d8..c1fdfb271 100644 --- a/tests/test_emissions_tracker.py +++ b/tests/test_emissions_tracker.py @@ -425,6 +425,224 @@ def test_output_methods_overrides_save_to_flags( ) ) + def test_csv_run_name_enables_interval_file_output( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + from codecarbon.output_methods.emissions_data import EmissionsData + from codecarbon.output_methods.file import FileOutput + + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV], + csv_run_name="interval_emissions.csv", + api_call_interval=1, + measure_power_secs=1, + allow_multiple_runs=True, + ) + + live_handlers = [ + h + for h in tracker._output_handlers + if isinstance(h, FileOutput) and h.enable_live_out + ] + self.assertEqual(len(live_handlers), 1) + self.assertEqual(live_handlers[0].output_file_name, "interval_emissions.csv") + + sample = EmissionsData( + timestamp="2023-01-01T00:00:00", + project_name="project_foo", + run_id=str(tracker.run_id), + experiment_id="test_experiment_id", + duration=10, + emissions=0.5, + emissions_rate=0.05, + cpu_power=20, + gpu_power=30, + ram_power=5, + cpu_energy=200, + gpu_energy=300, + ram_energy=50, + energy_consumed=550, + water_consumed=0.1, + country_name="Testland", + country_iso_code="TS", + region="Test Region", + cloud_provider="N/A", + cloud_region="N/A", + os="TestOS", + python_version="3.8", + codecarbon_version="2.0", + cpu_count=4, + cpu_model="Test CPU", + gpu_count=1, + gpu_model="Test GPU", + longitude=0, + latitude=0, + ram_total_size=16, + tracking_mode="machine", + on_cloud="N", + pue=1.0, + wue=0.0, + ) + for handler in tracker._output_handlers: + handler.live_out(sample, sample) + handler.live_out(sample, sample) + + interval_path = self.temp_path / "interval_emissions.csv" + self.assertTrue(interval_path.exists()) + df = pd.read_csv(interval_path) + self.assertEqual(len(df), 2) + + # Default emissions.csv stays final-only (live_out is a no-op). + self.assertFalse(self.emissions_file_path.exists()) + + def test_csv_run_name_auto_uses_run_id( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + from codecarbon.output_methods.file import FileOutput + + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV], + csv_run_name="auto", + allow_multiple_runs=True, + ) + + expected = f"emissions_{tracker.run_id}.csv" + live_handlers = [ + h + for h in tracker._output_handlers + if isinstance(h, FileOutput) and h.enable_live_out + ] + self.assertEqual(len(live_handlers), 1) + self.assertEqual(live_handlers[0].output_file_name, expected) + + def test_csv_run_name_same_as_output_file_enables_live_on_primary( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + from codecarbon.output_methods.file import FileOutput + + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV], + output_file="emissions.csv", + csv_run_name="emissions.csv", + allow_multiple_runs=True, + ) + + file_handlers = [ + h for h in tracker._output_handlers if isinstance(h, FileOutput) + ] + self.assertEqual(len(file_handlers), 1) + self.assertTrue(file_handlers[0].enable_live_out) + + def test_csv_run_name_empty_string_uses_run_id( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + from codecarbon.output_methods.file import FileOutput + + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV], + csv_run_name="", + allow_multiple_runs=True, + ) + + expected = f"emissions_{tracker.run_id}.csv" + live_handlers = [ + h + for h in tracker._output_handlers + if isinstance(h, FileOutput) and h.enable_live_out + ] + self.assertEqual(len(live_handlers), 1) + self.assertEqual(live_handlers[0].output_file_name, expected) + + def test_csv_run_name_default_keeps_csv_final_only( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + from codecarbon.output_methods.file import FileOutput + + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV], + allow_multiple_runs=True, + ) + + file_handlers = [ + h for h in tracker._output_handlers if isinstance(h, FileOutput) + ] + self.assertEqual(len(file_handlers), 1) + self.assertFalse(file_handlers[0].enable_live_out) + + def test_csv_run_name_without_csv_method_still_enables_interval_file( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + from codecarbon.output_methods.file import FileOutput + + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[], + save_to_file=False, + csv_run_name="interval_only.csv", + allow_multiple_runs=True, + ) + + live_handlers = [ + h + for h in tracker._output_handlers + if isinstance(h, FileOutput) and h.enable_live_out + ] + self.assertEqual(len(live_handlers), 1) + self.assertEqual(live_handlers[0].output_file_name, "interval_only.csv") + def test_output_methods_parsed_from_config_string( self, mock_cli_setup,