Google Floods Part 1: Adding backend + creating Gauge Points; COUNTRY=cambodia

api/app/googleflood.py

@@ -0,0 +1,57 @@
+"""Get data from Humanitarian Data Cube (HDC) API"""
+import logging
+from os import getenv
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+GOOGLE_FLOODS_API_KEY = getenv("GOOGLE_FLOODS_API_KEY", "")
+if GOOGLE_FLOODS_API_KEY == "":
+    logger.warning("Missing backend parameter: GOOGLE_FLOODS_API_KEY")
+
+def format_to_geojson(data):
+    geojson = {
+        "type": "Feature",
+        "geometry": {
+            "type": "Point",
+            "coordinates": [data["gaugeLocation"]["longitude"], data["gaugeLocation"]["latitude"]]
+        },
+        "properties": {
+            "gaugeId": data["gaugeId"],
+            "issuedTime": data["issuedTime"],
+            # "forecastTimeRange": data["forecastTimeRange"],
+            # "forecastChange": data["forecastChange"],
+            # "forecastTrend": data["forecastTrend"],
+            "severity": data["severity"],
+            "source": data["source"],
+            "qualityVerified": data["qualityVerified"]
+        }
+    }
+    if "inundationMapSet" in data:
+        geojson["properties"]["inundationMapSet"] = data["inundationMapSet"]
+    return geojson
+
+
+def get_google_floods_gauges(
+    iso2: str,
+    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(
+        URL,
+        json={'regionCode': iso2}
+    ).json().get('floodStatuses', [])
+
+    if asGeojson:
+        geojson_feature_collection = {
+            "type": "FeatureCollection",
+            "features": [format_to_geojson(data) for data in response]
+        }
+        return geojson_feature_collection
+    return response
+
+

api/app/main.py

@@ -14,6 +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.hdc import get_hdc_stats
 from app.kobo import get_form_dates, get_form_responses, parse_datetime_params
 from app.models import AcledRequest, RasterGeotiffModel
@@ -412,3 +413,10 @@ def post_raster_geotiff(raster_geotiff: RasterGeotiffModel):
     return JSONResponse(
         content={"download_url": presigned_download_url}, status_code=200
     )
+
+
+@app.get("/google-floods-gauges")
+def get_google_floods_gauges_api(
+    iso2: str,
+):
+    return get_google_floods_gauges(iso2)

api/app/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+log_cli = true
+log_cli_level = INFO

api/app/tests/test_google_floods_api.py

@@ -0,0 +1,20 @@
+from app.googleflood import get_google_floods_gauges
+from app.main import app
+from fastapi.testclient import TestClient
+import logging, json
+
+client = TestClient(app)
+
+def test_get_google_floods_gauges():
+    gauges = get_google_floods_gauges("BD")
+    assert len(gauges) > 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
+
+