amadinm/MicrogridDummy

1# backend/app.py2from __future__ import annotations3 4from datetime import datetime, timedelta, timezone5from typing import Dict, Any, List, Optional6 7import math8 9from fastapi import FastAPI, HTTPException10 11app = FastAPI(title="Microgrid MVP Backend")12 13# In-memory storage for tariffs (by site_id)14TARIFFS: Dict[str, Dict[str, Any]] = {}15 16 17# ---------------------------18# Helpers19# ---------------------------20 21def _parse_iso(ts: str) -> datetime:22    """Parse ISO string, force UTC."""23    return datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone(timezone.utc)24 25 26def _build_tou_price_vector(start_ts: str, horizon: int, interval_min: int) -> Dict[str, Any]:27    """28    Simple TOU (time-of-use) price curve:29    - Off-peak (0–7, 19–24): $0.14/kWh30    - Peak (7–19):          $0.28/kWh31    """32    start = _parse_iso(start_ts)33    step = timedelta(minutes=interval_min)34    values: List[float] = []35 36    for i in range(horizon):37        t = start + i * step38        if 7 <= t.hour < 19:39            price = 0.2840        else:41            price = 0.1442        values.append(price)43 44    return {45        "start_ts": start.isoformat(),46        "interval_min": interval_min,47        "values_usd_per_kwh": values,48    }49 50 51def _synthetic_load_and_pv(52    horizon: int, interval_min: int, application: str = "commercial"53) -> Dict[str, Dict[str, Any]]:54    """55    Generate synthetic load (kW) and PV (kW) profiles, similar to your56    Streamlit fallback.57    """58    step = timedelta(minutes=interval_min)59    start = datetime.now(timezone.utc).replace(second=0, microsecond=0)60    app_lower = (application or "").lower()61 62    base_kw = 2.0 if "res" in app_lower else (20.0 if "com" in app_lower else 100.0)63 64    load_vals: List[float] = []65    pv_vals: List[float] = []66 67    for i in range(horizon):68        t = start + i * step69        # normalized position in the day70        frac = (t.hour + t.minute / 60.0) / 24.0 * 2 * math.pi71 72        # daily cycle73        day_cycle = 0.6 + 0.5 * math.sin(frac - math.pi / 2)74        evening_bump = 0.25 * math.exp(-0.5 * ((t.hour - 19) / 2.0) ** 2)75        load = max(0.1, base_kw * (0.7 + 0.5 * day_cycle) + base_kw * evening_bump)76 77        # PV peak at 13:0078        pv_peak_kw = 0.6 if "res" in app_lower else 5.079        pv_shape = math.exp(-0.5 * ((t.hour + t.minute / 60.0 - 13.0) / 3.0) ** 2)80        pv = max(0.0, pv_peak_kw * pv_shape)81 82        load_vals.append(float(load))83        pv_vals.append(float(pv))84 85    load = {86        "start_ts": start.isoformat(),87        "interval_min": interval_min,88        "values_kw": load_vals,89    }90    pv = {91        "start_ts": start.isoformat(),92        "interval_min": interval_min,93        "values_kw": pv_vals,94    }95 96    return {"load": load, "pv": pv}97 98 99# ---------------------------100# Tariff endpoints101# ---------------------------102 103@app.post("/tariff/set")104async def set_tariff(payload: Dict[str, Any]):105    """106    Store a tariff definition. The UI sends:107    {108      "site_id": "...",109      "timezone": "UTC",110      "weekday_periods": [...],111      "weekend_periods": null / [...],112      "demand_charge_lambda": ...,113      "diesel_cost_usd_per_kwh": ...114    }115    """116    site_id = payload.get("site_id") or "site"117    TARIFFS[site_id] = payload118    return {"status": "ok", "site_id": site_id}119 120 121@app.get("/tariff/price_vector")122async def get_price_vector(123    site_id: str,124    start_ts: str,125    horizon: int = 24,126    interval_min: int = 15,127):128    """129    Return a price vector for this site_id. For now, we ignore the stored130    tariff and just return a simple TOU curve. The Streamlit UI only needs131    a vector of prices.132    """133    if horizon <= 0:134        raise HTTPException(status_code=400, detail="horizon must be > 0")135    if interval_min <= 0:136        raise HTTPException(status_code=400, detail="interval_min must be > 0")137 138    return _build_tou_price_vector(start_ts=start_ts, horizon=horizon, interval_min=interval_min)139 140 141# ---------------------------142# Forecast endpoints143# ---------------------------144 145@app.post("/forecast/load")146async def forecast_load(payload: Dict[str, Any]):147    """148    Request body example:149    {150      "site_id": "north_america_01803_commercial",151      "horizon": 24,152      "interval_min": 15153    }154    """155    site_id = payload.get("site_id", "site")156    horizon = int(payload.get("horizon", 24))157    interval_min = int(payload.get("interval_min", 15))158 159    if horizon <= 0 or interval_min <= 0:160        raise HTTPException(status_code=400, detail="Invalid horizon or interval_min")161 162    # For now we ignore site_id and application in forecasts and just use a generic pattern.163    synthetic = _synthetic_load_and_pv(horizon=horizon, interval_min=interval_min, application=site_id)164    return synthetic["load"]165 166 167@app.post("/forecast/pv")168async def forecast_pv(payload: Dict[str, Any]):169    """170    Same request shape as /forecast/load.171    """172    site_id = payload.get("site_id", "site")173    horizon = int(payload.get("horizon", 24))174    interval_min = int(payload.get("interval_min", 15))175 176    if horizon <= 0 or interval_min <= 0:177        raise HTTPException(status_code=400, detail="Invalid horizon or interval_min")178 179    synthetic = _synthetic_load_and_pv(horizon=horizon, interval_min=interval_min, application=site_id)180    return synthetic["pv"]181 182 183# ---------------------------184# Optimization endpoint185# ---------------------------186 187@app.post("/optimize/plan")188async def optimize_plan(payload: Dict[str, Any]):189    """190    Request body example:191    {192      "site_id": "site",193      "horizon": 96,194      "interval_min": 15,195      "price_usd_per_kwh": [...],196      "load_kw": [...],197      "pv_kw": [...]198    }199    Simplest possible "optimization":200    - Use PV first.201    - Remaining load comes from the grid.202    - No battery or diesel for now (but we structure schedule so UI can expand later).203    """204    site_id = payload.get("site_id", "site")205    horizon = int(payload.get("horizon", 0))206    interval_min = int(payload.get("interval_min", 15))207 208    prices = payload.get("price_usd_per_kwh") or []209    load_kw = payload.get("load_kw") or []210    pv_kw = payload.get("pv_kw") or []211 212    H = min(len(prices), len(load_kw), len(pv_kw))213    if horizon > 0:214        H = min(H, horizon)215 216    if H <= 0:217        raise HTTPException(status_code=400, detail="No valid horizon from input data.")218 219    schedule: List[Dict[str, Any]] = []220    soc = 0.5  # dummy state of charge221 222    for i in range(H):223        load = float(load_kw[i])224        pv = float(pv_kw[i])225        # First use PV to cover load; we ignore feed-in for now226        grid = max(load - pv, 0.0)227 228        row = {229            "t_idx": i,230            "p_grid_kw": grid,231            "p_batt_kw": 0.0,   # no battery yet232            "p_diesel_kw": 0.0, # no generator233            "soc": soc,234        }235        schedule.append(row)236 237    start_ts = datetime.now(timezone.utc).replace(second=0, microsecond=0).isoformat()238 239    return {240        "site_id": site_id,241        "start_ts": start_ts,242        "interval_min": interval_min,243        "schedule": schedule,244    }245 246 247# ---------------------------248# Savings endpoint249# ---------------------------250 251@app.post("/analyze/savings")252async def analyze_savings(payload: Dict[str, Any]):253    """254    Request body example (from UI):255    {256      "site_id": "...",257      "interval_min": 15,258      "price_usd_per_kwh": [...],259      "load_kw": [...],260      "pv_kw": [...],261      "schedule": [...],262      "diesel_cost_usd_per_kwh": 0.35,263      "demand_charge_lambda": 5.0,264      "degr_cost_usd_per_kwh": 0.02265    }266    """267    prices = payload.get("price_usd_per_kwh") or []268    load_kw = payload.get("load_kw") or []269    pv_kw = payload.get("pv_kw") or []270    schedule = payload.get("schedule") or []271    interval_min = int(payload.get("interval_min", 15))272 273    H = min(len(prices), len(load_kw), len(pv_kw), len(schedule))274    if H <= 0:275        raise HTTPException(status_code=400, detail="Insufficient data to analyze savings.")276 277    step_hours = interval_min / 60.0278 279    baseline_cost = 0.0280    optimized_cost = 0.0281 282    for i in range(H):283        price = float(prices[i])284        load = float(load_kw[i])285        pv = float(pv_kw[i])286 287        # Baseline: assume grid supplies load minus PV (no storage)288        base_grid = max(load - pv, 0.0)289        baseline_cost += base_grid * price * step_hours290 291        sch = schedule[i]292        grid_opt = float(sch.get("p_grid_kw", 0.0))293        optimized_cost += grid_opt * price * step_hours294 295    savings_abs = baseline_cost - optimized_cost296    savings_pct = (savings_abs / baseline_cost * 100.0) if baseline_cost > 0 else 0.0297 298    return {299        "baseline_cost": baseline_cost,300        "optimized_cost": optimized_cost,301        "savings_abs": savings_abs,302        "savings_pct": savings_pct,303    }304