From 0979c97ac2dd182baaf54c41787c75f057f0ce54 Mon Sep 17 00:00:00 2001 From: Jammy2211 Date: Fri, 30 Oct 2026 14:51:57 +0000 Subject: [PATCH 1/2] feat(campaign): Green Algorithms carbon estimate and end-of-campaign report (hpc-campaign p4) autofit.campaign.carbon: E = t x (n_c P_c u_c + n_m x 0.3725 + n_g P_g) x PUE, C = E x CI, in four cases - low (low inputs, measured usage, peak memory), central (usage basis), allocation (u = 1, requested memory) and high (high inputs on the allocation basis); site constants from autonerves (tdp per partition, gpu power, PUE, carbon_intensity bases with ci_basis central and the spread of bases as bounds, ruling F6); no world-average default, a missing input is reported and its figure left out; successful vs failed or repeated work (earlier attempts charged from allocated CPU time); per stage, per completed task, per object. forecast() for planned stages reproduces the report 1.4 worked DR1 estimate exactly (1,239 / 1,717 / 4,494 kWh, 124 / 223 / 854 kgCO2e). Optional, cached NESO period intensity and a 48 h --begin hint, network only on request. autofit.campaign.report: report.md + report.json (compute, energy and carbon with inputs and sources, failure census, stage table, results-index counts, a paper paragraph that says estimate, gives the range and names the basis). CLI: carbon, report. Refs #1746 Co-Authored-By: Claude Opus 5.5 --- autofit/campaign/__init__.py | 6 +- autofit/campaign/__main__.py | 113 +++++ autofit/campaign/carbon.py | 706 +++++++++++++++++++++++++++ autofit/campaign/report.py | 350 +++++++++++++ test_autofit/campaign/test_carbon.py | 247 ++++++++++ test_autofit/campaign/test_report.py | 132 +++++ 6 files changed, 1552 insertions(+), 2 deletions(-) create mode 100644 autofit/campaign/carbon.py create mode 100644 autofit/campaign/report.py create mode 100644 test_autofit/campaign/test_carbon.py create mode 100644 test_autofit/campaign/test_report.py diff --git a/autofit/campaign/__init__.py b/autofit/campaign/__init__.py index a385983d2..e25c8d24a 100644 --- a/autofit/campaign/__init__.py +++ b/autofit/campaign/__init__.py @@ -12,9 +12,11 @@ - ``retro``: the digest of a finished campaign from a ``sacct`` dump alone; - ``evidence``: profiling / inference summaries, read and graded per stage; - ``plan``: proposed resources, wall time, ETA and disk, every number graded; -- ``pilot``: a stratified subset of the manifest to measure first. +- ``pilot``: a stratified subset of the manifest to measure first; +- ``carbon``: the energy and carbon estimate (Green Algorithms; four cases); +- ``report``: the end-of-campaign report, with a paragraph for a paper. -Command line: ``python -m autofit.campaign {validate,submit,status,resubmit,retro,plan,pilot}``; +Command line: ``python -m autofit.campaign {validate,submit,status,resubmit,retro,plan,pilot,carbon,report}``; in Python, ``autofit.campaign.submit.submit(campaign, scheduler)`` and ``autofit.campaign.retro.retro(dump, ...)``. diff --git a/autofit/campaign/__main__.py b/autofit/campaign/__main__.py index fc83878e8..f93ac7492 100644 --- a/autofit/campaign/__main__.py +++ b/autofit/campaign/__main__.py @@ -28,6 +28,15 @@ [--seed S] [--output DIR] Write ``pilot.json``: a stratified subset of the manifest to measure first. Nothing is submitted. +carbon SOURCE [--site NAME] [--objects N] [--ci-basis B] [--pue ...] [--tdp ...] [--ci ...] + [--neso [--neso-region ID]] [--begin-hint HOURS] + The energy and carbon estimate (Green Algorithms) of a digest folder or + ``campaign.yaml``. ``--neso`` fetches the grid's recorded intensity for the + campaign's period (network, cached); ``--begin-hint`` suggests a start time for a + short, deferrable array from the 48 h forecast (network). Offline otherwise. +report SOURCE [--site NAME] [--objects N] [--results-index DIR] [--output DIR] [carbon options] + Write ``report.md`` + ``report.json``: compute, energy and carbon, failure census, + stage tables, results counts and a paragraph for a paper. """ import argparse import json @@ -227,6 +236,100 @@ def _pilot(arguments) -> int: return 0 +def _carbon_inputs(arguments): + """ + The digest folder, site and carbon inputs a ``carbon`` / ``report`` call names. + """ + from pathlib import Path + + from autofit.campaign.carbon import CarbonInputs, apply_neso, neso_intensity, period_of + from autofit.campaign.digest import read_outcomes + + def bounds(text): + if text is None: + return None + values = [float(v) for v in text.split(",")] + return values[0] if len(values) == 1 else tuple(values) + + source = Path(arguments.source) + site = None + if source.suffix in (".yaml", ".yml"): + campaign = Campaign.load(source) + directory = campaign.directory + site = campaign.site if campaign.site_name else None + else: + directory = source + if arguments.site: + from autonerves.site_config import load_site, site_path + + site = load_site(site_path(arguments.site)) + inputs = CarbonInputs.from_site( + site, + ci_basis=arguments.ci_basis, + pue=bounds(arguments.pue), + tdp_w_per_cpu=bounds(arguments.tdp), + ci=bounds(arguments.ci), + ) + if arguments.neso: + period = period_of(read_outcomes(Path(directory) / "tasks.jsonl")) + if period is None: + raise CampaignException("No task start/end times: cannot fetch NESO intensity.") + inputs = apply_neso(inputs, neso_intensity(*period, region_id=arguments.neso_region)) + return directory, site, inputs + + +def _carbon(arguments) -> int: + from pathlib import Path + + from autofit.campaign.carbon import begin_hint, estimate, neso_forecast + from autofit.campaign.digest import read_outcomes + + if arguments.begin_hint is not None: + hint = begin_hint(neso_forecast(arguments.neso_region), arguments.begin_hint * 3600) + print(json.dumps(hint, indent=1)) + return 0 + try: + directory, _, inputs = _carbon_inputs(arguments) + except CampaignException as exception: + print(exception, file=sys.stderr) + return 1 + result = estimate(read_outcomes(Path(directory) / "tasks.jsonl"), inputs, objects=arguments.objects) + print(json.dumps(result, indent=1, default=str)) + return 1 if "error" in result else 0 + + +def _report(arguments) -> int: + from autofit.campaign.report import build_report, report_markdown, write_report + + try: + directory, site, inputs = _carbon_inputs(arguments) + except CampaignException as exception: + print(exception, file=sys.stderr) + return 1 + report = build_report( + directory, + site=site, + inputs=inputs, + objects=arguments.objects, + results_index=arguments.results_index, + ) + write_report(directory, report, output=arguments.output) + print(report_markdown(report), end="") + return 0 + + +def _add_carbon_options(command): + command.add_argument("source", nargs="?", help="A digest folder or campaign.yaml.") + command.add_argument("--site", help="Site layer (autonerves site file).") + command.add_argument("--objects", type=int, help="Objects analysed, for per-object values.") + command.add_argument("--ci-basis", help="carbon_intensity basis to use centrally.") + command.add_argument("--pue", help="PUE: value, low,high or low,central,high.") + command.add_argument("--tdp", help="W per scheduler CPU: value, low,high or low,central,high.") + command.add_argument("--ci", help="gCO2e/kWh: value, low,high or low,central,high.") + command.add_argument("--neso", action="store_true", help="Fetch the period's grid intensity (network).") + command.add_argument("--neso-region", type=int, help="NESO region id (12: South England).") + + def main(argv=None) -> int: parser = argparse.ArgumentParser(prog="python -m autofit.campaign") commands = parser.add_subparsers(dest="command", required=True) @@ -285,6 +388,16 @@ def main(argv=None) -> int: pilot.add_argument("--seed", type=int, default=0) pilot.add_argument("--output", help="Write pilot.json here.") pilot.set_defaults(handler=_pilot) + carbon = commands.add_parser("carbon", help="Energy and carbon estimate of a campaign.") + _add_carbon_options(carbon) + carbon.add_argument("--begin-hint", type=float, help="Hours of a deferrable job (network).") + carbon.set_defaults(handler=_carbon) + + report = commands.add_parser("report", help="End-of-campaign report.") + _add_carbon_options(report) + report.add_argument("--results-index", help="A results index folder (autofit.results).") + report.add_argument("--output", help="Write report.md / report.json here.") + report.set_defaults(handler=_report) arguments = parser.parse_args(argv) return arguments.handler(arguments) diff --git a/autofit/campaign/carbon.py b/autofit/campaign/carbon.py new file mode 100644 index 000000000..0e0fb30d2 --- /dev/null +++ b/autofit/campaign/carbon.py @@ -0,0 +1,706 @@ +""" +Energy and carbon of a campaign, estimated with the Green Algorithms model +(Lannelongue, Grealey & Inouye 2021, Adv. Sci.; arXiv 2007.07610): + + E [kWh] = t × (n_c × P_c × u_c + n_m × P_m + n_g × P_g) × PUE / 1000 + C [gCO2e] = E × CI + +with ``t`` the wall time in hours, ``n_c`` the CPUs, ``P_c`` the power per CPU (W), +``u_c`` the CPU usage (0-1), ``n_m`` the memory in GB, ``P_m`` = 0.3725 W/GB, ``n_g`` +the GPUs, ``P_g`` their power (W, at full use) and ``CI`` the grid carbon intensity. + +This is an **estimate**: CPU power is not measured on a shared node (RAPL is root-only +and package-wide; Slurm energy accounting is off at most sites), so the per-CPU power, +the PUE and the carbon intensity come from the site file (``autonerves.site_config``) +as low / central / high values. Every estimate is given as four cases: + +- ``low``: low power, PUE and carbon intensity; measured CPU usage; peak memory; +- ``central``: central inputs; measured CPU usage; requested memory (the usage basis); +- ``allocation``: central inputs; every reserved CPU charged at full use (``u = 1``) + and requested memory (the allocation basis, which is what reserving costs); +- ``high``: high inputs on the allocation basis — the upper bound. + +The carbon-intensity range is the spread of the site's bases (e.g. regional, national, +DESNZ; ruling F6): the central value is the site's ``ci_basis``, the bounds the lowest +and highest of the bases' central values. Half-hourly spreads within one basis are not +used, since a campaign averages over many half-hours. + +Nothing falls back to Green Algorithms' world-average defaults (PUE 1.67, 475 g/kWh, +10-12 W per core), which overstate a modern site by an order of magnitude: an input the +site does not give is reported missing and the figure that needs it is left out. + +The network is never used unless asked: ``neso_intensity`` fetches the GB grid's +recorded carbon intensity for a campaign's period from the NESO Carbon Intensity API +(cached on disk), and ``begin_hint`` suggests a start time for a short, deferrable array +from a 48 h forecast. +""" +import json +import math +import urllib.request +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Callable, Dict, Iterable, List, Optional, Sequence, Tuple + +from autofit.campaign.digest import TaskOutcome +from autofit.non_linear.status.states import RunState + +SCHEMA = 1 + +P_MEMORY_W_PER_GB = 0.3725 +"""Green Algorithms' power draw of memory, W per GB.""" + +CASES = ("low", "central", "allocation", "high") + +NESO_API = "https://api.carbonintensity.org.uk" +NESO_MAX_DAYS = 13 +"""The API serves at most 14 days per request.""" + + +@dataclass +class Bounds: + """ + A low / central / high triple of one input. + """ + + low: Optional[float] + central: Optional[float] + high: Optional[float] + + @classmethod + def of(cls, value) -> Optional["Bounds"]: + """ + From an ``autonerves.site_config.SiteValue``, a number, a 2- or 3-sequence or + ``None``. + """ + if value is None: + return None + if isinstance(value, Bounds): + return value + if isinstance(value, (int, float)): + return cls(float(value), float(value), float(value)) + if isinstance(value, (list, tuple)): + if len(value) == 2: + return cls(value[0], (value[0] + value[1]) / 2, value[1]) + return cls(*value) + central = value.central + if central is None: + return None + low, high = value.range + return cls( + low if low is not None else central, + central, + high if high is not None else central, + ) + + def case(self, case: str) -> Optional[float]: + """ + The value a case uses: ``low`` → low, ``high`` → high, else central. + """ + return {"low": self.low, "high": self.high}.get(case, self.central) + + def as_list(self) -> List[Optional[float]]: + return [self.low, self.central, self.high] + + +@dataclass +class CarbonInputs: + """ + The site constants of an estimate, each a ``Bounds``. + """ + + tdp_w_per_cpu: Dict[str, Bounds] = field(default_factory=dict) + """Power per Slurm CPU (W), by partition.""" + gpu_w: Dict[str, Bounds] = field(default_factory=dict) + """Power per GPU at full use (W), by partition.""" + pue: Optional[Bounds] = None + ci_g_per_kwh: Optional[Bounds] = None + ci_basis: Optional[str] = None + ci_bases: Dict[str, float] = field(default_factory=dict) + """The central carbon intensity of every basis the site gives.""" + ci_source: Optional[str] = None + unverified: List[str] = field(default_factory=list) + site: Optional[str] = None + + @classmethod + def from_site( + cls, + site=None, + ci_basis: Optional[str] = None, + pue=None, + tdp_w_per_cpu=None, + ci=None, + ) -> "CarbonInputs": + """ + The inputs from an ``autonerves`` ``Site`` (may be ``None``), with optional + overrides: ``pue``, ``tdp_w_per_cpu`` (all partitions) and ``ci`` (gCO2e/kWh), + each a number, a ``(low, high)`` or a ``(low, central, high)``. + """ + inputs = cls(site=getattr(site, "name", None)) + if site is not None: + for name, partition in site.partitions.items(): + tdp = Bounds.of(partition.values.get("tdp_w_per_cpu")) + if tdp is not None: + inputs.tdp_w_per_cpu[name] = tdp + gpu = Bounds.of(partition.values.get("gpu_tdp_w")) + if gpu is not None: + inputs.gpu_w[name] = gpu + inputs.pue = Bounds.of(site.pue) + table = getattr(site, "carbon_intensity", {}) or {} + basis = ci_basis or site.ci_basis + if table and basis in table: + inputs.ci_basis = basis + inputs.ci_bases = { + name: float(value.central) + for name, value in table.items() + if value.central is not None + } + centrals = list(inputs.ci_bases.values()) + inputs.ci_g_per_kwh = Bounds( + min(centrals), inputs.ci_bases[basis], max(centrals) + ) + inputs.ci_source = ( + f"site {site.name}: {basis} basis, bounds the spread of bases " + f"{', '.join(f'{k} {v:g}' for k, v in inputs.ci_bases.items())}" + ) + inputs.unverified = list(site.unverified()) + if pue is not None: + inputs.pue = Bounds.of(pue) + if tdp_w_per_cpu is not None: + bounds = Bounds.of(tdp_w_per_cpu) + inputs.tdp_w_per_cpu = {name: bounds for name in inputs.tdp_w_per_cpu} + inputs.tdp_w_per_cpu["*"] = bounds + if ci is not None: + inputs.ci_g_per_kwh = Bounds.of(ci) + inputs.ci_basis = ci_basis or "given" + inputs.ci_source = "given on the command line" + return inputs + + def tdp_for(self, partition: Optional[str]) -> Tuple[Optional[Bounds], bool]: + """ + The per-CPU power of ``partition`` and whether it is a fallback: a partition + the site does not list (or a task on several) takes the widest range of the + partitions it does list. + """ + if partition in self.tdp_w_per_cpu: + return self.tdp_w_per_cpu[partition], False + if "*" in self.tdp_w_per_cpu: + return self.tdp_w_per_cpu["*"], False + if not self.tdp_w_per_cpu: + return None, True + values = list(self.tdp_w_per_cpu.values()) + return ( + Bounds( + min(v.low for v in values), + sum(v.central for v in values) / len(values), + max(v.high for v in values), + ), + True, + ) + + def missing(self) -> List[str]: + missing = [] + if not self.tdp_w_per_cpu: + missing.append("tdp_w_per_cpu") + if self.pue is None: + missing.append("pue") + if self.ci_g_per_kwh is None: + missing.append("carbon_intensity") + return missing + + def to_dict(self) -> Dict: + return { + "site": self.site, + "tdp_w_per_cpu": {k: v.as_list() for k, v in self.tdp_w_per_cpu.items()}, + "gpu_w": {k: v.as_list() for k, v in self.gpu_w.items()}, + "pue": None if self.pue is None else self.pue.as_list(), + "ci_g_per_kwh": None + if self.ci_g_per_kwh is None + else self.ci_g_per_kwh.as_list(), + "ci_basis": self.ci_basis, + "ci_bases": self.ci_bases, + "ci_source": self.ci_source, + "unverified": self.unverified, + "missing": self.missing(), + } + + +def power_w( + cpus: float, + tdp_w_per_cpu: float, + usage: float, + memory_gb: float, + gpus: float = 0, + gpu_w: float = 0.0, +) -> float: + """ + The Green Algorithms power draw (W) of one task, before PUE. + """ + return cpus * tdp_w_per_cpu * usage + memory_gb * P_MEMORY_W_PER_GB + gpus * gpu_w + + +def gpu_count(gres: Optional[str]) -> int: + """ + The number of GPUs a ``--gres`` / TRES request asks for (``gres/gpu=2``, + ``gpu:a100:2``, ``gpu:1``; a bare ``gpu`` is one). + """ + if not gres: + return 0 + total = 0 + for item in str(gres).replace(" ", ",").split(","): + item = item.strip().lower() + if not item: + continue + name = item.split("=", 1)[0].split(":", 1)[0] + if name not in ("gpu", "gpus", "gres/gpu") and not name.startswith("gres/gpu"): + continue + if "=" in item: + count = item.split("=", 1)[1] + else: + parts = item.split(":") + count = parts[-1] if len(parts) > 1 else "1" + try: + total += int(count) + except ValueError: + total += 1 + return total + + +def _empty() -> Dict[str, float]: + return {case: 0.0 for case in CASES} + + +@dataclass +class _Tally: + kwh: Dict[str, float] = field(default_factory=_empty) + successful_kwh: Dict[str, float] = field(default_factory=_empty) + tasks: int = 0 + completed: int = 0 + + def add(self, energy: Dict[str, float], successful: bool): + for case in CASES: + self.kwh[case] += energy[case] + if successful: + self.successful_kwh[case] += energy[case] + + +def _usage(outcome: TaskOutcome) -> Optional[float]: + efficiency = outcome.cpu_efficiency + if efficiency is None: + return None + return min(max(efficiency, 0.0), 1.0) + + +def _attempt_energy( + hours: float, + cpus: float, + usage: float, + memory_requested_gb: float, + memory_peak_gb: float, + gpus: int, + tdp: Bounds, + gpu: Optional[Bounds], + pue: Bounds, +) -> Dict[str, float]: + energy = {} + for case in CASES: + u = 1.0 if case in ("allocation", "high") else usage + memory = memory_peak_gb if case == "low" else memory_requested_gb + gpu_w = gpu.case(case) if (gpu is not None and gpus) else 0.0 + watts = power_w(cpus, tdp.case(case), u, memory, gpus, gpu_w) + energy[case] = hours * watts * pue.case(case) / 1000.0 + return energy + + +def _carbon(kwh: Dict[str, float], ci: Optional[Bounds]) -> Optional[Dict[str, float]]: + if ci is None: + return None + return {case: kwh[case] * ci.case(case) / 1000.0 for case in CASES} + + +def _round_cases(values: Optional[Dict[str, float]], digits=6): + if values is None: + return None + return {case: round(value, digits) for case, value in values.items()} + + +def _summary(tally: _Tally, ci: Optional[Bounds]) -> Dict: + failed = {case: tally.kwh[case] - tally.successful_kwh[case] for case in CASES} + central = tally.kwh["central"] + summary = { + "tasks": tally.tasks, + "completed": tally.completed, + "kwh": _round_cases(tally.kwh), + "kg_co2e": _round_cases(_carbon(tally.kwh, ci)), + "successful_kwh": _round_cases(tally.successful_kwh), + "failed_or_repeated_kwh": _round_cases(failed), + "failed_or_repeated_share": round(failed["central"] / central, 3) + if central + else None, + } + if tally.completed: + per_task = {case: tally.kwh[case] / tally.completed for case in CASES} + summary["per_completed_task"] = { + "kwh": _round_cases(per_task, 6), + "g_co2e": _round_cases( + None + if ci is None + else {c: per_task[c] * ci.case(c) for c in CASES}, + 1, + ), + } + return summary + + +def estimate( + outcomes: Iterable[TaskOutcome], + inputs: CarbonInputs, + objects: Optional[int] = None, +) -> Dict: + """ + The energy and carbon of a campaign's tasks (live or retrospective), per stage and + in total, in the four cases (module docstring), split into successful work (the + final attempt of each completed task) and failed or repeated work (every other + attempt and every task that did not complete). + + Earlier attempts of a task are known only by their allocated CPU time; they are + charged with the final attempt's CPUs, memory and usage. A task with no CPU time + recorded is charged at full use (``u = 1``) in every case and counted in + ``assumptions``. GPUs are charged at full use, as Green Algorithms does. + + Parameters + ---------- + objects + The number of objects (e.g. lenses) the campaign set out to analyse; when + given, the totals are also divided per object. + """ + missing = [m for m in inputs.missing() if m != "carbon_intensity"] + result = { + "schema": SCHEMA, + "model": "Green Algorithms (Lannelongue et al. 2021)", + "cases": { + "low": "low inputs, measured CPU usage, peak memory", + "central": "central inputs, measured CPU usage, requested memory (usage basis)", + "allocation": "central inputs, every reserved CPU at full use, requested memory (allocation basis)", + "high": "high inputs on the allocation basis", + }, + "inputs": inputs.to_dict(), + } + if missing: + result["error"] = ( + f"no energy estimate: the site gives no {', '.join(missing)} " + f"(pass them explicitly; no world-average default is used)" + ) + return result + + ci = inputs.ci_g_per_kwh + stages: Dict[str, _Tally] = {} + total = _Tally() + counts = {"usage_missing": 0, "partition_fallback": 0, "gpu_tasks": 0, "no_time": 0} + + for outcome in outcomes: + tally = stages.setdefault(outcome.stage, _Tally()) + tally.tasks += 1 + total.tasks += 1 + completed = outcome.state == str(RunState.COMPLETED) + if completed: + tally.completed += 1 + total.completed += 1 + cpus = outcome.cpus or 0 + if not cpus or outcome.elapsed_s is None: + counts["no_time"] += 1 + continue + + usage = _usage(outcome) + if usage is None: + counts["usage_missing"] += 1 + usage = 1.0 + tdp, fallback = inputs.tdp_for(outcome.partition) + if fallback: + counts["partition_fallback"] += 1 + gpus = gpu_count(outcome.gres) + if gpus: + counts["gpu_tasks"] += 1 + gpu = inputs.gpu_w.get(outcome.partition) if gpus else None + + requested_gb = (outcome.req_mem_mb or 0.0) / 1024.0 + peak_gb = ( + outcome.max_rss_mb / 1024.0 if outcome.max_rss_mb else requested_gb + ) + + # The final attempt, then any earlier ones (known by allocated CPU time only). + earlier_cpu_s = (outcome.cpu_s_all_attempts or 0.0) - outcome.elapsed_s * cpus + attempts = [(outcome.elapsed_s / 3600.0, completed and outcome.success is not False)] + if earlier_cpu_s > 1e-6: + attempts.append((earlier_cpu_s / cpus / 3600.0, False)) + for hours, successful in attempts: + energy = _attempt_energy( + hours, cpus, usage, requested_gb, peak_gb, gpus, tdp, gpu, inputs.pue + ) + tally.add(energy, successful) + total.add(energy, successful) + + result["total"] = _summary(total, ci) + result["stages"] = {name: _summary(t, ci) for name, t in sorted(stages.items())} + if objects: + result["objects"] = objects + result["per_object"] = { + "kwh": _round_cases({c: total.kwh[c] / objects for c in CASES}, 6), + "g_co2e": None + if ci is None + else _round_cases( + {c: total.kwh[c] * ci.case(c) / objects for c in CASES}, 1 + ), + } + result["assumptions"] = counts + if ci is None: + result["note"] = ( + "energy only: the site gives no carbon_intensity, so no carbon figure " + "(pass --ci; no world-average default is used)" + ) + return result + + +@dataclass +class StageForecast: + """ + A planned stage for ``forecast``: ``tasks`` tasks of ``hours`` (low, central, high) + wall time on ``cpus`` CPUs at CPU ``usage``, holding ``memory_gb`` (low = peak, + central = requested, high = the largest reservation considered). + """ + + name: str + tasks: int + hours: Sequence[float] + cpus: int + usage: float + memory_gb: Sequence[float] + gpus: int = 0 + partition: Optional[str] = None + + +def forecast(stages: Iterable[StageForecast], inputs: CarbonInputs) -> Dict: + """ + The energy and carbon of a planned campaign, in the same four cases as + ``estimate``: ``low`` takes each stage's low hours and memory at its usage, + ``central`` the central ones, ``allocation`` the central ones at full use and + ``high`` the high ones at full use. Used by planning, and to reproduce the worked + estimate of the hpc-campaign report (§1.4). + """ + missing = [m for m in inputs.missing() if m != "carbon_intensity"] + if missing: + return {"error": f"no energy forecast: the site gives no {', '.join(missing)}"} + ci = inputs.ci_g_per_kwh + result = {"stages": {}, "inputs": inputs.to_dict()} + total = _empty() + for stage in stages: + tdp, _ = inputs.tdp_for(stage.partition) + gpu = inputs.gpu_w.get(stage.partition) if stage.gpus else None + hours = Bounds.of(tuple(stage.hours)) + memory = Bounds.of(tuple(stage.memory_gb)) + kwh = {} + for case in CASES: + u = 1.0 if case in ("allocation", "high") else stage.usage + h, m = hours.case(case), memory.case(case) + gpu_w = gpu.case(case) if gpu is not None else 0.0 + watts = power_w(stage.cpus, tdp.case(case), u, m, stage.gpus, gpu_w) + kwh[case] = stage.tasks * h * watts * inputs.pue.case(case) / 1000.0 + total[case] += kwh[case] + result["stages"][stage.name] = { + "watts_central": round( + power_w( + stage.cpus, + tdp.central, + stage.usage, + memory.central, + stage.gpus, + gpu.central if gpu is not None else 0.0, + ), + 2, + ), + "kwh": _round_cases(kwh), + "kg_co2e": _round_cases(_carbon(kwh, ci), 2), + } + result["total"] = {"kwh": _round_cases(total), "kg_co2e": _round_cases(_carbon(total, ci), 2)} + return result + + +# --- NESO Carbon Intensity API (network only when asked) ------------------------- + + +def _fetch_json(url: str, timeout: float = 30.0) -> Dict: + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + + +def _iso(moment: datetime) -> str: + return moment.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%MZ") + + +def _points(document: Dict) -> List[float]: + data = document.get("data") + if isinstance(data, dict): + data = data.get("data", []) + elif isinstance(data, list) and data and isinstance(data[0], dict) and "data" in data[0]: + data = [point for region in data for point in region.get("data", [])] + values = [] + for point in data or []: + intensity = point.get("intensity") or {} + value = intensity.get("actual") + if value is None: + value = intensity.get("forecast") + if value is not None: + values.append(float(value)) + return values + + +def default_cache_dir() -> Path: + return Path.home() / ".cache" / "autofit" / "carbon" + + +def neso_intensity( + start: datetime, + end: datetime, + region_id: Optional[int] = None, + cache_dir: Optional[Path] = None, + fetch: Callable[[str], Dict] = _fetch_json, +) -> Dict: + """ + The mean GB grid carbon intensity (gCO2e/kWh) over ``start``-``end`` from the NESO + Carbon Intensity API (CC BY 4.0, no key): national actuals, or the regional series + of ``region_id`` (12 is South England). Requests are split into windows the API + accepts and each response is cached under ``cache_dir`` (default + ``~/.cache/autofit/carbon``), so a period is fetched once. + + Returns ``{"mean": ..., "points": n, "basis": ..., "start": ..., "end": ...}``. + """ + cache_dir = Path(cache_dir) if cache_dir is not None else default_cache_dir() + values = [] + window_start = start + while window_start < end: + window_end = min(window_start + timedelta(days=NESO_MAX_DAYS), end) + span = f"{_iso(window_start)}/{_iso(window_end)}" + if region_id is None: + url = f"{NESO_API}/intensity/{span}" + else: + url = f"{NESO_API}/regional/intensity/{span}/regionid/{region_id}" + cache = cache_dir / (url[len(NESO_API) + 1 :].replace("/", "_").replace(":", "") + ".json") + if cache.exists(): + document = json.loads(cache.read_text()) + else: + document = fetch(url) + cache.parent.mkdir(parents=True, exist_ok=True) + cache.write_text(json.dumps(document)) + values += _points(document) + window_start = window_end + if not values: + raise ValueError(f"NESO returned no intensity for {start} → {end}") + return { + "mean": sum(values) / len(values), + "points": len(values), + "basis": "national (NESO actual)" if region_id is None else f"NESO region {region_id}", + "start": _iso(start), + "end": _iso(end), + } + + +def begin_hint( + series: Sequence[Tuple[datetime, float]], + duration_s: float, + now: Optional[datetime] = None, +) -> Optional[Dict]: + """ + The start, within a forecast ``series`` of ``(half-hour start, gCO2e/kWh)``, that + minimises the mean forecast intensity over a job of ``duration_s``, against + starting now. Only worth acting on for a short array that can wait: over a + campaign of days the window averages out (hpc-campaign report §1.4). + """ + series = sorted(series) + if not series: + return None + slots = max(1, math.ceil(duration_s / 1800.0)) + if len(series) < slots: + return None + now = now or series[0][0] + + def mean_from(i): + return sum(v for _, v in series[i : i + slots]) / slots + + first = next((i for i, (t, _) in enumerate(series) if t >= now), 0) + best = min(range(first, len(series) - slots + 1), key=mean_from) + return { + "begin": series[best][0].strftime("%Y-%m-%dT%H:%M:%S"), + "mean_g_per_kwh": round(mean_from(best), 1), + "now_mean_g_per_kwh": round(mean_from(first), 1), + "slurm": f"--begin={series[best][0].strftime('%Y-%m-%dT%H:%M:%S')}", + } + + +def neso_forecast( + region_id: Optional[int] = None, + now: Optional[datetime] = None, + fetch: Callable[[str], Dict] = _fetch_json, +) -> List[Tuple[datetime, float]]: + """ + The 48 h carbon-intensity forecast from now (national, or ``region_id``). + """ + now = now or datetime.now(timezone.utc) + if region_id is None: + url = f"{NESO_API}/intensity/{_iso(now)}/fw48h" + else: + url = f"{NESO_API}/regional/intensity/{_iso(now)}/fw48h/regionid/{region_id}" + document = fetch(url) + data = document.get("data") + if isinstance(data, dict): + data = data.get("data", []) + series = [] + for point in data or []: + value = (point.get("intensity") or {}).get("forecast") + if value is None or not point.get("from"): + continue + moment = datetime.strptime(point["from"], "%Y-%m-%dT%H:%MZ").replace( + tzinfo=timezone.utc + ) + series.append((moment, float(value))) + return series + + +def apply_neso(inputs: CarbonInputs, measured: Dict) -> CarbonInputs: + """ + ``inputs`` with the central carbon intensity replaced by a NESO period mean; the + bounds widen to include it. + """ + mean = measured["mean"] + ci = inputs.ci_g_per_kwh + low = mean if ci is None else min(ci.low, mean) + high = mean if ci is None else max(ci.high, mean) + inputs.ci_g_per_kwh = Bounds(low, mean, high) + inputs.ci_basis = measured["basis"] + inputs.ci_bases = {**inputs.ci_bases, measured["basis"]: round(mean, 1)} + inputs.ci_source = ( + f"{measured['basis']} mean over {measured['start']} → {measured['end']} " + f"({measured['points']} half-hours)" + ) + return inputs + + +def period_of(outcomes: Iterable[TaskOutcome]) -> Optional[Tuple[datetime, datetime]]: + """ + The first start and last end of ``outcomes`` (naive times taken as UTC). + """ + starts, ends = [], [] + for outcome in outcomes: + for value, into in ((outcome.start, starts), (outcome.end, ends)): + if not value: + continue + try: + moment = datetime.fromisoformat(value) + except ValueError: + continue + if moment.tzinfo is None: + moment = moment.replace(tzinfo=timezone.utc) + into.append(moment) + if not starts or not ends: + return None + return min(starts), max(ends) diff --git a/autofit/campaign/report.py b/autofit/campaign/report.py new file mode 100644 index 000000000..c6178a085 --- /dev/null +++ b/autofit/campaign/report.py @@ -0,0 +1,350 @@ +""" +The end-of-campaign report: ``python -m autofit.campaign report``. + +Built from a campaign's digest folder (``digest.json`` + ``tasks.jsonl``, written by +``status`` or ``retro``) and the site, it writes ``report.md`` and ``report.json``: + +- compute: tasks, period, CPU-hours (successful and failed or repeated), peak CPUs; +- energy and carbon (``autofit.campaign.carbon``): the four cases, the share spent on + failed or repeated work, per stage, per completed task and, given the number of + objects, per object, with every input, its source and whether it is unverified; +- the failure census and a per-stage table of wall time, memory and CPU efficiency; +- results: the counts of a results index (``python -m autofit.results index``) when + one is given; +- a paste-ready paragraph for a paper, which calls the figures an estimate, gives the + range and names the carbon-intensity basis. + +The report describes what was used. It compares against nothing and claims no +reduction: a comparison belongs to the reader, against the same analysis on other +settings. +""" +import json +from pathlib import Path +from typing import Dict, List, Optional + +from autofit.campaign.carbon import CarbonInputs, estimate, period_of +from autofit.campaign.digest import read_outcomes +from autofit.campaign.units import format_time +from autofit.tools.util import open_atomic + + +def _significant(value: float) -> str: + return f"{value:.2g}" if value < 10 else f"{value:,.0f}" + + +def _scaled(values: Dict[str, float], small: str, large: str) -> Dict[str, str]: + """ + The central, low and high of ``values`` in one unit: ``large`` (a thousand + ``small``) when the central value reaches a thousand, else ``small``. + """ + scale, unit = (1000.0, large) if values["central"] >= 1000 else (1.0, small) + return { + case: _significant(values[case] / scale) + for case in ("low", "central", "high") + } | {"unit": unit} + + +def _number(value) -> str: + if value is None: + return "-" + if isinstance(value, float) and not value.is_integer(): + return f"{value:.3g}" if abs(value) < 10 else f"{value:,.1f}" + return f"{int(value):,}" + + +def results_counts(index_dir) -> Optional[Dict]: + """ + The result-state counts of a results index (``/index.json``), or + ``None`` if there is none. + """ + if index_dir is None: + return None + path = Path(index_dir) / "index.json" + if not path.exists(): + return None + manifest = json.loads(path.read_text()) + return { + "index": str(Path(index_dir)), + "updated": manifest.get("updated"), + "counts": manifest.get("counts", {}), + } + + +def paper_paragraph(digest: Dict, carbon: Dict) -> Optional[str]: + """ + A paragraph for the paper's computing statement, or ``None`` without an energy + estimate. + """ + if "error" in carbon: + return None + totals = digest["totals"] + inputs = carbon["inputs"] + kwh = carbon["total"]["kwh"] + kg = carbon["total"]["kg_co2e"] + share = carbon["total"]["failed_or_repeated_share"] + period = digest.get("period") or {} + when = ( + f" between {period['first_submit'][:10]} and {period['last_end'][:10]}" + if period.get("first_submit") and period.get("last_end") + else "" + ) + tdp = sorted({tuple(v) for v in inputs["tdp_w_per_cpu"].values()}) + tdp_text = "; ".join(f"{c:g} W (range {lo:g}–{hi:g} W)" for lo, c, hi in tdp) + pue = inputs["pue"] + site = f" on the {inputs['site']} site" if inputs.get("site") else "" + + text = ( + f"The {digest['campaign']} campaign ran {totals['tasks']:,} scheduler tasks" + f"{site}{when}, allocating {totals['cpu_h_all_attempts']:,.0f} CPU-hours. " + f"Following the Green Algorithms methodology (Lannelongue et al. 2021), with " + f"wall time, CPU time and requested memory from scheduler accounting, a power " + f"draw of {tdp_text} per scheduler CPU, a data-centre PUE of {pue[1]:g} " + f"(range {pue[0]:g}–{pue[2]:g})" + ) + if kg is not None: + ci = inputs["ci_g_per_kwh"] + bases = ", ".join( + f"{k} {v:g}" for k, v in inputs["ci_bases"].items() if k != inputs["ci_basis"] + ) + text += ( + f" and a grid carbon intensity of {ci[1]:g} gCO2e/kWh ({inputs['ci_basis']} " + f"basis; range {ci[0]:g}–{ci[2]:g} across bases" + + (f": {bases}" if bases else "") + + ")" + ) + energy = _scaled(kwh, "kWh", "MWh") + text += ( + f", we estimate the campaign used ≈{energy['central']} {energy['unit']} " + f"(range {energy['low']}–{energy['high']} {energy['unit']})" + ) + if kg is not None: + mass = _scaled(kg, "kgCO2e", "tCO2e") + text += ( + f", or ≈{mass['central']} {mass['unit']} (range {mass['low']}–" + f"{mass['high']} {mass['unit']})" + ) + per_object = carbon.get("per_object") + if per_object and per_object.get("g_co2e"): + text += ( + f", ≈{per_object['g_co2e']['central'] / 1000:.2g} kgCO2e per object " + f"over {carbon['objects']:,} objects" + ) + text += "." + if share is not None: + text += ( + f" About {share:.0%} of the central estimate was spent on failed, " + f"timed-out or repeated runs." + ) + text += ( + " The upper bound charges every reserved CPU at full use and the requested " + "memory (allocation basis) with the high inputs. The estimate excludes embodied " + "hardware emissions, storage, idle capacity held for the campaign and " + "development runs." + ) + if inputs.get("unverified"): + text += ( + " Site constants not yet confirmed against the cluster are used as ranges." + ) + return text + + +def _cases_row(name: str, kwh: Dict, kg: Optional[Dict]) -> str: + cells = [name] + for case in ("low", "central", "allocation", "high"): + value = _number(kwh[case]) + if kg is not None: + value += f" / {_number(kg[case])}" + cells.append(value) + return "| " + " | ".join(cells) + " |" + + +def build_report( + directory, + site=None, + inputs: Optional[CarbonInputs] = None, + objects: Optional[int] = None, + results_index=None, +) -> Dict: + """ + The report of the campaign whose digest is in ``directory``. + """ + directory = Path(directory) + digest = json.loads((directory / "digest.json").read_text()) + outcomes = read_outcomes(directory / "tasks.jsonl") + inputs = inputs or CarbonInputs.from_site(site) + carbon = estimate(outcomes, inputs, objects=objects) + period = period_of(outcomes) + return { + "campaign": digest.get("campaign"), + "digest": digest, + "carbon": carbon, + "results": results_counts(results_index), + "period": None + if period is None + else [period[0].isoformat(), period[1].isoformat()], + "paragraph": paper_paragraph(digest, carbon), + } + + +def report_markdown(report: Dict) -> str: + digest = report["digest"] + carbon = report["carbon"] + totals = digest["totals"] + lines: List[str] = [f"# Campaign report: {digest['campaign']}", ""] + + period = digest.get("period") or {} + lines += ["## Compute", ""] + lines.append( + f"- {totals['tasks']:,} scheduler tasks: " + + ", ".join(f"{k} {v:,}" for k, v in totals["states"].items()) + ) + if period: + lines.append(f"- Period: {period.get('first_submit')} → {period.get('last_end')}") + cpu_all, cpu_ok = totals["cpu_h_all_attempts"], totals["cpu_h_successful"] + share = 1 - cpu_ok / cpu_all if cpu_all else 0 + lines.append( + f"- CPU-hours allocated: {cpu_all:,.0f}, of which {cpu_ok:,.0f} in completed " + f"tasks ({share:.0%} failed, timed-out or repeated)" + ) + peak = digest.get("peak_cpus") + if isinstance(peak, dict) and peak.get("cpus"): + lines.append(f"- Peak CPUs held: {peak['cpus']:,} ({peak.get('at')})") + lines.append("") + + lines += ["## Energy and carbon (estimate)", ""] + if "error" in carbon: + lines += [f"Not estimated: {carbon['error']}.", ""] + else: + has_carbon = carbon["total"]["kg_co2e"] is not None + unit = "kWh / kgCO2e" if has_carbon else "kWh" + lines.append(f"| stage ({unit}) | low | central | allocation | high |") + lines.append("|---|---|---|---|---|") + for name, stage in carbon["stages"].items(): + lines.append(_cases_row(name, stage["kwh"], stage["kg_co2e"])) + lines.append( + _cases_row("**total**", carbon["total"]["kwh"], carbon["total"]["kg_co2e"]) + ) + lines.append("") + for case, meaning in carbon["cases"].items(): + lines.append(f"- **{case}**: {meaning}") + total = carbon["total"] + lines.append( + f"- Failed or repeated work: {_number(total['failed_or_repeated_kwh']['central'])} " + f"kWh of {_number(total['kwh']['central'])} central " + f"({total['failed_or_repeated_share']:.0%})" + ) + rows = ["", "| stage | per completed task, central | failed or repeated share |", "|---|---|---|"] + for name, stage in carbon["stages"].items(): + per = stage.get("per_completed_task") + if per: + text = f"{_number(per['kwh']['central'])} kWh" + if per["g_co2e"]: + text += f", {_number(per['g_co2e']['central'])} gCO2e" + else: + text = "-" + share_text = ( + "-" + if stage["failed_or_repeated_share"] is None + else f"{stage['failed_or_repeated_share']:.0%}" + ) + rows.append(f"| {name} | {text} | {share_text} |") + lines += rows + if carbon.get("per_object"): + per = carbon["per_object"] + text = f"{_number(per['kwh']['central'])} kWh" + if per["g_co2e"]: + text += f", {_number(per['g_co2e']['central'])} gCO2e" + lines += ["", f"Per object ({carbon['objects']:,} objects), central: {text}."] + if carbon.get("note"): + lines += ["", carbon["note"] + "."] + lines.append("") + + inputs = carbon["inputs"] + lines += ["### Inputs", "", "| input | low | central | high | source |", "|---|---|---|---|---|"] + for name, bounds in inputs["tdp_w_per_cpu"].items(): + lines.append( + f"| W per CPU ({name}) | {bounds[0]} | {bounds[1]} | {bounds[2]} | site {inputs['site']} |" + ) + for name, bounds in inputs["gpu_w"].items(): + lines.append(f"| W per GPU ({name}) | {bounds[0]} | {bounds[1]} | {bounds[2]} | site {inputs['site']} |") + if inputs["pue"]: + p = inputs["pue"] + lines.append(f"| PUE | {p[0]} | {p[1]} | {p[2]} | site {inputs['site']} |") + if inputs["ci_g_per_kwh"]: + c = inputs["ci_g_per_kwh"] + lines.append(f"| gCO2e/kWh | {c[0]} | {c[1]} | {c[2]} | {inputs['ci_source']} |") + lines.append("| memory | | 0.3725 W/GB | | Green Algorithms |") + lines.append("") + if inputs.get("unverified"): + lines.append(f"Unverified site values: {', '.join(inputs['unverified'])}.") + assumptions = carbon.get("assumptions", {}) + notes = [] + if assumptions.get("usage_missing"): + notes.append(f"{assumptions['usage_missing']:,} tasks without CPU time charged at full use") + if assumptions.get("partition_fallback"): + notes.append(f"{assumptions['partition_fallback']:,} tasks on partitions the site does not list (widest range used)") + if assumptions.get("gpu_tasks"): + notes.append(f"{assumptions['gpu_tasks']:,} GPU tasks, GPUs charged at full power") + if assumptions.get("no_time"): + notes.append(f"{assumptions['no_time']:,} tasks with no wall time (not charged)") + if notes: + lines.append("Assumptions: " + "; ".join(notes) + ".") + lines.append("") + + lines += ["## Failure census", "", "| stage | tasks | states | failure classes |", "|---|---|---|---|"] + for name, stage in digest["stages"].items(): + states = ", ".join(f"{k} {v:,}" for k, v in stage["states"].items()) + classes = ", ".join(f"{k} {v:,}" for k, v in stage.get("failure_classes", {}).items()) or "-" + lines.append(f"| {name} | {stage['tasks']:,} | {states} | {classes} |") + lines.append("") + + lines += [ + "## Stages", + "", + "| stage | wall p50 / p95 / max | requested time | peak memory p95 / requested (MB) | CPU efficiency p50 | CPU-h ok / all |", + "|---|---|---|---|---|---|", + ] + for name, stage in digest["stages"].items(): + wall = stage["wall_s"] + lines.append( + f"| {name} | {format_time(wall['p50']) or '-'} / {format_time(wall['p95']) or '-'} / " + f"{format_time(wall['max']) or '-'} | {format_time(stage['request']['time_s']) or '-'} | " + f"{_number(stage['rss_mb']['p95'])} / {_number(stage['request']['mem_mb'])} | " + f"{stage['cpu_efficiency_p50'] if stage['cpu_efficiency_p50'] is not None else '-'} | " + f"{_number(stage['cpu_h']['successful'])} / {_number(stage['cpu_h']['all_attempts'])} |" + ) + flags = [ + f"- {name}: {flag['flag']} — {flag['detail']}" + for name, stage in digest["stages"].items() + for flag in stage.get("flags", []) + ] + if flags: + lines += ["", "Flags:"] + flags + lines.append("") + + if report.get("results"): + results = report["results"] + lines += ["## Results", ""] + lines.append( + f"Results index `{results['index']}` (updated {results['updated']}): " + + ", ".join(f"{k} {v:,}" for k, v in results["counts"].items()) + ) + lines.append("") + + if report.get("paragraph"): + lines += ["## For the paper", "", report["paragraph"], ""] + return "\n".join(lines) + + +def write_report(directory, report: Dict, output=None) -> Path: + """ + Write ``report.md`` and ``report.json`` (the report without the digest, which is + beside it) into ``output`` (default: the digest folder). + """ + output = Path(output or directory) + output.mkdir(parents=True, exist_ok=True) + with open_atomic(output / "report.md", "w") as f: + f.write(report_markdown(report)) + with open_atomic(output / "report.json", "w") as f: + json.dump({k: v for k, v in report.items() if k != "digest"}, f, indent=1, default=str) + return output diff --git a/test_autofit/campaign/test_carbon.py b/test_autofit/campaign/test_carbon.py new file mode 100644 index 000000000..a6ed19981 --- /dev/null +++ b/test_autofit/campaign/test_carbon.py @@ -0,0 +1,247 @@ +""" +The Green Algorithms estimate: the worked DR1 example of the hpc-campaign report +(§1.4), the four cases, failed or repeated work, missing inputs, GPUs and the NESO +helpers (with the network mocked). +""" +from datetime import datetime, timedelta, timezone + +import pytest + +from autofit.campaign.carbon import ( + Bounds, + CarbonInputs, + StageForecast, + apply_neso, + begin_hint, + estimate, + forecast, + gpu_count, + neso_intensity, + power_w, +) +from autofit.campaign.digest import TaskOutcome + +# Report §1.4 / survey carbon.md §6: low 2.8 W, PUE 1.20, 100 g; central 3.5 W, 1.31, +# 130 g; high 4.4 W, 1.45, 190 g. +REPORT_INPUTS = dict(pue=(1.20, 1.31, 1.45), tdp_w_per_cpu=(2.8, 3.5, 4.4), ci=(100, 130, 190)) + +DR1_FORECAST = [ + StageForecast("vis_pix", 4600, (9.1, 9.1, 9.1), 8, 0.83, (11, 11, 64)), + StageForecast("vis_lp", 5000, (1.0, 1.5, 2.0), 8, 0.53, (4, 12, 64)), + StageForecast("sed", 5000, (11 / 60,) * 3, 8, 0.72, (4, 8, 16)), +] + + +def _outcome(state="completed", elapsed_s=3600, cpus=8, total_cpu_s=None, **kwargs): + defaults = dict( + stage="vis_lp", + key="1_0", + state=state, + elapsed_s=elapsed_s, + cpus=cpus, + req_mem_mb=12288.0, + max_rss_mb=4096.0, + total_cpu_s=total_cpu_s if total_cpu_s is not None else 0.5 * elapsed_s * cpus, + partition="ral", + cpu_s_all_attempts=elapsed_s * cpus, + ) + defaults.update(kwargs) + return TaskOutcome(**defaults) + + +def test_power_is_the_green_algorithms_formula(): + # Report §1.4: vis_pix central draw 8 × 3.5 × 0.83 + 11 × 0.3725 = 27.3 W. + assert power_w(8, 3.5, 0.83, 11) == pytest.approx(27.34, abs=0.01) + assert power_w(4, 3.5, 1.0, 0, gpus=1, gpu_w=300) == pytest.approx(314.0) + + +def test_report_worked_dr1_estimate_is_reproduced(): + result = forecast(DR1_FORECAST, CarbonInputs.from_site(None, **REPORT_INPUTS)) + + stages = result["stages"] + assert stages["vis_pix"]["watts_central"] == pytest.approx(27.3, abs=0.05) + assert stages["vis_lp"]["watts_central"] == pytest.approx(19.3, abs=0.05) + assert stages["sed"]["watts_central"] == pytest.approx(23.1, abs=0.05) + + # kWh low / central / high per stage and in total, as in the report's table. + expected = { + "vis_pix": (1140, 1499, 3584), + "vis_lp": (80, 190, 856), + "sed": (19, 28, 55), + } + for name, (low, central, high) in expected.items(): + kwh = stages[name]["kwh"] + assert (kwh["low"], kwh["central"], kwh["high"]) == pytest.approx( + (low, central, high), abs=1 + ) + total = result["total"] + assert (total["kwh"]["low"], total["kwh"]["central"], total["kwh"]["high"]) == pytest.approx( + (1239, 1717, 4494), abs=1 + ) + # ≈0.22 tCO2e, range 0.12–0.85 t. + kg = total["kg_co2e"] + assert (kg["low"], kg["central"], kg["high"]) == pytest.approx((124, 223, 854), abs=1) + # The allocation basis of vis_lp (u = 1) is the survey's 319 kWh. + assert stages["vis_lp"]["kwh"]["allocation"] == pytest.approx(319, abs=1) + + +def test_cases_order_and_bases(): + inputs = CarbonInputs.from_site(None, **REPORT_INPUTS) + result = estimate([_outcome()], inputs) + kwh = result["total"]["kwh"] + + assert kwh["low"] < kwh["central"] < kwh["allocation"] < kwh["high"] + # Usage basis: 1 h × (8 × 3.5 × 0.5 + 12 × 0.3725) × 1.31. + assert kwh["central"] == pytest.approx((14 + 4.47) * 1.31 / 1000, abs=1e-4) + # Allocation basis charges every reserved CPU at full use. + assert kwh["allocation"] == pytest.approx((28 + 4.47) * 1.31 / 1000, abs=1e-4) + # Low uses peak memory (4 GB). + assert kwh["low"] == pytest.approx((8 * 2.8 * 0.5 + 4 * 0.3725) * 1.20 / 1000, abs=1e-4) + assert result["total"]["kg_co2e"]["central"] == pytest.approx( + kwh["central"] * 0.130, abs=1e-4 + ) + + +def test_failed_and_repeated_work_is_split_out(): + inputs = CarbonInputs.from_site(None, **REPORT_INPUTS) + once = _outcome(key="a") + twice = _outcome(key="b", attempts=2, cpu_s_all_attempts=2 * 3600 * 8) + timeout = _outcome(key="c", state="timeout") + failed_check = _outcome(key="d", success=False) + + result = estimate([once, twice, timeout, failed_check], inputs) + total = result["total"] + per_attempt = estimate([once], inputs)["total"]["kwh"]["central"] + + assert total["kwh"]["central"] == pytest.approx(5 * per_attempt, abs=1e-3) + assert total["successful_kwh"]["central"] == pytest.approx(2 * per_attempt, abs=1e-3) + assert total["failed_or_repeated_kwh"]["central"] == pytest.approx(3 * per_attempt, abs=1e-3) + assert total["failed_or_repeated_share"] == pytest.approx(0.6) + assert total["completed"] == 3 + + +def test_per_stage_per_task_and_per_object(): + inputs = CarbonInputs.from_site(None, **REPORT_INPUTS) + outcomes = [_outcome(key=str(i)) for i in range(4)] + [ + _outcome(stage="sed", key="s", elapsed_s=600) + ] + result = estimate(outcomes, inputs, objects=2) + + assert set(result["stages"]) == {"vis_lp", "sed"} + assert result["stages"]["vis_lp"]["per_completed_task"]["kwh"]["central"] == pytest.approx( + result["stages"]["vis_lp"]["kwh"]["central"] / 4, abs=1e-3 + ) + assert result["per_object"]["kwh"]["central"] == pytest.approx( + result["total"]["kwh"]["central"] / 2, abs=1e-3 + ) + + +def test_missing_cpu_time_is_charged_at_full_use(): + inputs = CarbonInputs.from_site(None, **REPORT_INPUTS) + outcome = _outcome() + outcome.total_cpu_s = None + result = estimate([outcome], inputs) + + assert result["assumptions"]["usage_missing"] == 1 + assert result["total"]["kwh"]["central"] == result["total"]["kwh"]["allocation"] + + +def test_gpu_tasks_add_their_gpus(): + site_like = CarbonInputs.from_site(None, **REPORT_INPUTS) + site_like.gpu_w["gpu"] = Bounds(250, 300, 400) + cpu = estimate([_outcome(partition="gpu")], site_like) + gpu = estimate([_outcome(partition="gpu", gres="gres/gpu=1")], site_like) + + assert gpu["assumptions"]["gpu_tasks"] == 1 + assert gpu["total"]["kwh"]["central"] - cpu["total"]["kwh"]["central"] == pytest.approx( + 300 * 1.31 / 1000, abs=1e-3 + ) + assert gpu_count("gres/gpu=2") == 2 + assert gpu_count("gpu:a100:4") == 4 + assert gpu_count("cpu=8,mem=12G") == 0 + assert gpu_count(None) == 0 + + +def test_ral_site_inputs(ral): + inputs = CarbonInputs.from_site(ral) + + assert inputs.missing() == [] + assert inputs.ci_basis == "regional" + # Ruling F6: regional central, the other bases as the bounds. + assert inputs.ci_g_per_kwh.as_list() == [129, 188, 188] + assert inputs.pue.as_list() == [1.14, 1.31, 1.45] + assert inputs.tdp_w_per_cpu["ral"].as_list() == [2.8, 3.5, 4.4] + assert inputs.gpu_w["gpu"].central == 300 + assert "carbon_intensity.regional" in inputs.unverified + + national = CarbonInputs.from_site(ral, ci_basis="national") + assert national.ci_g_per_kwh.central == 129 + + +def test_no_site_means_no_estimate_and_no_world_default(): + result = estimate([_outcome()], CarbonInputs.from_site(None)) + assert "error" in result + assert "tdp_w_per_cpu" in result["error"] and "pue" in result["error"] + + energy_only = estimate( + [_outcome()], CarbonInputs.from_site(None, pue=1.3, tdp_w_per_cpu=3.5) + ) + assert energy_only["total"]["kwh"]["central"] > 0 + assert energy_only["total"]["kg_co2e"] is None + assert "no world-average default" in energy_only["note"] + + +def test_unlisted_partition_takes_the_widest_range(ral): + inputs = CarbonInputs.from_site(ral) + result = estimate([_outcome(partition="other")], inputs) + assert result["assumptions"]["partition_fallback"] == 1 + + +def test_neso_intensity_is_cached_and_chunked(tmp_path): + calls = [] + + def fetch(url): + calls.append(url) + return {"data": [{"intensity": {"forecast": 150, "actual": 100}}, {"intensity": {"forecast": 200, "actual": None}}]} + + start = datetime(2026, 9, 1, tzinfo=timezone.utc) + end = start + timedelta(days=30) + measured = neso_intensity(start, end, cache_dir=tmp_path, fetch=fetch) + + assert len(calls) == 3 # 13 + 13 + 4 days + assert measured["mean"] == pytest.approx(150) + assert measured["points"] == 6 + + neso_intensity(start, end, cache_dir=tmp_path, fetch=fetch) + assert len(calls) == 3 # served from the cache + + regional = neso_intensity( + start, + start + timedelta(days=1), + region_id=12, + cache_dir=tmp_path, + fetch=lambda url: {"data": [{"regionid": 12, "data": [{"intensity": {"forecast": 180}}]}]}, + ) + assert regional["mean"] == 180 + assert regional["basis"] == "NESO region 12" + + +def test_apply_neso_widens_the_bounds(): + inputs = CarbonInputs.from_site(None, ci=(129, 188, 188)) + apply_neso(inputs, {"mean": 210.0, "points": 4, "basis": "NESO region 12", "start": "a", "end": "b"}) + assert inputs.ci_g_per_kwh.as_list() == [129, 210.0, 210.0] + assert inputs.ci_basis == "NESO region 12" + + +def test_begin_hint_picks_the_lowest_window(): + now = datetime(2026, 10, 30, 12, 0, tzinfo=timezone.utc) + values = [200, 190, 120, 110, 130, 210, 220, 230] + series = [(now + timedelta(minutes=30 * i), v) for i, v in enumerate(values)] + + hint = begin_hint(series, duration_s=3600, now=now) + + assert hint["begin"] == "2026-10-30T13:00:00" + assert hint["mean_g_per_kwh"] == 115.0 + assert hint["now_mean_g_per_kwh"] == 195.0 + assert hint["slurm"] == "--begin=2026-10-30T13:00:00" + assert begin_hint([], 3600) is None diff --git a/test_autofit/campaign/test_report.py b/test_autofit/campaign/test_report.py new file mode 100644 index 000000000..a4aa34c97 --- /dev/null +++ b/test_autofit/campaign/test_report.py @@ -0,0 +1,132 @@ +""" +``campaign report`` and ``campaign carbon`` over a digest folder: the report's +sections, the paper paragraph's wording and the command line (offline). +""" +import json +import re + +from autofit.campaign.__main__ import main +from autofit.campaign.digest import TaskOutcome, build_digest, write_digest +from autofit.campaign.report import build_report, paper_paragraph, report_markdown + + +def _digest_folder(tmp_path): + outcomes = [ + TaskOutcome( + stage="vis_lp", + key=f"1_{i}", + state="completed", + elapsed_s=5400, + cpus=8, + req_mem_mb=12288.0, + max_rss_mb=4000.0, + total_cpu_s=0.53 * 5400 * 8, + partition="ral", + cpu_s_all_attempts=5400 * 8, + start="2026-09-11T10:00:00", + end="2026-09-11T11:30:00", + ) + for i in range(30) + ] + [ + TaskOutcome( + stage="sed", + key="2_0", + state="timeout", + elapsed_s=10800, + cpus=8, + req_mem_mb=16384.0, + max_rss_mb=3000.0, + total_cpu_s=0.3 * 10800 * 8, + partition="ral", + cpu_s_all_attempts=10800 * 8, + start="2026-09-12T10:00:00", + end="2026-09-12T13:00:00", + ) + ] + digest = build_digest("demo", outcomes) + digest["period"] = {"first_submit": "2026-09-11T10:00:00", "last_end": "2026-09-12T13:00:00"} + folder = tmp_path / "digest" + write_digest(folder, digest, outcomes) + return folder + + +def test_report_sections_and_paragraph(tmp_path, ral): + folder = _digest_folder(tmp_path) + index = tmp_path / "index" + index.mkdir() + (index / "index.json").write_text( + json.dumps({"updated": "2026-10-30", "counts": {"completed": 29, "corrupt_zip": 1}}) + ) + + report = build_report(folder, site=ral, objects=30, results_index=index) + text = report_markdown(report) + + for heading in ( + "## Compute", + "## Energy and carbon (estimate)", + "### Inputs", + "## Failure census", + "## Stages", + "## Results", + "## For the paper", + ): + assert heading in text + assert "corrupt_zip 1" in text + assert "Per object (30 objects)" in text + + paragraph = report["paragraph"] + assert "estimate" in paragraph + assert "Green Algorithms" in paragraph + assert "regional basis" in paragraph + assert re.search(r"range [\d.,]+–[\d.,]+ (k|M)Wh", paragraph) + assert "failed, timed-out or repeated" in paragraph + assert "saving" not in text.lower() + + +def test_paragraph_without_carbon_intensity(tmp_path): + folder = _digest_folder(tmp_path) + from autofit.campaign.carbon import CarbonInputs + + report = build_report(folder, inputs=CarbonInputs.from_site(None, pue=1.3, tdp_w_per_cpu=3.5)) + assert "gCO2e" not in report["paragraph"] + assert "energy only" in report_markdown(report) + + no_inputs = build_report(folder) + assert no_inputs["paragraph"] is None + assert "Not estimated" in report_markdown(no_inputs) + assert paper_paragraph(no_inputs["digest"], no_inputs["carbon"]) is None + + +def test_cli_report_and_carbon(tmp_path, capsys): + folder = _digest_folder(tmp_path) + out = tmp_path / "out" + + assert main(["report", str(folder), "--site", "ral", "--output", str(out)]) == 0 + assert (out / "report.md").exists() and (out / "report.json").exists() + assert "## For the paper" in capsys.readouterr().out + + assert main(["carbon", str(folder), "--site", "ral", "--ci-basis", "national"]) == 0 + result = json.loads(capsys.readouterr().out) + assert result["inputs"]["ci_g_per_kwh"][1] == 129 + + assert main(["carbon", str(folder), "--pue", "1.2,1.3,1.4", "--tdp", "3.5"]) == 0 + result = json.loads(capsys.readouterr().out) + assert result["total"]["kg_co2e"] is None + + assert main(["carbon", str(folder)]) == 1 + capsys.readouterr() + + +def test_cli_report_from_campaign_file(campaign_file, capsys): + from autofit.campaign.spec import Campaign + + campaign = Campaign.load(campaign_file) + outcomes = [ + TaskOutcome(stage="vis_lp", key="1_0", state="completed", elapsed_s=3600, cpus=4, + req_mem_mb=12288.0, total_cpu_s=7200.0, partition="ral", cpu_s_all_attempts=14400.0) + ] + write_digest(campaign.directory, build_digest("demo", outcomes), outcomes) + + assert main(["report", str(campaign_file)]) == 0 + assert "regional basis" in capsys.readouterr().out + assert (campaign.directory / "report.md").exists() From 8e7a1f5b4d2d8972505b9be7195d3cf83af4edb2 Mon Sep 17 00:00:00 2001 From: Jammy2211 Date: Fri, 30 Oct 2026 15:04:49 +0000 Subject: [PATCH 2/2] fix(campaign): carbon/report review findings (hpc-campaign p4) - A site whose ci_basis basis has no value (or an all-null carbon_intensity table) gave a KeyError / ValueError; it now gives energy only, with the reason in the note. - Tasks pending, running or timeout_pending get their own 'in progress' bucket: excluded from the failed-or-repeated share (now of finished work only) and named in the paragraph as unfinished, never as wasted. - A report with no finished energy (share None) no longer crashes. - --ci given with a site drops the site's bases: the basis is 'given' and the paragraph names no bases the range no longer comes from. - carbon / report with no source is a usage error, not a TypeError. Refs #1746 Co-Authored-By: Claude Opus 5.5 --- autofit/campaign/__main__.py | 6 ++ autofit/campaign/carbon.py | 99 +++++++++++++++++++--------- autofit/campaign/report.py | 44 ++++++++----- test_autofit/campaign/test_carbon.py | 66 +++++++++++++++++++ test_autofit/campaign/test_report.py | 46 +++++++++++++ 5 files changed, 215 insertions(+), 46 deletions(-) diff --git a/autofit/campaign/__main__.py b/autofit/campaign/__main__.py index f93ac7492..c6429f0b2 100644 --- a/autofit/campaign/__main__.py +++ b/autofit/campaign/__main__.py @@ -400,6 +400,12 @@ def main(argv=None) -> int: report.set_defaults(handler=_report) arguments = parser.parse_args(argv) + if ( + arguments.handler in (_carbon, _report) + and arguments.source is None + and getattr(arguments, "begin_hint", None) is None + ): + parser.error(f"{arguments.command}: give a digest folder or campaign.yaml") return arguments.handler(arguments) diff --git a/autofit/campaign/carbon.py b/autofit/campaign/carbon.py index 0e0fb30d2..b20bfaea4 100644 --- a/autofit/campaign/carbon.py +++ b/autofit/campaign/carbon.py @@ -119,6 +119,8 @@ class CarbonInputs: ci_bases: Dict[str, float] = field(default_factory=dict) """The central carbon intensity of every basis the site gives.""" ci_source: Optional[str] = None + ci_missing_reason: Optional[str] = None + """Why there is no carbon intensity, when the site has a table but no usable value.""" unverified: List[str] = field(default_factory=list) site: Optional[str] = None @@ -148,20 +150,26 @@ def from_site( inputs.pue = Bounds.of(site.pue) table = getattr(site, "carbon_intensity", {}) or {} basis = ci_basis or site.ci_basis - if table and basis in table: + bases = { + name: float(value.central) + for name, value in table.items() + if value.central is not None + } + if basis in bases: inputs.ci_basis = basis - inputs.ci_bases = { - name: float(value.central) - for name, value in table.items() - if value.central is not None - } - centrals = list(inputs.ci_bases.values()) - inputs.ci_g_per_kwh = Bounds( - min(centrals), inputs.ci_bases[basis], max(centrals) - ) + inputs.ci_bases = bases + centrals = list(bases.values()) + inputs.ci_g_per_kwh = Bounds(min(centrals), bases[basis], max(centrals)) inputs.ci_source = ( f"site {site.name}: {basis} basis, bounds the spread of bases " - f"{', '.join(f'{k} {v:g}' for k, v in inputs.ci_bases.items())}" + f"{', '.join(f'{k} {v:g}' for k, v in bases.items())}" + ) + elif table: + # The chosen basis has no value: no carbon figure rather than one from + # a basis nobody chose. + inputs.ci_missing_reason = ( + f"site {site.name} gives no value for the {basis!r} " + f"carbon_intensity basis" ) inputs.unverified = list(site.unverified()) if pue is not None: @@ -172,8 +180,10 @@ def from_site( inputs.tdp_w_per_cpu["*"] = bounds if ci is not None: inputs.ci_g_per_kwh = Bounds.of(ci) - inputs.ci_basis = ci_basis or "given" - inputs.ci_source = "given on the command line" + inputs.ci_basis = "given" + inputs.ci_bases = {} + inputs.ci_source = "given explicitly" + inputs.ci_missing_reason = None return inputs def tdp_for(self, partition: Optional[str]) -> Tuple[Optional[Bounds], bool]: @@ -220,6 +230,7 @@ def to_dict(self) -> Dict: "ci_basis": self.ci_basis, "ci_bases": self.ci_bases, "ci_source": self.ci_source, + "ci_missing_reason": self.ci_missing_reason, "unverified": self.unverified, "missing": self.missing(), } @@ -270,18 +281,26 @@ def _empty() -> Dict[str, float]: return {case: 0.0 for case in CASES} +IN_PROGRESS_STATES = ("pending", "running", str(RunState.TIMEOUT_PENDING)) +"""Tasks not finished yet: their energy so far is neither successful nor wasted.""" + +BUCKETS = ("successful", "in_progress", "failed_or_repeated") + + @dataclass class _Tally: kwh: Dict[str, float] = field(default_factory=_empty) - successful_kwh: Dict[str, float] = field(default_factory=_empty) + buckets: Dict[str, Dict[str, float]] = field( + default_factory=lambda: {bucket: _empty() for bucket in BUCKETS} + ) tasks: int = 0 completed: int = 0 + in_progress: int = 0 - def add(self, energy: Dict[str, float], successful: bool): + def add(self, energy: Dict[str, float], bucket: str): for case in CASES: self.kwh[case] += energy[case] - if successful: - self.successful_kwh[case] += energy[case] + self.buckets[bucket][case] += energy[case] def _usage(outcome: TaskOutcome) -> Optional[float]: @@ -325,17 +344,22 @@ def _round_cases(values: Optional[Dict[str, float]], digits=6): def _summary(tally: _Tally, ci: Optional[Bounds]) -> Dict: - failed = {case: tally.kwh[case] - tally.successful_kwh[case] for case in CASES} - central = tally.kwh["central"] + successful = tally.buckets["successful"] + failed = tally.buckets["failed_or_repeated"] + # The share of *finished* work that was failed or repeated: energy of tasks still + # in progress is neither, so it is left out of both sides. + finished = successful["central"] + failed["central"] summary = { "tasks": tally.tasks, "completed": tally.completed, + "in_progress": tally.in_progress, "kwh": _round_cases(tally.kwh), "kg_co2e": _round_cases(_carbon(tally.kwh, ci)), - "successful_kwh": _round_cases(tally.successful_kwh), + "successful_kwh": _round_cases(successful), + "in_progress_kwh": _round_cases(tally.buckets["in_progress"]), "failed_or_repeated_kwh": _round_cases(failed), - "failed_or_repeated_share": round(failed["central"] / central, 3) - if central + "failed_or_repeated_share": round(failed["central"] / finished, 3) + if finished > 0 else None, } if tally.completed: @@ -360,8 +384,10 @@ def estimate( """ The energy and carbon of a campaign's tasks (live or retrospective), per stage and in total, in the four cases (module docstring), split into successful work (the - final attempt of each completed task) and failed or repeated work (every other - attempt and every task that did not complete). + final attempt of each completed task), work in progress (the latest attempt of a + pending, running or ``timeout_pending`` task, so far) and failed or repeated work + (every earlier attempt and every finished task that did not complete). + ``failed_or_repeated_share`` is of finished work only. Earlier attempts of a task are known only by their allocated CPU time; they are charged with the final attempt's CPUs, memory and usage. A task with no CPU time @@ -403,9 +429,13 @@ def estimate( tally.tasks += 1 total.tasks += 1 completed = outcome.state == str(RunState.COMPLETED) + in_progress = outcome.state in IN_PROGRESS_STATES if completed: tally.completed += 1 total.completed += 1 + if in_progress: + tally.in_progress += 1 + total.in_progress += 1 cpus = outcome.cpus or 0 if not cpus or outcome.elapsed_s is None: counts["no_time"] += 1 @@ -430,15 +460,21 @@ def estimate( # The final attempt, then any earlier ones (known by allocated CPU time only). earlier_cpu_s = (outcome.cpu_s_all_attempts or 0.0) - outcome.elapsed_s * cpus - attempts = [(outcome.elapsed_s / 3600.0, completed and outcome.success is not False)] + if completed and outcome.success is not False: + final = "successful" + elif in_progress: + final = "in_progress" + else: + final = "failed_or_repeated" + attempts = [(outcome.elapsed_s / 3600.0, final)] if earlier_cpu_s > 1e-6: - attempts.append((earlier_cpu_s / cpus / 3600.0, False)) - for hours, successful in attempts: + attempts.append((earlier_cpu_s / cpus / 3600.0, "failed_or_repeated")) + for hours, bucket in attempts: energy = _attempt_energy( hours, cpus, usage, requested_gb, peak_gb, gpus, tdp, gpu, inputs.pue ) - tally.add(energy, successful) - total.add(energy, successful) + tally.add(energy, bucket) + total.add(energy, bucket) result["total"] = _summary(total, ci) result["stages"] = {name: _summary(t, ci) for name, t in sorted(stages.items())} @@ -454,9 +490,10 @@ def estimate( } result["assumptions"] = counts if ci is None: + reason = inputs.ci_missing_reason or "the site gives no carbon_intensity" result["note"] = ( - "energy only: the site gives no carbon_intensity, so no carbon figure " - "(pass --ci; no world-average default is used)" + f"energy only: {reason}, so no carbon figure " + f"(pass --ci; no world-average default is used)" ) return result diff --git a/autofit/campaign/report.py b/autofit/campaign/report.py index c6178a085..d753fb9ce 100644 --- a/autofit/campaign/report.py +++ b/autofit/campaign/report.py @@ -103,15 +103,17 @@ def paper_paragraph(digest: Dict, carbon: Dict) -> Optional[str]: ) if kg is not None: ci = inputs["ci_g_per_kwh"] - bases = ", ".join( - f"{k} {v:g}" for k, v in inputs["ci_bases"].items() if k != inputs["ci_basis"] - ) - text += ( - f" and a grid carbon intensity of {ci[1]:g} gCO2e/kWh ({inputs['ci_basis']} " - f"basis; range {ci[0]:g}–{ci[2]:g} across bases" - + (f": {bases}" if bases else "") - + ")" - ) + others = { + k: v for k, v in (inputs.get("ci_bases") or {}).items() if k != inputs["ci_basis"] + } + if others: + detail = ( + f"{inputs['ci_basis']} basis; range {ci[0]:g}–{ci[2]:g} across bases: " + + ", ".join(f"{k} {v:g}" for k, v in others.items()) + ) + else: + detail = f"{inputs['ci_basis']}; range {ci[0]:g}–{ci[2]:g}" + text += f" and a grid carbon intensity of {ci[1]:g} gCO2e/kWh ({detail})" energy = _scaled(kwh, "kWh", "MWh") text += ( f", we estimate the campaign used ≈{energy['central']} {energy['unit']} " @@ -132,9 +134,16 @@ def paper_paragraph(digest: Dict, carbon: Dict) -> Optional[str]: text += "." if share is not None: text += ( - f" About {share:.0%} of the central estimate was spent on failed, " + f" About {share:.0%} of the energy of finished work was spent on failed, " f"timed-out or repeated runs." ) + in_progress = (carbon["total"].get("in_progress_kwh") or {}).get("central") or 0 + if in_progress > 0: + text += ( + f" The campaign was still running: {carbon['total']['in_progress']:,} " + f"unfinished tasks account for ≈{_significant(in_progress)} kWh so far, " + f"included in the totals but in neither share." + ) text += ( " The upper bound charges every reserved CPU at full use and the requested " "memory (allocation basis) with the high inputs. The estimate excludes embodied " @@ -201,10 +210,8 @@ def report_markdown(report: Dict) -> str: if period: lines.append(f"- Period: {period.get('first_submit')} → {period.get('last_end')}") cpu_all, cpu_ok = totals["cpu_h_all_attempts"], totals["cpu_h_successful"] - share = 1 - cpu_ok / cpu_all if cpu_all else 0 lines.append( - f"- CPU-hours allocated: {cpu_all:,.0f}, of which {cpu_ok:,.0f} in completed " - f"tasks ({share:.0%} failed, timed-out or repeated)" + f"- CPU-hours allocated: {cpu_all:,.0f}, of which {cpu_ok:,.0f} in completed tasks" ) peak = digest.get("peak_cpus") if isinstance(peak, dict) and peak.get("cpus"): @@ -228,11 +235,18 @@ def report_markdown(report: Dict) -> str: for case, meaning in carbon["cases"].items(): lines.append(f"- **{case}**: {meaning}") total = carbon["total"] + share = total["failed_or_repeated_share"] lines.append( f"- Failed or repeated work: {_number(total['failed_or_repeated_kwh']['central'])} " - f"kWh of {_number(total['kwh']['central'])} central " - f"({total['failed_or_repeated_share']:.0%})" + f"kWh of {_number(total['kwh']['central'])} central" + + (f" ({share:.0%} of finished work)" if share is not None else "") ) + in_progress = (total.get("in_progress_kwh") or {}).get("central") or 0 + if in_progress > 0: + lines.append( + f"- In progress: {total['in_progress']:,} unfinished tasks, " + f"{_number(in_progress)} kWh so far (in neither share)" + ) rows = ["", "| stage | per completed task, central | failed or repeated share |", "|---|---|---|"] for name, stage in carbon["stages"].items(): per = stage.get("per_completed_task") diff --git a/test_autofit/campaign/test_carbon.py b/test_autofit/campaign/test_carbon.py index a6ed19981..557278d8b 100644 --- a/test_autofit/campaign/test_carbon.py +++ b/test_autofit/campaign/test_carbon.py @@ -245,3 +245,69 @@ def test_begin_hint_picks_the_lowest_window(): assert hint["now_mean_g_per_kwh"] == 195.0 assert hint["slurm"] == "--begin=2026-10-30T13:00:00" assert begin_hint([], 3600) is None + + +def _site_with_ci(table, basis="regional"): + from autonerves.site_config import Site + + return Site( + { + "name": "toy", + "partitions": {"ral": {"gpu_only": False, "tdp_w_per_cpu": {"low": 2.8, "central": 3.5, "high": 4.4}}}, + "pue": {"low": 1.2, "central": 1.31, "high": 1.45}, + "ci_basis": basis, + "carbon_intensity": table, + } + ) + + +def test_empty_ci_basis_gives_energy_only_never_a_crash(): + # Review finding 1: a null chosen basis (or an all-null table) used to raise. + for table in ( + {"regional": {"value": None, "unverified": True}, "national": 129}, + {"regional": {"value": None}, "national": {"value": None}}, + ): + inputs = CarbonInputs.from_site(_site_with_ci(table)) + assert inputs.ci_g_per_kwh is None + assert "'regional'" in inputs.ci_missing_reason + result = estimate([_outcome()], inputs) + assert result["total"]["kg_co2e"] is None + assert "no value for the 'regional'" in result["note"] + + national = CarbonInputs.from_site( + _site_with_ci({"regional": {"value": None}, "national": 129}), ci_basis="national" + ) + assert national.ci_g_per_kwh.as_list() == [129, 129, 129] + + +def test_in_progress_work_is_its_own_bucket(): + # Review finding 3: running / pending tasks are not failed work. + inputs = CarbonInputs.from_site(None, **REPORT_INPUTS) + done = _outcome(key="a") + running = _outcome(key="b", state="running") + flushing = _outcome(key="c", state="timeout_pending") + result = estimate([done, running, flushing], inputs) + total = result["total"] + one = estimate([done], inputs)["total"]["kwh"]["central"] + + assert total["in_progress"] == 2 + assert total["in_progress_kwh"]["central"] == pytest.approx(2 * one, abs=1e-6) + assert total["failed_or_repeated_kwh"]["central"] == 0 + assert total["failed_or_repeated_share"] == 0.0 + + only_running = estimate([running], inputs)["total"] + assert only_running["failed_or_repeated_share"] is None + + # An earlier attempt of a running task is still repeated work. + retried = _outcome(key="d", state="running", attempts=2, cpu_s_all_attempts=2 * 3600 * 8) + total = estimate([done, retried], inputs)["total"] + assert total["failed_or_repeated_kwh"]["central"] == pytest.approx(one, abs=1e-6) + assert total["failed_or_repeated_share"] == pytest.approx(0.5) + + +def test_given_ci_replaces_the_site_bases(ral): + # Review finding 4: --ci with a site must not name bases the range no longer uses. + inputs = CarbonInputs.from_site(ral, ci=(100, 130, 190)) + assert inputs.ci_basis == "given" + assert inputs.ci_bases == {} + assert inputs.ci_g_per_kwh.as_list() == [100, 130, 190] diff --git a/test_autofit/campaign/test_report.py b/test_autofit/campaign/test_report.py index a4aa34c97..f5f3a88d8 100644 --- a/test_autofit/campaign/test_report.py +++ b/test_autofit/campaign/test_report.py @@ -130,3 +130,49 @@ def test_cli_report_from_campaign_file(campaign_file, capsys): assert main(["report", str(campaign_file)]) == 0 assert "regional basis" in capsys.readouterr().out assert (campaign.directory / "report.md").exists() + + +def _live_folder(tmp_path, outcomes): + folder = tmp_path / "live" + write_digest(folder, build_digest("live", outcomes), outcomes) + return folder + + +def test_report_with_only_unstarted_or_running_tasks(tmp_path, ral): + # Review finding 2: zero finished energy must not crash the report. + pending = TaskOutcome(stage="vis_lp", key="1_0", state="pending", cpus=8, partition="ral") + report = build_report(_live_folder(tmp_path, [pending]), site=ral) + text = report_markdown(report) + assert "Failed or repeated work: 0 kWh of 0 central" in text + assert "of finished work" not in text + + running = TaskOutcome( + stage="vis_lp", key="1_1", state="running", elapsed_s=1800, cpus=8, + req_mem_mb=12288.0, total_cpu_s=7200.0, partition="ral", cpu_s_all_attempts=14400.0, + ) + report = build_report(_live_folder(tmp_path / "b", [running]), site=ral) + text = report_markdown(report) + assert "In progress: 1 unfinished tasks" in text + # Review finding 3: never claims running work was wasted. + assert "was spent on failed" not in report["paragraph"] + assert "still running" in report["paragraph"] + + +def test_paragraph_with_given_ci_names_no_site_bases(tmp_path, ral): + from autofit.campaign.carbon import CarbonInputs + + folder = _digest_folder(tmp_path) + inputs = CarbonInputs.from_site(ral, ci=(100, 130, 190)) + paragraph = build_report(folder, site=ral, inputs=inputs)["paragraph"] + assert "130 gCO2e/kWh (given; range 100–190)" in paragraph + assert "national" not in paragraph and "regional" not in paragraph + + +def test_cli_without_source_is_a_usage_error(capsys): + import pytest + + for command in ("report", "carbon"): + with pytest.raises(SystemExit) as exit_info: + main([command]) + assert exit_info.value.code == 2 + assert "give a digest folder or campaign.yaml" in capsys.readouterr().err