#!/usr/bin/env python3 """ AI Energy Impact Analyzer Calculates energy consumption and carbon footprint for AI model usage, implementing an "energy tax" system for sustainable AI development practices. Methodology (v2, 2026 refresh — see ai/README.md for full citations): - Per-token energy constants are calibrated against published full-stack inference measurements: Epoch AI's ~0.3 Wh per GPT-4o-class query (https://epoch.ai/gradient-updates/how-much-energy-does-chatgpt-use), Google's Gemini production measurement of 0.24 Wh median text prompt (https://arxiv.org/abs/2508.15734), the ML.Energy leaderboard (https://ml.energy/leaderboard/), and the cross-model benchmark "How Hungry is AI?" (https://arxiv.org/abs/2505.09598). - Cloud inference figures include host CPU/DRAM and data-center PUE overhead, which typically add 30-70% on top of GPU-only measurements. - Prompt-cache reads skip prefill compute; the 0.10x energy multiplier mirrors Anthropic's 0.1x cache-read pricing signal. Cache writes carry a 1.25x overhead, mirroring the 1.25x cache-write pricing signal. - When only dollar costs are available (claude/claude_api_cost_*.csv exports), token counts are reverse-estimated from Anthropic's published per-model pricing (https://platform.claude.com/docs/en/about-claude/pricing) before any energy math. Reports label this estimation explicitly. Anthropic publishes no per-model energy figures, so all Wh constants are research-calibrated estimates, not vendor measurements. Treat totals as order-of-magnitude indicators, not billing-grade numbers. Author: Justin Murry Organization: unparty LLC Version: v1.0.1 License: MIT LLM: Claude Code """ import argparse import csv import importlib.util import json import os import re import sys import time from datetime import datetime from pathlib import Path # ============================================================================= # UNPARTY BRAND STYLING # ============================================================================= # Reimplemented here (rather than sourced from unparty_styles.sh, a # bash-only library) so this module works standalone from a release # download with no extra file to bundle. ai_energy_cli.py imports these # same helpers rather than duplicating them, since it already depends on # this module for AIEnergyImpactAnalyzer. Palette/wordmark/tagline mirror # unparty_styles.sh's BRAND_COLOR, show_unparty_logo(), and # show_unparty_execution() — see that file for the canonical bash version. BRAND_COLOR = '\033[38;2;249;194;46m' # #f9c22e — official unparty yellow _CYAN = '\033[0;36m' _NC = '\033[0m' _UNPARTY_LOGO = ( "██╗ ██╗███╗ ██╗██████╗ █████╗ ██████╗ ████████╗██╗ ██╗", "██║ ██║████╗ ██║██╔══██╗██╔══██╗██╔══██╗╚══██╔══╝╚██╗ ██╔╝", "██║ ██║██╔██╗ ██║██████╔╝███████║██████╔╝ ██║ ╚████╔╝ ", "██║ ██║██║╚██╗██║██╔═══╝ ██╔══██║██╔══██╗ ██║ ╚██╔╝ ", "╚██████╔╝██║ ╚████║██║ ██║ ██║██║ ██║ ██║ ██║ ", " ╚═════╝ ╚═╝ ╚═══╝╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝ ╚═╝ ╚═╝ ", ) _UNPARTY_TAGLINE = "WHERE INNOVATION MEETS EXECUTION" def _brand_enabled() -> bool: """Only animate/colorize for a real interactive terminal. Piped output, CI, and this repo's core-directory performance monitor (which invokes this script non-interactively — see performance/performance_monitor.py) all get plain text instead: animation there would just be sleep-cost and log noise, not a feature.""" if os.environ.get('NO_COLOR') or os.environ.get('CI') or os.environ.get('UNPARTY_NO_BRAND'): return False return sys.stdout.isatty() def show_unparty_banner() -> None: """Print the unparty wordmark — animated line-by-line on a real terminal, a single plain-text tag otherwise.""" if not _brand_enabled(): print(f"[unparty] {_UNPARTY_TAGLINE}") return print(BRAND_COLOR, end='') for line in _UNPARTY_LOGO: print(line) time.sleep(0.03) print(_NC, end='') print(f"{_CYAN}{_UNPARTY_TAGLINE}{_NC}\n") def brand(text: str) -> str: """Wrap text in the brand color when attached to a real terminal.""" return f"{BRAND_COLOR}{text}{_NC}" if _brand_enabled() else text def brand_divider(width: int = 62) -> str: """Brand-colored divider line, matching unparty_styles.sh's unparty_divider().""" return brand('=' * width) # Energy consumption constants (Wh per 1M tokens, full-stack cloud inference # including host overhead and PUE). Central estimates per model tier: # small (Haiku-class) ~150 Wh/1M output tokens (range 50-300) # medium (Sonnet-class) ~600 Wh/1M output tokens (range 300-1,000) # large (Opus/Fable) ~2,000 Wh/1M output tokens (range 1,000-5,500) # Input (prefill) tokens are compute-parallel and cost roughly 10% of decode # energy per token. Newer generations get modestly lower figures, reflecting # newer accelerators and efficiency gains (Google reported a 33x median-prompt # reduction in 12 months; per-token frontier costs fall far more slowly). MODEL_ENERGY_PROFILES = { 'claude-haiku-3-5': { 'input_wh_per_1m_tokens': 15.0, 'output_wh_per_1m_tokens': 150.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-haiku-4-5': { 'input_wh_per_1m_tokens': 12.0, 'output_wh_per_1m_tokens': 120.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-sonnet-3-5': { 'input_wh_per_1m_tokens': 60.0, 'output_wh_per_1m_tokens': 600.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-sonnet-3-7': { 'input_wh_per_1m_tokens': 60.0, 'output_wh_per_1m_tokens': 600.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-sonnet-4': { 'input_wh_per_1m_tokens': 50.0, 'output_wh_per_1m_tokens': 500.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-sonnet-4-5': { 'input_wh_per_1m_tokens': 45.0, 'output_wh_per_1m_tokens': 450.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-sonnet-5': { 'input_wh_per_1m_tokens': 40.0, 'output_wh_per_1m_tokens': 400.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-opus-3': { 'input_wh_per_1m_tokens': 250.0, 'output_wh_per_1m_tokens': 2500.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-opus-4': { 'input_wh_per_1m_tokens': 220.0, 'output_wh_per_1m_tokens': 2200.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-opus-4-1': { 'input_wh_per_1m_tokens': 220.0, 'output_wh_per_1m_tokens': 2200.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-opus-4-5': { 'input_wh_per_1m_tokens': 180.0, 'output_wh_per_1m_tokens': 1800.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-opus-5': { 'input_wh_per_1m_tokens': 160.0, 'output_wh_per_1m_tokens': 1600.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-fable-5': { 'input_wh_per_1m_tokens': 200.0, 'output_wh_per_1m_tokens': 2000.0, 'cache_read_multiplier': 0.10, 'cache_write_multiplier': 1.25, }, 'claude-fable-5-1': { 'input_wh_per_1m_tokens': 200.0, 'output_wh_per_1m_tokens': 2000.0, 'cache_read_multiplier': 0.025, # Fable 5.1 cache reads cost 0.025x input 'cache_write_multiplier': 1.25, }, } # Anthropic API pricing (USD per 1M tokens), September 2026. # Source: https://platform.claude.com/docs/en/about-claude/pricing # Standard multipliers: 5-minute cache write = 1.25x input, cache read = 0.1x # input (Fable 5.1 cache read is a special 0.025x rate). Used to reverse- # estimate token counts when only dollar costs are available. MODEL_PRICING_USD_PER_1M_TOKENS = { 'claude-haiku-3-5': { 'input_no_cache': 0.80, 'input_cache_write': 1.00, 'input_cache_read': 0.08, 'output': 4.00, }, 'claude-haiku-4-5': { 'input_no_cache': 1.00, 'input_cache_write': 1.25, 'input_cache_read': 0.10, 'output': 5.00, }, 'claude-sonnet-3-5': { 'input_no_cache': 3.00, 'input_cache_write': 3.75, 'input_cache_read': 0.30, 'output': 15.00, }, 'claude-sonnet-3-7': { 'input_no_cache': 3.00, 'input_cache_write': 3.75, 'input_cache_read': 0.30, 'output': 15.00, }, 'claude-sonnet-4': { 'input_no_cache': 3.00, 'input_cache_write': 3.75, 'input_cache_read': 0.30, 'output': 15.00, }, 'claude-sonnet-4-5': { 'input_no_cache': 3.00, 'input_cache_write': 3.75, 'input_cache_read': 0.30, 'output': 15.00, }, 'claude-sonnet-5': { 'input_no_cache': 2.00, 'input_cache_write': 2.50, 'input_cache_read': 0.20, 'output': 10.00, }, 'claude-opus-3': { 'input_no_cache': 15.00, 'input_cache_write': 18.75, 'input_cache_read': 1.50, 'output': 75.00, }, 'claude-opus-4': { 'input_no_cache': 15.00, 'input_cache_write': 18.75, 'input_cache_read': 1.50, 'output': 75.00, }, 'claude-opus-4-1': { 'input_no_cache': 15.00, 'input_cache_write': 18.75, 'input_cache_read': 1.50, 'output': 75.00, }, 'claude-opus-4-5': { 'input_no_cache': 5.00, 'input_cache_write': 6.25, 'input_cache_read': 0.50, 'output': 25.00, }, 'claude-opus-5': { 'input_no_cache': 5.00, 'input_cache_write': 6.25, 'input_cache_read': 0.50, 'output': 25.00, }, 'claude-fable-5': { 'input_no_cache': 10.00, 'input_cache_write': 12.50, 'input_cache_read': 1.00, 'output': 50.00, }, 'claude-fable-5-1': { 'input_no_cache': 10.00, 'input_cache_write': 12.50, 'input_cache_read': 0.25, 'output': 50.00, }, } # Model families recognized by the resolver. 'mythos' is Anthropic's # relaxed-safeguard Fable variant and shares Fable's profiles. MODEL_FAMILIES = ('haiku', 'sonnet', 'opus', 'fable', 'mythos') FAMILY_ALIASES = {'mythos': 'fable'} # Documented default for genuinely unrecognized model strings: the current # mid-tier model. Usage of the default is tracked and surfaced in reports # instead of being silently absorbed (this replaced a silent Sonnet guess). DEFAULT_MODEL_KEY = 'claude-sonnet-5' # Carbon intensity scenarios (gCO2e per kWh): # - us_grid_average: EPA eGRID national output emission rate, 2022 data # (latest official release): https://www.epa.gov/egrid # - cloud_market_based: implied by Google's Aug 2025 Gemini paper # (0.03 gCO2e / 0.24 Wh): https://arxiv.org/abs/2508.15734 # - us_datacenter_location_based: Virginia/us-east region estimates: # https://github.com/GoogleCloudPlatform/region-carbon-info CARBON_INTENSITY_G_PER_KWH = { 'us_grid_average': 386, 'cloud_market_based': 125, 'us_datacenter_location_based': 395, } DEFAULT_CARBON_SCENARIO = 'us_grid_average' # Electricity cost scenarios (USD per kWh): # - us_commercial_average: EIA Electric Power Monthly, 2025 average # https://www.eia.gov/electricity/monthly/epm_table_grapher.php?t=table_5_03 # - us_industrial_average / renewable_ppa: unofficial estimates ENERGY_COST_RATES_USD_PER_KWH = { 'us_commercial_average': 0.134, 'us_industrial_average': 0.08, 'renewable_ppa': 0.06, } DEFAULT_ENERGY_COST_SCENARIO = 'us_commercial_average' # Internal carbon price used for the "energy tax": $30/tonne CO2 = $0.03/kg, # a conservative figure in line with California cap-and-trade auction prices. CARBON_PRICE_USD_PER_KG = 0.03 # Source data older than this is flagged as stale in reports and by # `--check-freshness` (source CSVs are monthly exports; a month of data plus # two weeks of grace keeps reports honest without being noisy). DEFAULT_STALENESS_THRESHOLD_DAYS = 45 METHODOLOGY_VERSION = '2.0' # Any CSV dropped in claude/ is classified by its header row, not its # filename — a Console export can be renamed however it downloads. A file # qualifies for a schema if its header is a superset of these columns. # Token schema is checked first: measured tokens always win over a dollar # export that happens to also carry these columns. TOKEN_SCHEMA_REQUIRED_COLUMNS = { 'usage_date_utc', 'model_version', 'usage_input_tokens_no_cache', 'usage_output_tokens', } COST_SCHEMA_REQUIRED_COLUMNS = {'usage_date_utc', 'model', 'cost_usd'} class AIEnergyImpactAnalyzer: def __init__(self, repo_root=None, carbon_intensity_g_per_kwh=None, energy_cost_usd_per_kwh=None, staleness_max_age_days=DEFAULT_STALENESS_THRESHOLD_DAYS): self.repo_root = Path(repo_root) if repo_root else Path(__file__).resolve().parent.parent self.output_dir = self.repo_root / 'docs' self.data_dir = self.repo_root / 'data' self.output_dir.mkdir(exist_ok=True) # Energy consumption constants (Wh per 1M tokens) — see module-level # MODEL_ENERGY_PROFILES for tiers, calibration anchors and citations. self.model_energy_consumption = { key: dict(profile) for key, profile in MODEL_ENERGY_PROFILES.items() } self.model_pricing = { key: dict(rates) for key, rates in MODEL_PRICING_USD_PER_1M_TOKENS.items() } # Carbon intensity (gCO2/kWh) scenarios; override the selected value # per-region/provider via the constructor argument. self.carbon_intensity = dict(CARBON_INTENSITY_G_PER_KWH) self.carbon_intensity_g_per_kwh = ( carbon_intensity_g_per_kwh if carbon_intensity_g_per_kwh is not None else self.carbon_intensity[DEFAULT_CARBON_SCENARIO] ) # Energy cost rates ($/kWh) scenarios; override via the constructor. self.energy_cost_rates = dict(ENERGY_COST_RATES_USD_PER_KWH) self.energy_cost_usd_per_kwh = ( energy_cost_usd_per_kwh if energy_cost_usd_per_kwh is not None else self.energy_cost_rates[DEFAULT_ENERGY_COST_SCENARIO] ) # Freshness policy for ingested source data self.staleness_max_age_days = staleness_max_age_days # Data-quality tracking, surfaced in every report self.unrecognized_models = set() self.generation_fallbacks = {} self.skipped_source_rows = 0 self.token_data_source = 'unknown' # Every CSV found in claude/, classified by schema, and which of # them actually fed the report — measured token exports always win # over cost exports even when both exist for the same period, so # "found" and "used" are tracked separately and both surfaced in # every report's Data Sources & Quality section. The previous # filename-only matching also failed *silently* when a # correctly-formatted export had an unexpected name; `ignored` # makes that visible instead. self.source_files = {'token_files_found': [], 'cost_files_found': [], 'ignored_files': []} self.active_source_files = [] def _classify_csv_schema(self, fieldnames): """Classify a CSV's schema from its header row alone. Content-sniffing (instead of requiring an exact filename in an exact directory) is what lets a Console export be dropped into claude/ under whatever name it downloads with and still be found. """ columns = set(fieldnames or []) if TOKEN_SCHEMA_REQUIRED_COLUMNS <= columns: return 'token' if COST_SCHEMA_REQUIRED_COLUMNS <= columns: return 'cost' return None def _discover_claude_csvs(self): """Find every CSV that looks like a Claude usage/cost export. Scans claude/ — the single drop folder — plus the legacy data/claude_api_tokens_master.csv path documented by earlier versions of this tool, so files created under the old convention still work. Returns (token_paths, cost_paths, ignored_paths). """ candidates = sorted((self.repo_root / 'claude').glob('*.csv')) legacy_master = self.data_dir / 'claude_api_tokens_master.csv' if legacy_master.exists() and legacy_master not in candidates: candidates.append(legacy_master) token_paths, cost_paths, ignored_paths = [], [], [] for path in candidates: try: with open(path, 'r', newline='') as f: fieldnames = csv.DictReader(f).fieldnames except (OSError, csv.Error): ignored_paths.append(path) continue schema = self._classify_csv_schema(fieldnames) if schema == 'token': token_paths.append(path) elif schema == 'cost': cost_paths.append(path) else: ignored_paths.append(path) return token_paths, cost_paths, ignored_paths def load_claude_usage_data(self): """Load Claude API usage data from available sources. Every CSV in claude/ is classified by its header row and merged with every other file of the same schema — there is no single "master" filename to hand-maintain and no directory convention to memorize, so a fresh Console export just needs to be dropped in claude/ under any name. """ token_paths, cost_paths, ignored_paths = self._discover_claude_csvs() self.source_files = { 'token_files_found': [self._relpath(p) for p in token_paths], 'cost_files_found': [self._relpath(p) for p in cost_paths], 'ignored_files': [self._relpath(p) for p in ignored_paths], } # Measured token counts always win, even when cost exports also # exist for the same (or a newer) period — reverse-estimating # tokens from dollars is strictly less accurate than reading them # directly, so it's never worth mixing the two or preferring # whichever file is newer. usage_data = [] for path in token_paths: with open(path, 'r') as f: for row in csv.DictReader(f): usage_data.append(row) if usage_data: self.token_data_source = 'measured' self.active_source_files = [self._relpath(p) for p in token_paths] return usage_data # Fallback: derive token estimates from LLM baseline CSV metrics llm_usage = self._load_usage_from_llm_baseline() if llm_usage: self.token_data_source = 'estimated_from_cost (LLM baseline)' self.active_source_files = ['LLM/LLM.py (reads claude/claude_api_cost_*.csv)'] return llm_usage # Last-resort fallback when LLM baseline dependencies are unavailable cost_usage = self._load_usage_from_claude_cost_csv(cost_paths) if cost_usage: self.token_data_source = 'estimated_from_cost (claude_api_cost_*.csv)' self.active_source_files = [self._relpath(p) for p in cost_paths] return cost_usage def _relpath(self, path): """Repo-relative path for report display (never leak local usernames or absolute paths into a report that may be shared).""" try: return str(Path(path).relative_to(self.repo_root)) except ValueError: return str(path) def _normalize_model_version(self, model_name): """Map a raw model string to a canonical registry key. Handles console display names ("Claude Sonnet 4", "Claude Haiku 3.5"), API IDs old and new ("claude-3-5-sonnet-20240620", "claude-opus-4-5-20251101") and bare family names. Returns None for strings with no recognizable model family (e.g. '--' rows) so callers can track them instead of silently guessing. """ raw = (model_name or '').strip().lower() if not raw or raw == '--': return None family = next((f for f in MODEL_FAMILIES if f in raw), None) if family is None: return None family = FAMILY_ALIASES.get(family, family) generation = self._extract_generation(raw, family) if generation is None: # Family recognized but no version number: assume the newest # profile we have for that family. return self._latest_family_key(family) candidate = f"claude-{family}-{generation.replace('.', '-')}" if candidate in self.model_energy_consumption: return candidate # Known family, unknown generation (e.g. a brand-new release): fall # back to the newest profile for the family and record the assumption. fallback = self._latest_family_key(family) self.generation_fallbacks[model_name] = fallback return fallback def _extract_generation(self, raw, family): """Extract a generation like '3.5', '4', '4.5' adjacent to the family keyword, without being fooled by date suffixes such as -20250514.""" text = re.sub(r'[\s_]+', '-', raw) # Generation before the family keyword, e.g. claude-3-5-sonnet match = re.search(rf'(\d(?:[.\-]\d)?)(?!\d)[.\-]*{family}', text) if not match: # Generation after the family keyword, e.g. sonnet-4-5-20250929 match = re.search(rf'{family}[.\-]*(\d(?:[.\-]\d)?)(?!\d)', text) if not match: return None return match.group(1).replace('-', '.') def _latest_family_key(self, family): """Newest registry key for a model family (numeric generation sort).""" prefix = f'claude-{family}-' def generation_tuple(key): parts = key[len(prefix):].split('-') return tuple(int(p) for p in parts if p.isdigit()) candidates = sorted( (key for key in self.model_energy_consumption if key.startswith(prefix)), key=generation_tuple, ) return candidates[-1] if candidates else DEFAULT_MODEL_KEY def _resolve_energy_profile(self, model_name): """Resolve a model string to (registry_key, profile, recognized). Unrecognized strings use the documented DEFAULT_MODEL_KEY profile and are tracked in self.unrecognized_models so reports can disclose the guess (previously everything non-haiku/sonnet silently fell through to a Sonnet profile — including Opus, whose profile was dead code). """ key = self._normalize_model_version(model_name) if key is None: if model_name not in self.unrecognized_models: print(f"⚠️ Unrecognized model '{model_name}' — using default profile '{DEFAULT_MODEL_KEY}'") self.unrecognized_models.add(model_name) return DEFAULT_MODEL_KEY, self.model_energy_consumption[DEFAULT_MODEL_KEY], False return key, self.model_energy_consumption[key], True def _estimate_tokens_from_cost(self, cost_usd, model_version, token_type): """Reverse-estimate token counts from dollar cost using the published per-model pricing table. Returns 0 for non-positive costs or unknown token types. Unknown models use DEFAULT_MODEL_KEY pricing.""" rates = self.model_pricing.get(model_version) or self.model_pricing[DEFAULT_MODEL_KEY] rate = rates.get(token_type) if not rate or cost_usd <= 0: return 0 return int((cost_usd / rate) * 1_000_000) def _load_usage_from_llm_baseline(self): llm_file = self.repo_root / 'LLM' / 'LLM.py' if not llm_file.exists(): return [] try: spec = importlib.util.spec_from_file_location("llm_baseline_module", llm_file) if not spec or not spec.loader: return [] llm_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(llm_module) baseline = llm_module.LLMBaseline( claude_dir=str(self.repo_root / 'claude'), llm_dir=str(self.repo_root / 'LLM') ) csv_metrics = baseline.process_csv_data() except Exception: return [] usage_data = [] for metric in csv_metrics: token_type_raw = (getattr(metric, 'token_type', '') or '').lower() if 'cache_read' in token_type_raw: token_type = 'input_cache_read' field_name = 'usage_input_tokens_cache_read' elif 'cache_write' in token_type_raw: token_type = 'input_cache_write' field_name = 'usage_input_tokens_cache_write_5m' elif 'output' in token_type_raw: token_type = 'output' field_name = 'usage_output_tokens' else: token_type = 'input_no_cache' field_name = 'usage_input_tokens_no_cache' raw_model = getattr(metric, 'model', '') model_version = self._normalize_model_version(raw_model) or raw_model cost_usd = float(getattr(metric, 'cost_usd', 0) or 0) token_estimate = self._estimate_tokens_from_cost(cost_usd, model_version, token_type) date_value = getattr(metric, 'date', '') date_str = date_value.isoformat() if hasattr(date_value, 'isoformat') else str(date_value) row = { 'usage_date_utc': date_str, 'model_version': model_version, 'usage_input_tokens_no_cache': 0, 'usage_input_tokens_cache_write_5m': 0, 'usage_input_tokens_cache_read': 0, 'usage_output_tokens': 0, } row[field_name] = token_estimate usage_data.append(row) if usage_data: print(f"ℹ️ Using {len(usage_data)} LLM baseline usage records (estimated tokens)") return usage_data def _load_usage_from_claude_cost_csv(self, cost_files=None): usage_data = [] files = ( cost_files if cost_files is not None else sorted((self.repo_root / 'claude').glob('claude_api_cost_*.csv')) ) for csv_file in files: try: with open(csv_file, 'r') as f: reader = csv.DictReader(f) for row in reader: # Skip non-token charges (e.g. web_search fees exported # with model '--'); they carry no token usage and were # previously mis-estimated as Sonnet input tokens. cost_type = (row.get('cost_type', '') or '').strip().lower() raw_model = (row.get('model', '') or '').strip() if (cost_type and cost_type != 'token') or raw_model in ('', '--'): self.skipped_source_rows += 1 continue token_type_raw = (row.get('token_type', '') or '').lower() if 'cache_read' in token_type_raw: token_type = 'input_cache_read' field_name = 'usage_input_tokens_cache_read' elif 'cache_write' in token_type_raw: token_type = 'input_cache_write' field_name = 'usage_input_tokens_cache_write_5m' elif 'output' in token_type_raw: token_type = 'output' field_name = 'usage_output_tokens' else: token_type = 'input_no_cache' field_name = 'usage_input_tokens_no_cache' model_version = self._normalize_model_version(raw_model) or raw_model cost_usd = float(row.get('cost_usd', 0) or 0) token_estimate = self._estimate_tokens_from_cost(cost_usd, model_version, token_type) usage_row = { 'usage_date_utc': row.get('usage_date_utc', ''), 'model_version': model_version, 'usage_input_tokens_no_cache': 0, 'usage_input_tokens_cache_write_5m': 0, 'usage_input_tokens_cache_read': 0, 'usage_output_tokens': 0, } usage_row[field_name] = token_estimate usage_data.append(usage_row) except Exception: continue if usage_data: print(f"ℹ️ Using {len(usage_data)} Claude cost CSV fallback records (estimated tokens)") return usage_data def calculate_energy_consumption(self, usage_row): """Calculate energy consumption for a single usage record""" model = usage_row.get('model_version', '') # Resolve the model to an energy profile. This has an explicit branch # for every registered family (including Opus, which previously fell # through to the Sonnet profile) and tracks unrecognized models # instead of silently defaulting. profile_key, energy_profile, recognized = self._resolve_energy_profile(model) # Extract token counts (convert to millions) input_no_cache = self._to_float(usage_row.get('usage_input_tokens_no_cache', 0)) / 1_000_000 input_cache_write = self._to_float(usage_row.get('usage_input_tokens_cache_write_5m', 0)) / 1_000_000 input_cache_read = self._to_float(usage_row.get('usage_input_tokens_cache_read', 0)) / 1_000_000 output_tokens = self._to_float(usage_row.get('usage_output_tokens', 0)) / 1_000_000 # Calculate energy consumption (Wh) energy_consumption = { 'input_no_cache_wh': input_no_cache * energy_profile['input_wh_per_1m_tokens'], 'input_cache_write_wh': (input_cache_write * energy_profile['input_wh_per_1m_tokens'] * energy_profile['cache_write_multiplier']), 'input_cache_read_wh': (input_cache_read * energy_profile['input_wh_per_1m_tokens'] * energy_profile['cache_read_multiplier']), 'output_wh': output_tokens * energy_profile['output_wh_per_1m_tokens'] } total_energy_wh = sum(energy_consumption.values()) total_energy_kwh = total_energy_wh / 1000 # Calculate carbon footprint (gCO2) carbon_footprint_g = total_energy_kwh * self.carbon_intensity_g_per_kwh carbon_footprint_kg = carbon_footprint_g / 1000 # Calculate energy cost energy_cost_usd = total_energy_kwh * self.energy_cost_usd_per_kwh return { 'date': usage_row.get('usage_date_utc', ''), 'model': model, 'energy_profile': profile_key, 'profile_recognized': recognized, 'total_energy_wh': round(total_energy_wh, 4), 'total_energy_kwh': round(total_energy_kwh, 6), 'carbon_footprint_g': round(carbon_footprint_g, 2), 'carbon_footprint_kg': round(carbon_footprint_kg, 4), 'energy_cost_usd': round(energy_cost_usd, 4), 'breakdown': energy_consumption, 'cache_efficiency': self._calculate_cache_efficiency(usage_row) } @staticmethod def _to_float(value): """Parse numeric CSV fields defensively ('' and None become 0.0).""" try: return float(value) except (TypeError, ValueError): return 0.0 def _calculate_cache_efficiency(self, usage_row): """Calculate cache efficiency for energy optimization insights""" total_input = (self._to_float(usage_row.get('usage_input_tokens_no_cache', 0)) + self._to_float(usage_row.get('usage_input_tokens_cache_write_5m', 0)) + self._to_float(usage_row.get('usage_input_tokens_cache_read', 0))) if total_input == 0: return 0 cache_read = self._to_float(usage_row.get('usage_input_tokens_cache_read', 0)) cache_efficiency = (cache_read / total_input) * 100 return round(cache_efficiency, 2) def calculate_energy_tax(self, energy_results): """Calculate energy tax based on consumption and efficiency""" total_carbon_kg = sum(result['carbon_footprint_kg'] for result in energy_results) total_energy_kwh = sum(result['total_energy_kwh'] for result in energy_results) # Energy tax calculation (carbon pricing + efficiency penalty) carbon_price_per_kg = CARBON_PRICE_USD_PER_KG # $30/tonne CO2 = $0.03/kg base_energy_tax = total_carbon_kg * carbon_price_per_kg # Calculate average cache efficiency cache_efficiencies = [r['cache_efficiency'] for r in energy_results if r['cache_efficiency'] > 0] avg_cache_efficiency = sum(cache_efficiencies) / len(cache_efficiencies) if cache_efficiencies else 0 # Efficiency bonus/penalty (encourage cache usage) if avg_cache_efficiency > 60: efficiency_multiplier = 0.8 # 20% discount for high efficiency elif avg_cache_efficiency > 30: efficiency_multiplier = 1.0 # No penalty else: efficiency_multiplier = 1.3 # 30% penalty for low efficiency energy_tax = base_energy_tax * efficiency_multiplier return { 'total_carbon_kg': round(total_carbon_kg, 4), 'total_energy_kwh': round(total_energy_kwh, 4), 'base_carbon_tax_usd': round(base_energy_tax, 4), 'efficiency_multiplier': efficiency_multiplier, 'final_energy_tax_usd': round(energy_tax, 4), 'avg_cache_efficiency': round(avg_cache_efficiency, 2), 'carbon_price_per_kg_usd': carbon_price_per_kg } def generate_energy_recommendations(self, energy_results): """Generate recommendations for reducing energy consumption""" recommendations = [] # Calculate total energy by model model_usage = {} for result in energy_results: model = result['model'] if model not in model_usage: model_usage[model] = {'energy_kwh': 0, 'carbon_kg': 0, 'usage_count': 0} model_usage[model]['energy_kwh'] += result['total_energy_kwh'] model_usage[model]['carbon_kg'] += result['carbon_footprint_kg'] model_usage[model]['usage_count'] += 1 # Cache efficiency analysis low_efficiency_sessions = [r for r in energy_results if r['cache_efficiency'] < 30] if low_efficiency_sessions: recommendations.append({ 'category': 'Cache Optimization', 'priority': 'High', 'recommendation': f'Improve cache utilization - {len(low_efficiency_sessions)} sessions have <30% cache efficiency', 'potential_savings_kwh': sum(r['total_energy_kwh'] * 0.7 for r in low_efficiency_sessions), 'potential_savings_usd': sum(r['energy_cost_usd'] * 0.7 for r in low_efficiency_sessions) }) # Model efficiency recommendations (family-based so any Haiku/Sonnet/ # Opus generation participates, not just the legacy hardcoded IDs) haiku_stats = {m: s for m, s in model_usage.items() if 'haiku' in m.lower()} heavy_stats = {m: s for m, s in model_usage.items() if 'haiku' not in m.lower()} if haiku_stats and heavy_stats: haiku_efficiency = min( s['energy_kwh'] / s['usage_count'] for s in haiku_stats.values() if s['usage_count'] ) heavy_energy_kwh = sum(s['energy_kwh'] for s in heavy_stats.values()) heavy_sessions = sum(s['usage_count'] for s in heavy_stats.values()) heavy_efficiency = heavy_energy_kwh / heavy_sessions if heavy_sessions else 0 if heavy_efficiency > haiku_efficiency * 2: recommendations.append({ 'category': 'Model Selection', 'priority': 'Medium', 'recommendation': 'Consider using Haiku for simpler tasks to reduce energy consumption', 'potential_savings_kwh': heavy_energy_kwh * 0.4, 'potential_savings_usd': sum( r['energy_cost_usd'] for r in energy_results if 'haiku' not in r['model'].lower() ) * 0.4 }) return recommendations def analyze_energy_trends(self, energy_results): """Analyze energy consumption trends over time""" # Group by date daily_energy = {} for result in energy_results: date = result['date'] if date not in daily_energy: daily_energy[date] = { 'energy_kwh': 0, 'carbon_kg': 0, 'cost_usd': 0, 'sessions': 0, 'avg_cache_efficiency': 0 } daily_energy[date]['energy_kwh'] += result['total_energy_kwh'] daily_energy[date]['carbon_kg'] += result['carbon_footprint_kg'] daily_energy[date]['cost_usd'] += result['energy_cost_usd'] daily_energy[date]['sessions'] += 1 daily_energy[date]['avg_cache_efficiency'] += result['cache_efficiency'] # Calculate averages for date_data in daily_energy.values(): if date_data['sessions'] > 0: date_data['avg_cache_efficiency'] /= date_data['sessions'] return daily_energy def check_data_freshness(self, usage_data, now=None): """Check how old the newest ingested usage record is. Returns a dict describing the newest usage date, its age in days and whether it exceeds the staleness threshold, so reports are honest about the data they are built on (source CSVs are manual monthly exports and can silently go stale). """ now = now or datetime.now() latest_date = None for row in usage_data: raw = str(row.get('usage_date_utc', '') or '').strip() try: parsed = datetime.strptime(raw[:10], '%Y-%m-%d') except ValueError: continue if latest_date is None or parsed > latest_date: latest_date = parsed if latest_date is None: return { 'status': 'unknown', 'is_stale': True, 'latest_usage_date': None, 'age_days': None, 'threshold_days': self.staleness_max_age_days, 'message': 'No parseable usage dates found in source data', } age_days = (now - latest_date).days is_stale = age_days > self.staleness_max_age_days if is_stale: message = ( f'Newest usage record is {age_days} days old ' f'(threshold: {self.staleness_max_age_days} days). ' f'Export a fresh usage or cost CSV from the Anthropic Console ' f'and drop it into claude/ before trusting this report.' ) else: message = f'Source data is fresh (newest record {age_days} days old).' return { 'status': 'stale' if is_stale else 'fresh', 'is_stale': is_stale, 'latest_usage_date': latest_date.strftime('%Y-%m-%d'), 'age_days': age_days, 'threshold_days': self.staleness_max_age_days, 'message': message, } def generate_report(self): """Generate comprehensive AI energy impact report""" print("🌱 AI Energy Impact Analysis Starting...") # Load usage data usage_data = self.load_claude_usage_data() if not usage_data: print("❌ No Claude usage data found") return print(f"📊 Analyzing {len(usage_data)} usage records...") # Check whether the source data is stale before doing any math data_freshness = self.check_data_freshness(usage_data) if data_freshness['is_stale']: print(f"⚠️ STALE DATA: {data_freshness['message']}") # Calculate energy consumption for each usage record energy_results = [] for usage_row in usage_data: energy_result = self.calculate_energy_consumption(usage_row) energy_results.append(energy_result) # Calculate energy tax energy_tax_data = self.calculate_energy_tax(energy_results) # Generate recommendations recommendations = self.generate_energy_recommendations(energy_results) # Analyze trends daily_trends = self.analyze_energy_trends(energy_results) # Calculate summary statistics total_energy_kwh = sum(r['total_energy_kwh'] for r in energy_results) total_carbon_kg = sum(r['carbon_footprint_kg'] for r in energy_results) total_energy_cost = sum(r['energy_cost_usd'] for r in energy_results) avg_cache_efficiency = sum(r['cache_efficiency'] for r in energy_results) / len(energy_results) # Generate comprehensive report report = { 'analysis_metadata': { 'generated_at': datetime.now().isoformat(), 'methodology_version': METHODOLOGY_VERSION, 'total_sessions_analyzed': len(energy_results), 'date_range': { 'start': min(r['date'] for r in energy_results), 'end': max(r['date'] for r in energy_results) }, 'data_freshness': data_freshness, 'data_quality': { 'token_data_source': self.token_data_source, 'skipped_non_token_rows': self.skipped_source_rows, 'unrecognized_models': sorted(str(m) for m in self.unrecognized_models), 'generation_fallbacks': dict(self.generation_fallbacks), 'source_files': self.source_files, 'active_source_files': self.active_source_files, }, 'assumptions': { 'carbon_intensity_g_per_kwh': self.carbon_intensity_g_per_kwh, 'energy_cost_usd_per_kwh': self.energy_cost_usd_per_kwh, 'carbon_price_usd_per_kg': CARBON_PRICE_USD_PER_KG, 'default_model_profile': DEFAULT_MODEL_KEY, }, }, 'energy_summary': { 'total_energy_consumption_kwh': round(total_energy_kwh, 4), 'total_carbon_footprint_kg': round(total_carbon_kg, 4), 'total_energy_cost_usd': round(total_energy_cost, 4), 'average_cache_efficiency_percent': round(avg_cache_efficiency, 2), 'energy_per_session_kwh': round(total_energy_kwh / len(energy_results), 6), 'carbon_per_session_kg': round(total_carbon_kg / len(energy_results), 6) }, 'energy_tax_calculation': energy_tax_data, 'model_breakdown': self._calculate_model_breakdown(energy_results), 'daily_trends': daily_trends, 'optimization_recommendations': recommendations, 'detailed_sessions': energy_results } # Save report report_file = self.output_dir / f'ai_energy_impact_analysis_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json' with open(report_file, 'w') as f: json.dump(report, f, indent=2) # Generate markdown summary self._generate_markdown_summary(report) print(f"\n🌱 AI Energy Impact Analysis Complete!") print(f"📊 Total Energy Consumption: {total_energy_kwh:.4f} kWh") print(f"🌍 Carbon Footprint: {total_carbon_kg:.4f} kg CO2") print(f"💰 Energy Cost: ${total_energy_cost:.4f}") print(f"🔋 Energy Tax: ${energy_tax_data['final_energy_tax_usd']:.4f}") print(f"⚡ Cache Efficiency: {avg_cache_efficiency:.1f}%") if self.unrecognized_models: print(f"⚠️ Unrecognized models (default profile applied): {sorted(self.unrecognized_models)}") print(f"📄 Report saved: {report_file}") return report def _calculate_model_breakdown(self, energy_results): """Calculate energy breakdown by model""" model_stats = {} for result in energy_results: model = result['model'] if model not in model_stats: model_stats[model] = { 'sessions': 0, 'total_energy_kwh': 0, 'total_carbon_kg': 0, 'total_cost_usd': 0, 'avg_cache_efficiency': 0 } stats = model_stats[model] stats['sessions'] += 1 stats['total_energy_kwh'] += result['total_energy_kwh'] stats['total_carbon_kg'] += result['carbon_footprint_kg'] stats['total_cost_usd'] += result['energy_cost_usd'] stats['avg_cache_efficiency'] += result['cache_efficiency'] # Calculate averages for model, stats in model_stats.items(): if stats['sessions'] > 0: stats['avg_cache_efficiency'] /= stats['sessions'] stats['energy_per_session'] = stats['total_energy_kwh'] / stats['sessions'] stats['carbon_per_session'] = stats['total_carbon_kg'] / stats['sessions'] return model_stats def _generate_markdown_summary(self, report): """Generate markdown summary report""" metadata = report['analysis_metadata'] freshness = metadata['data_freshness'] data_quality = metadata['data_quality'] assumptions = metadata['assumptions'] freshness_banner = '' if freshness['is_stale']: freshness_banner = f"""> ⚠️ **STALE DATA WARNING**: {freshness['message']} """ md_content = f"""# AI Energy Impact Analysis Report *Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} · Methodology v{metadata['methodology_version']}* {freshness_banner}## 🌱 Executive Summary Our AI development practices consumed **{report['energy_summary']['total_energy_consumption_kwh']:.4f} kWh** of energy across {report['analysis_metadata']['total_sessions_analyzed']} development sessions, generating **{report['energy_summary']['total_carbon_footprint_kg']:.4f} kg** of CO2 emissions. ### 💰 Energy Tax Assessment - **Base Carbon Tax**: ${report['energy_tax_calculation']['base_carbon_tax_usd']:.4f} - **Efficiency Multiplier**: {report['energy_tax_calculation']['efficiency_multiplier']:.1f}x - **Final Energy Tax**: ${report['energy_tax_calculation']['final_energy_tax_usd']:.4f} - **Cache Efficiency**: {report['energy_tax_calculation']['avg_cache_efficiency']:.1f}% ## 📊 Energy Consumption Breakdown | Metric | Value | |--------|--------| | Total Energy Consumption | {report['energy_summary']['total_energy_consumption_kwh']:.4f} kWh | | Carbon Footprint | {report['energy_summary']['total_carbon_footprint_kg']:.4f} kg CO2 | | Energy Cost | ${report['energy_summary']['total_energy_cost_usd']:.4f} | | Average per Session | {report['energy_summary']['energy_per_session_kwh']:.6f} kWh | | Cache Efficiency | {report['energy_summary']['average_cache_efficiency_percent']:.1f}% | ## 🔋 Model Performance Analysis """ for model, stats in report['model_breakdown'].items(): md_content += f"""### {model} - **Sessions**: {stats['sessions']} - **Energy**: {stats['total_energy_kwh']:.4f} kWh ({stats['energy_per_session']:.6f} kWh/session) - **Carbon**: {stats['total_carbon_kg']:.4f} kg CO2 ({stats['carbon_per_session']:.6f} kg/session) - **Cache Efficiency**: {stats['avg_cache_efficiency']:.1f}% """ md_content += """## 🎯 Optimization Recommendations """ for rec in report['optimization_recommendations']: potential_savings = rec.get('potential_savings_usd', 0) md_content += f"""### {rec['category']} ({rec['priority']} Priority) {rec['recommendation']} - **Potential Energy Savings**: {rec.get('potential_savings_kwh', 0):.4f} kWh - **Potential Cost Savings**: ${potential_savings:.4f} """ md_content += f"""## 📈 Sustainability Insights 1. **Energy Efficiency**: Our cache utilization of {report['energy_summary']['average_cache_efficiency_percent']:.1f}% {'exceeds' if report['energy_summary']['average_cache_efficiency_percent'] > 50 else 'falls below'} optimal levels 2. **Carbon Impact**: Each development session generates {report['energy_summary']['carbon_per_session_kg']:.6f} kg CO2 3. **Cost Optimization**: Energy tax of ${report['energy_tax_calculation']['final_energy_tax_usd']:.4f} reflects our environmental impact ## 🔎 Data Sources & Quality | Item | Value | |------|-------| | Data freshness | {freshness['status']} — {freshness['message']} | | Newest usage record | {freshness['latest_usage_date'] or 'unknown'} | | Token counts | {data_quality['token_data_source']} | | **Files actually used this run** | {', '.join(data_quality['active_source_files']) if data_quality['active_source_files'] else 'none'} | | Skipped non-token rows | {data_quality['skipped_non_token_rows']} | | Unrecognized models | {', '.join(data_quality['unrecognized_models']) if data_quality['unrecognized_models'] else 'none'} | | Generation fallbacks | {', '.join(f'{k} → {v}' for k, v in data_quality['generation_fallbacks'].items()) if data_quality['generation_fallbacks'] else 'none'} | | Token export files found | {', '.join(data_quality['source_files']['token_files_found']) if data_quality['source_files']['token_files_found'] else 'none'} | | Cost export files found | {', '.join(data_quality['source_files']['cost_files_found']) if data_quality['source_files']['cost_files_found'] else 'none'} | | CSVs in claude/ ignored (unrecognized schema) | {', '.join(data_quality['source_files']['ignored_files']) if data_quality['source_files']['ignored_files'] else 'none'} | Precedence rule: **measured token exports always win over cost exports**, even when both exist for the same or an overlapping period — cost-based estimates are only used when no token export is found at all. ## 📐 Methodology & Assumptions All energy figures are **research-calibrated estimates** — Anthropic publishes no per-model energy data. When token counts are estimated from dollar costs, the result is an estimate built on an estimate; treat totals as order-of-magnitude indicators. | Assumption | Value | Source | |------------|-------|--------| | Carbon intensity | {assumptions['carbon_intensity_g_per_kwh']} gCO2/kWh | EPA eGRID US grid average (https://www.epa.gov/egrid) | | Electricity rate | ${assumptions['energy_cost_usd_per_kwh']:.3f}/kWh | EIA US commercial average (https://www.eia.gov/electricity/monthly/) | | Carbon price | ${assumptions['carbon_price_usd_per_kg']:.2f}/kg CO2 ($30/tonne) | Internal energy-tax policy | | Per-token energy | Model-tier estimates | Epoch AI (https://epoch.ai/gradient-updates/how-much-energy-does-chatgpt-use), Google Gemini paper (https://arxiv.org/abs/2508.15734), ML.Energy (https://ml.energy/leaderboard/) | | Token estimation | Anthropic published pricing | https://platform.claude.com/docs/en/about-claude/pricing | | Default profile for unknown models | {assumptions['default_model_profile']} | Tracked and disclosed above | ## 🌍 Environmental Commitment This analysis supports our commitment to sustainable AI development by: - Quantifying energy consumption and carbon footprint - Implementing energy-based cost adjustments - Providing actionable optimization recommendations - Encouraging efficient AI tool usage patterns --- *Analysis includes {len(report['detailed_sessions'])} development sessions from {report['analysis_metadata']['date_range']['start']} to {report['analysis_metadata']['date_range']['end']}* """ # Save markdown report md_file = self.output_dir / f'ai_energy_impact_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.md' with open(md_file, 'w') as f: f.write(md_content) print(f"📄 Markdown report saved: {md_file}") def main(argv=None): """Main execution function""" parser = argparse.ArgumentParser( description='AI Energy Impact Analyzer — energy, carbon and energy-tax ' 'estimates for Claude API usage.' ) parser.add_argument( '--max-age-days', type=int, default=DEFAULT_STALENESS_THRESHOLD_DAYS, help='Age (days) after which the newest ingested usage record is ' f'considered stale (default: {DEFAULT_STALENESS_THRESHOLD_DAYS})', ) parser.add_argument( '--check-freshness', action='store_true', help='Only check source-data freshness; exit 1 if stale (for CI)', ) args = parser.parse_args(argv) analyzer = AIEnergyImpactAnalyzer(staleness_max_age_days=args.max_age_days) if args.check_freshness: # Kept terse and script-friendly for CI — no banner here. usage_data = analyzer.load_claude_usage_data() freshness = analyzer.check_data_freshness(usage_data) icon = '⚠️' if freshness['is_stale'] else '✅' print(f"{icon} {freshness['message']}") return 1 if freshness['is_stale'] else 0 show_unparty_banner() report = analyzer.generate_report() return 0 if report else 1 if __name__ == "__main__": sys.exit(main())