""" Evaluation Metrics — Loss reduction, CO₂, cost, voltage, speedup. All metrics are computed deterministically from before/after power flow results. """ from __future__ import annotations from config import CFG def compute_impact( baseline: dict, optimized: dict, hours_per_year: int | None = None, emission_factor: float | None = None, electricity_price: float | None = None, ) -> dict: """Compute the full impact comparison between baseline and optimized results. Parameters ---------- baseline, optimized : dict Output of ``power_flow.extract_results()``. hours_per_year : int, optional emission_factor : float, optional (kg CO₂ / kWh) electricity_price : float, optional (USD / kWh) Returns ------- dict with all impact metrics. """ cfg = CFG.impact hours = hours_per_year or cfg.hours_per_year ef = emission_factor or cfg.emission_factor ep = electricity_price or cfg.electricity_price base_kw = baseline["total_loss_kw"] opt_kw = optimized["total_loss_kw"] saved_kw = base_kw - opt_kw loss_reduction_pct = (saved_kw / base_kw * 100) if base_kw > 0 else 0.0 # Annualised values saved_kwh_year = saved_kw * hours saved_mwh_year = saved_kwh_year / 1000 co2_saved_kg_year = saved_kwh_year * ef co2_saved_tonnes_year = co2_saved_kg_year / 1000 cost_saved_year = saved_kwh_year * ep # Voltage improvement base_violations = baseline["voltage_violations"] opt_violations = optimized["voltage_violations"] voltage_improvement = base_violations - opt_violations return { # --- Loss reduction --- "baseline_loss_kw": round(base_kw, 2), "optimized_loss_kw": round(opt_kw, 2), "loss_reduction_kw": round(saved_kw, 2), "loss_reduction_pct": round(loss_reduction_pct, 2), # --- Annualised impact --- "energy_saved_mwh_year": round(saved_mwh_year, 2), "co2_saved_tonnes_year": round(co2_saved_tonnes_year, 2), "cost_saved_usd_year": round(cost_saved_year, 2), # --- Voltage --- "baseline_voltage_violations": base_violations, "optimized_voltage_violations": opt_violations, "voltage_violations_fixed": voltage_improvement, "baseline_min_voltage": baseline["min_voltage_pu"], "optimized_min_voltage": optimized["min_voltage_pu"], # --- Equivalences (for presentation) --- "equivalent_trees_planted": int(co2_saved_kg_year / 21), # ~21 kg CO₂/tree/year "equivalent_cars_removed": round(co2_saved_tonnes_year / 4.6, 1), # ~4.6 t CO₂/car/year } def compute_speedup(classical_time_sec: float, hybrid_time_sec: float) -> dict: """Compute speedup metrics between classical and hybrid solvers.""" speedup = classical_time_sec / hybrid_time_sec if hybrid_time_sec > 0 else float("inf") return { "classical_time_sec": round(classical_time_sec, 4), "hybrid_time_sec": round(hybrid_time_sec, 4), "speedup_factor": round(speedup, 1), } # --------------------------------------------------------------------------- # Solution Energy Footprint # --------------------------------------------------------------------------- def compute_solution_footprint( computation_time_sec: float, server_tdp_watts: float = 350.0, emission_factor: float | None = None, ) -> dict: """Estimate the energy and CO₂ cost of running the optimisation itself. Parameters ---------- computation_time_sec : float Wall-clock time of the optimisation run (seconds). server_tdp_watts : float Thermal Design Power of the server (CPU + GPU). Default 350 W is a conservative estimate for a workstation with GPU. emission_factor : float, optional kg CO₂ per kWh. Falls back to global average from config. Returns ------- dict with solution energy (kWh), CO₂ (kg), and context. """ cfg = CFG.impact ef = emission_factor or cfg.emission_factor energy_kwh = (server_tdp_watts * computation_time_sec) / 3_600_000 co2_kg = energy_kwh * ef return { "computation_time_sec": round(computation_time_sec, 4), "server_tdp_watts": server_tdp_watts, "solution_energy_kwh": round(energy_kwh, 6), "solution_co2_kg": round(co2_kg, 6), "emission_factor_used": ef, } def compute_net_benefit( impact: dict, footprint: dict, ) -> dict: """Frame the solution's impact as waste elimination, not solution-vs-cost. The correct comparison is: - Before: the grid wastes X kWh/year as heat in distribution lines. - After: the grid wastes (X - saved) kWh/year. - The solution itself consumes negligible energy (software on a server). Parameters ---------- impact : dict Output of ``compute_impact()``. footprint : dict Output of ``compute_solution_footprint()``. Returns ------- dict with waste elimination framing, solution overhead, and trustworthiness. """ cfg = CFG.impact baseline_waste_kwh_year = impact["baseline_loss_kw"] * cfg.hours_per_year optimized_waste_kwh_year = impact["optimized_loss_kw"] * cfg.hours_per_year saved_kwh_year = impact["energy_saved_mwh_year"] * 1000 waste_eliminated_pct = impact["loss_reduction_pct"] solution_kwh_per_run = footprint["solution_energy_kwh"] # Dynamic reconfiguration runs every 15 minutes runs_per_year = 365 * 24 * 4 # 35,040 runs total_solution_kwh_year = solution_kwh_per_run * runs_per_year # Solution overhead as % of savings (should be negligible) overhead_pct = (total_solution_kwh_year / saved_kwh_year * 100) if saved_kwh_year > 0 else 0 co2_saved_kg = impact["co2_saved_tonnes_year"] * 1000 co2_cost_kg = footprint["solution_co2_kg"] * runs_per_year return { # --- Waste elimination framing --- "baseline_waste_kwh_year": round(baseline_waste_kwh_year, 0), "optimized_waste_kwh_year": round(optimized_waste_kwh_year, 0), "waste_eliminated_kwh_year": round(saved_kwh_year, 0), "waste_eliminated_pct": round(waste_eliminated_pct, 2), # --- Solution overhead (negligible) --- "solution_energy_kwh_year": round(total_solution_kwh_year, 2), "solution_overhead_pct_of_savings": round(overhead_pct, 4), "runs_per_year": runs_per_year, # --- CO₂ --- "co2_eliminated_kg_year": round(co2_saved_kg, 2), "solution_co2_kg_year": round(co2_cost_kg, 4), # --- Trustworthiness --- "trustworthiness": ( "Energy savings are computed from pandapower's Newton-Raphson AC " "power flow — an industry-standard, physics-validated solver used " "by grid operators worldwide. The loss values are derived from " "Kirchhoff's laws and validated line impedances, not approximations. " "Annualisation assumes constant load; real-world savings are " "~60-80% of this figure due to load variation. " f"Solution computational overhead is {overhead_pct:.4f}% of savings " "(effectively zero)." ), } # --------------------------------------------------------------------------- # Business Model / Pricing # --------------------------------------------------------------------------- def compute_business_model( impact: dict, n_feeders_pilot: int = 10, n_feeders_city: int = 5000, ) -> dict: """Compute pricing and revenue projections for a utility deployment. Parameters ---------- impact : dict Output of ``compute_impact()`` for a single feeder. n_feeders_pilot : int Number of feeders in Phase 1 pilot. n_feeders_city : int Number of feeders in a city-wide deployment (Cairo estimate). Returns ------- dict with pricing models, revenue projections, and competitive analysis. """ eg = CFG.egypt savings_per_feeder_year_real = impact["energy_saved_mwh_year"] * 1000 * eg.electricity_price_real savings_per_feeder_year_sub = impact["energy_saved_mwh_year"] * 1000 * eg.electricity_price_subsidised return { "usage_model": { "type": "Recurring SaaS — NOT one-time", "unit": "Per feeder (a feeder is one radial distribution circuit, " "typically 20-40 buses, serving 500-5,000 customers)", "frequency": "Continuous — runs every 15-60 minutes with live SCADA data", "why_recurring": ( "Load patterns change hourly (morning peak, evening peak), " "seasonally (summer AC in Egypt doubles demand), and with new " "connections. The optimal switch configuration changes with load. " "Static one-time reconfiguration captures only ~40% of the benefit " "vs dynamic recurring optimisation." ), }, "savings_per_feeder": { "energy_saved_kwh_year": round(impact["energy_saved_mwh_year"] * 1000, 0), "cost_saved_year_subsidised_usd": round(savings_per_feeder_year_sub, 0), "cost_saved_year_real_cost_usd": round(savings_per_feeder_year_real, 0), "co2_saved_tonnes_year": impact["co2_saved_tonnes_year"], }, "pricing_models": { "model_a_saas": { "name": "SaaS Subscription", "price_per_feeder_month_usd": 200, "price_per_feeder_year_usd": 2400, "value_proposition": ( f"Feeder saves ${savings_per_feeder_year_real:,.0f}/year at real cost. " f"License costs $2,400/year = {2400/savings_per_feeder_year_real*100:.1f}% of savings. " "Payback: immediate." ), }, "model_b_revenue_share": { "name": "Revenue Share", "share_pct": 15, "revenue_per_feeder_year_usd": round(savings_per_feeder_year_real * 0.15, 0), "value_proposition": "No upfront cost. Utility pays 15% of verified savings.", }, "model_c_enterprise": { "name": "Enterprise License", "price_per_year_usd": 500_000, "covers_feeders_up_to": 1000, "effective_per_feeder_usd": 500, "value_proposition": "Flat annual license for large utilities.", }, }, "revenue_projections": { "pilot_phase": { "n_feeders": n_feeders_pilot, "annual_revenue_saas": n_feeders_pilot * 2400, "annual_savings_to_utility_real": round( n_feeders_pilot * savings_per_feeder_year_real, 0 ), }, "city_phase_cairo": { "n_feeders": n_feeders_city, "annual_revenue_saas": n_feeders_city * 2400, "annual_savings_to_utility_real": round( n_feeders_city * savings_per_feeder_year_real, 0 ), }, }, "comparison_to_alternatives": { "manual_switching": { "method": "Operator manually changes switch positions quarterly/yearly", "loss_reduction": "5-10%", "cost": "Zero software cost, but high labour + suboptimal results", "limitation": "Cannot adapt to load changes. Human error. Slow.", }, "full_adms": { "method": "ABB/Siemens/GE Advanced Distribution Management System", "loss_reduction": "15-25%", "cost": "$5-50 million for full deployment + annual maintenance", "limitation": ( "Massive CAPEX. 12-24 month deployment. Requires new SCADA " "hardware. Reconfiguration is one small module in a huge platform." ), }, "optiq": { "method": "OptiQ Hybrid Quantum-AI-Classical SaaS", "loss_reduction": "28-32% (matches published global optimal)", "cost": "$200/feeder/month or 15% revenue share", "advantage": ( "Software-only — works on existing SCADA infrastructure. " "No CAPEX. Deploys in weeks, not years. Achieves global " "optimum via physics-informed AI + quantum-inspired search, " "while ADMS typically uses simple heuristics. " "10-100x cheaper than full ADMS deployment." ), }, }, } # --------------------------------------------------------------------------- # Egypt / Scaling Impact # --------------------------------------------------------------------------- def compute_egypt_impact( loss_reduction_pct: float, ) -> dict: """Extrapolate IEEE 33-bus loss reduction to Egypt and global scale. Parameters ---------- loss_reduction_pct : float Percentage loss reduction achieved on the benchmark (e.g. 31.15). Returns ------- dict with Egypt-specific and global impact projections. """ eg = CFG.egypt reduction_frac = loss_reduction_pct / 100.0 # --- Egypt --- egypt_dist_loss_twh = eg.total_generation_twh * eg.dist_loss_fraction egypt_savings_twh = egypt_dist_loss_twh * reduction_frac egypt_savings_gwh = egypt_savings_twh * 1000 egypt_savings_kwh = egypt_savings_twh * 1e9 # 1 TWh = 1e9 kWh egypt_co2_saved_mt = egypt_savings_kwh * eg.emission_factor / 1e9 # million tonnes egypt_cost_saved_subsidised = egypt_savings_kwh * eg.electricity_price_subsidised egypt_cost_saved_real = egypt_savings_kwh * eg.electricity_price_real # Cairo-specific cairo_savings_twh = egypt_savings_twh * eg.cairo_consumption_share cairo_co2_saved_mt = egypt_co2_saved_mt * eg.cairo_consumption_share # As a percentage of Egypt total generation egypt_impact_pct = (egypt_savings_twh / eg.total_generation_twh) * 100 # --- Global --- global_dist_loss_twh = eg.global_generation_twh * eg.global_dist_loss_fraction global_savings_twh = global_dist_loss_twh * reduction_frac global_savings_kwh = global_savings_twh * 1e9 # 1 TWh = 1e9 kWh global_co2_saved_mt = global_savings_kwh * CFG.impact.emission_factor / 1e9 global_impact_pct = (global_savings_twh / eg.global_generation_twh) * 100 return { "loss_reduction_pct_applied": round(loss_reduction_pct, 2), # --- Egypt --- "egypt": { "total_generation_twh": eg.total_generation_twh, "distribution_losses_twh": round(egypt_dist_loss_twh, 2), "potential_savings_twh": round(egypt_savings_twh, 2), "potential_savings_gwh": round(egypt_savings_gwh, 1), "co2_saved_million_tonnes": round(egypt_co2_saved_mt, 3), "cost_saved_usd_subsidised": round(egypt_cost_saved_subsidised, 0), "cost_saved_usd_real": round(egypt_cost_saved_real, 0), "impact_pct_of_generation": round(egypt_impact_pct, 2), "emission_factor": eg.emission_factor, }, "cairo": { "potential_savings_twh": round(cairo_savings_twh, 3), "co2_saved_million_tonnes": round(cairo_co2_saved_mt, 4), "share_of_national": eg.cairo_consumption_share, }, # --- Global --- "global": { "total_generation_twh": eg.global_generation_twh, "distribution_losses_twh": round(global_dist_loss_twh, 1), "potential_savings_twh": round(global_savings_twh, 1), "co2_saved_million_tonnes": round(global_co2_saved_mt, 1), "impact_pct_of_generation": round(global_impact_pct, 3), }, # --- Implementation plan (Egypt-specific) --- "implementation_plan": { "target_partners": [ "North Cairo Electricity Distribution Company (NCEDC) — " "already deploying 500,000 smart meters with Iskraemeco", "South Cairo Electricity Distribution Company", "Egyptian Electricity Holding Company (EEHC) — parent of all 9 regional companies", ], "phase_0_mvp": { "timeline": "Now (completed)", "deliverable": "IEEE benchmark validated, matches published global optimal", "cost": "$0 (open-source tools, no hardware)", }, "phase_1_pilot": { "timeline": "3-6 months", "scope": "5-10 feeders in one NCEDC substation", "steps": [ "1. Partner with NCEDC (they already have SCADA + smart meters)", "2. Get read-only access to SCADA data for 5-10 feeders " "(bus loads, switch states, voltage readings)", "3. Map their feeder topology to pandapower format " "(line impedances from utility records, bus loads from SCADA)", "4. Run OptiQ in shadow mode: compute optimal switch positions " "but do NOT actuate — compare recommendations vs operator decisions", "5. After 1 month of shadow mode proving accuracy, " "actuate switches on 1-2 feeders with motorised switches", ], "hardware_needed": "None — uses existing SCADA. Runs on a standard cloud VM.", "cost": "$10,000-20,000 (cloud hosting + integration labour)", }, "phase_2_district": { "timeline": "6-12 months after pilot", "scope": "100+ feeders across one distribution company", "steps": [ "1. Automate SCADA data pipeline (real-time feed every 15 min)", "2. Deploy on all feeders in one NCEDC district", "3. Add motorised switches where manual-only exists (~$2,000 per switch)", "4. Measure and verify savings against utility billing data", ], "cost": "$50,000-100,000 (software + switch upgrades where needed)", }, "phase_3_city": { "timeline": "1-2 years", "scope": "City-wide Cairo (~5,000+ feeders across NCEDC + SCEDC)", "cost": "$500,000-1,000,000 (enterprise license + integration)", }, "phase_4_national": { "timeline": "2-3 years", "scope": "All 9 distribution companies across Egypt", "cost": "$2-5 million (national enterprise license)", }, }, } def count_dependent_variables(net=None) -> dict: """Count all variables the solution depends on. Returns a structured breakdown of physical, algorithmic, and external variables for the hackathon validation question. """ physical = { "bus_loads_p": 33, "bus_loads_q": 33, "line_resistance": 37, "line_reactance": 37, "switch_states_binary": 5, "bus_voltages_state": 33, } algorithmic = { "quantum_reps": 1, "quantum_shots": 1, "quantum_top_k": 1, "quantum_penalties": 2, "quantum_sa_iters": 1, "quantum_sa_restarts": 1, "quantum_sa_temperature": 2, "gnn_hidden_dim": 1, "gnn_layers": 1, "gnn_dropout": 1, "gnn_lr": 1, "gnn_epochs": 1, "gnn_batch_size": 1, "physics_loss_weights": 3, "dual_lr": 1, "n_scenarios": 1, } external = { "emission_factor": 1, "electricity_price": 1, "hours_per_year": 1, } total_physical = sum(physical.values()) total_algo = sum(algorithmic.values()) total_ext = sum(external.values()) return { "physical_variables": physical, "algorithmic_hyperparameters": algorithmic, "external_assumptions": external, "totals": { "physical": total_physical, "algorithmic": total_algo, "external": total_ext, "grand_total": total_physical + total_algo + total_ext, }, "decision_variables": 5, "note": ( "Of ~200 total variables, only 5 are decision variables " "(which lines to open/close). The rest are grid physics " "parameters (~178) and tunable hyperparameters (~20)." ), }