Energy-Efficient AI & Green Data Centers 2026: Reduce Power Consumption by 70% Guide

from dataclasses import dataclass
from typing import Dict, List
from datetime import datetime

@dataclass
class EnergyMetrics:
    timestamp: datetime
    model_name: str
    operation_type: str  # 'training', 'inference', 'fine_tuning'
    gpu_type: str
    gpu_hours: float
    power_draw_watts: float
    pue: float  # Power Usage Effectiveness
    carbon_intensity: float  # gCO2/kWh
    region: str

class CarbonFootprintCalculator:
    """Calculate and track AI system carbon emissions"""

    # GPU power consumption (TDP in watts)
    GPU_POWER = {
        'H100': 700,
        'A100': 400,
        'L4': 72,
        'T4': 70,
        'V100': 300
    }

    # Average carbon intensity by region (gCO2/kWh)
    CARBON_INTENSITY = {
        'us-west': 350,      # California (high renewable)
        'us-east': 450,      # East coast
        'europe-north': 50,  # Nordic (hydro/wind)
        'europe-west': 300,  # Western Europe
        'asia-pacific': 600, # Avg coal-heavy
        'global-avg': 475
    }

    def __init__(self):
        self.energy_log: List[EnergyMetrics] = []

    def calculate_training_emissions(
        self,
        model_name: str,
        gpu_type: str,
        num_gpus: int,
        training_hours: float,
        region: str,
        pue: float = 1.6
    ) -> Dict:
        """Calculate emissions for model training"""

        # Get GPU power draw
        power_per_gpu = self.GPU_POWER.get(gpu_type, 400)
        total_power_kw = (power_per_gpu * num_gpus) / 1000

        # Account for data center overhead (PUE)
        actual_power_kw = total_power_kw * pue

        # Calculate energy consumption
        energy_kwh = actual_power_kw * training_hours

        # Get carbon intensity for region
        carbon_intensity = self.CARBON_INTENSITY.get(region, 475)

        # Calculate emissions
        emissions_kg_co2 = (energy_kwh * carbon_intensity) / 1000
        emissions_tonnes_co2 = emissions_kg_co2 / 1000

        # Log metrics
        self.energy_log.append(EnergyMetrics(
            timestamp=datetime.now(),
            model_name=model_name,
            operation_type='training',
            gpu_type=gpu_type,
            gpu_hours=training_hours * num_gpus,
            power_draw_watts=power_per_gpu,
            pue=pue,
            carbon_intensity=carbon_intensity,
            region=region
        ))

        return {
            'model_name': model_name,
            'gpu_type': gpu_type,
            'num_gpus': num_gpus,
            'training_hours': training_hours,
            'energy_kwh': energy_kwh,
            'emissions_kg_co2': emissions_kg_co2,
            'emissions_tonnes_co2': emissions_tonnes_co2,
            'equivalent_households_year': emissions_tonnes_co2 / 4.6,  # Avg US household
            'cost_at_12c_kwh': energy_kwh * 0.12,
            'region': region,
            'pue': pue
        }

    def calculate_inference_emissions_per_million(
        self,
        model_name: str,
        avg_latency_ms: float,
        requests_per_day: int,
        gpu_type: str,
        region: str,
        days: int = 30
    ) -> Dict:
        """Calculate emissions for production inference"""

        # Get GPU power
        power_watts = self.GPU_POWER.get(gpu_type, 400)

        # Calculate energy per request
        energy_per_request_wh = (power_watts * (avg_latency_ms / 1000)) / 3600
        energy_per_million_kwh = (energy_per_request_wh * 1_000_000) / 1000

        # Monthly energy
        total_requests = requests_per_day * days
        monthly_energy_kwh = (energy_per_request_wh * total_requests) / 1000

        # Emissions
        carbon_intensity = self.CARBON_INTENSITY.get(region, 475)
        monthly_emissions_kg = (monthly_energy_kwh * carbon_intensity) / 1000

        return {
            'model_name': model_name,
            'requests_per_day': requests_per_day,
            'avg_latency_ms': avg_latency_ms,
            'energy_per_million_requests_kwh': energy_per_million_kwh,
            'monthly_energy_kwh': monthly_energy_kwh,
            'monthly_emissions_kg_co2': monthly_emissions_kg,
            'emissions_per_million_requests_kg': (energy_per_million_kwh * carbon_intensity) / 1000,
            'monthly_cost_at_12c_kwh': monthly_energy_kwh * 0.12
        }

    def generate_compliance_report(self, year: int) -> Dict:
        """Generate annual sustainability report for compliance"""

        year_logs = [
            log for log in self.energy_log
            if log.timestamp.year == year
        ]

        # Calculate totals
        total_gpu_hours = sum(log.gpu_hours for log in year_logs)
        total_energy_kwh = sum(
            (log.power_draw_watts * log.gpu_hours * log.pue) / 1000
            for log in year_logs
        )
        total_emissions_tonnes = sum(
            (log.power_draw_watts * log.gpu_hours * log.pue * log.carbon_intensity) / 1_000_000_000
            for log in year_logs
        )

        # Break down by operation type
        by_operation = {}
        for log in year_logs:
            if log.operation_type not in by_operation:
                by_operation[log.operation_type] = {
                    'gpu_hours': 0,
                    'energy_kwh': 0,
                    'emissions_tonnes': 0
                }

            energy = (log.power_draw_watts * log.gpu_hours * log.pue) / 1000
            emissions = (energy * log.carbon_intensity) / 1000

            by_operation[log.operation_type]['gpu_hours'] += log.gpu_hours
            by_operation[log.operation_type]['energy_kwh'] += energy
            by_operation[log.operation_type]['emissions_tonnes'] += emissions / 1000

        return {
            'reporting_year': year,
            'total_gpu_hours': total_gpu_hours,
            'total_energy_consumption_kwh': total_energy_kwh,
            'total_emissions_tonnes_co2': total_emissions_tonnes,
            'equivalent_households': total_emissions_tonnes / 4.6,
            'breakdown_by_operation': by_operation,
            'compliance_frameworks': ['EU AI Act', 'CSRD', 'GHG Protocol Scope 2']
        }

