feat/migrate windwatts-data to app backend

windwatts-api/Dockerfile

@@ -19,15 +19,6 @@ RUN apt-get update && apt-get upgrade -y && apt-get install -y \
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
-# Download and install windwatts_data.whl from s3
-ARG WINDWATTS_DATA_URL
-ARG WINDWATTS_DATA_VERSION=1.0.4
-ARG WINDWATTS_DATA_FILE=windwatts_data-${WINDWATTS_DATA_VERSION}-py3-none-any.whl
-ENV WINDWATTS_DATA_URL=${WINDWATTS_DATA_URL}
-RUN curl -o /tmp/windwatts_data-${WINDWATTS_DATA_VERSION}-py3-none-any.whl ${WINDWATTS_DATA_URL}${WINDWATTS_DATA_FILE} && \
-    pip install --no-cache-dir /tmp/${WINDWATTS_DATA_FILE} && \
-    rm /tmp/${WINDWATTS_DATA_FILE}
-
 # Copy the FastAPI application
 COPY . .
 

windwatts-api/app/config/model_config.py

@@ -75,6 +75,8 @@
         "schema": "quantile_yearly",
         "years": {"full": list(range(2013, 2024)), "sample": [2020, 2021, 2022, 2023]},
         "heights": {"windspeed": [30, 40, 50, 60, 80, 100], "winddirection": []},
+        "interpolation": True,
+        "grid": "era5",
         "grid_info": {
             "min_lat": 23.402,
             "min_long": -137.725,
@@ -96,6 +98,8 @@
             "windspeed": [40, 60, 80, 100, 120, 140, 160, 200],
             "winddirection": [],
         },
