diff --git a/pyproject.toml b/pyproject.toml index bd58c1e..a206eaa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ authors = [ classifiers = ["Programming Language :: Python :: 3"] dependencies = [ "fastapi >= 0.105.0", + "pytz >= 2023.3", "structlog >= 23.2.0", "uvicorn >= 0.24.0", ] diff --git a/src/india_api/internal/__init__.py b/src/india_api/internal/__init__.py index 185c34d..f347f35 100644 --- a/src/india_api/internal/__init__.py +++ b/src/india_api/internal/__init__.py @@ -2,7 +2,7 @@ from .models import ( DatabaseInterface, - PredictedYield, + PredictedPower, ) from . import ( @@ -11,7 +11,7 @@ ) __all__ = [ - "PredictedYield", + "PredictedPower", "DatabaseInterface", "inputs", "service", diff --git a/src/india_api/internal/inputs/dummydb/client.py b/src/india_api/internal/inputs/dummydb/client.py index e3e5a63..365fecf 100644 --- a/src/india_api/internal/inputs/dummydb/client.py +++ b/src/india_api/internal/inputs/dummydb/client.py @@ -8,7 +8,7 @@ from ._models import DummyDBPredictedYield # step defines the time interval between each data point -step: dt.timedelta = dt.timedelta(minutes=5) +step: dt.timedelta = dt.timedelta(minutes=15) class Client(internal.DatabaseInterface): @@ -17,7 +17,7 @@ class Client(internal.DatabaseInterface): def get_predicted_solar_yields_for_location( self, location: str, - ) -> list[internal.PredictedYield]: + ) -> list[internal.PredictedPower]: """Gets the predicted solar yields for a location. Args: @@ -26,24 +26,24 @@ def get_predicted_solar_yields_for_location( # Get the window start, end = _getWindow() numSteps = int((end - start) / step) - yields: list[internal.PredictedYield] = [] + values: list[internal.PredictedPower] = [] for i in range(numSteps): time = start + i * step _yield = _basicSolarYieldFunc(int(time.timestamp())) - yields.append( - internal.PredictedYield( + values.append( + internal.PredictedPower( Time=time, - YieldKW=int(_yield.YieldKW), + PowerKW=int(_yield.YieldKW), ), ) - return yields + return values def get_predicted_wind_yields_for_location( self, location: str, - ) -> list[internal.PredictedYield]: + ) -> list[internal.PredictedPower]: """Gets the predicted wind yields for a location. Args: @@ -52,57 +52,57 @@ def get_predicted_wind_yields_for_location( # Get the window start, end = _getWindow() numSteps = int((end - start) / step) - yields: list[internal.PredictedYield] = [] + values: list[internal.PredictedPower] = [] for i in range(numSteps): time = start + i * step _yield = _basicWindYieldFunc(int(time.timestamp())) - yields.append( - internal.PredictedYield( + values.append( + internal.PredictedPower( Time=time, - YieldKW=int(_yield.YieldKW), + PowerKW=int(_yield.YieldKW), ), ) - return yields + return values - def get_actual_solar_yields_for_location(self, location: str) -> list[internal.PredictedYield]: + def get_actual_solar_yields_for_location(self, location: str) -> list[internal.PredictedPower]: """Gets the actual solar yields for a location.""" # Get the window start, end = _getWindow() numSteps = int((end - start) / step) - yields: list[internal.PredictedYield] = [] + values: list[internal.PredictedPower] = [] for i in range(numSteps): time = start + i * step _yield = _basicSolarYieldFunc(int(time.timestamp())) - yields.append( - internal.PredictedYield( + values.append( + internal.PredictedPower( Time=time, - YieldKW=int(_yield.YieldKW), + PowerKW=int(_yield.YieldKW), ), ) - return yields + return values - def get_actual_wind_yields_for_location(self, location: str) -> list[internal.PredictedYield]: + def get_actual_wind_yields_for_location(self, location: str) -> list[internal.PredictedPower]: """Gets the actual wind yields for a location.""" # Get the window start, end = _getWindow() numSteps = int((end - start) / step) - yields: list[internal.PredictedYield] = [] + values: list[internal.PredictedPower] = [] for i in range(numSteps): time = start + i * step _yield = _basicWindYieldFunc(int(time.timestamp())) - yields.append( - internal.PredictedYield( + values.append( + internal.PredictedPower( Time=time, - YieldKW=int(_yield.YieldKW), + PowerKW=int(_yield.YieldKW), ), ) - return yields + return values def get_wind_regions(self) -> list[str]: """Gets the valid wind regions.""" diff --git a/src/india_api/internal/models.py b/src/india_api/internal/models.py index fb20a81..cf22cee 100644 --- a/src/india_api/internal/models.py +++ b/src/india_api/internal/models.py @@ -6,17 +6,24 @@ from pydantic import BaseModel -class PredictedYield(BaseModel): - """Defines the model for a predicted yield returned by the API.""" +class PredictedPower(BaseModel): + """Defines the data structure for a predicted power value returned by the API.""" - YieldKW: int + PowerKW: int Time: dt.datetime + def to_timezone(self, tz: dt.timezone) -> "PredictedPower": + """Converts the time of this predicted power value to the given timezone.""" + return PredictedPower( + PowerKW=self.PowerKW, + Time=self.Time.astimezone(tz=tz), + ) -class ActualYield(BaseModel): - """Defines the model for an actual yield returned by the API.""" - YieldKW: int +class ActualPower(BaseModel): + """Defines the data structure for an actual power value returned by the API.""" + + PowerKW: int Time: dt.datetime @@ -24,22 +31,22 @@ class DatabaseInterface(abc.ABC): """Defines the interface for a generic database connection.""" @abc.abstractmethod - def get_predicted_solar_yields_for_location(self, location: str) -> list[PredictedYield]: + def get_predicted_solar_yields_for_location(self, location: str) -> list[PredictedPower]: """Returns a list of predicted solar yields for a given location.""" pass @abc.abstractmethod - def get_actual_solar_yields_for_location(self, location: str) -> list[ActualYield]: + def get_actual_solar_yields_for_location(self, location: str) -> list[ActualPower]: """Returns a list of actual solar yields for a given location.""" pass @abc.abstractmethod - def get_predicted_wind_yields_for_location(self, location: str) -> list[PredictedYield]: + def get_predicted_wind_yields_for_location(self, location: str) -> list[PredictedPower]: """Returns a list of predicted wind yields for a given location.""" pass @abc.abstractmethod - def get_actual_wind_yields_for_location(self, location: str) -> list[ActualYield]: + def get_actual_wind_yields_for_location(self, location: str) -> list[ActualPower]: """Returns a list of actual wind yields for a given location.""" pass diff --git a/src/india_api/internal/service/server.py b/src/india_api/internal/service/server.py index 93f3180..7de3d42 100644 --- a/src/india_api/internal/service/server.py +++ b/src/india_api/internal/service/server.py @@ -1,6 +1,7 @@ """Defines the routes of the API.""" import datetime as dt +import pytz from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, status @@ -8,17 +9,23 @@ from india_api.internal import ( DatabaseInterface, - PredictedYield, + PredictedPower, ) + +local_tz = pytz.timezone("Asia/Kolkata") + server = FastAPI() + def get_db_client() -> DatabaseInterface: """Dependency injection for the database client.""" return DatabaseInterface() + DBClientDependency = Annotated[DatabaseInterface, Depends(get_db_client)] + def validate_source(source: str) -> str: """Validate the source parameter.""" if source not in ["wind", "solar"]: @@ -28,15 +35,34 @@ def validate_source(source: str) -> str: ) return source + ValidSourceDependency = Annotated[str, Depends(validate_source)] + +route_tags = [ + { + "name": "Forecast Routes", + "description": "Routes for fetching forecast power values.", + }, + { + "name": "API Information", + "description": "Routes pertaining to the usage and status of the API.", + }, +] + + +# === API ROUTES ============================================================== + + class GetHealthResponse(BaseModel): """Model for the health endpoint response.""" status: int + @server.get( "/health", + tags=["API Information"], status_code=status.HTTP_200_OK, ) def get_health_route() -> GetHealthResponse: @@ -44,85 +70,85 @@ def get_health_route() -> GetHealthResponse: return GetHealthResponse(status=status.HTTP_200_OK) -class GetHistoricTimeseriesResponse(BaseModel): - """Model for the historic timeseries endpoint response.""" +class GetHistoricGenerationResponse(BaseModel): + """Model for the historic generation endpoint response.""" - yields: list[PredictedYield] + values: list[PredictedPower] @server.get( - "/{source}/{region}/historic_timeseries", + "/{source}/{region}/historic_generation", + tags=["Forecast Routes"], status_code=status.HTTP_200_OK, ) def get_historic_timeseries_route( source: ValidSourceDependency, region: str, db: DBClientDependency, -) -> GetHistoricTimeseriesResponse: - """Function for the historic timeseries route.""" - yields: list[PredictedYield] = [] +) -> GetHistoricGenerationResponse: + """Function for the historic generation route.""" + values: list[PredictedPower] = [] try: if source == "wind": - yields = db.get_predicted_wind_yields_for_location(location=region) + values = db.get_predicted_wind_yields_for_location(location=region) elif source == "solar": - yields = db.get_predicted_solar_yields_for_location(location=region) + values = db.get_predicted_solar_yields_for_location(location=region) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error getting solar yields: {e}", ) from e - return GetHistoricTimeseriesResponse( - yields=[ - y for y in yields - if y.Time < dt.datetime.now(tz=dt.UTC) - ], + return GetHistoricGenerationResponse( + values=[y.to_timezone(tz=local_tz) for y in values if y.Time < dt.datetime.now(tz=dt.UTC)], ) -class GetForecastTimeseriesResponse(BaseModel): - """Model for the forecast timeseries endpoint response.""" +class GetForecastGenerationResponse(BaseModel): + """Model for the forecast generation endpoint response.""" + + values: list[PredictedPower] - yields: list[PredictedYield] @server.get( - "/{source}/{region}/forecast_timeseries", + "/{source}/{region}/forecast_generation", + tags=["Forecast Routes"], status_code=status.HTTP_200_OK, ) def get_forecast_timeseries_route( source: ValidSourceDependency, region: str, db: DBClientDependency, -) -> GetForecastTimeseriesResponse: - """Function for the forecast timeseries route.""" - yields: list[PredictedYield] = [] +) -> GetForecastGenerationResponse: + """Function for the forecast generation route.""" + values: list[PredictedPower] = [] try: if source == "wind": - yields = db.get_predicted_wind_yields_for_location(location=region) + values = db.get_predicted_wind_yields_for_location(location=region) elif source == "solar": - yields = db.get_predicted_solar_yields_for_location(location=region) + values = db.get_predicted_solar_yields_for_location(location=region) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error getting yields: {e}", ) from e - return GetForecastTimeseriesResponse( - yields=[ - y for y in yields - if y.Time >= dt.datetime.now(tz=dt.UTC) - ], + return GetForecastGenerationResponse( + values=[y.to_timezone(tz=local_tz) for y in values if y.Time >= dt.datetime.now(tz=dt.UTC)], ) + class GetSourcesResponse(BaseModel): """Model for the sources endpoint response.""" sources: list[str] + @server.get( "/sources", + tags=["API Information"], status_code=status.HTTP_200_OK, ) def get_sources_route() -> GetSourcesResponse: @@ -135,8 +161,10 @@ class GetRegionsResponse(BaseModel): regions: list[str] + @server.get( "/{source}/regions", + tags=["API Information"], status_code=status.HTTP_200_OK, ) def get_regions_route( @@ -149,4 +177,3 @@ def get_regions_route( elif source == "solar": regions = db.get_solar_regions() return GetRegionsResponse(regions=regions) -