# Usage
calculator = CarbonFootprintCalculator()

# Calculate training emissions for GPT-sized model
training_report = calculator.calculate_training_emissions(
    model_name="llm-v1",
    gpu_type="A100",
    num_gpus=256,
    training_hours=720,  # 30 days
    region="us-west",
    pue=1.4  # Efficient data center
)

print("=== TRAINING CARBON FOOTPRINT ===")
print(f"Model: {training_report['model_name']}")
print(f"Energy consumed: {training_report['energy_kwh']:,.0f} kWh")
print(f"CO₂ emissions: {training_report['emissions_tonnes_co2']:.1f} tonnes")
print(f"Equivalent to: {training_report['equivalent_households_year']:.1f} household-years")
print(f"Energy cost: ${training_report['cost_at_12c_kwh']:,.2f}")

# Calculate monthly inference emissions
inference_report = calculator.calculate_inference_emissions_per_million(
    model_name="llm-v1-prod",
    avg_latency_ms=150,
    requests_per_day=1_000_000,
    gpu_type="L4",
    region="us-west",
    days=30
)

print("\n=== MONTHLY INFERENCE FOOTPRINT ===")
print(f"Requests per day: {inference_report['requests_per_day']:,}")
print(f"Monthly energy: {inference_report['monthly_energy_kwh']:,.0f} kWh")
print(f"Monthly emissions: {inference_report['monthly_emissions_kg_co2']:.1f} kg CO₂")
print(f"Per million requests: {inference_report['emissions_per_million_requests_kg']:.2f} kg CO₂")

import subprocess
from typing import Dict
import time

