Add partial success for ibtrace.

Author: Rup-Narayan-Rajbanshi

pystac_monty/sources/ibtracs.py

@@ -2,6 +2,9 @@
 
 import csv
 import io
+import itertools
+import logging
+import typing
 from typing import Dict, List, Union
 
 import pytz
@@ -11,6 +14,10 @@
 from pystac_monty.extension import HazardDetail, MontyEstimateType, MontyExtension
 from pystac_monty.hazard_profiles import MontyHazardProfiles
 from pystac_monty.sources.common import MontyDataSource, MontyDataTransformer
+from pystac_monty.validators.ibtracs import IBTracsdataValidator
+
+logger = logging.getLogger(__name__)
+
 
 STAC_EVENT_ID_PREFIX = "ibtracs-event-"
 STAC_HAZARD_ID_PREFIX = "ibtracs-hazard-"
@@ -47,15 +54,14 @@ def _parse_csv(self) -> List[Dict[str, str]]:
             csv_data.append(row)
         return csv_data
 
-    def get_storm_ids(self) -> List[str]:
-        """Get a list of unique storm IDs from the data."""
-        data = self.get_data()
-        return list(set(row.get("SID", "").strip() for row in data if row.get("SID")))
+    def parse_row_data(self, rows: list[dict]):
+        validated_data = []
+        for row in rows:
+            obj = IBTracsdataValidator.validate_event(row)
+            if obj:
+                validated_data.append(obj)
 
-    def get_storm_data(self, storm_id: str) -> List[Dict[str, str]]:
-        """Get all data rows for a specific storm ID."""
-        data = self.get_data()
-        return [row for row in data if row.get("SID", "").strip() == storm_id]
+        return validated_data
 
 
 class IBTrACSTransformer(MontyDataTransformer[IBTrACSDataSource]):
@@ -64,37 +70,41 @@ class IBTrACSTransformer(MontyDataTransformer[IBTrACSDataSource]):
     hazard_profiles = MontyHazardProfiles()
     source_name = 'ibtracs'
 
-    def make_items(self) -> List[Item]:
-        """Create STAC Items from IBTrACS data.
+    # FIXME: This is deprecated
+    def make_items(self):
+        return list(self.get_stac_items())
 
-        Returns:
-            List of STAC Items (events and hazards)
-        """
-        items = []
+    def get_stac_items(self) -> typing.Generator[Item, None, None]:
+        csv_data = self.data_source._parse_csv()
+        csv_data.sort(key=lambda x: x.get("SID"))
 
-        storm_ids = self.data_source.get_storm_ids()
+        grouped_rows = {}
+        for sid, group in itertools.groupby(csv_data, key=lambda x: x.get("SID")):
+            grouped_rows[sid] = list(group)
 
-        # Create event items (one per storm)
-        for storm_id in storm_ids:
-            try:
-                event_hazard_items = []
-                event_item = self.make_source_event_items(storm_id)
-                event_hazard_items.append(event_item)
-                hazard_items = self.make_hazard_items(event_item)
-                event_hazard_items.extend(hazard_items)
+        # # TODO: Use sax xml parser for memory efficient usage
+        failed_items_count = 0
+        total_items_count = 0
 
-                yield event_hazard_items
+        for storm_id, storm_data in grouped_rows.items():
+            total_items_count += 1
+            try:
+                storm_data = self.data_source.parse_row_data(storm_data)
+                if event_item := self.make_source_event_items(storm_id, storm_data):
+                    yield event_item
+                    yield from self.make_hazard_items(event_item, storm_data)
             except Exception:
-                logger.info("Transformation failed", exc_info=True)
+                failed_items_count += 1
+                logger.error("Failed to process desinventar", exc_info=True)
+
+        print(failed_items_count)
 
-    def make_source_event_items(self, storm_id) -> List[Item]:
+    def make_source_event_items(self, storm_id, storm_data) -> List[Item]:
         """Create source event items from IBTrACS data.
 
         Returns:
             List of event STAC Items
         """
-        storm_data = self.data_source.get_storm_data(storm_id)
-
         if not storm_data:
             return
 
