feat: carbon-aware codecarbon wait (forecast + greenest-window selection)

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():
     """

codecarbon/cli/wait.py

@@ -0,0 +1,132 @@
+"""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.electricitymaps_api import resolve_token
+    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)
+
+    token = resolve_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)

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