"""FastAPI backend for G2ELin — P7 (web UI) first slice, plus a network
editor / from-scratch builder layer.
Wraps the real ``g2elin_core`` compute functions (no mocking): static power
flow, linear modal analysis, time-series load flow, and nonlinear EMT
time-domain simulation. Two ways to get a ``Network`` to run these on:
- ``/api/presets/{id}/...`` — one of the fixed, named presets in
``presets.py``.
- ``/api/network/...`` (``network_routes.py``) — an arbitrary ``Network``
JSON body, e.g. a preset the frontend cloned and edited, or one built
from scratch in the drag-and-drop canvas. Nothing here is persisted
server-side; the client holds the edited network and resends it whole
each call.
Both sets of endpoints delegate to the same functions in ``analysis.py``
(each takes a ``Network`` and returns the exact response model its
endpoint returns) so the actual analysis logic exists in exactly one
place. This module's own handlers are just "resolve preset_id -> Network
(404 on unknown id), then call into analysis.py."
Also serves the plain-JS frontend in ``web/`` as static files alongside the
JSON API, and the built Sphinx documentation (if present) at ``/manual``
for the frontend's Documentation tab — see ``docs/sphinx/api_and_web.md``'s
"Mount layout" section for why ``/manual`` and not ``/docs`` (FastAPI's own
Swagger UI already lives there), and why mount order matters (specific
routes/mounts before the catch-all ``/``).
"""
from __future__ import annotations
import logging
import os
import threading
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from fastapi.staticfiles import StaticFiles
from g2elin_core.network.schema import Network
from g2elin_core.operating_point import gfl_params, gfm_params, gfm_retuned, sm_params
from . import analysis
from .network_routes import router as network_router
from .presets import PRESETS, get_preset
from .schemas import (
ModelLevelsResponse,
SolversResponse,
BatchPowerFlowRequest,
BatchPowerFlowResponse,
EmtRequest,
EmtResponse,
FreeResponseRequest,
FreeResponseResponse,
ModalResponse,
ModeShapeRequest,
ModeShapeResponse,
PowerFlowRequest,
PowerFlowResponse,
PresetSummary,
SensitivityRequest,
SensitivityResponse,
StatesResponse,
StepResponseRequest,
StepResponseResponse,
TimeSeriesResponse,
TopologyResponse,
UnitDefaultsRequest,
UnitRetuneRequest,
UnitRetuneResponse,
UnitDefaultsResponse,
)
logger = logging.getLogger("g2elin_api")
app = FastAPI(title="G2ELin API", description="Power-system analysis over G2ELin's ported presets.")
@app.middleware("http")
async def _revalidate_static(request: Request, call_next):
"""The web UI's HTML/CSS/JS are served without a cache lifetime, so
browsers may reuse a stale copy after an update. ``no-cache`` makes them
revalidate each load (a cheap 304 via StaticFiles' ETag when unchanged).
"""
response = await call_next(request)
if not request.url.path.startswith("/api/"):
response.headers.setdefault("Cache-Control", "no-cache")
return response
# Presets whose models are pre-built at startup when G2ELIN_WARMUP=1 (set in
# the Docker image): between them they cover every unit type (SM, GFM, GFL,
# infinite bus), so the one-off symbolic derivation per type happens before
# the first visitor asks for a modal analysis or an EMT run, not during it.
_WARMUP_PRESETS = ("wscc9_3sm", "wscc9_1sm_1gfm_1gfl", "sm_smib")
@app.on_event("startup")
def _warm_up_models() -> None:
if os.environ.get("G2ELIN_WARMUP") != "1":
return
def run() -> None:
for preset_id in _WARMUP_PRESETS:
try:
net = get_preset(preset_id).build()
analysis.build_modal_from_network(net)
analysis.build_nonlinear_model_from_network(net)
logger.info("warm-up: built models for %s", preset_id)
except Exception: # noqa: BLE001 -- best effort; a real request reports the error properly
logger.exception("warm-up failed for %s", preset_id)
threading.Thread(target=run, name="g2elin-warmup", daemon=True).start()
def _resolve_preset_network(preset_id: str) -> Network:
try:
return get_preset(preset_id).build()
except KeyError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
[docs]
@app.get("/api/presets", response_model=list[PresetSummary])
def list_presets() -> list[PresetSummary]:
summaries = []
for info in sorted(PRESETS.values(), key=lambda p: p.id):
net = info.build()
summaries.append(
PresetSummary(
id=info.id,
name=info.name,
description=info.description,
n_buses=len(net.buses),
n_lines=len(net.lines),
n_transformers=len(net.transformers),
n_loads=len(net.loads),
n_der_units=len(net.der_units),
der_unit_types=sorted({d.unit_type.value for d in net.der_units}),
)
)
return summaries
[docs]
@app.get("/api/presets/{preset_id}/network", response_model=Network)
def get_preset_network(preset_id: str) -> Network:
"""The preset's own ``Network`` definition, verbatim -- how the web UI
clones a preset into an editable copy (see ``/api/network/*`` below).
"""
return _resolve_preset_network(preset_id)
[docs]
@app.get("/api/presets/{preset_id}/topology", response_model=TopologyResponse)
def get_topology(preset_id: str) -> TopologyResponse:
return analysis.topology_response(_resolve_preset_network(preset_id))
[docs]
@app.get("/api/powerflow/algorithms")
def list_powerflow_algorithms() -> dict[str, str]:
"""Solver ids accepted by ``PowerFlowOptions.algorithm`` -> display label."""
return analysis.POWERFLOW_ALGORITHMS
[docs]
@app.get("/api/solvers", response_model=SolversResponse)
def list_solvers() -> SolversResponse:
"""The time-domain integrators this build offers, with what each one is
good for, plus the default tolerances. Static, like /api/model-levels."""
return analysis.solvers_response()
[docs]
@app.get("/api/model-levels", response_model=ModelLevelsResponse)
def list_model_levels() -> ModelLevelsResponse:
"""The model-order reduction catalogue: every element type's named
levels and the state groups they are built from. Static, so the web UI
fetches it once and builds its level pickers and per-state controls
from it rather than hard-coding either."""
return analysis.model_levels_response()
[docs]
@app.post("/api/units/defaults", response_model=UnitDefaultsResponse)
def unit_defaults(req: UnitDefaultsRequest) -> UnitDefaultsResponse:
"""A unit type's default electrical/control parameters for given base
values -- no network needed, so the editor can show and edit a unit's
parameters before it's connected (or while the network is incomplete)."""
if not (req.sn_mva > 0 and req.f_hz > 0 and req.un_kv > 0):
raise HTTPException(status_code=422, detail="sn_mva, f_hz and un_kv must be positive")
base = dict(sn_mva=req.sn_mva, f_hz=req.f_hz, rt_pu=req.rt_pu, lt_pu=req.lt_pu)
if req.unit_type == "sm":
try:
params = sm_params(**base, exciter=req.exciter or "g2elin",
pss=req.pss or "g2elin", governor=req.governor or "g2elin")
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e)) from e
elif req.unit_type == "gfm":
try:
params = gfm_params(**base, un_kv=req.un_kv, controller=req.controller or "droop")
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e)) from e
elif req.unit_type == "gfl":
params = gfl_params(**base, un_kv=req.un_kv)
elif req.unit_type == "infinite_bus":
params = {}
else:
raise HTTPException(status_code=422, detail=f"unknown unit type {req.unit_type!r}")
return UnitDefaultsResponse(params=params)
[docs]
@app.post("/api/units/retune", response_model=UnitRetuneResponse)
def unit_retune(req: UnitRetuneRequest) -> UnitRetuneResponse:
"""A converter's outer-loop parameters, re-expressed.
Every grid-forming law's gains are written in terms of the same droop
tuning, which is what makes them comparable. So the editor can offer
``mp``, ``nq``, ``wf`` and the equivalent inertia ``H`` on any of them --
including the ones that carry no parameter by those names -- and can
swap a law without throwing the tuning away. The formulas live in
``operating_point`` and are not mirrored in the frontend.
"""
try:
params, tuning = gfm_retuned(
req.params, req.controller, req.to_controller, req.tuning or None
)
except (ValueError, KeyError, ZeroDivisionError) as e:
raise HTTPException(status_code=422, detail=str(e)) from e
return UnitRetuneResponse(params=params, tuning=tuning)
[docs]
@app.post("/api/presets/{preset_id}/powerflow", response_model=PowerFlowResponse)
def run_powerflow(preset_id: str, req: PowerFlowRequest | None = None) -> PowerFlowResponse:
# The body is optional: a bare POST solves with pandapower's defaults.
return analysis.powerflow_response(_resolve_preset_network(preset_id), req.options if req else None)
[docs]
@app.post("/api/presets/{preset_id}/powerflow/batch", response_model=BatchPowerFlowResponse)
def run_powerflow_batch(preset_id: str, req: BatchPowerFlowRequest) -> BatchPowerFlowResponse:
return analysis.batch_powerflow_response(_resolve_preset_network(preset_id), req)
[docs]
@app.post("/api/presets/{preset_id}/modal", response_model=ModalResponse)
def run_modal(preset_id: str) -> ModalResponse:
return analysis.modal_response(_resolve_preset_network(preset_id))
[docs]
@app.post("/api/presets/{preset_id}/modal/sensitivity", response_model=SensitivityResponse)
def run_modal_sensitivity(preset_id: str, req: SensitivityRequest) -> SensitivityResponse:
return analysis.modal_sensitivity_response(_resolve_preset_network(preset_id), req)
[docs]
@app.post("/api/presets/{preset_id}/modal/mode_shape", response_model=ModeShapeResponse)
def run_modal_mode_shape(preset_id: str, req: ModeShapeRequest) -> ModeShapeResponse:
return analysis.modal_mode_shape_response(_resolve_preset_network(preset_id), req)
[docs]
@app.post("/api/presets/{preset_id}/modal/free_response", response_model=FreeResponseResponse)
def run_modal_free_response(preset_id: str, req: FreeResponseRequest) -> FreeResponseResponse:
return analysis.modal_free_response_response(_resolve_preset_network(preset_id), req)
[docs]
@app.post("/api/presets/{preset_id}/modal/step_response", response_model=StepResponseResponse)
def run_modal_step_response(preset_id: str, req: StepResponseRequest) -> StepResponseResponse:
return analysis.modal_step_response_response(_resolve_preset_network(preset_id), req)
[docs]
@app.post("/api/presets/{preset_id}/timeseries", response_model=TimeSeriesResponse)
def run_timeseries(preset_id: str) -> TimeSeriesResponse:
return analysis.timeseries_response(_resolve_preset_network(preset_id))
[docs]
@app.get("/api/presets/{preset_id}/states", response_model=StatesResponse)
def list_states(preset_id: str) -> StatesResponse:
"""State names for the nonlinear (EMT) model — lets the frontend
populate its state pickers without hard-coding per-preset names (which
differ: e.g. only synchronous machines have ``dw_r_*`` states, and
which DER id is the slack varies by preset).
"""
return analysis.states_response(_resolve_preset_network(preset_id))
[docs]
@app.post("/api/presets/{preset_id}/emt", response_model=EmtResponse)
def run_emt(preset_id: str, req: EmtRequest) -> EmtResponse:
return analysis.emt_response(_resolve_preset_network(preset_id), req)
[docs]
@app.post("/api/presets/{preset_id}/emt/live")
def run_emt_live(preset_id: str, req: EmtRequest, request: Request) -> StreamingResponse:
"""Newline-delimited JSON, one line per solver step -- see
``analysis.emt_live_stream``'s own docstring for the exact line
shapes. Validation (``prepare_emt_live``) runs here, synchronously,
*before* the streaming response is constructed, so a bad request
still gets a normal 422 instead of a broken stream.
"""
plan = analysis.prepare_emt_live(_resolve_preset_network(preset_id), req)
return StreamingResponse(
analysis.emt_live_stream(plan, req.perturb_kind, request), media_type="application/x-ndjson"
)
app.include_router(network_router)
_repo_root = Path(__file__).resolve().parents[2]
_docs_dir = _repo_root / "docs" / "sphinx" / "_build" / "html"
if _docs_dir.is_dir():
app.mount("/manual", StaticFiles(directory=str(_docs_dir), html=True), name="manual")
_web_dir = _repo_root / "web"
if _web_dir.is_dir():
app.mount("/", StaticFiles(directory=str(_web_dir), html=True), name="web")