Spaces:
Sleeping
Sleeping
File size: 5,695 Bytes
7477316 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """
Simulate endpoint — Manual switch control and power flow simulation.
"""
from __future__ import annotations
import json
from typing import Optional
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel, Field
from src.grid.loader import load_network, clone_network, get_line_info
from src.grid.power_flow import apply_topology, run_power_flow, extract_results, check_radial_connected, check_topology_valid
from src.evaluation.metrics import compute_impact
from api.auth import optional_auth, require_auth, FirebaseUser
from api.database import log_usage, log_audit
router = APIRouter()
class SimulateRequest(BaseModel):
system: str = Field(default="case33bw", description="IEEE test system name")
open_lines: list[int] = Field(description="List of line indices to set as OPEN")
load_multiplier: float = Field(default=1.0, ge=0.5, le=2.0, description="Load scaling factor")
@router.post("/simulate")
def simulate(req: SimulateRequest, user: FirebaseUser = Depends(optional_auth)):
"""Simulate power flow with custom switch configuration.
Allows manual open/close of switches and recomputes losses.
"""
try:
net = load_network(req.system)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
# Scale loads if multiplier != 1.0
if req.load_multiplier != 1.0:
net.load["p_mw"] *= req.load_multiplier
net.load["q_mvar"] *= req.load_multiplier
# Get baseline first
net_baseline = clone_network(net)
if not run_power_flow(net_baseline):
raise HTTPException(status_code=500, detail="Baseline power flow did not converge")
baseline = extract_results(net_baseline)
# Check connectivity (required) — radiality is a soft warning, not a hard block
if not check_topology_valid(net, req.open_lines, require_radial=False):
raise HTTPException(
status_code=400,
detail="Invalid topology: all buses must remain connected"
)
# Apply topology and run power flow
net_sim = apply_topology(net, req.open_lines)
if not run_power_flow(net_sim):
raise HTTPException(status_code=500, detail="Power flow did not converge for this configuration")
simulated = extract_results(net_sim)
simulated["open_lines"] = req.open_lines
# Compute impact
impact = compute_impact(baseline, simulated)
# Log if authenticated
if user:
try:
# Calculate annualized values
hours_per_year = 8760
saved_kw = baseline["total_loss_kw"] - simulated["total_loss_kw"]
saved_kwh = saved_kw * hours_per_year
co2_saved = saved_kwh * 0.50 # Egypt factor
money_saved = saved_kwh * 0.08
log_usage(
firebase_uid=user.uid,
system=req.system,
method="simulate",
baseline_loss_kw=baseline["total_loss_kw"],
optimized_loss_kw=simulated["total_loss_kw"],
energy_saved_kwh=saved_kwh,
co2_saved_kg=co2_saved,
money_saved_usd=money_saved,
computation_time_sec=0.0,
load_multiplier=req.load_multiplier,
switches_changed=json.dumps(req.open_lines),
)
except Exception:
pass # Don't fail request if logging fails
return {
"system": req.system,
"load_multiplier": req.load_multiplier,
"baseline": baseline,
"simulated": simulated,
"impact": impact,
"open_lines": req.open_lines,
}
class ToggleSwitchRequest(BaseModel):
system: str = Field(default="case33bw")
line_id: int = Field(description="Line index to toggle")
current_open_lines: list[int] = Field(description="Current list of open line indices")
@router.post("/simulate/toggle")
def toggle_switch(req: ToggleSwitchRequest, user: FirebaseUser = Depends(optional_auth)):
"""Toggle a single switch and return the new configuration validity."""
try:
net = load_network(req.system)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
# Compute new open lines
if req.line_id in req.current_open_lines:
new_open_lines = [l for l in req.current_open_lines if l != req.line_id]
action = "closed"
else:
new_open_lines = req.current_open_lines + [req.line_id]
action = "opened"
# Check connectivity (hard requirement) — loops are allowed
is_connected = check_topology_valid(net, new_open_lines, require_radial=False)
is_distribution = "33" in req.system
is_radial = check_topology_valid(net, new_open_lines, require_radial=True) if is_connected else False
result = {
"line_id": req.line_id,
"action": action,
"new_open_lines": new_open_lines,
"is_valid": is_connected,
}
if is_connected:
# Run power flow — pandapower handles meshed grids fine
net_sim = apply_topology(net, new_open_lines)
if run_power_flow(net_sim):
simulated = extract_results(net_sim)
result["power_flow"] = simulated
if is_distribution and not is_radial:
result["warnings"] = ["Configuration has loops — distribution grids are normally operated radially."]
else:
result["is_valid"] = False
result["error"] = "Power flow did not converge"
else:
result["error"] = "Configuration disconnects the network — all buses must remain reachable"
return result
|