Add geocoder service url for emdat.

.github/workflows/ci.yml

@@ -25,6 +25,7 @@ env:
   EMDAT_AUTHORIZATION_KEY: dummy-value
   IDMC_CLIENT_ID: dummy-value
   IFRC_DATA_URL: dummy-url
+  GEOCODER_URL: dummy-url
   EOAPI_DOMAIN: https://montandon-eoapi.dummy.com
   GFD_CREDENTIAL: dummy-value
   GFD_SERVICE_ACCOUNT: dummy-value

apps/etl/transform/sources/emdat.py

@@ -1,15 +1,24 @@
 import json
 import logging
+import uuid
 
 from celery import shared_task
+from django.conf import settings
+from pystac_monty.geocoding import GAULGeocoder
 from pystac_monty.sources.emdat import EMDATDataSource, EMDATTransformer
 
-from apps.etl.models import ExtractionData
-from apps.etl.transform.sources.base import transform_data
+from apps.etl.models import ExtractionData, PyStacLoadData, Transform
 from apps.etl.utils import read_file_data
+from main.managers import BulkCreateManager
 
 logger = logging.getLogger(__name__)
 
+collection_and_item_type_map = {
+    "emdat-events": PyStacLoadData.ItemType.EVENT,
+    "emdat-hazards": PyStacLoadData.ItemType.HAZARD,
+    "emdat-impacts": PyStacLoadData.ItemType.IMPACT,
+}
+
 
 @shared_task
 def transform_emdat_data(extraction_id, **kwargs):
@@ -32,3 +41,60 @@ def transform_emdat_data(extraction_id, **kwargs):
         extraction_id,
         json_data,
     )
+
+
+@shared_task
+def transform_data(source, transformer, data_source, extraction_id, data):
+    logger.info(f"Transformation started for {source} data")
+    ext_instance = ExtractionData.objects.get(id=extraction_id)
+
+    # create transform object
+    transform_obj = Transform.objects.create(
+        extraction=ext_instance,
+        status=Transform.Status.PENDING,
+    )
+
+    # initialize bulk manager to create the PyStacLoadData in bulk.
+    bulk_mgr = BulkCreateManager(chunk_size=1000)
+
+    # XXX: Local file is used
+    # geocoder = GAULGeocoder(gpkg_path="/code/gaul2014_2015.gpkg")
+    geocoder = GAULGeocoder(service_base_url=settings.GEOCODER_URL)
+
+    try:
+        # Get transformer for each source and transform it to stac item..
+        transformer = transformer(data=data_source(source_url=ext_instance.url, data=data), geocoder=geocoder)
+        transformed_items = transformer.make_items()
+
+        # update transformation status to success
+        transform_obj.status = Transform.Status.SUCCESS
+        transform_obj.save(update_fields=["status"])
+    except Exception as e:
+        logger.error("Transformation failed", exc_info=True, extra={"extraction_id": ext_instance.id, "source": source})
+        # update transformation status to success
+        transform_obj.status = Transform.Status.FAILED
+        transform_obj.save(update_fields=["status"])
+        # FIXME: Check if this creates duplicate entry in Sentry. if yes, remove this.
+        raise e
+
+    # create objects into PyStacLoadData table
+    for item in transformed_items:
+        item_type = collection_and_item_type_map[item.collection_id]
+        transformed_item_dict = item.to_dict()
+        transformed_item_dict["properties"]["monty:etl_id"] = str(uuid.uuid4())
+        bulk_mgr.add(
+            PyStacLoadData(
+                transform_id=transform_obj,
+                item=transformed_item_dict,
+                collection_id=item.collection_id,
+                item_type=item_type,
+                load_status=PyStacLoadData.LoadStatus.PENDING,
+            )
+        )
+
+    bulk_mgr.done()
+
+    transform_obj.is_loaded = True
+    transform_obj.save(update_fields=["is_loaded"])
+
+    logger.info("Transformation ended for emdat data")

docker-compose.yml

@@ -24,6 +24,7 @@ x-server: &base_server_setup
       GFD_CREDENTIAL: ${GFD_CREDENTIAL?error}
       GFD_SERVICE_ACCOUNT: ${GFD_SERVICE_ACCOUNT?error}
       IFRC_DATA_URL: ${IFRC_DATA_URL?error}
+      GEOCODER_URL: ${GEOCODER_URL?error}
       # ETL Load
       EOAPI_DOMAIN: ${EOAPI_DOMAIN?error}
   volumes:

helm/templates/config/secret.yaml

@@ -21,6 +21,8 @@ stringData:
   GFD_CREDENTIAL: {{ required "secrets.GFD_CREDENTIAL" .Values.secrets.GFD_CREDENTIAL | quote }}
   GFD_SERVICE_ACCOUNT: {{ required "secrets.GFD_SERVICE_ACCOUNT" .Values.secrets.GFD_SERVICE_ACCOUNT | quote }}
   IFRC_DATA_URL: {{ required "secrets.IFRC_DATA_URL" .Values.secrets.IFRC_DATA_URL | quote }}
+  GEOCODER_URL: {{ required "secrets.GEOCODER_URL" .Values.secrets.GEOCODER_URL | quote }}
+
 
   # Database
   {{- if .Values.postgresql.enabled }}

helm/values-test.yaml

@@ -39,6 +39,7 @@ secrets:
   GFD_CREDENTIAL: dummy-gfd-cred
   GFD_SERVICE_ACCOUNT: dummy-gfd-service-ac
   IFRC_DATA_URL: dummy-url
+  GEOCODER_URL: dummy-url
 secretsAdditional:
   ENABLE_MAGIC_SECRET: "true"
   MAGIC_KEY: to-much-fun

helm/values.yaml

@@ -182,6 +182,7 @@ secrets:
   GFD_CREDENDIAL:
   GFD_SERVICE_ACCOUNT:
   IFRC_DATA_URL:
+  GEOCODER_URL:
 # NOTE: Used to pass additional secrets to api/worker containers
 # NOTE: Not used by azure vault
 secretsAdditional:

main/settings.py

@@ -75,8 +75,11 @@
     IFRC_DATA_URL=str,
     # ETL Load configs
     EOAPI_DOMAIN=str,  # http://montandon-eoapi.ifrc.org
+    GEOCODER_URL=str,
 )
 
+GEOCODER_URL = env("GEOCODER_URL")
+
 GFD_SERVICE_ACCOUNT = env("GFD_SERVICE_ACCOUNT")
 
 GFD_CREDENTIAL = env("GFD_CREDENTIAL")