Add transformer for Glide.

Author: Rup-Narayan-Rajbanshi

pystac_monty/collections/__init__.py

@@ -0,0 +1,19 @@
+glide_event_collection = {
+    "stac_version": "1.0.0",
+    "stac_extensions": ["https://ifrcgo.github.io/monty/v0.1.0/schema.json"],
+    "type": "Collection",
+    "id": "glide-events",
+    "title": "Glide Source Events",
+    "description": "GLobal IDEntifier Number (GLIDE)",
+    "license": "MIT",
+    "roles": ["event", "source"],
+    "providers": [{"name": "Glide", "roles": ["producer"], "url": "https://www.glidenumber.net/", "email": "gliderep@adrc.asia"}],
+    "extent": {"spatial": {"bbox": [[-180, -90, 180, 90]]}, "temporal": {"interval": [["2000-01-01T00:00:00Z", ""]]}},
+    "summaries": {
+        "datetime": {"minimum": "2015-06-23T00:00:00Z", "maximum": "2019-07-10T13:44:56Z"},
+        "roles": ["event", "source"],
+        "monty:country_codes": [],
+        "monty:hazard_codes": [],
+    },
+    "links": [],
+}

pystac_monty/collections/glide_events.py

@@ -87,7 +87,7 @@ def make_items(self) -> list[Item]:
         source_event_item = self.make_source_event_item()
         items.append(source_event_item)
 
-        """ 2. Create the hazard item """
+        # """ 2. Create the hazard item """
         hazard_event_item = self.make_hazard_event_item()
         items.append(hazard_event_item)
 

pystac_monty/sources/gdacs.py

@@ -0,0 +1,102 @@
+import json
+from datetime import datetime
+from typing import Any
+
+from pystac import Collection, Item, Link
+from shapely.geometry import Point, mapping
+
+from pystac_monty.collections.glide_events import glide_event_collection
+from pystac_monty.extension import MontyExtension
+from pystac_monty.sources.common import MontyDataSource
+
+
+class GlideDataSource(MontyDataSource):
+    def __init__(self, source_url: str, data: Any):
+        super().__init__(source_url, data)
+        self.data = json.loads(data)
+
+
+class GlideTransformer:
+    """
+    Transforms Glide event data into STAC Items
+    """
+
+    def __init__(self, data: list[GlideDataSource]) -> None:
+        self.data = data
+
+    def make_items(self) -> list[Item]:
+        items = []
+
+        # validate data for glide transformation
+        glide_events = self.validate_glide_data()
+
+        for obj in glide_events:
+            glide_data = obj.data.get("glideset")
+            if not glide_data == []:
+                for data in glide_data:
+                    glide_id = data.get("event") + "-" + data.get("number") + "-" + data.get("geocode")
+                    latitude = float(data.get("latitude"))
+                    longitude = float(data.get("longitude"))
+                    event_date = {"year": data.get("year"), "month": data.get("month"), "day": data.get("day")}
+
+                    point = Point(longitude, latitude)
+                    geometry = mapping(point)
+                    bbox = [longitude, latitude, longitude, latitude]
+
+                    item = Item(
+                        id=glide_id,
+                        geometry=geometry,
+                        bbox=bbox,
+                        datetime=self.make_date(event_date),
+                        properties={
+                            "title": data.get("title", ""),
+                            "description": data.get("comments", ""),
+                            "magnitude": data.get("magnitude", ""),
+                            "location": data.get("location", ""),
+                            "source": data.get("source", ""),
+                            "docid": data.get("docid", ""),
+                            "status": data.get("status", ""),
+                        },
+                    )
+
+                    item.set_collection(self.get_collection())
+
+                    MontyExtension.add_to(item)
+                    monty = MontyExtension.ext(item)
+                    monty.hazard_codes = data.get("event")
+                    monty.country_codes = data.get("geocode")
+
+                    item.add_link(Link("via", obj.get_source_url(), "application/json", "Glide Event Data"))
+
+                    items.append(item)
+
+        return items
+
+    def make_date(self, data: dict):
+        year = data["year"]
+        month = data["month"]
+        day = data["day"]
+
+        dt = datetime(year, month, day)
+        formatted_date = dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+        date = datetime.fromisoformat(formatted_date.replace("Z", "+00:00"))
+        return date
+
+    def get_collection(self) -> Collection:
+        data = glide_event_collection
+        return Collection.from_dict(data)
+
+    def validate_glide_data(self) -> list[GlideDataSource]:
+        glide_data_list = []
+        for index, obj in enumerate(self.data):
+            glideset = obj.get_data()["glideset"]
+            if glideset == []:
+                raise ValueError(f"No Glide data found in object at index {index}")
+            for obj in glideset:
+                required_fields = ["latitude", "longitude", "event", "number", "geocode"]
+                missing_fields = [field for field in required_fields if field not in obj]
+
+                if missing_fields:
+                    raise ValueError(f"Missing required fields {missing_fields} in object at index {index}")
+        return glide_data_list