Spaces:
Sleeping
Sleeping
| """ | |
| Optimize endpoint — Runs the hybrid Quantum-AI-Classical pipeline. | |
| Supports all solver modes: classical, quantum, ai, hybrid. | |
| """ | |
| from __future__ import annotations | |
| import time | |
| from fastapi import APIRouter, HTTPException | |
| from pydantic import BaseModel, Field | |
| from src.grid.loader import load_network | |
| from src.grid.power_flow import get_baseline, evaluate_topology | |
| from src.grid.reconfiguration import branch_exchange_search | |
| from src.hybrid.pipeline import run_hybrid_pipeline | |
| from src.evaluation.metrics import compute_impact, compute_speedup | |
| router = APIRouter() | |
| class OptimizeRequest(BaseModel): | |
| system: str = Field(default="case33bw", description="IEEE test system name") | |
| method: str = Field( | |
| default="hybrid", | |
| description="Optimization method: 'classical', 'quantum', 'ai', or 'hybrid'", | |
| ) | |
| quantum_iters: int = Field(default=300, description="SA iterations for quantum solver") | |
| quantum_restarts: int = Field(default=3, description="SA restarts") | |
| quantum_top_k: int = Field(default=5, description="Number of candidate topologies") | |
| open_lines: list[int] | None = Field( | |
| default=None, | |
| description="User's current open lines to use as starting point for optimisation", | |
| ) | |
| def optimize(req: OptimizeRequest): | |
| """Run optimization on the specified IEEE system.""" | |
| try: | |
| net = load_network(req.system) | |
| except ValueError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) | |
| baseline = get_baseline(net) | |
| if not baseline.get("converged"): | |
| raise HTTPException(status_code=500, detail="Baseline did not converge.") | |
| t_start = time.perf_counter() | |
| if req.method == "classical": | |
| result = branch_exchange_search(net, verbose=False, initial_open_lines=req.open_lines) | |
| if "error" in result: | |
| raise HTTPException(status_code=500, detail=result["error"]) | |
| optimized = evaluate_topology(net, result["best_open_lines"]) | |
| if not optimized.get("converged"): | |
| raise HTTPException(status_code=500, detail="Optimized topology did not converge.") | |
| t_end = time.perf_counter() | |
| impact = compute_impact(baseline, optimized) | |
| return { | |
| "system": req.system, | |
| "method": "classical", | |
| "baseline": baseline, | |
| "optimized": optimized, | |
| "impact": impact, | |
| "solver": { | |
| "method": "branch_exchange", | |
| "iterations": result["iterations"], | |
| "time_sec": result["time_sec"], | |
| }, | |
| "total_time_sec": round(t_end - t_start, 4), | |
| } | |
| elif req.method in ("quantum", "hybrid", "ai"): | |
| use_quantum = req.method in ("quantum", "hybrid") | |
| use_ai = req.method in ("ai", "hybrid") | |
| pipeline_result = run_hybrid_pipeline( | |
| net, | |
| use_quantum=use_quantum, | |
| use_ai=use_ai, | |
| quantum_iters=req.quantum_iters, | |
| quantum_restarts=req.quantum_restarts, | |
| quantum_top_k=req.quantum_top_k, | |
| initial_open_lines=req.open_lines, | |
| ) | |
| if "error" in pipeline_result: | |
| raise HTTPException(status_code=500, detail=pipeline_result["error"]) | |
| return { | |
| "system": req.system, | |
| "method": pipeline_result["method"], | |
| "baseline": pipeline_result["baseline"], | |
| "optimized": pipeline_result["optimized"], | |
| "impact": pipeline_result["impact"], | |
| "candidates": pipeline_result.get("all_evaluated", []), | |
| "timings": pipeline_result["timings"], | |
| "total_time_sec": pipeline_result["timings"]["total_sec"], | |
| } | |
| else: | |
| raise HTTPException(status_code=400, detail=f"Unknown method: {req.method}") | |