class GPUPowerMonitor:
    """Monitor real-time GPU power consumption"""

    def __init__(self):
        self.measurements = []

    def get_gpu_power_usage(self) -> Dict:
        """Get current GPU power draw using nvidia-smi"""

        try:
            # Query GPU power usage
            result = subprocess.run(
                ['nvidia-smi', '--query-gpu=power.draw,power.limit,utilization.gpu,temperature.gpu',
                 '--format=csv,noheader,nounits'],
                capture_output=True,
                text=True
            )

            if result.returncode == 0:
                lines = result.stdout.strip().split('\n')
                gpu_data = []

                for idx, line in enumerate(lines):
                    power_draw, power_limit, util, temp = line.split(',')
                    gpu_data.append({
                        'gpu_id': idx,
                        'power_draw_watts': float(power_draw),
                        'power_limit_watts': float(power_limit),
                        'utilization_pct': float(util),
                        'temperature_c': float(temp)
                    })

                return {
                    'timestamp': time.time(),
                    'gpus': gpu_data,
                    'total_power_draw': sum(g['power_draw_watts'] for g in gpu_data)
                }

        except Exception as e:
            return {'error': str(e)}

    def monitor_training_session(
        self,
        duration_seconds: int = 60,
        sample_interval: int = 5
    ) -> Dict:
        """Monitor power during training session"""

        samples = []
        start_time = time.time()

        while time.time() - start_time < duration_seconds:
            power_data = self.get_gpu_power_usage()
            if 'error' not in power_data:
                samples.append(power_data)

            time.sleep(sample_interval)

        # Calculate statistics
        if not samples:
            return {'error': 'No samples collected'}

        total_powers = [s['total_power_draw'] for s in samples]
        avg_power = sum(total_powers) / len(total_powers)
        max_power = max(total_powers)
        min_power = min(total_powers)

        # Estimate energy consumption
        duration_hours = duration_seconds / 3600
        energy_kwh = (avg_power * duration_hours) / 1000

        return {
            'duration_seconds': duration_seconds,
            'samples_collected': len(samples),
            'avg_power_watts': avg_power,
            'max_power_watts': max_power,
            'min_power_watts': min_power,
            'energy_consumed_kwh': energy_kwh,
            'estimated_cost_at_12c_kwh': energy_kwh * 0.12
        }

# Mock usage (would work with actual nvidia-smi)
monitor = GPUPowerMonitor()
print("GPU power monitoring initialized")
# results = monitor.monitor_training_session(duration_seconds=300)

class PUECalculator:
    """Calculate Power Usage Effectiveness for data centers"""

    def __init__(self):
        self.measurements = []

    def calculate_pue(
        self,
        it_equipment_power_kw: float,
        cooling_power_kw: float,
        lighting_power_kw: float,
        networking_power_kw: float,
        other_facility_power_kw: float = 0
    ) -> Dict:
        """Calculate PUE and efficiency metrics"""

        total_facility_power = (
            it_equipment_power_kw +
            cooling_power_kw +
            lighting_power_kw +
            networking_power_kw +
            other_facility_power_kw
        )

        pue = total_facility_power / it_equipment_power_kw if it_equipment_power_kw > 0 else 0

        # Calculate efficiency
        overhead_power = total_facility_power - it_equipment_power_kw
        overhead_pct = (overhead_power / total_facility_power) * 100 if total_facility_power > 0 else 0

        # Determine rating
        if pue < 1.2:
            rating = "Excellent"
        elif pue < 1.5:
            rating = "Good"
        elif pue < 2.0:
            rating = "Average"
        else:
            rating = "Poor"

        return {
            'it_equipment_power_kw': it_equipment_power_kw,
            'cooling_power_kw': cooling_power_kw,
            'total_facility_power_kw': total_facility_power,
            'pue': pue,
            'efficiency_rating': rating,
            'overhead_percentage': overhead_pct,
            'wasted_power_kw': overhead_power,
            'potential_savings_at_pue_1.2': (total_facility_power - (it_equipment_power_kw * 1.2)) if pue > 1.2 else 0
        }

    def calculate_annual_cost_impact(
        self,
        current_pue: float,
        it_load_kw: float,
        electricity_cost_per_kwh: float = 0.12,
        hours_per_year: int = 8760
    ) -> Dict:
        """Calculate annual cost of PUE inefficiency"""

        # Current annual cost
        current_total_power = it_load_kw * current_pue
        current_annual_kwh = current_total_power * hours_per_year
        current_annual_cost = current_annual_kwh * electricity_cost_per_kwh

        # Best-in-class PUE
        target_pue = 1.2
        target_total_power = it_load_kw * target_pue
        target_annual_kwh = target_total_power * hours_per_year
        target_annual_cost = target_annual_kwh * electricity_cost_per_kwh

        # Savings potential
        annual_savings = current_annual_cost - target_annual_cost
        savings_percentage = (annual_savings / current_annual_cost) * 100 if current_annual_cost > 0 else 0

        return {
            'current_pue': current_pue,
            'target_pue': target_pue,
            'current_annual_cost': current_annual_cost,
            'target_annual_cost': target_annual_cost,
            'annual_savings_potential': annual_savings,
            'savings_percentage': savings_percentage,
            'roi_months': 24  # Typical payback for cooling upgrades
        }

