Fix pdc file open

  - Add start date from env variable for gdacs, glide, emdat, ifrc
  - Load pystac item with status equals to pending.
  - Set conrtab for idu historical data.
  - Fix glide url
  - Send data into pdc transformer in a temp file.

Author: Rup-Narayan-Rajbanshi

apps/etl/etl_tasks/emdat.py

@@ -1,4 +1,4 @@
-from celery import shared_task
+from celery import chain, shared_task
 
 from apps.etl.extraction.sources.emdat.extract import (
     extract_emdat_historical_data,
@@ -9,17 +9,9 @@
 
 @shared_task
 def ext_and_transform_emdat_historical_data(**kwargs):
-    # Extract the data from emdat
-    extraction_id = extract_emdat_historical_data()
-
-    # Transform the data from emdat
-    transform_emdat_data(extraction_id)
+    chain(extract_emdat_historical_data.s(), transform_emdat_data.s()).apply_async()
 
 
 @shared_task
 def ext_and_transform_emdat_latest_data(**kwargs):
-    # Extract the data from emdat
-    extraction_id = extract_emdat_latest_data()
-
-    # Transform the data from emdat
-    transform_emdat_data(extraction_id)
+    chain(extract_emdat_latest_data.s(), transform_emdat_data.s()).apply_async()

apps/etl/etl_tasks/gdacs.py

@@ -4,6 +4,7 @@
 
 import requests
 from celery import chain, shared_task
+from django.conf import settings
 
 from apps.etl.extraction.sources.base.extract import Extraction
 from apps.etl.extraction.sources.base.utils import store_extraction_data
@@ -40,7 +41,7 @@ def _ext_and_transform_data(hazard_type: str, hazard_type_str: str):
             from_date = ext_object.created_at.date()
         else:
             # Fetch data up to one week at the begining.
-            from_date = datetime.today() - timedelta(days=7)
+            from_date = datetime.strptime(settings.GDACS_START_DATE, "%Y-%m-%d").date()
 
         to_date = datetime.now().date()
         ext_and_transform_gdacs_data.delay(hazard_type, hazard_type_str, from_date, to_date)

apps/etl/etl_tasks/idu.py

@@ -8,7 +8,7 @@
 
 logger = logging.getLogger(__name__)
 
-HISTORICAL_DATA_URL = f"{settings.IDMC_DATA_URL}/external-api/idus/idus_all_retrieve"
+HISTORICAL_DATA_URL = f"{settings.IDMC_DATA_URL}/external-api/idus/all/"
 LATEST_DATA_URL = f"{settings.IDMC_DATA_URL}/external-api/idus/last-180-days/"
 
 

apps/etl/etl_tasks/ifrc_event.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime
 
 from celery import chain, shared_task
 from django.conf import settings
@@ -13,9 +13,6 @@
 
 @shared_task
 def ext_and_transform_ifrcevent_latest_data():
-    END_DATE = datetime.now().date()
-    START_DATE = END_DATE - timedelta(days=7)
-
     ext_object = (
         ExtractionData.objects.filter(
             source=ExtractionData.Source.DREF, status=ExtractionData.Status.SUCCESS, resp_data__isnull=False
@@ -26,6 +23,8 @@ def ext_and_transform_ifrcevent_latest_data():
 
     if ext_object:
         START_DATE = ext_object.created_at.date()
+    else:
+        START_DATE = datetime.strptime(settings.GLIDE_START_DATE, "%Y-%m-%d").date()
 
     LATEST_DATA_PARAMS = {
         "disaster_start_date__gte": START_DATE,

apps/etl/etl_tasks/usgs.py

@@ -5,7 +5,7 @@
 
 @shared_task
 def ext_and_transform_usgs_latest_data():
-    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.geojson"
+    url = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.geojson"
     ext_and_transform_data.delay(url)
 
 

apps/etl/extraction/sources/base/utils.py

@@ -1,8 +1,6 @@
 import hashlib
 import json
-import os
 
-from django.conf import settings
 from django.core.files.base import ContentFile
 
 from apps.etl.models import ExtractionData
@@ -107,15 +105,3 @@ def store_pdc_exposure_data(
     instance.resp_data.save(content_file.name, content_file)
 
     return instance
-
-
-def store_geojson_file(response, source=None, validate_source_func=None, instance_id=None, hazard_type=None, metadata=None):
-    file_extension = "geojson"
-    file_name = f"{instance_id}pdc.{file_extension}"
-    instance = ExtractionData.objects.get(id=instance_id.id)
-    file_path = os.path.join(settings.MEDIA_ROOT, "source_raw_data", file_name)
-    with open(file_path, "w") as f:
-        json.dump(response, f)
-    instance.metadata["geojson_file_path"] = file_path
-    instance.save()
-    return instance.id

apps/etl/extraction/sources/desinventar/extract.py

@@ -61,7 +61,7 @@ def handle_extraction(cls, url: str, params: dict, headers: dict, source: int) -
         try:
             cls._update_instance_status(instance, ExtractionData.Status.IN_PROGRESS)
 
-            response = requests.get(url, params=params, headers=headers, timeout=30)
+            response = requests.get(url, params=params, headers=headers, timeout=180)
             response.raise_for_status()
             instance.resp_code = response.status_code
 

apps/etl/extraction/sources/emdat/extract.py

@@ -15,7 +15,7 @@
 @shared_task
 def extract_emdat_latest_data():
     to_year = datetime.now().year
-    from_year = int(to_year) - 1
+    from_year = int(settings.EMDAT_START_YEAR)
     # ref: https://files.emdat.be/docs/emdat_api_cookbook.pdfhttps://files.emdat.be/docs/emdat_api_cookbook.pdf
     variables = {"limit": -1, "from": from_year, "to": to_year}
     return import_hazard_data(variables)

apps/etl/extraction/sources/glide/extract.py

@@ -1,8 +1,9 @@
 import logging
-from datetime import datetime, timedelta
+from datetime import datetime
 
 import requests
 from celery import shared_task
+from django.conf import settings
 
 from apps.etl.extraction.sources.base.extract import Extraction
 from apps.etl.extraction.sources.base.utils import store_extraction_data
@@ -26,11 +27,10 @@ def extract_glide_latest_data(hazard_type, hazard_type_str):
     if ext_object:
         from_date = ext_object.created_at.date()
     else:
-        # Fetch data up to one week at the begining.
-        from_date = datetime.today() - timedelta(days=7)
+        from_date = datetime.strptime(settings.GLIDE_START_DATE, "%Y-%m-%d").date()
 
     to_date = datetime.today().date()
-    url = f"https://www.glidenumber.net/glide/jsonglideset.jsp?fromyear={from_date.year}&frommonth={from_date.month}&fromday={from_date.day}&toyear={to_date.year}&frommonth={to_date.month}&to_date={to_date.day}&events={hazard_type}"  # noqa: E501
+    url = f"https://www.glidenumber.net/glide/jsonglideset.jsp?fromyear={from_date.year}&frommonth={from_date.month}&fromday={from_date.day}&toyear={to_date.year}&tomonth={to_date.month}&today={to_date.day}&events={hazard_type}"  # noqa: E501
     return import_glide_hazard_data(hazard_type, hazard_type_str, url)
 
 

apps/etl/extraction/sources/pdc/extract.py

@@ -8,7 +8,6 @@
 from apps.etl.extraction.sources.base.extract import Extraction
 from apps.etl.extraction.sources.base.utils import (
     store_extraction_data,
-    store_geojson_file,
     store_pdc_exposure_data,
 )
 from apps.etl.models import ExtractionData, HazardType
@@ -44,7 +43,16 @@ def get_hazard_details(self, extraction_id, **kwargs):
     for hazard in response_data:
         try:
             geo_json_file = geo.get_polygon(hazard["uuid"])
-            store_geojson_file(geo_json_file, instance_id=instance_id)
+
+            geo_json_data = store_pdc_exposure_data(
+                response=geo_json_file,
+                source=ExtractionData.Source.PDC,
+                validate_source_func=None,
+                parent_id=instance_id.id,
+                hazard_type=HAZARD_TYPE_MAP.get(hazard["type_ID"]),
+                metadata={},
+            )
+
             if hazard["type_ID"] not in HAZARD_TYPE_MAP.keys():
                 continue
             r = requests.get(
@@ -72,7 +80,7 @@ def get_hazard_details(self, extraction_id, **kwargs):
                     hazard_type=HAZARD_TYPE_MAP.get(hazard["type_ID"]),
                     metadata={"exposure_id": exposure_id, "uuid": hazard["uuid"]},
                 )
-                PDCTransformHandler.task(exposure_detail.id)
+                PDCTransformHandler.task(exposure_detail.id, geo_json_data.id)
         except Exception as exc:
             self.retry(exc=exc, kwargs={"instance_id": instance_id.id, "retry_count": self.request.retries})
 
@@ -105,7 +113,7 @@ def import_hazard_data(self, **kwargs):
     # Extract the data from api.
     pdc_extraction = Extraction(
         url=pdc_url,
-        headers={"Authorization": "Bearer {}".format(settings.PDC_AUTHORIZATION_KEY)},
+        headers={"Authorization": "Bearer {}".format(settings.PDC_AUTHORIZATION_KEY)},  # NOTE: Does this key expire??
     )
     response = None
     try:

apps/etl/load/sources/base.py

@@ -29,7 +29,7 @@ def load_data(django_command: BaseCommand | None = None):
     """Load data into STAC"""
     logger.info("Loading data into Stac")
 
-    transformed_items = PyStacLoadData.objects.exclude(load_status=PyStacLoadData.LoadStatus.SUCCESS)
+    transformed_items = PyStacLoadData.objects.filter(load_status=PyStacLoadData.LoadStatus.PENDING)
 
     bulk_mgr = BulkUpdateManager(["load_status"], chunk_size=1000)
     for item in transformed_items.iterator():

apps/etl/transform/sources/pdc.py

@@ -1,30 +1,84 @@
 import json
+import logging
+import tempfile
 
 from pystac_monty.sources.pdc import PDCDataSource, PDCTransformer
 
-from apps.etl.models import ExtractionData
+from apps.etl.models import ExtractionData, Transform
 from main.celery import app
 
 from .handler import BaseTransformerHandler
 
+logger = logging.getLogger(__name__)
+
 
 class PDCTransformHandler(BaseTransformerHandler):
     transformer = PDCTransformer
     transformer_schema = PDCDataSource
 
     @classmethod
-    def get_schema_data(cls, extraction_obj: ExtractionData):
+    def get_schema_data(cls, extraction_obj: ExtractionData, geo_json_obj: ExtractionData):
         source_url = extraction_obj.url
+
+        with extraction_obj.parent.resp_data.open("rb") as f:
+            file_content = f.read()
+        tmp_hazard_file = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
+        tmp_hazard_file.write(file_content)
+
+        with extraction_obj.resp_data.open("rb") as f:
+            file_content = f.read()
+        tmp_exposure_detail_file = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
+        tmp_exposure_detail_file.write(file_content)
+
+        with geo_json_obj.resp_data.open("rb") as f:
+            file_content = f.read()
+        tmp_geojson_file = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
+        tmp_geojson_file.write(file_content)
+
         data = {
-            "hazards_file_path": extraction_obj.parent.resp_data.path,
+            "hazards_file_path": tmp_hazard_file.name,
             "exposure_timestamp": extraction_obj.metadata["exposure_id"],
             "uuid": extraction_obj.metadata["uuid"],
-            "exposure_detail_file_path": extraction_obj.resp_data.path,
-            "geojson_file_path": extraction_obj.parent.metadata["geojson_file_path"] or None,
+            "exposure_detail_file_path": tmp_exposure_detail_file.name,
+            "geojson_file_path": tmp_geojson_file.name,
         }
+
         return cls.transformer_schema(source_url=source_url, data=json.dumps(data))
 
+    @classmethod
+    def handle_transformation(cls, extraction_id, geo_json_id):
+        logger.info("Transformation started")
+        extraction_obj = ExtractionData.objects.filter(id=extraction_id).first()
+        geo_json_obj = ExtractionData.objects.filter(id=geo_json_id).first()
+        if not extraction_obj.resp_data:
+            logger.info("Transformation ended due to no data")
+            return
+
+        transform_obj = Transform.objects.create(
+            extraction=extraction_obj,
+            status=Transform.Status.PENDING,
+        )
+
+        try:
+            schema = cls.get_schema_data(extraction_obj, geo_json_obj)
+            transformer = cls.transformer(schema)
+            transformed_items = transformer.make_items()
+
+            transform_obj.status = Transform.Status.SUCCESS
+            transform_obj.save(update_fields=["status"])
+
+            cls.load_stac_item_to_queue(transformed_items, transform_obj.id)
+
+            logger.info("Transformation ended")
+
+        except Exception as e:
+            logger.error("Transformation failed", exc_info=True, extra={"extraction_id": extraction_obj.id})
+            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
+
     @staticmethod
     @app.task
-    def task(extraction_id):
-        return PDCTransformHandler().handle_transformation(extraction_id)
+    def task(extraction_id, geo_json_id):
+        return PDCTransformHandler().handle_transformation(extraction_id, geo_json_id)

helm/values.yaml

@@ -118,7 +118,7 @@ worker:
     # NOTE: Make sure keys are lowercase
     default:
       enabled: true
-      replicaCount: 2
+      replicaCount: 10
       celeryArgs:
         - "-Q"
         - "celery"

libs/pystac-monty

@@ -71,6 +71,11 @@
     EMDAT_AUTHORIZATION_KEY=str,
     IDMC_CLIENT_ID=str,
     IDMC_DATA_URL=(str, "https://helix-tools-api.idmcdb.org"),
+    # Default start date for latest data extraction
+    GLIDE_START_DATE=(str, "2025-01-01"),
+    IFRCEVENT_START_DATE=(str, "2025-01-01"),
+    GDACS_START_DATE=(str, "2025-01-01"),
+    EMDAT_START_YEAR=(str, "2024"),
     # ETL Load configs
     EOAPI_DOMAIN=str,  # http://montandon-eoapi.ifrc.org
     GFD_CREDENTIAL=str,
@@ -85,6 +90,13 @@
     ARC_USERNAME=str,
     ARC_PASSWORD=str,
 )
+GLIDE_START_DATE = env("GLIDE_START_DATE")
+
+GDACS_START_DATE = env("GDACS_START_DATE")
+
+IFRCEVENT_START_DATE = env("IFRCEVENT_START_DATE")
+
+EMDAT_START_YEAR = env("EMDAT_START_YEAR")
 
 DESINVENTAR_DATA_URL = env("DESINVENTAR_DATA_URL")