#!/usr/bin/env python3 """ AI Energy Dashboard Integrator Integrates energy impact analysis with existing development intelligence dashboard to provide comprehensive AI cost and environmental impact insights. Author: Justin Murry Organization: unparty LLC License: MIT LLM: Claude Code """ import json import os import csv from datetime import datetime from pathlib import Path class AIEnergyDashboardIntegrator: def __init__(self, repo_root=None): self.repo_root = Path(repo_root) if repo_root else Path(__file__).resolve().parent.parent self.docs_dir = self.repo_root / 'docs' self.docs_dir.mkdir(exist_ok=True) def load_latest_energy_analysis(self): """Load the most recent energy analysis""" energy_files = list(self.docs_dir.glob('ai_energy_impact_analysis_*.json')) if not energy_files: print("❌ No energy analysis files found") return None latest_file = max(energy_files, key=lambda f: f.stat().st_mtime) print(f"📄 Loading energy analysis: {latest_file.name}") with open(latest_file, 'r') as f: return json.load(f) def load_claude_cost_data(self): """Load Claude API cost data for financial correlation""" cost_data = [] cost_file = self.repo_root / 'data' / 'claude_api_cost_master.csv' if os.path.exists(cost_file): with open(cost_file, 'r') as f: reader = csv.DictReader(f) for row in reader: cost_data.append(row) return cost_data def calculate_comprehensive_impact(self, energy_data, cost_data): """Calculate comprehensive AI impact metrics""" if not energy_data: return None # Financial impact from cost data total_financial_cost = sum(float(row.get('cost_usd', 0)) for row in cost_data) # Energy impact from energy analysis energy_summary = energy_data['energy_summary'] energy_tax = energy_data['energy_tax_calculation'] # Calculate total AI impact comprehensive_impact = { 'financial_metrics': { 'direct_api_cost_usd': round(total_financial_cost, 4), 'energy_cost_usd': energy_summary['total_energy_cost_usd'], 'energy_tax_usd': energy_tax['final_energy_tax_usd'], 'total_ai_cost_usd': round(total_financial_cost + energy_summary['total_energy_cost_usd'] + energy_tax['final_energy_tax_usd'], 4) }, 'environmental_metrics': { 'total_energy_kwh': energy_summary['total_energy_consumption_kwh'], 'carbon_footprint_kg': energy_summary['total_carbon_footprint_kg'], 'cache_efficiency_percent': energy_summary['average_cache_efficiency_percent'], 'energy_per_dollar_kwh': round(energy_summary['total_energy_consumption_kwh'] / total_financial_cost if total_financial_cost > 0 else 0, 6) }, 'efficiency_metrics': { 'cost_per_kwh_usd': round(total_financial_cost / energy_summary['total_energy_consumption_kwh'] if energy_summary['total_energy_consumption_kwh'] > 0 else 0, 4), 'carbon_intensity_per_dollar': round(energy_summary['total_carbon_footprint_kg'] / total_financial_cost if total_financial_cost > 0 else 0, 6), 'sessions_analyzed': energy_data['analysis_metadata']['total_sessions_analyzed'], 'avg_energy_per_session_kwh': energy_summary['energy_per_session_kwh'] } } return comprehensive_impact def generate_sustainability_recommendations(self, comprehensive_impact, energy_data): """Generate sustainability-focused recommendations""" recommendations = [] env_metrics = comprehensive_impact['environmental_metrics'] fin_metrics = comprehensive_impact['financial_metrics'] eff_metrics = comprehensive_impact['efficiency_metrics'] # Cache efficiency recommendations if env_metrics['cache_efficiency_percent'] < 50: recommendations.append({ 'category': 'Critical - Cache Optimization', 'impact': 'High Environmental Impact', 'recommendation': f"Cache efficiency at {env_metrics['cache_efficiency_percent']:.1f}% is below optimal. Improving to 70% could reduce energy by ~{env_metrics['total_energy_kwh'] * 0.3:.2f} kWh", 'potential_savings': { 'energy_kwh': round(env_metrics['total_energy_kwh'] * 0.3, 4), 'cost_usd': round(fin_metrics['energy_cost_usd'] * 0.3, 4), 'carbon_kg': round(env_metrics['carbon_footprint_kg'] * 0.3, 4) } }) # Model efficiency recommendations (family-based so it works for any # model generation reported by the analyzer, not just legacy IDs) model_breakdown = energy_data['model_breakdown'] haiku_models = {k: v for k, v in model_breakdown.items() if 'haiku' in k.lower()} heavy_models = {k: v for k, v in model_breakdown.items() if 'sonnet' in k.lower() or 'opus' in k.lower()} if haiku_models and heavy_models: haiku_energy = min(v['energy_per_session'] for v in haiku_models.values()) heaviest_key = max(heavy_models, key=lambda k: heavy_models[k]['energy_per_session']) heavy_stats = heavy_models[heaviest_key] heavy_energy = heavy_stats['energy_per_session'] if haiku_energy > 0 and heavy_energy > haiku_energy * 50: # Significant difference recommendations.append({ 'category': 'Model Selection Optimization', 'impact': 'Medium Environmental Impact', 'recommendation': f"{heaviest_key} uses {heavy_energy/haiku_energy:.1f}x more energy than Haiku per session. Consider Haiku for simpler tasks.", 'potential_savings': { 'energy_kwh': round(heavy_stats['total_energy_kwh'] * 0.3, 4), 'cost_usd': round(heavy_stats['total_cost_usd'] * 0.3, 4) if 'total_cost_usd' in heavy_stats else 0, 'carbon_kg': round(heavy_stats['total_carbon_kg'] * 0.3, 4) } }) # Financial efficiency recommendations if eff_metrics['carbon_intensity_per_dollar'] > 0.02: # High carbon per dollar recommendations.append({ 'category': 'Cost-Carbon Efficiency', 'impact': 'Medium Financial & Environmental', 'recommendation': f"Carbon intensity of {eff_metrics['carbon_intensity_per_dollar']:.6f} kg CO2/$1 suggests inefficient AI usage patterns.", 'potential_savings': { 'energy_kwh': round(env_metrics['total_energy_kwh'] * 0.2, 4), 'cost_usd': round(fin_metrics['total_ai_cost_usd'] * 0.15, 4), 'carbon_kg': round(env_metrics['carbon_footprint_kg'] * 0.2, 4) } }) # Session optimization if eff_metrics['avg_energy_per_session_kwh'] > 0.5: recommendations.append({ 'category': 'Session Optimization', 'impact': 'Low Environmental Impact', 'recommendation': f"Average session energy of {eff_metrics['avg_energy_per_session_kwh']:.6f} kWh suggests longer development sessions. Consider shorter, focused sessions.", 'potential_savings': { 'energy_kwh': round(env_metrics['total_energy_kwh'] * 0.1, 4), 'cost_usd': round(fin_metrics['total_ai_cost_usd'] * 0.1, 4), 'carbon_kg': round(env_metrics['carbon_footprint_kg'] * 0.1, 4) } }) return recommendations def create_energy_dashboard_summary(self): """Create integrated energy dashboard summary""" print("🌱 Creating AI Energy Dashboard Integration...") # Load data energy_data = self.load_latest_energy_analysis() cost_data = self.load_claude_cost_data() if not energy_data: print("❌ Cannot proceed without energy analysis data") return None # Calculate comprehensive impact comprehensive_impact = self.calculate_comprehensive_impact(energy_data, cost_data) # Generate recommendations recommendations = self.generate_sustainability_recommendations(comprehensive_impact, energy_data) # Create integrated dashboard data dashboard_data = { 'dashboard_metadata': { 'generated_at': datetime.now().isoformat(), 'energy_analysis_source': 'ai_energy_impact_analyzer.py', 'integration_version': '1.0', 'analysis_period': energy_data['analysis_metadata']['date_range'] }, 'comprehensive_ai_impact': comprehensive_impact, 'sustainability_recommendations': recommendations, 'energy_analysis_summary': { 'total_sessions': energy_data['analysis_metadata']['total_sessions_analyzed'], 'model_breakdown': energy_data['model_breakdown'], 'daily_trends_count': len(energy_data['daily_trends']), 'optimization_opportunities': len(energy_data['optimization_recommendations']) }, 'key_insights': self._extract_key_insights(comprehensive_impact, energy_data) } # Save integrated dashboard data dashboard_file = self.docs_dir / f'ai_energy_dashboard_integration_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json' with open(dashboard_file, 'w') as f: json.dump(dashboard_data, f, indent=2) # Generate markdown dashboard summary self._generate_dashboard_markdown(dashboard_data) print(f"✅ Energy dashboard integration complete!") print(f"💰 Total AI Cost: ${comprehensive_impact['financial_metrics']['total_ai_cost_usd']:.4f}") print(f"🌍 Carbon Footprint: {comprehensive_impact['environmental_metrics']['carbon_footprint_kg']:.4f} kg CO2") print(f"⚡ Energy Consumption: {comprehensive_impact['environmental_metrics']['total_energy_kwh']:.4f} kWh") print(f"🔋 Cache Efficiency: {comprehensive_impact['environmental_metrics']['cache_efficiency_percent']:.1f}%") print(f"📊 Recommendations: {len(recommendations)} optimization opportunities") print(f"📄 Dashboard saved: {dashboard_file}") return dashboard_data def _extract_key_insights(self, comprehensive_impact, energy_data): """Extract key insights for dashboard""" insights = [] env_metrics = comprehensive_impact['environmental_metrics'] fin_metrics = comprehensive_impact['financial_metrics'] # Energy tax insight direct_cost = fin_metrics['direct_api_cost_usd'] if direct_cost > 0: energy_tax_percent = (fin_metrics['energy_tax_usd'] / direct_cost) * 100 insights.append(f"Energy tax represents {energy_tax_percent:.1f}% of direct API costs") else: insights.append("No direct API cost data available for energy tax comparison") # Cache efficiency insight if env_metrics['cache_efficiency_percent'] > 60: insights.append(f"Good cache efficiency at {env_metrics['cache_efficiency_percent']:.1f}% reducing energy consumption") else: insights.append(f"Cache efficiency at {env_metrics['cache_efficiency_percent']:.1f}% has room for improvement") # Model usage insight model_breakdown = energy_data['model_breakdown'] most_used_model = max(model_breakdown.keys(), key=lambda k: model_breakdown[k]['sessions']) most_energy_model = max(model_breakdown.keys(), key=lambda k: model_breakdown[k]['total_energy_kwh']) if most_used_model != most_energy_model: insights.append(f"Most used model ({most_used_model}) differs from highest energy model ({most_energy_model})") # Financial efficiency insight if env_metrics['total_energy_kwh'] > 0: cost_per_kwh = fin_metrics['direct_api_cost_usd'] / env_metrics['total_energy_kwh'] insights.append(f"AI development efficiency: ${cost_per_kwh:.2f} per kWh of energy consumption") return insights def _generate_dashboard_markdown(self, dashboard_data): """Generate markdown dashboard summary""" impact = dashboard_data['comprehensive_ai_impact'] recommendations = dashboard_data['sustainability_recommendations'] insights = dashboard_data['key_insights'] md_content = f"""# AI Energy Impact Dashboard *Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}* ## 🎯 Executive Dashboard ### 💰 Comprehensive AI Cost Analysis | Component | Cost (USD) | |-----------|------------| | Direct API Usage | ${impact['financial_metrics']['direct_api_cost_usd']:.4f} | | Energy Cost | ${impact['financial_metrics']['energy_cost_usd']:.4f} | | Energy Tax | ${impact['financial_metrics']['energy_tax_usd']:.4f} | | **Total AI Investment** | **${impact['financial_metrics']['total_ai_cost_usd']:.4f}** | ### 🌍 Environmental Impact | Metric | Value | |--------|--------| | Energy Consumption | {impact['environmental_metrics']['total_energy_kwh']:.4f} kWh | | Carbon Footprint | {impact['environmental_metrics']['carbon_footprint_kg']:.4f} kg CO2 | | Cache Efficiency | {impact['environmental_metrics']['cache_efficiency_percent']:.1f}% | | Energy per Dollar | {impact['environmental_metrics']['energy_per_dollar_kwh']:.6f} kWh/$1 | ### ⚡ Efficiency Metrics | Metric | Value | |--------|--------| | Cost per kWh | ${impact['efficiency_metrics']['cost_per_kwh_usd']:.4f} | | Carbon per Dollar | {impact['efficiency_metrics']['carbon_intensity_per_dollar']:.6f} kg CO2/$1 | | Avg Energy per Session | {impact['efficiency_metrics']['avg_energy_per_session_kwh']:.6f} kWh | | Sessions Analyzed | {impact['efficiency_metrics']['sessions_analyzed']} | ## 🎯 Sustainability Recommendations """ for i, rec in enumerate(recommendations, 1): savings = rec.get('potential_savings', {}) md_content += f"""### {i}. {rec['category']} **Impact Level**: {rec['impact']} {rec['recommendation']} **Potential Savings:** - Energy: {savings.get('energy_kwh', 0):.4f} kWh - Cost: ${savings.get('cost_usd', 0):.4f} - Carbon: {savings.get('carbon_kg', 0):.4f} kg CO2 """ md_content += f"""## 💡 Key Insights """ for insight in insights: md_content += f"- {insight}\n" md_content += f""" ## 📊 Integration Summary This dashboard integrates: - **Energy Analysis**: {dashboard_data['energy_analysis_summary']['total_sessions']} AI sessions analyzed - **Model Performance**: {len(dashboard_data['energy_analysis_summary']['model_breakdown'])} different models tracked - **Optimization Opportunities**: {dashboard_data['energy_analysis_summary']['optimization_opportunities']} technical recommendations - **Sustainability Focus**: {len(recommendations)} sustainability-focused improvements ## 🌱 Environmental Commitment Our AI energy tax system encourages: - ✅ High cache efficiency (>60% gets 20% discount) - ✅ Model selection optimization - ✅ Session efficiency improvements - ✅ Environmental impact transparency --- *This dashboard provides comprehensive AI cost and environmental impact analysis to support sustainable development practices.* """ # Save markdown dashboard md_file = self.docs_dir / f'ai_energy_dashboard_summary_{datetime.now().strftime("%Y%m%d_%H%M%S")}.md' with open(md_file, 'w') as f: f.write(md_content) print(f"📄 Dashboard summary saved: {md_file}") def main(): """Main execution function""" integrator = AIEnergyDashboardIntegrator() dashboard_data = integrator.create_energy_dashboard_summary() return dashboard_data if __name__ == "__main__": main()