# Usage
pue_calc = PUECalculator()

# Calculate PUE for data center
pue_result = pue_calc.calculate_pue(
    it_equipment_power_kw=1000,  # 1 MW of GPU/server power
    cooling_power_kw=450,         # Cooling systems
    lighting_power_kw=30,
    networking_power_kw=70,
    other_facility_power_kw=50
)

print("=== DATA CENTER PUE ANALYSIS ===")
print(f"PUE: {pue_result['pue']:.2f}")
print(f"Rating: {pue_result['efficiency_rating']}")
print(f"Overhead: {pue_result['overhead_percentage']:.1f}%")
print(f"Wasted power: {pue_result['wasted_power_kw']:.0f} kW")

# Calculate cost impact
cost_impact = pue_calc.calculate_annual_cost_impact(
    current_pue=pue_result['pue'],
    it_load_kw=1000,
    electricity_cost_per_kwh=0.12
)

print(f"\n=== ANNUAL COST IMPACT ===")
print(f"Current annual cost: ${cost_impact['current_annual_cost']:,.0f}")
print(f"Potential savings: ${cost_impact['annual_savings_potential']:,.0f} ({cost_impact['savings_percentage']:.1f}%)")

import numpy as np

class ModelEnergyAnalyzer:
    """Compare energy consumption across model architectures"""

    # Energy per parameter (relative units)
    ARCHITECTURE_EFFICIENCY = {
        'transformer_dense': 1.0,      # Baseline
        'transformer_sparse': 0.4,     # MoE, sparse attention
        'linear_transformer': 0.6,     # Linear complexity
        'distilbert': 0.5,             # Distilled model
        'mobilenet_style': 0.3,        # Mobile-optimized
        'quantized_int8': 0.35,        # 8-bit quantization
    }

    def estimate_training_energy(
        self,
        architecture: str,
        num_parameters_b: float,  # billions
        training_tokens_b: float,  # billions
        gpu_type: str = 'A100'
    ) -> Dict:
        """Estimate training energy consumption"""

        base_efficiency = self.ARCHITECTURE_EFFICIENCY.get(architecture, 1.0)

        # FLOPs calculation (simplified)
        # Training: 6 * params * tokens (forward + backward pass)
        flops_e18 = 6 * num_parameters_b * training_tokens_b  # EFLOPs

        # GPU efficiency (TFLOPS)
        gpu_tflops = {'H100': 1979, 'A100': 312, 'V100': 125}.get(gpu_type, 312)

        # Calculate GPU hours
        gpu_hours = (flops_e18 * 1e9) / (gpu_tflops * 3600)

        # Apply architecture efficiency
        actual_gpu_hours = gpu_hours * base_efficiency

        # Power consumption
        gpu_power_kw = {'H100': 0.7, 'A100': 0.4, 'V100': 0.3}.get(gpu_type, 0.4)
        energy_kwh = actual_gpu_hours * gpu_power_kw

        return {
            'architecture': architecture,
            'parameters_billions': num_parameters_b,
            'training_tokens_billions': training_tokens_b,
            'estimated_flops_eflops': flops_e18,
            'gpu_hours': actual_gpu_hours,
            'energy_kwh': energy_kwh,
            'efficiency_multiplier': base_efficiency,
            'co2_kg_at_450g_kwh': energy_kwh * 0.45,
            'cost_at_12c_kwh': energy_kwh * 0.12
        }

    def compare_architectures(
        self,
        num_parameters_b: float,
        training_tokens_b: float
    ) -> List[Dict]:
        """Compare energy across different architectures"""

        architectures = [
            'transformer_dense',
            'transformer_sparse',
            'linear_transformer',
            'distilbert',
            'quantized_int8'
        ]

        comparisons = []
        for arch in architectures:
            result = self.estimate_training_energy(
                arch, num_parameters_b, training_tokens_b
            )
            comparisons.append(result)

        return sorted(comparisons, key=lambda x: x['energy_kwh'])

