fixup! chore(extraction): Extract impact data for gdacs.

Author: Rup-Narayan-Rajbanshi

apps/etl/etl_tasks/gdacs.py

@@ -18,12 +18,12 @@
 
 HAZARDS = [
     (HazardType.EARTHQUAKE, 64),
-    # (HazardType.CYCLONE, 64),
-    # (HazardType.FLOOD, 64),
-    # (HazardType.DROUGHT, 64),
-    # (HazardType.WILDFIRE, 64),
-    # (HazardType.VOLCANO, 64),
-    # (HazardType.TSUNAMI, 64),
+    (HazardType.CYCLONE, 64),
+    (HazardType.FLOOD, 64),
+    (HazardType.DROUGHT, 64),
+    (HazardType.WILDFIRE, 64),
+    (HazardType.VOLCANO, 64),
+    (HazardType.TSUNAMI, 64),
 ]
 
 URL = f"{etl_config.GDACS_URL}/gdacsapi/api/events/geteventlist/SEARCH"
@@ -42,12 +42,10 @@ def ext_and_transform_gdacs_latest_data():
         .order_by("-created_at")
         .first()
     )
-    # if ext_object:
-    #     start_date = ext_object.created_at.date()
-    # else:
-    #     start_date = etl_config.GDACS_START_DATE
-
-    start_date = etl_config.GDACS_START_DATE
+    if ext_object:
+        start_date = ext_object.created_at.date()
+    else:
+        start_date = etl_config.GDACS_START_DATE
 
     end_date = dt.today().date()
     for hazard, size in HAZARDS:

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

@@ -367,7 +367,6 @@ def init_extraction(
         add_to_queue: bool = True,
         queue_name: str | None = None,
     ) -> ExtractionData:
-        print("INIT Extraction ", metadata.model_dump())
         extraction_obj = ExtractionData.objects.create(
             source=cls.source_enum,
             url=metadata.url,  # type: ignore[reportAttributeAccessIssue] TODO
@@ -379,7 +378,6 @@ def init_extraction(
             attempt_no=0,
             resp_code=0,
         )
-        print("Ext obj---------:", extraction_obj.id)
 
         if add_to_queue:
             cls.task.apply_async([extraction_obj.pk], queue=queue_name or cls.DEFAULT_CELERY_QUEUE)

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

@@ -206,6 +206,7 @@ def handle_type_episode(self, retrigger: bool):
                 parent_extraction=self.extraction_object,
                 add_to_queue=False,
             )
+            # Extract Impact data
             for obj in impact_list:
                 # source = obj["source"]
                 impact_url = obj["resource"]["impact"]
@@ -219,6 +220,8 @@ def handle_type_episode(self, retrigger: bool):
                     parent_extraction=self.extraction_object,
                     add_to_queue=False,
                 )
+                GdacsExtraction.task(impact_obj.id)
+
         else:
             geo_obj = ExtractionData.objects.filter(
                 url=geometry_episode_url,

apps/etl/transform/sources/gdacs.py

@@ -22,6 +22,8 @@ class GDACSTransformHandler(BaseTransformerHandler[GDACSTransformer, GDACSDataSo
 
     @classmethod
     def get_schema_data(cls, extraction_object):
+        from apps.etl.extraction.sources.gdacs.extract import GdacsExtractionMetadataType
+
         with extraction_object.resp_data.open("rb") as f:
             file_content = f.read()
         data_file = write_into_temp_file(file_content)
@@ -39,7 +41,9 @@ def get_schema_data(cls, extraction_object):
                     source_url=episode_obj.url, input_data=File(path=episode_data_temp_file.name, data_type=DataType.FILE)
                 ),
             )
-            geometry_object = episode_obj.child_extractions.all().first()
+            geometry_object = (
+                episode_obj.child_extractions.all().filter(metadata__type=GdacsExtractionMetadataType.GEOMETRY).first()
+            )
 
             with geometry_object.resp_data.open("rb") as f:
                 file_content = f.read()
@@ -52,7 +56,21 @@ def get_schema_data(cls, extraction_object):
                 ),
             )
 
-            episode_data_tuple = (event_episode_data, geometry_episode_data)
+            impact_object = (
+                episode_obj.child_extractions.all().filter(metadata__type=GdacsExtractionMetadataType.IMPACT).first()
+            )
+
+            with impact_object.resp_data.open("rb") as f:
+                file_content = f.read()
+            impact_detail_temp_file = write_into_temp_file(file_content)
+            impact_episode_data = GdacsEpisodes(
+                type=GDACSDataSourceType.IMPACT,
+                data=GenericDataSource(
+                    source_url=impact_object.url,
+                    input_data=File(path=impact_detail_temp_file.name, data_type=DataType.FILE),
+                ),
+            )
+            episode_data_tuple = (event_episode_data, geometry_episode_data, impact_episode_data)
             episodes.append(episode_data_tuple)
 
         result = cls.transformer_schema(

libs/pystac-monty

@@ -217,8 +217,8 @@ scalar JSON @specifiedBy(url: "https://ecma-international.org/wp-content/uploads
 type Mutation {
   login(username: String!, password: String!): UserMeType!
   logout: Boolean!
-  retriggerPipeline(data: PipelineRetriggerInput!): StrMutationResponseType!
-  retriggerTransform(data: TransformRetriggerInput!): StrMutationResponseType!
+  retriggerPipeline(data: PipelineRetriggerInput!): PipelineRetriggerTypeMutationResponseType!
+  retriggerTransform(data: TransformRetriggerInput!): PipelineRetriggerTypeMutationResponseType!
 }
 
 type OffsetPaginationInfo {
@@ -244,6 +244,17 @@ input PipelineRetriggerInput {
   traceIds: [ID!]!
 }
 
+type PipelineRetriggerType {
+  taskId: ID!
+  status: String!
+}
+
+type PipelineRetriggerTypeMutationResponseType {
+  ok: Boolean!
+  errors: CustomErrorType
+  result: PipelineRetriggerType
+}
+
 enum PyStacLoadDataItemTypeEnum {
   EVENT
   HAZARD
@@ -482,12 +493,6 @@ input StrFilterLookup {
   iRegex: String
 }
 
-type StrMutationResponseType {
-  ok: Boolean!
-  errors: CustomErrorType
-  result: String
-}
-
 input TransformDataFilter {
   createdAt: DatetimeDatetimeFilterLookup
   status: DataStatusTypeEnum
@@ -564,4 +569,4 @@ type ValStatusSourceCount {
   noDataCount: Int!
   noChangeCount: Int!
   noValidationCount: Int!
-}
+}