Add MSAVI index to backend

backend/requirements.txt

@@ -2,5 +2,5 @@ earthengine-api
 python-dotenv
 pandas
 plotly
-fastapi
+fastapi[standard]
 uvicorn

backend/src/constants.py

@@ -29,6 +29,7 @@ class Unit(str, Enum):
     MM = "mm"
     PERCENT = "%"
     M3 = "m³/m³"
+    DIMENSIONLESS = "Dimensionless"
 
 
 class LocationPolygon(Enum):

backend/src/routes/ndvi_router.py → backend/src/controller/sat_index_controller.py

@@ -1,35 +1,36 @@
 from datetime import datetime, timezone
 
-from fastapi import APIRouter, Query
+from fastapi import Query
 from fastapi.responses import JSONResponse
 from src.constants import (
+    IndexType,
     AggregationMethod,
     LocationName,
     TemporalResolution,
     Unit,
 )
-from src.service import ndvi_service
+from src.service import sat_index_service
 from src.utils.temporal import get_optimistic_rounding
-from src.validation.models import NDVIResponse
 from src.validation.utils import (
     validate_timestamp_in_range_of_S2_imagery,
     validate_timestamp_start_date_before_end_date,
 )
 
-ndvi_router = APIRouter()
 
-
-@ndvi_router.get("/ndvi", response_model=NDVIResponse)
-async def get_temperature_data(
+async def sat_index_controller(
+    sat_index_type: IndexType, 
     startDate: int = Query(...,
+
                            description="Start date as UNIX timestamp in seconds"),
     endDate: int = Query(...,
+
                          description="End date as UNIX timestamp in seconds"),
     location: LocationName = Query(..., description="Location name"),
     temporalResolution: TemporalResolution = Query(
         ..., description="Temporal resolution"
     ),
     aggregation: AggregationMethod = Query(...,
+
                                            description="Aggregation method"),
 ):
 
@@ -42,7 +43,7 @@ async def get_temperature_data(
         start_date_dt, end_date_dt, temporalResolution
     )
 
-    data = ndvi_service(
+    data = sat_index_service(
         location=location,
         temporal_resolution=temporalResolution,
         aggregation_method=aggregation,
@@ -62,4 +63,4 @@ async def get_temperature_data(
         "data": data,
     }
 
-    return JSONResponse(content=response)
+    return JSONResponse(content=response)

backend/src/gee/caching_script.py

@@ -0,0 +1,30 @@
+import json
+
+content = [
+    {"timestamp": 1493251200, "value": 0.6265267304234295},
+    {"timestamp": 1494720000, "value": 0.68603163673333},
+    {"timestamp": 1494979200, "value": 0.755257128311451},
+]
+var_name = "msavi_daily_cache"
+file_name = "../cache/temp_cache.py"
+INDENT = "    "
+
+def combine_chache_with_update():
+    # update existing cache
+    return
+
+def write_year(file, year, results):
+    file.write(f"{INDENT}# Year {year}\n")
+
+def write_results_to_cache(results: list, var_name: str, file_name: str):
+    with open(file_name, "w") as file:
+        file.write(f"{var_name} = [\n")
+        for day in results:
+            file.write(f"{INDENT}{json.dumps(day)},\n")
+        file.write("]\n")
+    file.close()
+
+    return
+
+
+# write_results_to_cache(content, var_name, file_name)

backend/src/gee/ndvi.py → backend/src/gee/sat_index_info.py

@@ -1,19 +1,38 @@
 import ee
+from ..constants import IndexType
 
 INDEX_FEATURE_LABEL = "indexing_value"
 TIMESTAMP_FEATURE_LABEL = "start_of_day_timestamp"
 INDEX_NULL_STRING = "NULL"
 
 
-def calculate_mean_ndvi_GEE_SERVER(image: ee.Image, aoi: ee.Geometry.Polygon):
-    ndvi = image.normalizedDifference(["B8", "B4"]).rename("NDVI")
-    mean_ndvi = ndvi.reduceRegion(
+def get_index_image_by_index_type(index_type: IndexType, image: ee.Image):
+    match index_type:
+        case IndexType.NDVI:
+            return image.normalizedDifference(["B8", "B4"]).rename("NDVI")
+        case IndexType.MSAVI:
+            return image.expression(
+                expression="((2 * NIR + 1) - ((2 * NIR + 1)**2 - 8 * (NIR - RED))**0.5) / 2",
+                opt_map={
+                    "NIR": image.select("B4"),
+                    "RED": image.select("B8"),
+                },
+            ).rename("MSAVI")
+        case _:
+            return None
+
+
+def calculate_mean_index_GEE_SERVER(image: ee.Image, aoi: ee.Geometry.Polygon, index_type: IndexType):
+    index_image = get_index_image_by_index_type(index_type, image)
+    mean_index = index_image.reduceRegion(
         reducer=ee.Reducer.mean(), geometry=aoi, scale=10, maxPixels=1e8
-    ).get("NDVI")
-    mean_ndvi = ee.Algorithms.If(
-        ee.Algorithms.IsEqual(mean_ndvi, None), INDEX_NULL_STRING, mean_ndvi
+    ).get(index_type.value)
+
+    mean_index = ee.Algorithms.If(
+        ee.Algorithms.IsEqual(mean_index, None), INDEX_NULL_STRING, mean_index
     )
-    return image.set(INDEX_FEATURE_LABEL, mean_ndvi)
+    return image.set(INDEX_FEATURE_LABEL, mean_index)
+
 
 
 def calculate_start_of_day_timestamp_GEE_SERVER(image: ee.Image):
@@ -27,28 +46,28 @@ def calculate_start_of_day_timestamp_GEE_SERVER(image: ee.Image):
     return image.set(TIMESTAMP_FEATURE_LABEL, start_of_day)
 
 
-def get_ndvi_info(
-    image_collection: ee.ImageCollection, coordinates: ee.Geometry.Polygon
+def get_sat_index_info(
+    image_collection: ee.ImageCollection, coordinates: ee.Geometry.Polygon, index_type: IndexType
 ):
     aoi = ee.Geometry.Polygon(coordinates)
 
     # Setting indexing values Server side
-    image_collection_with_ndvi = image_collection.map(
-        lambda img: calculate_mean_ndvi_GEE_SERVER(img, aoi)
+    image_collection_with_index = image_collection.map(
+        lambda img: calculate_mean_index_GEE_SERVER(img, aoi, index_type)
     )
 
     # Setting timestamps Server side
-    image_collection_with_timestamp_and_ndvi = image_collection_with_ndvi.map(
+    image_collection_with_timestamp_and_index = image_collection_with_index.map(
         lambda img: calculate_start_of_day_timestamp_GEE_SERVER(img)
     )
 
     #  Getting indexing values to the client side
-    index_value_list = image_collection_with_timestamp_and_ndvi.aggregate_array(
+    index_value_list = image_collection_with_timestamp_and_index.aggregate_array(
         INDEX_FEATURE_LABEL
     ).getInfo()
 
     # Getting timestamps to the client
-    timestamp_list = image_collection_with_timestamp_and_ndvi.aggregate_array(
+    timestamp_list = image_collection_with_timestamp_and_index.aggregate_array(
         TIMESTAMP_FEATURE_LABEL
     ).getInfo()
 

backend/src/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from src.routes.ndvi_router import ndvi_router
+from src.routes.sat_index_router import sat_index_router
 
 from .weather.router import router as weather_router
 
@@ -17,10 +17,11 @@
     allow_headers=["*"],
 )
 
+
 @app.get("/")
 def read_root():
     return {"Hello": "World"}
 
 
 app.include_router(weather_router, prefix="/weather", tags=["Weather Data"])
-app.include_router(ndvi_router, prefix="/index", tags=["NDVI Data"])
+app.include_router(sat_index_router, prefix="/index", tags=["Vegetation Indices"])

backend/src/routes/sat_index_router.py

@@ -0,0 +1,126 @@
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Query
+from fastapi.responses import JSONResponse
+from src.constants import (
+    AggregationMethod,
+    LocationName,
+    TemporalResolution,
+    Unit,
+    IndexType,
+)
+from src.service import sat_index_service
+from src.utils.temporal import get_optimistic_rounding
+from src.validation.models import NDVIResponse, MSAVIResponse
+from src.validation.utils import (
+    validate_timestamp_in_range_of_S2_imagery,
+    validate_timestamp_start_date_before_end_date,
+)
+
+sat_index_router = APIRouter()
+
+
+@sat_index_router.get("/ndvi", response_model=NDVIResponse)
+async def get_ndvi_data(
+    startDate: int = Query(
+        ...,
+        description="First date of requested date range in UNIX timestamp as seconds",
+    ),
+    endDate: int = Query(
+        ...,
+        description="Last date of requested date range in UNIX timestamp as seconds",
+    ),
+    location: LocationName = Query(..., description="Name of the requested location"),
+    temporalResolution: TemporalResolution = Query(
+        ...,
+        description="Time interval that a single data point should represent e.g. one month",
+    ),
+    aggregation: AggregationMethod = Query(
+        ...,
+        description="Method of aggregating available data into a single datapoint to represent the selected time interval e.g. mean average",
+    ),
+):
+    validate_timestamp_start_date_before_end_date(startDate, endDate)
+    validate_timestamp_in_range_of_S2_imagery(startDate, endDate)
+    start_date_dt = datetime.fromtimestamp(startDate, tz=timezone.utc)
+    end_date_dt = datetime.fromtimestamp(endDate, tz=timezone.utc)
+
+    rounded_start_date, rounded_end_date = get_optimistic_rounding(
+        start_date_dt, end_date_dt, temporalResolution
+    )
+
+    data = sat_index_service(
+        location=location,
+        temporal_resolution=temporalResolution,
+        aggregation_method=aggregation,
+        start_date=rounded_start_date,
+        end_date=rounded_end_date,
+        index_type=IndexType.NDVI,
+    )
+
+    response = {
+        "meta": {
+            "location": LocationName[location].value,
+            "startDate": int(rounded_start_date.timestamp()),
+            "endDate": int(rounded_end_date.timestamp()),
+            "temporalResolution": TemporalResolution[temporalResolution].value,
+            "aggregation": AggregationMethod[aggregation].value,
+            "unit": Unit.NORMALIZED_DIFFERENCE.value,
+        },
+        "data": data,
+    }
+
+    return JSONResponse(content=response)
+
+
+@sat_index_router.get("/msavi", response_model=MSAVIResponse)
+async def get_msavi_data(
+    startDate: int = Query(
+        ...,
+        description="First date of requested date range in UNIX timestamp as seconds",
+    ),
+    endDate: int = Query(
+        ...,
+        description="Last date of requested date range in UNIX timestamp as seconds",
+    ),
+    location: LocationName = Query(..., description="Name of the requested location"),
+    temporalResolution: TemporalResolution = Query(
+        ...,
+        description="Time interval that a single data point should represent e.g. one month",
+    ),
+    aggregation: AggregationMethod = Query(
+        ...,
+        description="Method of aggregating available data into a single datapoint to represent the selected time interval e.g. mean average",
+    ),
+):
+    validate_timestamp_start_date_before_end_date(startDate, endDate)
+    validate_timestamp_in_range_of_S2_imagery(startDate, endDate)
+    start_date_dt = datetime.fromtimestamp(startDate, tz=timezone.utc)
+    end_date_dt = datetime.fromtimestamp(endDate, tz=timezone.utc)
+
+    rounded_start_date, rounded_end_date = get_optimistic_rounding(
+        start_date_dt, end_date_dt, temporalResolution
+    )
+
+    data = sat_index_service(
+        location=location,
+        temporal_resolution=temporalResolution,
+        aggregation_method=aggregation,
+        start_date=rounded_start_date,
+        end_date=rounded_end_date,
+        index_type=IndexType.MSAVI,
+    )
+
+    response = {
+        "meta": {
+            "location": LocationName[location].value,
+            "startDate": int(rounded_start_date.timestamp()),
+            "endDate": int(rounded_end_date.timestamp()),
+            "temporalResolution": TemporalResolution[temporalResolution].value,
+            "aggregation": AggregationMethod[aggregation].value,
+            "unit": Unit.DIMENSIONLESS.value,
+        },
+        "data": data,
+    }
+
+    return JSONResponse(content=response)