# Usage
analyzer = ModelEnergyAnalyzer()

# Compare 7B parameter model training
comparisons = analyzer.compare_architectures(
    num_parameters_b=7.0,
    training_tokens_b=1000  # 1T tokens
)

print("=== MODEL ARCHITECTURE ENERGY COMPARISON (7B params, 1T tokens) ===\n")
baseline_energy = comparisons[-1]['energy_kwh']

for comp in comparisons:
    savings_pct = ((baseline_energy - comp['energy_kwh']) / baseline_energy) * 100
    print(f"{comp['architecture']:25} {comp['energy_kwh']:10,.0f} kWh  "
          f"${comp['cost_at_12c_kwh']:8,.2f}  "
          f"({savings_pct:+.0f}% vs dense)")

import torch
import time

class QuantizationEnergyBenchmark:
    """Benchmark energy savings from quantization"""

    def __init__(self, model, sample_input):
        self.model_fp32 = model
        self.sample_input = sample_input

    def quantize_model_int8(self):
        """Quantize model to INT8"""
        # Dynamic quantization (post-training)
        quantized_model = torch.quantization.quantize_dynamic(
            self.model_fp32,
            {torch.nn.Linear},  # Quantize linear layers
            dtype=torch.qint8
        )
        return quantized_model

    def benchmark_inference(
        self,
        model,
        num_iterations: int = 1000
    ) -> Dict:
        """Benchmark inference performance"""

        latencies = []

        # Warmup
        for _ in range(10):
            _ = model(self.sample_input)

        # Benchmark
        for _ in range(num_iterations):
            start = time.time()
            _ = model(self.sample_input)
            latencies.append(time.time() - start)

        return {
            'mean_latency_ms': np.mean(latencies) * 1000,
            'p50_latency_ms': np.percentile(latencies, 50) * 1000,
            'p95_latency_ms': np.percentile(latencies, 95) * 1000,
            'throughput_rps': 1 / np.mean(latencies)
        }

    def compare_fp32_vs_int8(self) -> Dict:
        """Compare FP32 vs INT8 quantized model"""

        # Benchmark FP32
        print("Benchmarking FP32 model...")
        fp32_results = self.benchmark_inference(self.model_fp32)

        # Quantize and benchmark INT8
        print("Quantizing to INT8...")
        quantized_model = self.quantize_model_int8()

        print("Benchmarking INT8 model...")
        int8_results = self.benchmark_inference(quantized_model)

        # Calculate improvements
        latency_improvement = (
            (fp32_results['mean_latency_ms'] - int8_results['mean_latency_ms']) /
            fp32_results['mean_latency_ms']
        ) * 100

        throughput_improvement = (
            (int8_results['throughput_rps'] - fp32_results['throughput_rps']) /
            fp32_results['throughput_rps']
        ) * 100

        # Energy estimation
        # INT8 uses ~35% of FP32 energy
        fp32_energy_per_inference = 100  # Baseline units
        int8_energy_per_inference = 35   # 65% savings

        return {
            'fp32_latency_ms': fp32_results['mean_latency_ms'],
            'int8_latency_ms': int8_results['mean_latency_ms'],
            'latency_improvement_pct': latency_improvement,
            'fp32_throughput_rps': fp32_results['throughput_rps'],
            'int8_throughput_rps': int8_results['throughput_rps'],
            'throughput_improvement_pct': throughput_improvement,
            'energy_savings_pct': 65,
            'model_size_reduction_pct': 75,  # 4 bytes -> 1 byte
        }

