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

codecarbon/cli/main.py

@@ -465,6 +465,64 @@ 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",
+    finish_by: Annotated[
+        Optional[str],
+        typer.Option(
+            help="Latest acceptable finish time, e.g. '8h'. Overrides --deadline."
+        ),
+    ] = None,
+    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,
+        finish_by=finish_by,
+        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/monitor.py

@@ -55,7 +55,7 @@ def run_and_monitor(
 
     # Get the command from remaining args (strip nested subcommand / `--` leftovers)
     command = list(getattr(ctx, "args", None) or [])
-    while command and command[0] in ("monitor", "--"):
+    while command and command[0] in ("monitor", "wait", "--"):
         command.pop(0)
 
     if not command:

codecarbon/cli/wait.py

@@ -0,0 +1,172 @@
+"""CodeCarbon CLI - Wait Command"""
+
+import re
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+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: str | None,
+):
+    """Return (start, intensity, now_intensity) or None when we should run now.
+
+    `now_intensity` is the first forecast point, i.e. the current period as the
+    forecast sees it. Asking the /latest endpoint for a live value instead
+    would be a second HTTP call for a number only used to print a percentage.
+    """
+    import pycountry
+
+    from codecarbon.core.config import get_hierarchical_config
+    from codecarbon.core.intensity_forecast import best_window, get_forecast
+    from codecarbon.external.geography import GeoMetadata
+    from codecarbon.input import DataSource
+
+    # The tracker honours a configured country, so the forecast must too:
+    # geolocating by IP would silently forecast the wrong grid.
+    config = get_hierarchical_config()
+    configured = config.get("country_iso_code")
+    country = (
+        pycountry.countries.get(alpha_3=configured.upper()) if configured else None
+    )
+    if country:
+        geo = GeoMetadata(
+            country_iso_code=country.alpha_3,
+            country_name=country.name,
+            region=config.get("region"),
+            country_2letter_iso_code=country.alpha_2,
+        )
+    else:
+        geo = GeoMetadata.from_geo_js(DataSource().geo_js_url)
+    # A window may start as late as the deadline, so the forecast must cover
+    # the deadline plus one job length.
+    forecast = get_forecast(
+        geo, token=token, horizon_hours=_ceil_hours(deadline + duration)
+    )
+    if forecast is None:
+        return None
+
+    now_intensity = forecast.points[0].g_co2e_per_kwh
+    now = datetime.now(timezone.utc)
+    start, intensity = best_window(forecast, duration, deadline=now + deadline)
+    return start, intensity, now_intensity
+
+
+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",
+    finish_by: str | None = None,
+    threshold: float | None = 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
+
+        # Bound the finish time instead of the start
+        codecarbon wait --finish-by 8h --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)
+        # --deadline bounds the start, --finish-by bounds the end; the search
+        # only ever needs the latest acceptable start.
+        if finish_by is not None:
+            max_delay = parse_duration(finish_by) - job_duration
+            if max_delay < timedelta(0):
+                raise ValueError(
+                    f"--finish-by {finish_by} is sooner than --duration {duration}."
+                )
+        else:
+            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 abort."
+        )
+        try:
+            time.sleep(delay_seconds)
+        except KeyboardInterrupt:
+            print("\n⚠️  Wait aborted, the command was not run.", file=sys.stderr)
+            raise typer.Exit(130)
+
+    run_and_monitor(ctx, log_level=log_level, **tracker_args)

codecarbon/core/electricitymaps_api.py

@@ -9,6 +9,7 @@
 from codecarbon.external.logger import logger
 
 URL: str = "https://api.electricitymaps.com/v3/carbon-intensity/latest"
+FORECAST_URL: str = "https://api.electricitymaps.com/v3/carbon-intensity/forecast"
 ELECTRICITYMAPS_API_TIMEOUT: int = 30
 
 # Grid carbon intensity is published hourly at best, while emissions are computed

codecarbon/core/intensity_forecast.py

@@ -0,0 +1,139 @@
+"""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.
+
+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 short TTL.
+"""
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import List, Optional, Tuple
+
+from codecarbon.core.electricitymaps_api import FORECAST_URL, location_params, request
+from codecarbon.external.geography import GeoMetadata
+from codecarbon.external.logger import logger
+
+
+@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
+
+
+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:
+        data = request(FORECAST_URL, location_params(geo), token)
+        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,
+        )
+    except Exception as e:
+        logger.error(
+            f"intensity_forecast.get_forecast: {type(e).__name__}: {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.
+
+    `deadline` is the latest acceptable *start* time -- the job may run past
+    it. Only windows starting at or after the first forecast point, at or
+    before `deadline`, and fully covered by the forecast 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. Gaps are not
+    # guaranteed uniform, so the smallest one is the safe assumption.
+    step = (
+        min(b.at - a.at for a, b in zip(points, points[1:]))
+        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 start.at > 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.
+        # Each point holds until the next one, so weight it by how much of its
+        # period falls inside the window: an hourly point half-covered by the
+        # window's end must not count as a full hour.
+        covered = [point for point in points[start_index:] if point.at < window_end]
+        ends = [point.at for point in covered[1:]] + [window_end]
+        weights = [
+            (min(end, window_end) - point.at).total_seconds()
+            for point, end in zip(covered, ends)
+        ]
+        mean = sum(
+            point.g_co2e_per_kwh * weight for point, weight in zip(covered, weights)
+        ) / sum(weights)
+        if best is None or mean < best[1]:
+            best = (start.at, mean)
+
+    return best or fallback