Added csv export endpoint

main.py

@@ -1,9 +1,10 @@
 from typing import Annotated, Dict, List, Optional, Tuple
 from urllib.parse import urlencode
-from fastapi import FastAPI, Query, Request
+from fastapi import FastAPI, Query, Request, Response
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import RedirectResponse
+from fastapi.responses import PlainTextResponse, RedirectResponse
 import pandas as pd
+from pydantic import Field, model_validator
 from capacity_factors import get_solar_capacity_factor
 from schema import CapacityFactor, Location, SolarPVAssumptions
 from model import calculate_cashflow_for_renewable_project, calculate_lcoe
@@ -22,28 +23,22 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+@app.get("/solarpv/lcoe")
+def get_lcoe(pv_assumptions: Annotated[SolarPVAssumptions, Query()]):
+    return calculate_lcoe(pv_assumptions)
 
-# @app.get("/")
-# async def read_main(request: Request):
-#     # Get query parameters as dictionary
-#     query_params = dict(request.query_params)
-
-#     # Create new URL with parameters
-#     print(request.url.components)
-#     print(request.url.scheme)
-#     print(request.url.netloc)
-#     print("hf.space" in request.url.netloc)
-#     redirect_url = (request.url.scheme if "hf.space" not in request.url.netloc else "https") + "://" + request.url.netloc + request.url.path + "ui"
-#     print(redirect_url)
-#     if query_params:
-#         redirect_url += "?" + urlencode(query_params)
 
-#     return RedirectResponse(redirect_url)
+class SolarPVAssumptionsWithLCOE(SolarPVAssumptions):
+    lcoe: Annotated[float, Field(title="Levelized cost of electricity (USD/MWh)", description="The levelized cost of electricity in USD/MWh")]
 
 
-@app.get("/solarpv/lcoe")
-def get_lcoe(pv_assumptions: Annotated[SolarPVAssumptions, Query()]):
-    return calculate_lcoe(pv_assumptions)
+@app.get("/solarpv/lcoe.json")
+def get_lcoe_json(
+    pv_assumptions: Annotated[SolarPVAssumptions, Query()]
+) -> SolarPVAssumptionsWithLCOE:
+    return SolarPVAssumptionsWithLCOE(
+        **{"lcoe": calculate_lcoe(pv_assumptions), **pv_assumptions.model_dump()}
+    )
 
 
 @app.get("/solarpv/capacity_factor.json")
@@ -54,4 +49,43 @@ def get_capacity_factor(pv_location: Annotated[Location, Query()]) -> CapacityFa
     )
 
 
+class CashflowParams(SolarPVAssumptions):
+    tariff: Annotated[
+        Optional[float],
+        Query(
+            title="Tariff (USD/MWh)",
+            description="If a tariff in USD/MWh is supplied, it will be used to calculate the cashflow. If not, the break-even tariff will be calculated from the assumptions.",
+            gt=0,
+        ),
+    ] = None
+    transpose: Annotated[
+        bool,
+        Query(
+            title="Transpose cashflows",
+            description="Transpose the cashflow table to have years as columns and cashflows as rows",
+            default=False,
+        ),
+    ] = False
+
+    # If tariff is not provided, calculate it from the assumptions
+    @model_validator(mode="after")
+    @classmethod
+    def calculate_tariff(cls, values):
+        if values.tariff is None:
+            values.tariff = calculate_lcoe(values)
+        return values
+
+
+@app.get("/solarpv/cashflow.csv", response_class=PlainTextResponse)
+def get_cashflow(params: Annotated[CashflowParams, Query()]) -> str:
+    cashflow = calculate_cashflow_for_renewable_project(
+        params, tariff=params.tariff, return_model=True
+    )[0]
+    if params.transpose:
+        cashflow = cashflow.to_pandas().T
+        cashflow.columns = cashflow.loc["Period"].astype(int).astype(str)
+        return cashflow.drop(["Period"]).to_csv(float_format="%.3f")
+    return cashflow.write_csv(float_precision=3)
+
+
 app = gr.mount_gradio_app(app, interface, path="/")

model.py

@@ -280,4 +280,4 @@ def calculate_lcoe(
         # LCOE is too low
         LCOE_guess += 5
         lcoe = calculate_lcoe(assumptions, LCOE_guess, iter_count=iter_count + 1)
-    return lcoe
+    return float(lcoe)

schema.py

@@ -4,19 +4,19 @@ from pydantic import BaseModel, Field, computed_field, field_validator, model_va
 
 class SolarPVAssumptions(BaseModel):
     capacity_mw: Annotated[
-        float, Field(ge=1, le=1000, title="Capacity (MW)", description="Capacity in MW")
+        float, Field(gt=0, le=1000, title="Capacity (MW)", description="Capacity in MW")
     ] = 30
     capacity_factor: Annotated[
         float,
         Field(
             ge=0,
-            le=0.6,
+            le=1,
             title="Capacity factor (%)",
             description="Capacity factor as a decimal, e.g., 0.2 for 20%",
         ),
     ] = 0.10
     capital_expenditure_per_kw: Annotated[
-        float, Field(ge=1e2, le=1e4, title="Capital expenditure ($/kW)")
+        float, Field(gt=0, title="Capital expenditure ($/kW)")
     ] = 670
     o_m_cost_pct_of_capital_cost: Annotated[
         float,
@@ -245,10 +245,6 @@ class CapacityFactor(Location):
     capacity_factor: Annotated[float, Field(
         title="Capacity Factor",
         description="The capacity factor at the given location",
-    )]
-
-    @field_validator("capacity_factor")
-    def check_capacity_factor(cls, value):
-        if value < 0 or value > 0.6:
-            raise ValueError("Capacity factor must be between 0 and 60%")
-        return value
+        ge=0,
+        le=1,
+    )]