# Mock usage example
class SimpleModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = torch.nn.Linear(512, 256)
        self.fc2 = torch.nn.Linear(256, 128)
        self.fc3 = torch.nn.Linear(128, 10)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)

model = SimpleModel()
sample_input = torch.randn(1, 512)

benchmark = QuantizationEnergyBenchmark(model, sample_input)
comparison = benchmark.compare_fp32_vs_int8()

print("\n=== FP32 vs INT8 QUANTIZATION COMPARISON ===")
print(f"Latency:      {comparison['fp32_latency_ms']:.2f}ms -> {comparison['int8_latency_ms']:.2f}ms ({comparison['latency_improvement_pct']:+.1f}%)")
print(f"Throughput:   {comparison['fp32_throughput_rps']:.0f} RPS -> {comparison['int8_throughput_rps']:.0f} RPS ({comparison['throughput_improvement_pct']:+.1f}%)")
print(f"Energy:       {comparison['energy_savings_pct']}% savings")
print(f"Model size:   {comparison['model_size_reduction_pct']}% reduction")

import asyncio
from collections import deque
from typing import List, Any
import time

class DynamicBatcher:
    """Dynamic request batching for energy-efficient inference"""

    def __init__(
        self,
        max_batch_size: int = 32,
        max_wait_ms: int = 50,
        model_inference_fn: callable = None
    ):
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.model_inference_fn = model_inference_fn
        self.pending_requests = deque()
        self.batch_stats = []

    async def add_request(self, request_data: Any) -> Any:
        """Add request to batch queue"""

        future = asyncio.Future()
        self.pending_requests.append({
            'data': request_data,
            'future': future,
            'timestamp': time.time()
        })

        # Trigger batch processing if we hit max batch size
        if len(self.pending_requests) >= self.max_batch_size:
            asyncio.create_task(self._process_batch())

        return await future

    async def _process_batch(self):
        """Process accumulated batch"""

        if not self.pending_requests:
            return

        # Collect batch (up to max_batch_size)
        batch = []
        futures = []

        while self.pending_requests and len(batch) < self.max_batch_size:
            req = self.pending_requests.popleft()
            batch.append(req['data'])
            futures.append(req['future'])

        if not batch:
            return

        # Process batch
        start_time = time.time()
        try:
            results = await self.model_inference_fn(batch)

            # Return results to individual futures
            for future, result in zip(futures, results):
                future.set_result(result)

            # Record batch statistics
            self._record_batch_stats(
                batch_size=len(batch),
                processing_time=time.time() - start_time
            )

        except Exception as e:
            # Propagate error to all futures
            for future in futures:
                future.set_exception(e)

    def _record_batch_stats(self, batch_size: int, processing_time: float):
        """Record batch performance metrics"""

        self.batch_stats.append({
            'batch_size': batch_size,
            'processing_time_ms': processing_time * 1000,
            'throughput_rps': batch_size / processing_time,
            'timestamp': time.time()
        })

    def calculate_energy_efficiency(self) -> Dict:
        """Calculate energy efficiency gains from batching"""

        if not self.batch_stats:
            return {'error': 'No batch data'}

        avg_batch_size = np.mean([s['batch_size'] for s in self.batch_stats])
        total_requests = sum(s['batch_size'] for s in self.batch_stats)

        # Energy model: Base cost + per-request cost
        # Batching amortizes base cost across requests
        base_energy_per_batch = 10  # Arbitrary units
        energy_per_request = 1

        # Batched energy
        batched_energy = len(self.batch_stats) * base_energy_per_batch + total_requests * energy_per_request

        # Individual request energy (no batching)
        individual_energy = total_requests * (base_energy_per_batch + energy_per_request)

        energy_savings_pct = ((individual_energy - batched_energy) / individual_energy) * 100

        return {
            'total_batches': len(self.batch_stats),
            'total_requests': total_requests,
            'avg_batch_size': avg_batch_size,
            'energy_savings_pct': energy_savings_pct,
            'batched_energy_units': batched_energy,
            'individual_energy_units': individual_energy
        }