@@ -287,7 +297,7 @@ def make_source_event_items(self, storm_id) -> List[Item]:
 
         return item
 
-    def make_hazard_items(self, event_item: Item) -> Item:
+    def make_hazard_items(self, event_item: Item, storm_data) -> list[Item] | None:
         """Create hazard items from IBTrACS data.
 
         Args:
@@ -299,7 +309,6 @@ def make_hazard_items(self, event_item: Item) -> Item:
         hazard_items = []
 
         storm_id = event_item.id
-        storm_data = self.data_source.get_storm_data(storm_id)
 
         if not storm_data:
             return
@@ -360,6 +369,8 @@ def make_hazard_items(self, event_item: Item) -> Item:
                 except (ValueError, TypeError):
                     wind = 0
 
+            wind = 0 if wind.strip() == "" else wind
+
             try:
                 pressure = float(row.USA_PRES if row.USA_PRES else 0)
             except (ValueError, TypeError):
@@ -559,14 +570,13 @@ def _get_countries_from_track(self, track_geometry: Union[LineString, Point]) ->
             if isinstance(track_geometry, LineString):
                 for point in track_geometry.coords:
                     lon, lat = point
-                    # country_code = self.geocoder.get_iso3_from_geometry(Point(lon, lat))
-                    country_code = "UNK"
+                    country_code = self.geocoder.get_iso3_from_geometry(Point(lon, lat))
                     if country_code:
                         countries.append(country_code)
             # For Point, check the single point
             elif isinstance(track_geometry, Point):
                 lon, lat = track_geometry.x, track_geometry.y
-                # country_code = self.geocoder.get_iso3_from_geometry(track_geometry)
+                country_code = self.geocoder.get_iso3_from_geometry(track_geometry)
                 country_code = "UNK"
                 if country_code:
                     countries.append(country_code)

pystac_monty/validators/glide.py

@@ -38,7 +38,33 @@ class GlideSetValidator(BaseModelWithExtra):
 
     @field_validator("event")
     def validate_enum(cls, value):
-        if not value in ["EQ", "TC", "FL", "DR", "WF", "VO", "TS", "CW", "EP", "EC", "ET", "FR", "FF", "HT", "IN","LS","MS","ST","SL","AV","SS","AC","TO","VW","WV"]:
+        if value not in [
+            "EQ",
+            "TC",
+            "FL",
+            "DR",
+            "WF",
+            "VO",
+            "TS",
+            "CW",
+            "EP",
+            "EC",
+            "ET",
+            "FR",
+            "FF",
+            "HT",
+            "IN",
+            "LS",
+            "MS",
+            "ST",
+            "SL",
+            "AV",
+            "SS",
+            "AC",
+            "TO",
+            "VW",
+            "WV",
+        ]:
             raise ValueError(f"Event type {value} is not valid.")
         return value
 
@@ -51,4 +77,4 @@ def validate_event(cls, data: dict) -> bool:
             logger.error(f"Validation failed: {e}")
             return False
         # If all field validators return True, we consider it valid
-        return True
+        return True

tests/extensions/test_ibtracs.py

@@ -299,7 +299,7 @@ def test_event_item_properties(self) -> None:
         data_source = IBTrACSDataSource("test_url", SAMPLE_IBTRACS_CSV)
         transformer = IBTrACSTransformer(data_source, geocoder)
 
-        #items = list(transformer.make_items())
+        # items = list(transformer.make_items())
 
         # Find the event item
         event_items = [item for item in transformer.make_items() if "event" in item.properties.get("roles", [])]
@@ -329,8 +329,6 @@ def test_hazard_item_properties(self) -> None:
         data_source = IBTrACSDataSource("test_url", SAMPLE_IBTRACS_CSV)
         transformer = IBTrACSTransformer(data_source, geocoder)
 
-        items = transformer.make_items()
-
         # Find hazard items
         hazard_items = [item for item in transformer.make_items() if "hazard" in item.properties.get("roles", [])]
         self.assertGreater(len(hazard_items), 0)