Adding Google Floods Inundation API

api/app/googleflood.py

@@ -1,12 +1,17 @@
 """Get data from Humanitarian Data Cube (HDC) API"""
 import logging
-from os import getenv
-
+import os
 import requests
+import geopandas as gpd
+import pandas as pd
+import uuid
+
+from fiona.drvsupport import supported_drivers
+supported_drivers['LIBKML'] = 'rw'
 
 logger = logging.getLogger(__name__)
 
-GOOGLE_FLOODS_API_KEY = getenv("GOOGLE_FLOODS_API_KEY", "")
+GOOGLE_FLOODS_API_KEY = os.getenv("GOOGLE_FLOODS_API_KEY", "")
 if GOOGLE_FLOODS_API_KEY == "":
     logger.warning("Missing backend parameter: GOOGLE_FLOODS_API_KEY")
 
@@ -20,6 +25,8 @@ def format_to_geojson(data):
         "properties": {
             "gaugeId": data["gaugeId"],
             "issuedTime": data["issuedTime"],
+            # not present on all data tested at this point
+            # handle in the future
             # "forecastTimeRange": data["forecastTimeRange"],
             # "forecastChange": data["forecastChange"],
             # "forecastTrend": data["forecastTrend"],
@@ -38,7 +45,6 @@ def get_google_floods_gauges(
     asGeojson: bool = True,
 ):
     """Get statistical charts data"""
-    logging.info(GOOGLE_FLOODS_API_KEY)
 
     URL = f'https://floodforecasting.googleapis.com/v1/floodStatus:searchLatestFloodStatusByArea?key={GOOGLE_FLOODS_API_KEY}'
     response = requests.post(
@@ -55,3 +61,45 @@ def get_google_floods_gauges(
     return response
 
 
+from pydantic import BaseModel
+from typing import List
+
+class InundationMap(BaseModel):
+    level: str
+    serializedPolygonId: str
+
+class InundationMapSet(BaseModel):
+    inundationMaps: List[InundationMap]
+
+
+def get_google_floods_inundations(
+    inundationMapSet: InundationMapSet,
+) -> gpd.GeoDataFrame:
+    """Get statistical charts data"""
+    level_to_kml = dict()
+    URL = 'https://floodforecasting.googleapis.com/v1/serializedPolygons/{serializedPolygonId}?key={key}'
+    for inundationMap in inundationMapSet:
+        response = requests.get(
+            URL.format(
+                serializedPolygonId=inundationMap['serializedPolygonId'],
+                key = GOOGLE_FLOODS_API_KEY
+            )
+        ).json()
+        level_to_kml[inundationMap['level']] = response['kml']
+
+    # Create a temp path for writing kmls
+    tmp_path = os.path.join(f'/tmp/google-floods/{str(uuid.uuid4())}')
+    if not os.path.exists(tmp_path):
+        os.makedirs(tmp_path)
+
+    gdf_buff = []
+    for level, kml in level_to_kml.items():
+        kml_path = os.path.join(tmp_path, f'{level}.kml')
+        with open(kml_path, 'w') as f:
+            f.write(kml)
+        gdf = gpd.read_file(kml_path, driver='KML')
+        gdf['level'] = level
+        gdf_buff.append(gdf)
+
+    gdf = pd.concat(gdf_buff)
+    return gdf

api/app/main.py

@@ -14,7 +14,7 @@
 from app.database.alert_model import AlchemyEncoder, AlertModel
 from app.database.database import AlertsDataBase
 from app.database.user_info_model import UserInfoModel
-from app.googleflood import get_google_floods_gauges
+from app.googleflood import get_google_floods_gauges, get_google_floods_inundations
 from app.hdc import get_hdc_stats
 from app.kobo import get_form_dates, get_form_responses, parse_datetime_params
 from app.models import AcledRequest, RasterGeotiffModel
@@ -419,4 +419,12 @@ def post_raster_geotiff(raster_geotiff: RasterGeotiffModel):
 def get_google_floods_gauges_api(
     iso2: str,
 ):
-    return get_google_floods_gauges(iso2)
+    return get_google_floods_gauges(iso2)
+
+
+@app.get('/google-floods-inundations')
+def get_google_floods_inundations_api(
+    inundationMapSet_JSONString: str,
+):
+    """Get statistical charts data"""
+    return get_google_floods_inundations(json.loads(inundationMapSet_JSONString)).to_json()

api/app/tests/test_google_floods_api.py

@@ -1,20 +1,32 @@
-from app.googleflood import get_google_floods_gauges
+from app.googleflood import get_google_floods_gauges, get_google_floods_inundations
 from app.main import app
 from fastapi.testclient import TestClient
 import logging, json
 
 client = TestClient(app)
 
+test_inundation_map_set = [{'level': 'HIGH', 'serializedPolygonId': '9631a3143f2543ae9664c4fba7298931'}, {'level': 'MEDIUM', 'serializedPolygonId': '93444f067f42409493fd907b2b0323b7'}, {'level': 'LOW', 'serializedPolygonId': '757ac60c06ea481aba384201a107d611'}]
+test_inundation_map_set_jsonstr = json.dumps(test_inundation_map_set)
+
 def test_get_google_floods_gauges():
     gauges = get_google_floods_gauges("BD")
     assert len(gauges) > 0
 
 
+def test_get_google_floods_inundations():
+    floods = get_google_floods_inundations(test_inundation_map_set)
+    assert floods.shape[0] > 0
+
+
 def test_get_google_floods_gauges_api():
     response = client.get("/google-floods-gauges?iso2=BD")
     assert response.status_code == 200
-    logging.info(json.dumps(response.json()))
-    # logging.info(response.json())
     assert len(response.json()) > 0
 
-
+
+def test_get_google_floods_inundation_api():
+    response = client.get(f"/google-floods-inundations?inundationMapSet_JSONString={test_inundation_map_set_jsonstr}")
+    assert response.status_code == 200
+    logging.info(json.dumps(response.json()))
+    # logging.info(response.json())
+    assert len(response.json()) > 0

api/pyproject.toml

@@ -7,7 +7,7 @@ license = "Apache 2.0"
 
 [tool.poetry.dependencies]
 PyJWT = "2.4.0"
-python = "<3.11,>=3.8.0"
+python = "<3.11,>=3.9.0"
 requests = "^2.32.0"
 playwright = "1.38.0"
 fastapi = "^0.109.1"
@@ -28,6 +28,7 @@ pystac = "^1.8.4"
 pytest-asyncio = "^0.21.1"
 pytest-playwright = "^0.4.3"
 rasterio = "^1.3.9"
+geopandas = "^1.0.1"
 
 [tool.poetry.group.dev.dependencies]
 black = "^24.3.0"