# Mock inference function
async def mock_model_inference(batch: List) -> List:
    await asyncio.sleep(0.02)  # 20ms processing
    return [{'prediction': 0.8} for _ in batch]

# Usage
batcher = DynamicBatcher(
    max_batch_size=32,
    max_wait_ms=50,
    model_inference_fn=mock_model_inference
)

print("Dynamic batching energy efficiency analysis initialized")

import hashlib
from typing import Optional, Tuple

class InferenceCache:
    """Cache inference results to reduce redundant computation"""

    def __init__(self, max_size_mb: int = 100):
        self.cache = {}
        self.max_size_bytes = max_size_mb * 1024 * 1024
        self.current_size_bytes = 0
        self.stats = {
            'hits': 0,
            'misses': 0,
            'energy_saved_kwh': 0
        }

    def _hash_input(self, input_data: Any) -> str:
        """Create hash of input for cache key"""
        input_str = str(input_data)
        return hashlib.sha256(input_str.encode()).hexdigest()

    def get(self, input_data: Any) -> Optional[Any]:
        """Retrieve cached result"""
        cache_key = self._hash_input(input_data)

        if cache_key in self.cache:
            self.stats['hits'] += 1

            # Estimate energy saved (GPU inference avoided)
            # Typical inference: 0.001 kWh per request
            self.stats['energy_saved_kwh'] += 0.001

            return self.cache[cache_key]['result']

        self.stats['misses'] += 1
        return None

    def put(self, input_data: Any, result: Any):
        """Store result in cache"""
        cache_key = self._hash_input(input_data)

        # Estimate result size (simplified)
        result_size = len(str(result))

        # Check if we need to evict
        while (self.current_size_bytes + result_size > self.max_size_bytes and
               len(self.cache) > 0):
            # Simple FIFO eviction
            oldest_key = next(iter(self.cache))
            evicted_size = self.cache[oldest_key]['size']
            del self.cache[oldest_key]
            self.current_size_bytes -= evicted_size

        # Store in cache
        self.cache[cache_key] = {
            'result': result,
            'size': result_size
        }
        self.current_size_bytes += result_size

    def get_cache_efficiency(self) -> Dict:
        """Calculate cache hit rate and energy savings"""

        total_requests = self.stats['hits'] + self.stats['misses']
        hit_rate = self.stats['hits'] / total_requests if total_requests > 0 else 0

        # Calculate cost savings
        # Cached response: negligible energy
        # GPU inference: ~0.001 kWh @ $0.12/kWh = $0.00012
        cost_savings = self.stats['energy_saved_kwh'] * 0.12

        return {
            'total_requests': total_requests,
            'cache_hits': self.stats['hits'],
            'cache_misses': self.stats['misses'],
            'hit_rate_pct': hit_rate * 100,
            'energy_saved_kwh': self.stats['energy_saved_kwh'],
            'cost_saved_dollars': cost_savings,
            'cache_size_mb': self.current_size_bytes / (1024 * 1024)
        }

