From 904602fe50d3f9a62db6664f0a6a836020fca5a8 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 17:52:27 +0200 Subject: [PATCH 1/3] feat: add carbon-aware `codecarbon wait` Fetch an Electricity Maps carbon intensity forecast, pick the window with the lowest mean intensity that still meets the deadline, and either report it (--dry-run) or sleep until it and delegate to run_and_monitor. Advisory/blocking only: no EmissionsData schema change, no decorator, and no static fallback profile. Without a token, get_forecast returns None and the job runs immediately -- a job is never blocked on a missing credential. get_forecast should become a method on the provider protocol once pluggable intensity providers land. Refs #1356 Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/cli/main.py | 51 +++++++ codecarbon/cli/wait.py | 135 +++++++++++++++++++ codecarbon/core/intensity_forecast.py | 148 ++++++++++++++++++++ tests/cli/test_wait.py | 169 +++++++++++++++++++++++ tests/test_intensity_forecast.py | 187 ++++++++++++++++++++++++++ 5 files changed, 690 insertions(+) create mode 100644 codecarbon/cli/wait.py create mode 100644 codecarbon/core/intensity_forecast.py create mode 100644 tests/cli/test_wait.py create mode 100644 tests/test_intensity_forecast.py diff --git a/codecarbon/cli/main.py b/codecarbon/cli/main.py index 93f627e5b..a5c7758e9 100644 --- a/codecarbon/cli/main.py +++ b/codecarbon/cli/main.py @@ -465,6 +465,57 @@ def signal_handler(signum, frame): raise e +@codecarbon.command( + "wait", + short_help="Wait for a low-carbon window, then run a command.", + context_settings={"allow_extra_args": True, "ignore_unknown_options": True}, +) +def wait( + ctx: typer.Context, + duration: Annotated[ + str, + typer.Option(help="Expected job length, e.g. '90m', '2h', '1h30m'."), + ] = "1h", + deadline: Annotated[ + str, + typer.Option(help="Maximum delay before the job must start."), + ] = "12h", + threshold: Annotated[ + Optional[float], + typer.Option(help="gCO2e/kWh at or below which we start immediately."), + ] = None, + dry_run: Annotated[ + bool, + typer.Option(help="Print the recommendation and exit without waiting."), + ] = False, + measure_power_secs: Annotated[ + int, + typer.Option(help="Interval between two measures."), + ] = 10, + log_level: Annotated[ + str, + typer.Option(help="Log level (critical, error, warning, info, debug)"), + ] = "error", +): + """Wait for the greenest window in the carbon intensity forecast, then run + a command under measurement. + + Requires an Electricity Maps API token; without one, the command runs + immediately rather than blocking. + """ + from codecarbon.cli.wait import wait_for_green_window + + return wait_for_green_window( + ctx, + duration=duration, + deadline=deadline, + threshold=threshold, + dry_run=dry_run, + log_level=log_level, + measure_power_secs=measure_power_secs, + ) + + @codecarbon.command("detect", short_help="Detect hardware and print information.") def detect(): """ diff --git a/codecarbon/cli/wait.py b/codecarbon/cli/wait.py new file mode 100644 index 000000000..e5dbc83c4 --- /dev/null +++ b/codecarbon/cli/wait.py @@ -0,0 +1,135 @@ +"""CodeCarbon CLI - Wait Command""" + +import re +import sys +import time +from datetime import datetime, timedelta, timezone +from typing import Optional + +import typer +from rich import print + +_DURATION_RE = re.compile(r"^(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$") + + +def parse_duration(value: str) -> timedelta: + """Parse "90m", "2h", "1h30m" or a plain number of seconds.""" + value = value.strip().lower() + if value.isdigit(): + return timedelta(seconds=int(value)) + match = _DURATION_RE.match(value) + if not match or not any(match.groups()): + raise ValueError(f"Invalid duration: {value!r}. Use e.g. '90m', '2h', '1h30m'.") + hours, minutes, seconds = (int(g or 0) for g in match.groups()) + return timedelta(hours=hours, minutes=minutes, seconds=seconds) + + +def find_green_window( + duration: timedelta, + deadline: timedelta, + token: Optional[str], +): + """Return (start, intensity, now_intensity) or None when we should run now.""" + from codecarbon.core.intensity_forecast import best_window, get_forecast + from codecarbon.external.geography import GeoMetadata + from codecarbon.input import DataSource + + geo = GeoMetadata.from_geo_js(DataSource().geo_js_url) + forecast = get_forecast(geo, token=token, horizon_hours=_ceil_hours(deadline)) + if forecast is None: + return None + + now = datetime.now(timezone.utc) + start, intensity = best_window(forecast, duration, deadline=now + deadline) + return start, intensity, forecast.points[0].g_co2e_per_kwh + + +def _ceil_hours(delta: timedelta) -> int: + return max(1, -(-int(delta.total_seconds()) // 3600)) + + +def wait_for_green_window( + ctx: typer.Context, + duration: str = "1h", + deadline: str = "12h", + threshold: Optional[float] = None, + dry_run: bool = False, + log_level: str = "error", + **tracker_args, +): + """Wait for the greenest window in the forecast, then run a command. + + This is a sleep, not a scheduler: it does not fork, daemonise or persist. + For deferral that must survive a reboot, use cron, systemd or Airflow. + + Examples: + + # Print the recommendation and exit + codecarbon wait --dry-run --deadline 24h --duration 90m + + # Block until the greenest window, then run under measurement + codecarbon wait --deadline 12h --duration 2h -- python train.py + """ + from codecarbon.cli.monitor import run_and_monitor + from codecarbon.core.config import get_hierarchical_config + from codecarbon.external.logger import set_logger_level + + set_logger_level(log_level) + + try: + job_duration = parse_duration(duration) + max_delay = parse_duration(deadline) + except ValueError as e: + print(f"ERROR: {e}", file=sys.stderr) + raise typer.Exit(1) + + config = get_hierarchical_config() + token = config.get("electricitymaps_api_token") or config.get( + "co2_signal_api_token" + ) + + window = find_green_window(job_duration, max_delay, token) + delay_seconds = 0.0 + if window is None: + print("šŸŒ± CodeCarbon: no forecast available, running now.") + else: + start, intensity, now_intensity = window + delay_seconds = max(0.0, (start - datetime.now(timezone.utc)).total_seconds()) + if threshold is not None and now_intensity <= threshold: + print( + f"šŸŒ± CodeCarbon: current intensity {now_intensity:.0f} gCO2e/kWh is " + f"at or below the {threshold:.0f} threshold, running now." + ) + delay_seconds = 0.0 + elif delay_seconds <= 0: + print( + f"šŸŒ± CodeCarbon: now is already the greenest window " + f"({now_intensity:.0f} gCO2e/kWh)." + ) + else: + saving = ( + 100 * (now_intensity - intensity) / now_intensity + if now_intensity + else 0 + ) + print( + f"šŸŒ± Best start: {start:%Y-%m-%d %H:%M} UTC " + f"({intensity:.0f} gCO2e/kWh, now: {now_intensity:.0f}) " + f"-> saves ~{saving:.0f}%" + ) + + if dry_run: + raise typer.Exit(0) + + if delay_seconds > 0: + print( + f" Waiting {delay_seconds / 3600:.1f}h before starting. Ctrl-C to run now." + ) + try: + time.sleep(delay_seconds) + except KeyboardInterrupt: + print("\nāš ļø Wait interrupted, starting now.", file=sys.stderr) + + # Strip our own subcommand name so `run_and_monitor` sees only the command. + ctx.args = [arg for arg in getattr(ctx, "args", []) if arg != "wait"] + run_and_monitor(ctx, log_level=log_level, **tracker_args) diff --git a/codecarbon/core/intensity_forecast.py b/codecarbon/core/intensity_forecast.py new file mode 100644 index 000000000..b1150f7ef --- /dev/null +++ b/codecarbon/core/intensity_forecast.py @@ -0,0 +1,148 @@ +"""Carbon intensity forecasts and greenest-window selection. + +The only provider able to serve a forecast today is Electricity Maps, and only +for users holding a token for it. When no provider can answer, `get_forecast` +returns ``None`` and every caller must degrade to "run now" -- a job is never +blocked on a missing credential. + +Once pluggable intensity providers land (see issue #1356), `get_forecast` +should become an optional `forecast()` method on the provider protocol rather +than a second HTTP client. +""" + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional, Tuple + +import requests + +from codecarbon.core.electricitymaps_api import ELECTRICITYMAPS_API_TIMEOUT +from codecarbon.external.geography import GeoMetadata +from codecarbon.external.logger import logger + +FORECAST_URL: str = "https://api.electricitymaps.com/v3/carbon-intensity/forecast" + + +@dataclass(frozen=True) +class IntensityPoint: + at: datetime # timezone-aware, UTC + g_co2e_per_kwh: float + + +@dataclass(frozen=True) +class Forecast: + zone: str + points: List[IntensityPoint] # ordered, typically hourly + source: str + fetched_at: datetime + + +def _location_params(geo: GeoMetadata) -> Dict[str, Any]: + """Build the Electricity Maps location query, as `get_emissions` does.""" + if geo.latitude: + return {"lat": geo.latitude, "lon": geo.longitude} + return {"countryCode": geo.country_2letter_iso_code} + + +def _parse_datetime(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def get_forecast( + geo: GeoMetadata, + *, + token: Optional[str] = None, + horizon_hours: int = 48, +) -> Optional[Forecast]: + """Return an intensity forecast, or None when no provider can supply one. + + Never raises: a forecast is an optimisation, not a requirement. + """ + if not token: + logger.warning( + "No Electricity Maps API token configured, cannot fetch a carbon " + "intensity forecast." + ) + return None + + try: + resp = requests.get( + FORECAST_URL, + params=_location_params(geo), + headers={"auth-token": token}, + timeout=ELECTRICITYMAPS_API_TIMEOUT, + ) + if resp.status_code != 200: + body = resp.json() + raise ValueError(body.get("error") or body.get("message") or resp.text) + + data = resp.json() + horizon_end = datetime.now(timezone.utc) + timedelta(hours=horizon_hours) + points = [ + IntensityPoint( + at=_parse_datetime(entry["datetime"]), + g_co2e_per_kwh=float(entry["carbonIntensity"]), + ) + for entry in data["forecast"] + if entry.get("carbonIntensity") is not None + ] + points = sorted( + (point for point in points if point.at <= horizon_end), + key=lambda point: point.at, + ) + if not points: + raise ValueError("No usable forecast points in response") + + return Forecast( + zone=data.get("zone", ""), + points=points, + source="electricitymaps", + fetched_at=datetime.now(timezone.utc), + ) + except Exception as e: + logger.error( + f"intensity_forecast.get_forecast: {e} >>> Falling back to running now." + ) + return None + + +def best_window( + forecast: Forecast, + duration: timedelta, + deadline: Optional[datetime] = None, +) -> Tuple[datetime, float]: + """Start time minimising mean intensity over `duration`, and that mean. + + Only windows that both start at or after the first forecast point and + finish before `deadline` are considered. Returns the earliest point and its + intensity when no complete window fits, so "just run it" is the default. + """ + points = forecast.points + fallback = (points[0].at, points[0].g_co2e_per_kwh) + + # The forecast covers up to one step past its last point. + step = points[1].at - points[0].at if len(points) > 1 else duration + covered_until = points[-1].at + step + + best: Optional[Tuple[datetime, float]] = None + for start_index, start in enumerate(points): + window_end = start.at + duration + if window_end > covered_until: + break + if deadline is not None and window_end > deadline: + break + # ponytail: linear rescan per start, fine for hourly points over a few + # days; use a running sum if horizons ever grow by orders of magnitude. + covered = [ + point.g_co2e_per_kwh + for point in points[start_index:] + if point.at < window_end + ] + mean = sum(covered) / len(covered) + if best is None or mean < best[1]: + best = (start.at, mean) + + return best or fallback diff --git a/tests/cli/test_wait.py b/tests/cli/test_wait.py new file mode 100644 index 000000000..ddc532a81 --- /dev/null +++ b/tests/cli/test_wait.py @@ -0,0 +1,169 @@ +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest +import typer + +from codecarbon.cli import wait as wait_module +from codecarbon.core.intensity_forecast import Forecast, IntensityPoint + + +def _forecast(values, start): + return Forecast( + zone="FR", + points=[ + IntensityPoint(at=start + timedelta(hours=i), g_co2e_per_kwh=v) + for i, v in enumerate(values) + ], + source="test", + fetched_at=start, + ) + + +@pytest.fixture +def no_network(monkeypatch): + """Never let the wait command reach geolocation or the intensity API.""" + monkeypatch.setattr( + "codecarbon.external.geography.GeoMetadata.from_geo_js", + classmethod(lambda cls, url: SimpleNamespace()), + ) + monkeypatch.setattr( + "codecarbon.core.config.get_hierarchical_config", + lambda: {"electricitymaps_api_token": "tok"}, + ) + + +def _patch_forecast(monkeypatch, values): + now = datetime.now(timezone.utc) + monkeypatch.setattr( + "codecarbon.core.intensity_forecast.get_forecast", + lambda geo, **kwargs: _forecast(values, now), + ) + + +@pytest.mark.parametrize( + "value,expected", + [ + ("90m", timedelta(minutes=90)), + ("2h", timedelta(hours=2)), + ("1h30m", timedelta(hours=1, minutes=30)), + ("45s", timedelta(seconds=45)), + ("3600", timedelta(hours=1)), + ], +) +def test_parse_duration(value, expected): + assert wait_module.parse_duration(value) == expected + + +@pytest.mark.parametrize("value", ["", "soon", "2 hours", "h", "-1h"]) +def test_parse_duration_rejects_garbage(value): + with pytest.raises(ValueError): + wait_module.parse_duration(value) + + +def test_dry_run_prints_recommendation_and_exits(monkeypatch, capsys, no_network): + _patch_forecast(monkeypatch, [300, 300, 100, 100, 300]) + slept = [] + monkeypatch.setattr(wait_module.time, "sleep", lambda s: slept.append(s)) + + with pytest.raises(typer.Exit) as exc: + wait_module.wait_for_green_window( + SimpleNamespace(args=[]), duration="2h", deadline="6h", dry_run=True + ) + + assert exc.value.exit_code == 0 + assert slept == [] + out = capsys.readouterr().out + assert "Best start" in out + assert "saves ~67%" in out + + +def test_invalid_duration_exits_with_error(monkeypatch, capsys, no_network): + with pytest.raises(typer.Exit) as exc: + wait_module.wait_for_green_window( + SimpleNamespace(args=[]), duration="whenever", dry_run=True + ) + assert exc.value.exit_code == 1 + + +def test_no_forecast_runs_now(monkeypatch, capsys, no_network): + monkeypatch.setattr( + "codecarbon.core.intensity_forecast.get_forecast", lambda geo, **kwargs: None + ) + slept = [] + monkeypatch.setattr(wait_module.time, "sleep", lambda s: slept.append(s)) + called = {} + monkeypatch.setattr( + "codecarbon.cli.monitor.run_and_monitor", + lambda ctx, **kwargs: called.setdefault("args", list(ctx.args)), + ) + + wait_module.wait_for_green_window( + SimpleNamespace(args=["wait", "--", "python", "train.py"]) + ) + + assert slept == [] + assert called["args"] == ["--", "python", "train.py"] + assert "no forecast available" in capsys.readouterr().out + + +def test_threshold_short_circuits_the_wait(monkeypatch, capsys, no_network): + _patch_forecast(monkeypatch, [120, 300, 50, 50]) + slept = [] + monkeypatch.setattr(wait_module.time, "sleep", lambda s: slept.append(s)) + monkeypatch.setattr( + "codecarbon.cli.monitor.run_and_monitor", lambda ctx, **kwargs: None + ) + + wait_module.wait_for_green_window( + SimpleNamespace(args=["python", "train.py"]), + duration="1h", + deadline="6h", + threshold=150, + ) + + assert slept == [] + assert "running now" in capsys.readouterr().out + + +def test_sleeps_until_the_green_window_then_delegates(monkeypatch, no_network): + _patch_forecast(monkeypatch, [300, 300, 100, 100, 300]) + slept = [] + monkeypatch.setattr(wait_module.time, "sleep", lambda s: slept.append(s)) + called = {} + monkeypatch.setattr( + "codecarbon.cli.monitor.run_and_monitor", + lambda ctx, **kwargs: called.update(kwargs, args=list(ctx.args)), + ) + + wait_module.wait_for_green_window( + SimpleNamespace(args=["wait", "python", "train.py"]), + duration="2h", + deadline="6h", + measure_power_secs=15, + ) + + assert len(slept) == 1 + assert 2 * 3600 - 60 < slept[0] <= 2 * 3600 + assert called["args"] == ["python", "train.py"] + assert called["measure_power_secs"] == 15 + + +def test_keyboard_interrupt_during_wait_runs_immediately(monkeypatch, no_network): + _patch_forecast(monkeypatch, [300, 300, 100, 100, 300]) + + def _interrupt(seconds): + raise KeyboardInterrupt + + monkeypatch.setattr(wait_module.time, "sleep", _interrupt) + called = {} + monkeypatch.setattr( + "codecarbon.cli.monitor.run_and_monitor", + lambda ctx, **kwargs: called.setdefault("ran", True), + ) + + wait_module.wait_for_green_window( + SimpleNamespace(args=["python", "train.py"]), duration="2h", deadline="6h" + ) + + assert called["ran"] is True diff --git a/tests/test_intensity_forecast.py b/tests/test_intensity_forecast.py new file mode 100644 index 000000000..849b8948f --- /dev/null +++ b/tests/test_intensity_forecast.py @@ -0,0 +1,187 @@ +import unittest +from datetime import datetime, timedelta, timezone + +import responses + +from codecarbon.core import intensity_forecast +from codecarbon.core.intensity_forecast import ( + Forecast, + IntensityPoint, + best_window, + get_forecast, +) +from codecarbon.external.geography import GeoMetadata + +BASE = datetime(2026, 8, 13, 0, 0, tzinfo=timezone.utc) + + +def _forecast(values): + return Forecast( + zone="FR", + points=[ + IntensityPoint(at=BASE + timedelta(hours=i), g_co2e_per_kwh=v) + for i, v in enumerate(values) + ], + source="test", + fetched_at=BASE, + ) + + +def _payload(values, start=None): + start = start or datetime.now(timezone.utc) + return { + "zone": "FR", + "forecast": [ + { + "datetime": (start + timedelta(hours=i)) + .isoformat() + .replace("+00:00", "Z"), + "carbonIntensity": v, + } + for i, v in enumerate(values) + ], + } + + +class TestGetForecast(unittest.TestCase): + def setUp(self) -> None: + self._geo = GeoMetadata( + country_iso_code="FRA", + country_name="France", + region=None, + country_2letter_iso_code="FR", + ) + self._geo_latlon = GeoMetadata( + country_iso_code="FRA", + country_name="France", + region=None, + country_2letter_iso_code="FR", + latitude=48.85, + longitude=2.35, + ) + + def test_no_token_returns_none_without_calling_api(self): + assert get_forecast(self._geo, token=None) is None + + @responses.activate + def test_parses_forecast(self): + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json=_payload([100, 200, 50]), + status=200, + ) + forecast = get_forecast(self._geo, token="tok") + assert forecast is not None + assert forecast.zone == "FR" + assert forecast.source == "electricitymaps" + assert [p.g_co2e_per_kwh for p in forecast.points] == [100, 200, 50] + assert all(p.at.tzinfo is not None for p in forecast.points) + assert responses.calls[0].request.headers["auth-token"] == "tok" + assert "countryCode=FR" in responses.calls[0].request.url + + @responses.activate + def test_uses_lat_lon_when_available(self): + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json=_payload([100]), + status=200, + ) + get_forecast(self._geo_latlon, token="tok") + url = responses.calls[0].request.url + assert "lat=48.85" in url and "lon=2.35" in url + + @responses.activate + def test_naive_timestamps_are_treated_as_utc(self): + payload = _payload([100]) + payload["forecast"][0]["datetime"] = "2999-01-01T03:00:00" + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json=payload, + status=200, + ) + forecast = get_forecast(self._geo, token="tok", horizon_hours=24 * 365 * 1000) + assert forecast.points[0].at.tzinfo == timezone.utc + + @responses.activate + def test_horizon_truncates_points(self): + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json=_payload([100, 200, 300, 400]), + status=200, + ) + forecast = get_forecast(self._geo, token="tok", horizon_hours=2) + assert len(forecast.points) <= 3 + + @responses.activate + def test_error_status_returns_none(self): + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json={"error": "no access"}, + status=403, + ) + assert get_forecast(self._geo, token="tok") is None + + @responses.activate + def test_malformed_payload_returns_none(self): + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json={"unexpected": True}, + status=200, + ) + assert get_forecast(self._geo, token="tok") is None + + @responses.activate + def test_empty_forecast_returns_none(self): + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json={"zone": "FR", "forecast": []}, + status=200, + ) + assert get_forecast(self._geo, token="tok") is None + + +class TestBestWindow(unittest.TestCase): + def test_picks_the_trough(self): + forecast = _forecast([300, 250, 100, 90, 280, 300]) + start, mean = best_window(forecast, timedelta(hours=2)) + assert start == BASE + timedelta(hours=2) + assert mean == 95 + + def test_flat_series_picks_now(self): + forecast = _forecast([200] * 5) + start, mean = best_window(forecast, timedelta(hours=2)) + assert start == BASE + assert mean == 200 + + def test_decreasing_series_picks_last_complete_window(self): + forecast = _forecast([500, 400, 300, 200, 100]) + start, _ = best_window(forecast, timedelta(hours=2)) + assert start == BASE + timedelta(hours=3) + + def test_deadline_shorter_than_duration_falls_back_to_now(self): + forecast = _forecast([300, 100, 100]) + start, mean = best_window( + forecast, timedelta(hours=2), deadline=BASE + timedelta(minutes=30) + ) + assert start == BASE + assert mean == 300 + + def test_deadline_restricts_the_search(self): + forecast = _forecast([300, 200, 50, 50]) + start, _ = best_window( + forecast, timedelta(hours=1), deadline=BASE + timedelta(hours=2) + ) + assert start == BASE + timedelta(hours=1) + + def test_duration_longer_than_horizon_falls_back_to_now(self): + forecast = _forecast([300, 100]) + start, mean = best_window(forecast, timedelta(hours=10)) + assert start == BASE + assert mean == 300 From df9b17adced3d1affcf5f8845cff5b3fe08d1496 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Thu, 13 Aug 2026 07:11:26 +0200 Subject: [PATCH 2/3] docs: document the codecarbon wait command Add a CLI reference section for `codecarbon wait` covering every flag and its default, the forecast requirements, and the run-now degradation when no forecast is available, plus one cross-link from the CLI tutorial. Co-Authored-By: Claude Opus 5 (1M context) --- docs/reference/cli.md | 64 +++++++++++++++++++++++++++++++++++++++++++ docs/tutorials/cli.md | 1 + 2 files changed, 65 insertions(+) diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 71ba05a85..de12a0e7d 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -88,6 +88,70 @@ codecarbon monitor -- node app.js --port 8080 Same options as `codecarbon monitor` apply (see above). +### `codecarbon wait -- ` + +Wait for the greenest window in the carbon intensity forecast, then run a command under measurement. + +**Usage:** +```bash +codecarbon wait [OPTIONS] -- +``` + +CodeCarbon fetches an hourly carbon intensity forecast for your location from +[Electricity Maps](https://api.electricitymaps.com), picks the start time that minimises the +average intensity over the expected job length, sleeps until then, and finally hands the command +to `codecarbon monitor` ā€” so measurement, CSV output and exit-code propagation are identical. + +This is a sleep, not a scheduler: the process stays in the foreground and does not fork, daemonise +or persist across a reboot. For deferral that must survive a reboot, use cron, systemd or Airflow. +Pressing `Ctrl+C` during the wait does not abort ā€” it starts the job immediately. The emissions +tracker only starts after the sleep, so a waiting process holds no lock. + +**Options:** + +| Option | Type | Default | Description | +|--------|------|---------|-------------| +| `--duration` | string | 1h | Expected job length, e.g. `90m`, `2h`, `1h30m`, or a plain number of seconds | +| `--deadline` | string | 12h | Maximum delay before the job must start | +| `--threshold` | float | - | gCO2e/kWh at or below which the job starts immediately, without waiting | +| `--dry-run` | flag | false | Print the recommendation and exit without waiting or running | +| `--measure-power-secs` | int | 10 | Interval between two measures | +| `--log-level` | choice | error | Log level: critical, error, warning, info, debug | + +**Examples:** +```bash +# Print the recommendation and exit +codecarbon wait --dry-run --deadline 24h --duration 90m + +# Block until the greenest window, then run under measurement +codecarbon wait --deadline 12h --duration 2h -- python train.py + +# Start straight away if the grid is already below 100 gCO2e/kWh +codecarbon wait --threshold 100 --deadline 6h -- bash benchmark.sh +``` + +The dry run prints the chosen window, for example: + +```console +$ codecarbon wait --dry-run --deadline 24h --duration 90m +šŸŒ± Best start: 2026-08-13 03:00 UTC (112 gCO2e/kWh, now: 341) -> saves ~67% +``` + +**Requirements:** + +A forecast is only available with an `electricitymaps_api_token` (the `co2_signal_api_token` key +is also accepted) ā€” see [Electricity Maps API Token](../how-to/configuration.md#electricity-maps-api-token). +The location is detected automatically from your IP address; there is no offline or +`--country-iso-code` option for this command. + +**When no forecast is available:** + +A forecast is an optimisation, never a requirement ā€” the job is never blocked on a missing +credential. If no token is configured, or the API returns an error, a malformed payload or an +empty forecast, CodeCarbon prints `no forecast available, running now.` and starts the command +straight away. The same applies when no complete window fits before the deadline, or when the +forecast says now is already the greenest moment. + ### `codecarbon detect` Detect and print hardware information. diff --git a/docs/tutorials/cli.md b/docs/tutorials/cli.md index 55f42ffb8..0e3528da8 100644 --- a/docs/tutorials/cli.md +++ b/docs/tutorials/cli.md @@ -134,6 +134,7 @@ You've now learned how to track emissions from the command line. Next steps: - **Track in Python**: Use the [Python API tutorial](python-api.md) for fine-grained tracking within your code. - **Send to Dashboard**: Learn how to [send data to the CodeCarbon dashboard](../how-to/cloud-api.md). - **Configure Details**: See the [configuration guide](../how-to/configuration.md) for advanced options like proxy setup. +- **Run When the Grid Is Green**: Defer a job to the cleanest hours with [`codecarbon wait`](../reference/cli.md#codecarbon-wait-command). ## See Also From e7022e626d8b0874782f3c815980dd655b7676b4 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Thu, 13 Aug 2026 07:28:11 +0200 Subject: [PATCH 3/3] refactor: route the forecast through the shared Electricity Maps client `codecarbon wait` had its own HTTP path to Electricity Maps. It now goes through `electricitymaps_api.request`, so the token lookup, the request plumbing and the exponential failure cooldown are shared with the current-intensity path: a failing API is backed off once, process-wide. The forecast response is deliberately not put in the intensity cache. That cache exists for a value refetched on every measurement tick with a 300 s TTL; a forecast is fetched once per `wait` invocation and has a completely different useful lifetime. `get_forecast` still never raises: a cooldown is just one more reason to return None and run the job now. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/cli/wait.py | 7 +- codecarbon/core/electricitymaps_api.py | 102 ++++++++++++++++--------- codecarbon/core/intensity_forecast.py | 39 ++++------ tests/test_intensity_forecast.py | 20 ++++- 4 files changed, 104 insertions(+), 64 deletions(-) diff --git a/codecarbon/cli/wait.py b/codecarbon/cli/wait.py index e5dbc83c4..50c789d62 100644 --- a/codecarbon/cli/wait.py +++ b/codecarbon/cli/wait.py @@ -71,7 +71,7 @@ def wait_for_green_window( codecarbon wait --deadline 12h --duration 2h -- python train.py """ from codecarbon.cli.monitor import run_and_monitor - from codecarbon.core.config import get_hierarchical_config + from codecarbon.core.electricitymaps_api import resolve_token from codecarbon.external.logger import set_logger_level set_logger_level(log_level) @@ -83,10 +83,7 @@ def wait_for_green_window( print(f"ERROR: {e}", file=sys.stderr) raise typer.Exit(1) - config = get_hierarchical_config() - token = config.get("electricitymaps_api_token") or config.get( - "co2_signal_api_token" - ) + token = resolve_token() window = find_green_window(job_duration, max_delay, token) delay_seconds = 0.0 diff --git a/codecarbon/core/electricitymaps_api.py b/codecarbon/core/electricitymaps_api.py index 57ff3668e..173ea661f 100644 --- a/codecarbon/core/electricitymaps_api.py +++ b/codecarbon/core/electricitymaps_api.py @@ -70,6 +70,67 @@ def _start_cooldown() -> None: _cooldown_until = time.monotonic() + _cooldown_duration +def location_params(geo: GeoMetadata) -> Dict[str, Any]: + """Build the Electricity Maps location query for a geography.""" + if geo.latitude: + return {"lat": geo.latitude, "lon": geo.longitude} + return {"countryCode": geo.country_2letter_iso_code} + + +def resolve_token() -> Optional[str]: + """Read the Electricity Maps token from the hierarchical configuration. + + Falls back to the deprecated ``co2_signal_api_token`` name. + """ + from codecarbon.core.config import get_hierarchical_config + + config = get_hierarchical_config() + return config.get("electricitymaps_api_token") or config.get("co2_signal_api_token") + + +def request(url: str, params: Dict[str, Any], token: str) -> Any: + """GET an Electricity Maps endpoint, sharing the failure cooldown. + + Every endpoint goes through here so that a failing API backs off once, + process-wide, instead of once per caller. + + Raises: + ElectricityMapsAPICooldownError: a previous request failed recently. + ElectricityMapsAPIError: the API answered with an error. + """ + with _lock: + cooldown_until = _cooldown_until + if time.monotonic() < cooldown_until: + raise ElectricityMapsAPICooldownError( + "Electricity Maps API is in cooldown after a previous failure, " + f"retrying in {cooldown_until - time.monotonic():.0f} seconds" + ) + + try: + resp = requests.get( + url, + params=params, + headers={"auth-token": token}, + timeout=ELECTRICITYMAPS_API_TIMEOUT, + ) + if resp.status_code != 200: + body = resp.json() + raise ElectricityMapsAPIError( + body.get("error") or body.get("message") or resp.text + ) + return resp.json() + except Exception: + _start_cooldown() + raise + + +def clear_cooldown() -> None: + """Mark the API as healthy again after a usable response.""" + global _cooldown_duration + with _lock: + _cooldown_duration = 0.0 + + def get_carbon_intensity( geo: GeoMetadata, electricitymaps_api_token: str = "" ) -> float: @@ -97,12 +158,7 @@ def get_carbon_intensity( If the Electricity Maps API request fails, returns an error, or is currently in a failure cooldown. """ - global _cooldown_duration - params: Dict[str, Any] - if geo.latitude: - params = {"lat": geo.latitude, "lon": geo.longitude} - else: - params = {"countryCode": geo.country_2letter_iso_code} + params = location_params(geo) key = _cache_key(params, electricitymaps_api_token) cached_carbon_intensity = _get_cached_carbon_intensity(key) @@ -113,37 +169,15 @@ def get_carbon_intensity( ) return cached_carbon_intensity - with _lock: - cooldown_until = _cooldown_until - if time.monotonic() < cooldown_until: - raise ElectricityMapsAPICooldownError( - "Electricity Maps API is in cooldown after a previous failure, " - f"retrying in {cooldown_until - time.monotonic():.0f} seconds" - ) - - try: - resp = requests.get( - URL, - params=params, - headers={"auth-token": electricitymaps_api_token}, - timeout=ELECTRICITYMAPS_API_TIMEOUT, - ) - if resp.status_code != 200: - message = resp.json().get("error") or resp.json().get("message") - raise ElectricityMapsAPIError(message) - - # API v3 response structure: carbonIntensity is at the root level - response_data = resp.json() - carbon_intensity_g_per_kWh = response_data.get("carbonIntensity") - - if carbon_intensity_g_per_kWh is None: - raise ElectricityMapsAPIError("No carbonIntensity data in response") - except Exception: + response_data = request(URL, params, electricitymaps_api_token) + # API v3 response structure: carbonIntensity is at the root level + carbon_intensity_g_per_kWh = response_data.get("carbonIntensity") + if carbon_intensity_g_per_kWh is None: _start_cooldown() - raise + raise ElectricityMapsAPIError("No carbonIntensity data in response") + clear_cooldown() with _lock: - _cooldown_duration = 0.0 _cache[key] = (time.monotonic(), carbon_intensity_g_per_kWh) return carbon_intensity_g_per_kWh diff --git a/codecarbon/core/intensity_forecast.py b/codecarbon/core/intensity_forecast.py index b1150f7ef..31a84476f 100644 --- a/codecarbon/core/intensity_forecast.py +++ b/codecarbon/core/intensity_forecast.py @@ -5,18 +5,25 @@ returns ``None`` and every caller must degrade to "run now" -- a job is never blocked on a missing credential. +HTTP goes through `codecarbon.core.electricitymaps_api.request`, so a failing +API backs off once for the whole process instead of once per caller. The +forecast response itself is not cached: it is fetched once per `codecarbon +wait` invocation, and its useful lifetime is nothing like the current +intensity's five-minute TTL. + Once pluggable intensity providers land (see issue #1356), `get_forecast` -should become an optional `forecast()` method on the provider protocol rather -than a second HTTP client. +should become an optional `forecast()` method on the provider protocol. """ from dataclasses import dataclass from datetime import datetime, timedelta, timezone -from typing import Any, Dict, List, Optional, Tuple - -import requests +from typing import List, Optional, Tuple -from codecarbon.core.electricitymaps_api import ELECTRICITYMAPS_API_TIMEOUT +from codecarbon.core.electricitymaps_api import ( + clear_cooldown, + location_params, + request, +) from codecarbon.external.geography import GeoMetadata from codecarbon.external.logger import logger @@ -37,13 +44,6 @@ class Forecast: fetched_at: datetime -def _location_params(geo: GeoMetadata) -> Dict[str, Any]: - """Build the Electricity Maps location query, as `get_emissions` does.""" - if geo.latitude: - return {"lat": geo.latitude, "lon": geo.longitude} - return {"countryCode": geo.country_2letter_iso_code} - - def _parse_datetime(value: str) -> datetime: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None: @@ -69,17 +69,7 @@ def get_forecast( return None try: - resp = requests.get( - FORECAST_URL, - params=_location_params(geo), - headers={"auth-token": token}, - timeout=ELECTRICITYMAPS_API_TIMEOUT, - ) - if resp.status_code != 200: - body = resp.json() - raise ValueError(body.get("error") or body.get("message") or resp.text) - - data = resp.json() + data = request(FORECAST_URL, location_params(geo), token) horizon_end = datetime.now(timezone.utc) + timedelta(hours=horizon_hours) points = [ IntensityPoint( @@ -96,6 +86,7 @@ def get_forecast( if not points: raise ValueError("No usable forecast points in response") + clear_cooldown() return Forecast( zone=data.get("zone", ""), points=points, diff --git a/tests/test_intensity_forecast.py b/tests/test_intensity_forecast.py index 849b8948f..f1ef46c68 100644 --- a/tests/test_intensity_forecast.py +++ b/tests/test_intensity_forecast.py @@ -3,7 +3,7 @@ import responses -from codecarbon.core import intensity_forecast +from codecarbon.core import electricitymaps_api, intensity_forecast from codecarbon.core.intensity_forecast import ( Forecast, IntensityPoint, @@ -45,6 +45,10 @@ def _payload(values, start=None): class TestGetForecast(unittest.TestCase): def setUp(self) -> None: + # The forecast shares the Electricity Maps failure cooldown with the + # current-intensity path, so a failing test must not starve the next. + electricitymaps_api.reset_cache() + self.addCleanup(electricitymaps_api.reset_cache) self._geo = GeoMetadata( country_iso_code="FRA", country_name="France", @@ -63,6 +67,20 @@ def setUp(self) -> None: def test_no_token_returns_none_without_calling_api(self): assert get_forecast(self._geo, token=None) is None + @responses.activate + def test_shared_cooldown_skips_the_request(self): + # A failure on the current-intensity path must back the forecast off + # too: no HTTP request, and still a None instead of a raise. + electricitymaps_api._start_cooldown() + responses.add( + responses.GET, + intensity_forecast.FORECAST_URL, + json=_payload([100]), + status=200, + ) + assert get_forecast(self._geo, token="tok") is None + assert len(responses.calls) == 0 + @responses.activate def test_parses_forecast(self): responses.add(