#!/usr/bin/env python """ OptiQ Benchmark — Compare all methods against published IEEE 33-bus results. Published benchmark values (Baran & Wu 1989, widely cited): Base case losses: ~202.7 kW Optimal (literature): ~139.55 kW (31.2% reduction) This script: 1. Runs all three methods (Classical, Quantum SA, Hybrid) on IEEE 33-bus 2. Compares against published optimal values 3. Tests across multiple load levels (multi-scenario) 4. Computes solution energy footprint and net benefit 5. Computes Egypt-specific and global scaling impact 6. Outputs a formatted results table for the hackathon presentation Usage: conda run -n projects python scripts/benchmark.py """ from __future__ import annotations import json import os import sys import time # Ensure project root is on path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pandapower as pp from config import CFG from src.grid.loader import load_network, clone_network, get_line_info from src.grid.power_flow import get_baseline, evaluate_topology from src.grid.reconfiguration import branch_exchange_search from src.quantum.qaoa_reconfig import solve_sa from src.hybrid.pipeline import run_hybrid_pipeline from src.evaluation.metrics import ( compute_impact, compute_speedup, compute_solution_footprint, compute_net_benefit, compute_egypt_impact, compute_business_model, count_dependent_variables, ) # --------------------------------------------------------------------------- # Published benchmark values # --------------------------------------------------------------------------- PUBLISHED = { "case33bw": { "base_loss_kw": 202.67, "optimal_loss_kw": 139.55, "optimal_reduction_pct": 31.15, "optimal_open_switches": "7, 9, 14, 32, 37", "source": "Baran & Wu 1989, widely reproduced (PSO, GA, MILP, Branch Exchange)", } } def divider(title: str) -> None: print(f"\n{'='*70}") print(f" {title}") print(f"{'='*70}") def run_single_system_benchmark(system: str = "case33bw") -> dict: """Run full benchmark on one system and return structured results.""" divider(f"IEEE {system} Benchmark") net = load_network(system) published = PUBLISHED.get(system, {}) # --- Baseline --- print("\n[1/4] Computing baseline...") t0 = time.perf_counter() baseline = get_baseline(net) baseline_time = time.perf_counter() - t0 if not baseline.get("converged"): print(" ERROR: Baseline power flow did not converge!") return {"error": "baseline_failed"} print(f" Baseline losses: {baseline['total_loss_kw']:.2f} kW") print(f" Published baseline: {published.get('base_loss_kw', 'N/A')} kW") print(f" Min voltage: {baseline['min_voltage_pu']:.4f} p.u.") print(f" Voltage violations: {baseline['voltage_violations']}") results = { "system": system, "baseline": baseline, "published": published, "methods": {}, } # --- Method 1: Classical Branch Exchange --- print("\n[2/4] Running Classical Branch Exchange...") t0 = time.perf_counter() classical = branch_exchange_search(net, verbose=True) t_classical = time.perf_counter() - t0 if "error" not in classical: ev = evaluate_topology(net, classical["best_open_lines"]) if ev.get("converged"): impact = compute_impact(baseline, ev) results["methods"]["classical"] = { "open_lines": classical["best_open_lines"], "loss_kw": ev["total_loss_kw"], "min_voltage": ev["min_voltage_pu"], "violations": ev["voltage_violations"], "reduction_pct": impact["loss_reduction_pct"], "time_sec": round(t_classical, 3), "impact": impact, } print(f" Best loss: {ev['total_loss_kw']:.2f} kW " f"({impact['loss_reduction_pct']:.2f}% reduction)") print(f" Open lines: {classical['best_open_lines']}") print(f" Time: {t_classical:.3f}s") else: print(f" ERROR: {classical.get('error')}") # --- Method 2: Quantum SA --- print("\n[3/4] Running Quantum-Inspired SA...") t0 = time.perf_counter() sa_result = solve_sa(net, n_iter=500, n_restarts=5, top_k=5) t_quantum = time.perf_counter() - t0 if "error" not in sa_result: ev = evaluate_topology(net, sa_result["best_open_lines"]) if ev.get("converged"): impact = compute_impact(baseline, ev) results["methods"]["quantum_sa"] = { "open_lines": sa_result["best_open_lines"], "loss_kw": ev["total_loss_kw"], "min_voltage": ev["min_voltage_pu"], "violations": ev["voltage_violations"], "reduction_pct": impact["loss_reduction_pct"], "time_sec": round(t_quantum, 3), "n_evaluated": sa_result["n_evaluated"], "impact": impact, } print(f" Best loss: {ev['total_loss_kw']:.2f} kW " f"({impact['loss_reduction_pct']:.2f}% reduction)") print(f" Open lines: {sa_result['best_open_lines']}") print(f" Evaluated: {sa_result['n_evaluated']} topologies") print(f" Time: {t_quantum:.3f}s") else: print(f" ERROR: {sa_result.get('error')}") # --- Method 3: Full Hybrid Pipeline --- print("\n[4/4] Running Full Hybrid Pipeline (Quantum + AI + Classical)...") t0 = time.perf_counter() hybrid = run_hybrid_pipeline( net, use_quantum=True, use_ai=True, verbose=True ) t_hybrid = time.perf_counter() - t0 if "error" not in hybrid: opt = hybrid["optimized"] impact = hybrid["impact"] results["methods"]["hybrid"] = { "open_lines": opt.get("open_lines"), "loss_kw": opt["total_loss_kw"], "min_voltage": opt["min_voltage_pu"], "violations": opt["voltage_violations"], "reduction_pct": impact["loss_reduction_pct"], "time_sec": round(t_hybrid, 3), "timings": hybrid.get("timings"), "impact": impact, } print(f" Best loss: {opt['total_loss_kw']:.2f} kW " f"({impact['loss_reduction_pct']:.2f}% reduction)") print(f" Open lines: {opt.get('open_lines')}") print(f" Time: {t_hybrid:.3f}s") else: print(f" NOTE: {hybrid.get('error')}") return results def run_multi_load_benchmark(system: str = "case33bw") -> dict: """Run optimisation across multiple load multipliers.""" divider("Multi-Load Scenario Testing") load_multipliers = [0.7, 0.85, 1.0, 1.15, 1.3] net_base = load_network(system) scenario_results = [] for mult in load_multipliers: net = clone_network(net_base) net.load["p_mw"] *= mult net.load["q_mvar"] *= mult baseline = get_baseline(net) if not baseline.get("converged"): print(f" Load x{mult}: Baseline FAILED") continue sa = solve_sa(net, n_iter=300, n_restarts=3, top_k=3) if "error" in sa: print(f" Load x{mult}: SA FAILED") continue ev = evaluate_topology(net, sa["best_open_lines"]) if not ev.get("converged"): print(f" Load x{mult}: Topology evaluation FAILED") continue impact = compute_impact(baseline, ev) entry = { "load_multiplier": mult, "baseline_loss_kw": baseline["total_loss_kw"], "optimized_loss_kw": ev["total_loss_kw"], "reduction_pct": impact["loss_reduction_pct"], "min_voltage_before": baseline["min_voltage_pu"], "min_voltage_after": ev["min_voltage_pu"], "open_lines": sa["best_open_lines"], } scenario_results.append(entry) print(f" Load x{mult:.2f}: {baseline['total_loss_kw']:.1f} -> " f"{ev['total_loss_kw']:.1f} kW ({impact['loss_reduction_pct']:.1f}% reduction)") return {"load_scenarios": scenario_results} def print_comparison_table(results: dict) -> None: """Print a formatted comparison table with published methods.""" divider("COMPARISON TABLE: OptiQ vs Published Algorithms (IEEE 33-bus)") published = results.get("published", {}) baseline = results.get("baseline", {}) methods = results.get("methods", {}) # --- Table A: All algorithms --- print(f"\n{'Method':<40} {'Loss (kW)':>10} {'Reduction':>10} {'Source':>12}") print("-" * 74) # Baseline bl_kw = baseline.get("total_loss_kw", 202.68) print(f"{'Baseline (no reconfiguration)':<40} {bl_kw:>10.2f} {'—':>10} {'[1]':>12}") # Published methods from literature (hardcoded from REFERENCES.md) lit_methods = [ ("Civanlar load-transfer (1988)", 146.0, 28.0, "[2]"), ("PSO (Sulaima 2014, local opt.)", 146.1, 27.9, "[5]"), ("Baran & Wu branch exch. (1989)", 139.55, 31.15, "[1]"), ("Goswami & Basu (1992)", 139.55, 31.15, "[3]"), ("GA (well-tuned, multiple)", 139.55, 31.15, "[7]"), ("MILP exact (Jabr 2012)", 139.55, 31.15, "[4]"), ("Br.Exch + Cluster (Pereira 2023)", 139.55, 31.15, "[6]"), ] for name, loss_kw, red_pct, source in lit_methods: print(f"{name:<40} {loss_kw:>10.2f} {red_pct:>9.2f}% {source:>12}") # Our methods print("-" * 74) for name, data in methods.items(): label = { "classical": "OptiQ Classical", "quantum_sa": "OptiQ Quantum SA", "hybrid": "OptiQ Hybrid", }.get(name, name) print(f"{label:<40} {data['loss_kw']:>10.2f} " f"{data['reduction_pct']:>9.2f}% {'this work':>12}") print() # --- Table B: Industry practice --- divider("COMPARISON TABLE: OptiQ vs Industry Practice") print(f"\n{'Solution':<40} {'Loss Reduction':>15} {'Cost':>20}") print("-" * 77) print(f"{'Manual switching (Egypt status quo)':<40} {'5-10% [9]':>15} {'$0 software':>20}") print(f"{'Basic ADMS (ABB/Siemens/GE)':<40} {'15-25% [9][22]':>15} {'$5-50M [22]':>20}") print(f"{'OptiQ':<40} {'28-32%':>15} {'$200/feeder/mo':>20}") print(f"\n Sources: see REFERENCES.md") print() def print_multi_load_table(multi: dict) -> None: """Print multi-load scenario results.""" divider("MULTI-LOAD SCENARIO RESULTS") scenarios = multi.get("load_scenarios", []) if not scenarios: print(" No scenarios completed.") return print(f"\n{'Load Mult':>10} {'Base Loss':>10} {'Opt Loss':>10} " f"{'Reduction':>10} {'V_min Before':>12} {'V_min After':>12}") print("-" * 68) for s in scenarios: print(f"{s['load_multiplier']:>10.2f} " f"{s['baseline_loss_kw']:>10.2f} " f"{s['optimized_loss_kw']:>10.2f} " f"{s['reduction_pct']:>9.2f}% " f"{s['min_voltage_before']:>12.4f} " f"{s['min_voltage_after']:>12.4f}") print() def print_impact_analysis(results: dict) -> None: """Print solution footprint and scaling impact.""" # Find the best method's results methods = results.get("methods", {}) best_method = None best_loss = float("inf") for name, data in methods.items(): if data["loss_kw"] < best_loss: best_loss = data["loss_kw"] best_method = name if not best_method: print(" No successful methods to analyse.") return data = methods[best_method] impact = data["impact"] # Solution footprint — framed as waste elimination divider("WASTE ELIMINATION ANALYSIS") footprint = compute_solution_footprint(data["time_sec"]) net_benefit = compute_net_benefit(impact, footprint) print(f"\n Best method: {best_method}") print(f"\n --- Energy Waste (per feeder) ---") print(f" Before OptiQ: {net_benefit['baseline_waste_kwh_year']:,.0f} kWh/year wasted as heat") print(f" After OptiQ: {net_benefit['optimized_waste_kwh_year']:,.0f} kWh/year wasted") print(f" Waste eliminated: {net_benefit['waste_eliminated_kwh_year']:,.0f} kWh/year " f"({net_benefit['waste_eliminated_pct']:.1f}%)") print(f"\n --- Solution Overhead ---") print(f" Computation time: {footprint['computation_time_sec']:.3f} s per run") print(f" Solution energy/year: {net_benefit['solution_energy_kwh_year']:.2f} kWh " f"({net_benefit['solution_overhead_pct_of_savings']:.4f}% of savings — effectively zero)") print(f" CO2 eliminated/year: {net_benefit['co2_eliminated_kg_year']:,.0f} kg") print(f" Solution CO2/year: {net_benefit['solution_co2_kg_year']:.2f} kg") print(f"\n --- Trustworthiness ---") print(f" {net_benefit['trustworthiness']}") # Egypt + Global scaling divider("EGYPT & GLOBAL SCALING IMPACT") loss_pct = impact["loss_reduction_pct"] egypt = compute_egypt_impact(loss_pct) print(f"\n Loss reduction achieved: {loss_pct:.2f}%") eg = egypt["egypt"] print(f"\n --- Egypt ---") print(f" Total generation: {eg['total_generation_twh']} TWh/year") print(f" Distribution losses: {eg['distribution_losses_twh']} TWh/year") print(f" Potential savings: {eg['potential_savings_twh']:.2f} TWh/year " f"({eg['potential_savings_gwh']:.0f} GWh)") print(f" CO2 saved: {eg['co2_saved_million_tonnes']:.3f} million tonnes/year") print(f" Cost saved (subsidised):{eg['cost_saved_usd_subsidised']:>15,.0f} USD/year") print(f" Cost saved (real cost): {eg['cost_saved_usd_real']:>15,.0f} USD/year") print(f" Impact (% of gen): {eg['impact_pct_of_generation']:.2f}%") ca = egypt["cairo"] print(f"\n --- Cairo ---") print(f" Share of national: {ca['share_of_national']*100:.0f}%") print(f" Potential savings: {ca['potential_savings_twh']:.3f} TWh/year") print(f" CO2 saved: {ca['co2_saved_million_tonnes']:.4f} million tonnes/year") gl = egypt["global"] print(f"\n --- Global ---") print(f" Total generation: {gl['total_generation_twh']:,.0f} TWh/year") print(f" Distribution losses: {gl['distribution_losses_twh']:,.0f} TWh/year") print(f" Potential savings: {gl['potential_savings_twh']:.1f} TWh/year") print(f" CO2 saved: {gl['co2_saved_million_tonnes']:.1f} million tonnes/year") print(f" Impact (% of gen): {gl['impact_pct_of_generation']:.3f}%") # Variables divider("DEPENDENT VARIABLES") vars_ = count_dependent_variables() totals = vars_["totals"] print(f"\n Physical variables: {totals['physical']}") print(f" Algorithmic hyperparams: {totals['algorithmic']}") print(f" External assumptions: {totals['external']}") print(f" Grand total: {totals['grand_total']}") print(f" Decision variables: {vars_['decision_variables']}") print(f"\n {vars_['note']}") # Implementation plan divider("REAL IMPLEMENTATION PLAN (EGYPT)") plan = egypt["implementation_plan"] print(f"\n Target partners:") for p in plan["target_partners"]: print(f" - {p}") for phase_key in ["phase_0_mvp", "phase_1_pilot", "phase_2_district", "phase_3_city", "phase_4_national"]: phase = plan[phase_key] print(f"\n {phase_key}:") print(f" Timeline: {phase['timeline']}") if "scope" in phase: print(f" Scope: {phase['scope']}") if "cost" in phase: print(f" Cost: {phase['cost']}") if "steps" in phase: for step in phase["steps"]: print(f" {step}") # Business model divider("BUSINESS MODEL & PRICING") biz = compute_business_model(impact) print(f"\n --- Usage Model ---") um = biz["usage_model"] print(f" Type: {um['type']}") print(f" Unit: {um['unit']}") print(f" Frequency: {um['frequency']}") print(f" Why recurring: {um['why_recurring']}") print(f"\n --- Savings Per Feeder ---") sf = biz["savings_per_feeder"] print(f" Energy saved: {sf['energy_saved_kwh_year']:,.0f} kWh/year") print(f" Cost saved (subsidised): ${sf['cost_saved_year_subsidised_usd']:,.0f}/year") print(f" Cost saved (real cost): ${sf['cost_saved_year_real_cost_usd']:,.0f}/year") print(f"\n --- Pricing Models ---") for model_key, model in biz["pricing_models"].items(): print(f"\n {model['name']}:") if "price_per_feeder_month_usd" in model: print(f" Price: ${model['price_per_feeder_month_usd']}/feeder/month " f"(${model['price_per_feeder_year_usd']}/year)") elif "share_pct" in model: print(f" Share: {model['share_pct']}% of verified savings " f"(~${model['revenue_per_feeder_year_usd']:,.0f}/feeder/year)") elif "price_per_year_usd" in model: print(f" Price: ${model['price_per_year_usd']:,.0f}/year " f"(up to {model['covers_feeders_up_to']} feeders)") print(f" {model['value_proposition']}") print(f"\n --- Revenue Projections ---") for phase_key, proj in biz["revenue_projections"].items(): print(f"\n {phase_key} ({proj['n_feeders']} feeders):") print(f" Annual revenue (SaaS): ${proj['annual_revenue_saas']:,.0f}") print(f" Annual savings to utility: ${proj['annual_savings_to_utility_real']:,.0f}") # Competitive analysis divider("COMPETITIVE ANALYSIS: WHY OPTIQ?") comp = biz["comparison_to_alternatives"] for name, alt in comp.items(): print(f"\n {alt['method']}:") print(f" Loss reduction: {alt['loss_reduction']}") print(f" Cost: {alt['cost']}") if "limitation" in alt: print(f" Limitation: {alt['limitation']}") if "advantage" in alt: print(f" Advantage: {alt['advantage']}") return { "footprint": footprint, "net_benefit": net_benefit, "egypt_impact": egypt, "variables": vars_, "business_model": biz, } def main(): print("=" * 70) print(" OptiQ Benchmark Suite") print(" Hybrid Quantum-AI-Classical Grid Optimization") print("=" * 70) # 1. Single-system benchmark with all methods results = run_single_system_benchmark("case33bw") if "error" in results: print("Benchmark failed.") return # 2. Comparison table print_comparison_table(results) # 3. Multi-load scenario testing multi = run_multi_load_benchmark("case33bw") print_multi_load_table(multi) # 4. Impact analysis analysis = print_impact_analysis(results) # 5. Save all results to JSON output = { "benchmark": { "system": results["system"], "published": results["published"], "baseline_loss_kw": results["baseline"]["total_loss_kw"], "methods": { name: { "loss_kw": d["loss_kw"], "reduction_pct": d["reduction_pct"], "time_sec": d["time_sec"], "open_lines": d["open_lines"], } for name, d in results["methods"].items() }, }, "multi_load": multi, } if analysis: output["footprint"] = analysis["footprint"] output["net_benefit"] = analysis["net_benefit"] output["egypt_impact"] = analysis["egypt_impact"] output["variables"] = analysis["variables"] output["business_model"] = analysis.get("business_model") out_path = os.path.join(os.path.dirname(__file__), "benchmark_results.json") with open(out_path, "w") as f: json.dump(output, f, indent=2, default=str) print(f"\n Results saved to: {out_path}") divider("BENCHMARK COMPLETE") if __name__ == "__main__": main()