# Usage
cache = InferenceCache(max_size_mb=100)

# Simulate requests
for i in range(1000):
    input_data = f"query_{i % 100}"  # 10% unique, 90% repeated

    # Check cache
    cached_result = cache.get(input_data)

    if cached_result is None:
        # Perform inference
        result = {'prediction': 0.85}
        cache.put(input_data, result)
    else:
        result = cached_result

efficiency = cache.get_cache_efficiency()
print("=== INFERENCE CACHE EFFICIENCY ===")
print(f"Cache hit rate: {efficiency['hit_rate_pct']:.1f}%")
print(f"Energy saved: {efficiency['energy_saved_kwh']:.3f} kWh")
print(f"Cost saved: ${efficiency['cost_saved_dollars']:.2f}")

class CloudProviderSustainability:
    """Compare cloud provider sustainability metrics"""

    PROVIDERS = {
        'aws': {
            'name': 'Amazon Web Services',
            'carbon_free_target': 2025,
            'carbon_free_target_pct': 100,
            'current_renewable_pct': 85,  # 2024
            'regions_renewable': ['us-west-2', 'eu-west-1', 'eu-north-1'],
            'pue': 1.2,
            'carbon_offset_program': True
        },
        'azure': {
            'name': 'Microsoft Azure',
            'carbon_negative_target': 2030,
            'carbon_negative_target_pct': 100,
            'current_renewable_pct': 90,
            'regions_renewable': ['west-europe', 'north-europe', 'west-us'],
            'pue': 1.18,
            'carbon_offset_program': True
        },
        'gcp': {
            'name': 'Google Cloud Platform',
            'carbon_free_target': 2030,
            'carbon_free_target_pct': 100,
            'current_renewable_pct': 95,  # Already highest
            'regions_renewable': ['us-central1', 'europe-west4', 'europe-north1'],
            'pue': 1.1,  # Industry-leading
            'carbon_offset_program': True
        }
    }

    def compare_providers(self) -> List[Dict]:
        """Compare sustainability across providers"""

        comparison = []

        for provider_id, data in self.PROVIDERS.items():
            comparison.append({
                'provider': provider_id,
                'name': data['name'],
                'renewable_pct_2024': data['current_renewable_pct'],
                'target_year': data.get('carbon_free_target') or data.get('carbon_negative_target'),
                'pue': data['pue'],
                'efficiency_rating': 'Excellent' if data['pue'] < 1.2 else 'Good'
            })

        return sorted(comparison, key=lambda x: x['renewable_pct_2024'], reverse=True)

    def recommend_region(
        self,
        provider: str,
        workload_type: str = 'training'
    ) -> Dict:
        """Recommend most sustainable region"""

        if provider not in self.PROVIDERS:
            return {'error': 'Provider not found'}

        provider_data = self.PROVIDERS[provider]
        recommended_regions = provider_data['regions_renewable']

        return {
            'provider': provider,
            'recommended_regions': recommended_regions,
            'reasoning': 'These regions have highest renewable energy percentage',
            'expected_carbon_savings_pct': 60  # vs coal-heavy regions
        }

# Usage
sustainability = CloudProviderSustainability()

comparison = sustainability.compare_providers()
print("=== CLOUD PROVIDER SUSTAINABILITY COMPARISON 2025 ===\n")

for provider in comparison:
    print(f"{provider['name']:30} Renewable: {provider['renewable_pct_2024']}%  "
          f"Target: {provider['target_year']}  PUE: {provider['pue']}")

# Get region recommendation
rec = sustainability.recommend_region('gcp', 'training')
print(f"\nRecommended GCP regions: {rec['recommended_regions']}")