+        "interpolation": True,
+        "grid": "wtk",
         "grid_info": {
             "min_lat": 7.75129,
             "min_long": -179.99918,
@@ -116,6 +120,8 @@
         "schema": "quantile_atemporal",
         "years": {"full": list(range(2013, 2024)), "sample": []},
         "heights": {"windspeed": [30, 40, 50, 60, 80, 100], "winddirection": []},
+        "interpolation": True,
+        "grid": "era5",
         "grid_info": {
             "min_lat": 23.402,
             "min_long": -137.725,
@@ -137,6 +143,8 @@
             "windspeed": [10, 30, 40, 50, 60, 80, 100],
             "winddirection": [10, 100],
         },
+        "interpolation": False,
+        "grid": "era5",
         "grid_info": {
             "min_lat": 23.402,
             "min_long": -137.725,

windwatts-api/app/config_manager.py

@@ -1,7 +1,6 @@
 import os
 import json
 import boto3
-import tempfile
 
 
 class ConfigManager:
@@ -21,45 +20,34 @@ def __init__(
         self.local_config_path = local_config_path
         self.client = boto3.client("secretsmanager", region_name=region_name)
 
-    def get_config(self) -> str:
+    def get_config(self) -> dict:
         """
         Retrieve the secret from AWS Secrets Manager, environment variables, or the local configuration file.
-        :return: The path to the configuration file.
+        :return: The dict with config keys.
         """
         # Try to retrieve the secret from AWS Secrets Manager
         if self.secret_arn:
             try:
                 response = self.client.get_secret_value(SecretId=self.secret_arn)
-                secret = response["SecretString"]
-                config_data = json.loads(secret)
-
-                # Save the secret to a temporary file
-                temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".json")
-                temp_file.close()
-                with open(temp_file.name, "w") as f:
-                    json.dump(config_data, f)
-                return temp_file.name
+                return json.loads(response["SecretString"])
             except self.client.exceptions.ClientError as e:
                 print(f"Unable to retrieve secret: {e}")
 
         # Try to retrieve config from environment variables
         env_config = self._get_config_from_env()
         if env_config:
-            temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".json")
-            temp_file.close()
-            with open(temp_file.name, "w") as f:
-                json.dump(env_config, f)
             print("Config loaded from environment variables.")
-            return temp_file.name
+            return env_config
 
         # Fallback: return the path to the local configuration file
         if self.local_config_path and os.path.exists(self.local_config_path):
-            print("Local configuration file found.")
-            return self.local_config_path
-        else:
-            raise FileNotFoundError(
-                "Local configuration file not found and unable to retrieve secret from AWS Secrets Manager or environment variables."
-            )
+            with open(self.local_config_path, "r") as f:
+                print(f"Loaded config from local path '{self.local_config_path}'")
+                return json.load(f)
+
+        raise FileNotFoundError(
+            "Local configuration file not found and unable to retrieve secret from AWS Secrets Manager or environment variables."
+        )
 
     def _get_config_from_env(self):
         """

windwatts-api/app/controllers/era5_data_controller.py

@@ -42,10 +42,10 @@
     # Initialize DataFetchers
     # s3_data_fetcher = S3DataFetcher("WINDWATTS_S3_BUCKET_NAME")
     athena_data_fetcher_era5 = AthenaDataFetcher(
-        athena_config=athena_config, source_key="era5"
+        athena_config=athena_config, model_key="era5-quantiles"
     )
     athena_data_fetcher_ensemble = AthenaDataFetcher(
-        athena_config=athena_config, source_key="ensemble"
+        athena_config=athena_config, model_key="ensemble-quantiles"
     )
     s3_data_fetcher_era5 = S3DataFetcher(
         bucket_name="windwatts-era5",

windwatts-api/app/controllers/wind_data_controller.py

@@ -26,6 +26,8 @@
 
 from app.power_curve.global_power_curve_manager import power_curve_manager
 
+from app.spatial.global_spatial_manager import init_spatial
+
 from app.schemas import (
     AvailableTurbinesResponse,
     RoseRequestPayload,
@@ -60,15 +62,18 @@
     )
     athena_config = config_manager.get_config()
 
+    # Initialize spatial
+    init_spatial()
+
     # Initialize Athena data fetchers
     athena_data_fetchers["era5-quantiles"] = AthenaDataFetcher(
-        athena_config=athena_config, source_key="era5"
+        athena_config=athena_config, model_key="era5-quantiles"
     )
     athena_data_fetchers["ensemble-quantiles"] = AthenaDataFetcher(
-        athena_config=athena_config, source_key="ensemble"
+        athena_config=athena_config, model_key="ensemble-quantiles"
     )
     athena_data_fetchers["wtk-timeseries"] = AthenaDataFetcher(
-        athena_config=athena_config, source_key="wtk"
+        athena_config=athena_config, model_key="wtk-timeseries"
     )
 
     # Initialize S3 data fetchers

windwatts-api/app/controllers/wtk_data_controller.py

@@ -44,7 +44,7 @@
         bucket_name="wtk-led", prefix="1224", grid="wtk", s3_key_template="wtk"
     )
     athena_data_fetcher_wtk = AthenaDataFetcher(
-        athena_config=athena_config, source_key="wtk"
+        athena_config=athena_config, model_key="wtk-timeseries"
     )
     # db_manager = DatabaseManager()
     # db_data_fetcher = DatabaseDataFetcher(db_manager=db_manager)

windwatts-api/app/data_fetchers/athena_data_fetcher.py

@@ -1,50 +1,71 @@
+import boto3
+import time
+import pandas as pd
+from io import StringIO
+from collections import OrderedDict
+
 from .abstract_data_fetcher import AbstractDataFetcher
-from windwatts_data import (
-    WindwattsWTKClient,
-    WindwattsERA5Client,
-    WindwattsEnsembleClient,
+from app.spatial.global_spatial_manager import spatial_manager
+from app.utils.wind_processing import (
+    resolve_heights,
+    interpolate_windspeed,
+    aggregate,
+    aggregate_quantile,
 )
+from app.config.model_config import MODEL_CONFIG, TEMPORAL_SCHEMAS
 
 
 class AthenaDataFetcher(AbstractDataFetcher):
-    def __init__(self, athena_config: str, source_key: str):
+    def __init__(self, athena_config: dict, model_key: str):
         """
-        Initializes the AthenaDataFetcher with a single source_key like 'wtk', 'era5', or 'era5_bc'.
-        We infer the base family ('wtk' or 'era5') from the part before the first underscore.
+        Initializes the AthenaDataFetcher with a single model_key like 'wtk-timeseries', 'era5-quantiles', or 'ensemble-quantiles' with its respective Athena config.
 
         Args:
-            athena_config (str): Path to the Athena configuration file.
-            source_key (str): Key in the config that specifies which athena source.
+            athena_config (str): Full athena config dict from ConfigManager.get_config()
+            model_key (str): Key into config["sources"], e.g. "wtk-timeseries", "era5-quantiles", "ensemble-quantiles". Same as MODEL_CONFIG keys.
         """
-        # self.data_type = data_type.lower()
-        self.source_key = source_key.lower()
-        self.base_type = self.source_key.split("_", 1)[
-            0
-        ]  # 'wtk' or 'era5' (from 'era5_bc' too)
-
-        if self.base_type == "wtk":
-            print(f"Initializing WTK Client with Source Key: {self.source_key}")
-            self.client = WindwattsWTKClient(
-                config_path=athena_config, source_key=self.source_key
-            )  # source_key  "wtk"
-        elif self.base_type == "era5":
-            print(f"Initializing ERA5 Client with Source Key: {self.source_key}")
-            self.client = WindwattsERA5Client(
-                config_path=athena_config, source_key=self.source_key
-            )  # source_key  "era5" or "era5_bc"
-        elif self.base_type == "ensemble":
-            print(f"Initializing Ensemble Client with Source Key: {self.source_key}")
-            self.client = WindwattsEnsembleClient(
-                config_path=athena_config, source_key=self.source_key
-            )  # source_key  "ensemble"
-        else:
-            raise ValueError(f"Unsupported base dataset: {self.base_type}")
+        print(f"Initializing Athene Data Fetcher for '{model_key}'")
+        self.model_key = model_key
+        source = athena_config["sources"][model_key]
+
+        self.database = athena_config["database"]
+        self.workgroup = athena_config["athena_workgroup"]
+        self.output_bucket = athena_config["output_bucket"]
+        self.output_location = athena_config["output_location"]
+        self.table = source["athena_table_name"]
+        self.alt_table = source.get("alt_athena_table_name", "")
+
+        self.athena = boto3.client("athena", region_name=athena_config["region_name"])
+        self.s3 = boto3.client("s3", region_name=athena_config["region_name"])
+
+        self._df_cache: OrderedDict[str, pd.DataFrame] = OrderedDict()
+        self._df_cache_maxsize = 100
+
+    def _schema(self) -> str:
+        return MODEL_CONFIG[self.model_key]["schema"]
+
+    def _available_heights(self) -> list[int]:
+        return MODEL_CONFIG[self.model_key]["heights"]["windspeed"]
+
+    def _cache_df(self, grid_idx: str) -> pd.DataFrame:
+        if grid_idx in self._df_cache:
+            self._df_cache.move_to_end(grid_idx)
+            return self._df_cache[grid_idx].copy()
+        query = f"SELECT * FROM {self.table} WHERE index = '{grid_idx}'"
+        df = self._execute_athena(query)
+        self._df_cache[grid_idx] = df
+        if len(self._df_cache) > self._df_cache_maxsize:
+            self._df_cache.popitem(last=False)
+        return df.copy()
 
     def fetch_data(
         self, lat: float, lng: float, height: int, period: str = "all"
     ) -> dict:
         """
-        Fetch aggregated wind data using the configured client.
+        Fetch aggregated wind data for a location.
+        Selects only the columns needed for the requested height and period.
+        Applies interpolation if height is not natively in the dataset.
+        Routes to the appropriate aggregation strategy (timeseries vs quantile).
 
         Args:
             lat (float): Latitude of the location.
@@ -56,28 +77,21 @@ def fetch_data(
 
         Returns:
             dict: Fetched aggregated wind data.
-
-        Raises:
-            ValueError: If the period is not supported for the selected client.
         """
-        if period == "all":
-            return self.client.fetch_global_avg_at_height(
-                lat=lat, long=lng, height=height
-            )
-        elif period == "annual":
-            return self.client.fetch_yearly_avg_at_height(
-                lat=lat, long=lng, height=height
-            )
-        elif period == "monthly":
-            return self.client.fetch_monthly_avg_at_height(
-                lat=lat, long=lng, height=height
-            )
-        elif period == "hourly":
-            return self.client.fetch_hourly_avg_at_height(
-                lat=lat, long=lng, height=height
+        grid_idx, _, _ = spatial_manager.find_nearest(lat, lng, self.model_key)
+        height_info = resolve_heights(height, self._available_heights())
+        df = self._cache_df(grid_idx)
+
+        if not height_info["exact"]:
+            df = interpolate_windspeed(
+                df, height, height_info["lower"], height_info["upper"]
             )
-        else:
-            raise ValueError(f"Invalid period: {period}")
+
+        schema = self._schema()
+        if schema in ("quantile_yearly", "quantile_atemporal"):
+            use_swi = TEMPORAL_SCHEMAS[schema]["processing"]["use_swi"]
+            return aggregate_quantile(df, height, period, use_swi=use_swi)
+        return aggregate(df, height, period)
 
     def fetch_raw(self, lat: float, lng: float, height: int):
         """
@@ -91,7 +105,16 @@ def fetch_raw(self, lat: float, lng: float, height: int):
         Returns:
             DataFrame: Raw wind data without aggregation.
         """
-        return self.client.fetch_df(lat=lat, long=lng, height=height)
+        grid_idx, _, _ = spatial_manager.find_nearest(lat, lng, self.model_key)
+        height_info = resolve_heights(height, self._available_heights())
+        df = self._cache_df(grid_idx)
+
+        if not height_info["exact"]:
+            df = interpolate_windspeed(
+                df, height, height_info["lower"], height_info["upper"]
+            )
+
+        return df
 
     def find_nearest_locations(self, lat: float, lng: float, n_neighbors: int = 1):
         """
@@ -112,5 +135,56 @@ def find_nearest_locations(self, lat: float, lng: float, n_neighbors: int = 1):
             :rtype: list[tuple[str, float, float]]
         """
         # A list of tuples where each tuple contains: (grid_index, latitude, longitude)
-        tuples = self.client.find_n_nearest_locations(lat, lng, n_neighbors)
+        tuples = spatial_manager.find_n_nearest(lat, lng, self.model_key, n_neighbors)
         return tuples
+
+    def _execute_athena(
+        self, query: str, params: list[str] | None = None
+    ) -> pd.DataFrame:
+        """Execute an Athena query and return results as a DataFrame.
+
+        Uses 7-day result reuse so repeated queries for the same location
+        resolve from cache server-side. Polls with exponential backoff,
+        checking immediately on the first attempt for fast cache hits.
+
+        Args:
+            query: SQL query string to execute.
+
+        Returns:
+            DataFrame parsed from the CSV result stored in S3.
+
+        Raises:
+            RuntimeError: If the query fails or is cancelled.
+        """
+        execution_id = self.athena.start_query_execution(
+            QueryString=query,
+            QueryExecutionContext={"Database": self.database},
+            ResultConfiguration={"OutputLocation": self.output_location},
+            ResultReuseConfiguration={
+                "ResultReuseByAgeConfiguration": {
+                    "Enabled": True,
+                    "MaxAgeInMinutes": 10080,
+                }
+            },
+            WorkGroup=self.workgroup,
+        )["QueryExecutionId"]
+
+        delay = 0
+        while True:
+            resp = self.athena.get_query_execution(QueryExecutionId=execution_id)
+            state = resp["QueryExecution"]["Status"]["State"]
+            if state == "SUCCEEDED":
+                break
+            if state in ("FAILED", "CANCELLED"):
+                reason = resp["QueryExecution"]["Status"].get("StateChangeReason", "")
+                raise RuntimeError(f"Athena query {state}: {reason}")
+            if delay == 0:
+                delay = 0.15
+            else:
+                delay = min(delay * 2, 5.0)
+            time.sleep(delay)
+
+        output = resp["QueryExecution"]["ResultConfiguration"]["OutputLocation"]
+        bucket, key = output.replace("s3://", "").split("/", 1)
+        obj = self.s3.get_object(Bucket=bucket, Key=key)
+        return pd.read_csv(StringIO(obj["Body"].read().decode("utf-8")))