Merge pull request #52 from IFRCGo/feature/partial-success-summary

Add support for partial transform success

Author: tnagorra

pystac_monty/sources/common.py

@@ -9,6 +9,36 @@
 from pystac_monty.geocoding import MontyGeoCoder
 
 
+class TransformSummaryInProgressException(Exception): ...
+
+
+@dataclass
+class TransformSummary:
+    total_rows: int = 0
+    failed_rows: int = 0
+    is_completed: bool = False
+
+    def increment_rows(self, increment=1):
+        self.total_rows += increment
+
+    def increment_failed_rows(self, increment=1):
+        self.failed_rows += increment
+
+    def mark_as_complete(self):
+        self.is_completed = True
+
+    def mark_as_started(self):
+        self.is_completed = False
+        self.total_rows = 0
+        self.failed_rows = 0
+
+    @property
+    def success_rows(self) -> int:
+        if not self.is_completed:
+            raise TransformSummaryInProgressException()
+        return self.total_rows - self.failed_rows
+
+
 @dataclass
 class MontyDataSource:
     source_url: str
@@ -60,6 +90,8 @@ def __init__(self, data_source: DataSource, geocoder: MontyGeoCoder):
 
         self.geocoder = geocoder
 
+        self.transform_summary = TransformSummary()
+
     def get_event_collection(self) -> Collection:
         """Get event collection"""
         if self._event_collection_cache is None:
@@ -108,5 +140,9 @@ def get_impact_collection(self) -> Collection:
             self._impact_collection_cache = collection
         return self._impact_collection_cache
 
+    # FIXME: This method is deprecated
     @abc.abstractmethod
     def make_items(self) -> list[Item]: ...
+
+    @abc.abstractmethod
+    def get_stac_items(self) -> typing.Generator[Item, None, None]: ...

pystac_monty/sources/desinventar.py

@@ -556,7 +556,6 @@ def _generate_geo_data_mapping(self, root: etree._Element) -> Dict[str, GeoDataE
                 "shapefile_data": shapefile_data
             }
 
-        print(geo_data)
         return geo_data
 
     @staticmethod
@@ -583,11 +582,9 @@ def get_stac_items(self) -> typing.Generator[Item, None, None]:
 
             events = root.xpath("//fichas/TR")
 
-            failed_items_count = 0
-            total_items_count = 0
-
+            self.transform_summary.mark_as_started()
             for event_row in events:
-                total_items_count += 1
+                self.transform_summary.increment_rows()
                 try:
                     if row_data := parse_row_data(
                         event_row,
@@ -599,12 +596,11 @@ def get_stac_items(self) -> typing.Generator[Item, None, None]:
                             yield event_item
                             yield from self._create_impact_items_from_row(row_data, event_item)
                         else:
-                            failed_items_count += 1
+                            self.transform_summary.increment_failed_rows()
                 except Exception:
-                    failed_items_count += 1
+                    self.transform_summary.increment_failed_rows()
                     logger.error('Failed to process desinventar', exc_info=True)
-
-            print(failed_items_count)
+            self.transform_summary.mark_as_complete()
 
     # FIXME: This is deprecated
     def make_items(self):

pystac_monty/sources/ibtracs.py

@@ -59,15 +59,13 @@ class IBTrACSTransformer(MontyDataTransformer[IBTrACSDataSource]):
     source_name = 'ibtracs'
 
     def get_stac_items(self) -> typing.Generator[Item, None, None]:
-        # # TODO: Use sax xml parser for memory efficient usage
-        failed_items_count = 0
-        total_items_count = 0
-
+        self.transform_summary.mark_as_started()
+        # TODO: Use sax xml parser for memory efficient usage
         csv_data = self.data_source._parse_csv()
         csv_data.sort(key=lambda x: x.get("SID", " "))
         for storm_id, storm_data_iterator in itertools.groupby(csv_data, key=lambda x: x.get("SID", " ")):
             storm_data = list(storm_data_iterator)
-            total_items_count += len(storm_data)
+            self.transform_summary.increment_rows(len(storm_data))
 
             try:
                 def parse_row_data(rows: list[dict]):
@@ -82,12 +80,11 @@ def parse_row_data(rows: list[dict]):
                     yield event_item
                     yield from self.make_hazard_items(event_item, storm_data)
                 else:
-                    failed_items_count += len(storm_data)
+                    self.transform_summary.increment_failed_rows(len(storm_data))
             except Exception:
-                failed_items_count += len(storm_data)
+                self.transform_summary.increment_failed_rows(len(storm_data))
                 logger.error("Failed to process ibtracs", exc_info=True)
-
-        print(failed_items_count)
+        self.transform_summary.mark_as_complete()
 
     # FIXME: This is deprecated
     def make_items(self):