71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import List, Dict, Any
|
|
|
|
from .dsl import Object, Morphism, Functor, TimeMonoid, Limit, Colimit
|
|
from .core import canonicalize_dsl
|
|
from .adapters import generate_adapters
|
|
from .solver import admm_lite
|
|
from .simulation import Sandbox
|
|
|
|
app = FastAPI(title="GridForge API", version="0.1.0")
|
|
|
|
|
|
class DSLPayload(BaseModel):
|
|
objects: List[Dict[str, Any]] = []
|
|
morphisms: List[Dict[str, Any]] = []
|
|
functors: List[Dict[str, Any]] = []
|
|
limits: List[Dict[str, Any]] = []
|
|
colimits: List[Dict[str, Any]] = []
|
|
time: Dict[str, Any] = {}
|
|
|
|
|
|
@app.post("/dsl/canonicalize")
|
|
def canonicalize(payload: DSLPayload):
|
|
# Naive canonicalization; in a fuller system we would instantiate domain models
|
|
canon = canonicalize_dsl(
|
|
[Object(**o) for o in payload.objects],
|
|
[Morphism(**m) for m in payload.morphisms],
|
|
[Functor(**f) for f in payload.functors],
|
|
[Limit(**l) for l in payload.limits],
|
|
[Colimit(**c) for c in payload.colimits],
|
|
TimeMonoid(**payload.time) if payload.time else TimeMonoid(id="t0", rounds=1)
|
|
)
|
|
return canon
|
|
|
|
|
|
@app.post("/adapters/generate")
|
|
def adapters_generate(payload: DSLPayload):
|
|
# Build a minimal payload dictionary and generate adapters to adapters/
|
|
d = {
|
|
"objects": payload.objects,
|
|
"morphisms": payload.morphisms,
|
|
"functors": payload.functors,
|
|
"limits": payload.limits,
|
|
"colimits": payload.colimits,
|
|
"time": payload.time or {"id": "t0", "rounds": 1, "mode": "sync"},
|
|
}
|
|
out = generate_adapters(d)
|
|
return {"adapters": out}
|
|
|
|
|
|
@app.post("/simulate")
|
|
def simulate(input: Dict[str, Any]):
|
|
# Simple toy ADMM-like call on a synthetic problem if provided
|
|
A = input.get("A")
|
|
b = input.get("b")
|
|
if A is None or b is None:
|
|
return {"status": "no-op", "message": "Provide A and b to run solver"}
|
|
import numpy as np
|
|
A = np.array(A, dtype=float)
|
|
b = np.array(b, dtype=float)
|
|
x = admm_lite(A, b)
|
|
return {"status": "ok", "x": x.tolist()}
|
|
|
|
|
|
@app.get("/")
|
|
def root():
|
|
return {"message": "GridForge API. Use /dsl/canonicalize, /adapters/generate, /simulate"}
|