feat(filter): add filtration classes and cronjob

Author: sandeshit

alert_system/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
 
-from .models import Connector, StacItems, EligibleItems
+from .models import Connector, EligibleItems, StacItems
 
 
 @admin.register(Connector)
@@ -16,6 +16,7 @@ class EventAdmin(admin.ModelAdmin):
     readonly_fields = ("connector",)
     search_fields = ("stac_id",)
 
+
 @admin.register(EligibleItems)
 class EligibleAdmin(admin.ModelAdmin):
     list_display = ("stac_id", "created_at", "collection")

alert_system/extraction/base_extraction.py

@@ -1,6 +1,5 @@
 import logging
 from datetime import datetime, timedelta, timezone
-from token import OP
 from typing import Dict, Generator, Optional
 
 import httpx
@@ -20,9 +19,7 @@ class BaseExtractionClass:
 
     def __init__(self, connector: Connector):
         if not self.event_endpoint:
-            raise NotImplementedError(
-                f"{self.__class__.__name__} must define event_endpoint, "
-            )
+            raise NotImplementedError(f"{self.__class__.__name__} must define event_endpoint, ")
         self.connector = connector
         self.base_url = connector.source_url.rstrip("/")
 
@@ -157,7 +154,7 @@ def process_hazard_items(self, stac_obj: StacItems) -> None:
         if not self.hazard_endpoint:
             return
         try:
-            hazard_features = self._fetch_items(self.hazard_endpoint,"filter_hazard", stac_obj.correlation_id)
+            hazard_features = self._fetch_items(self.hazard_endpoint, "filter_hazard", stac_obj.correlation_id)
         except Exception as e:
             logger.error(f"Failed to fetch hazards for event {stac_obj.stac_id}: {e}")
             raise

alert_system/extraction/gdacs_cyclone.py

@@ -1,5 +1,6 @@
 from alert_system.extraction.base_extraction import BaseExtractionClass
 
+
 class GdacsCycloneExtraction(BaseExtractionClass):
     event_endpoint = "/gdacs-events/items"
     hazard_endpoint = "/gdacs-hazards/items"

alert_system/extraction/gdacs_flood.py

@@ -1,7 +1,7 @@
 from alert_system.extraction.base_extraction import BaseExtractionClass
 
+
 class GdacsFloodExtraction(BaseExtractionClass):
     event_endpoint = "/gdacs-events/items"
     hazard_endpoint = "/gdacs-hazards/items"
     impact_endpoint = "/gdacs-impacts/items"
-

alert_system/extraction/usgs_earthquake.py

@@ -1,5 +1,6 @@
 from alert_system.extraction.base_extraction import BaseExtractionClass
 
+
 class USGSEarthquakeExtraction(BaseExtractionClass):
     event_endpoint = "/usgs-events/items"
     hazard_endpoint = "/usgs-hazards/items"

alert_system/filters/__init__.py

@@ -0,0 +1,35 @@
+from abc import ABC
+from typing import Dict
+
+from alert_system.models import StacItems
+
+
+class BaseFilter(ABC):
+
+    hazard_threshold: Dict
+    impact_threshold: Dict
+
+    def __init__(self, stac_obj: StacItems):
+        self.item = stac_obj
+
+    def is_hazard_eligible(self) -> bool:
+        """Return True if hazard passes filters."""
+        return (
+            self.item.severity_value > self.hazard_threshold["severity_value"]
+            and self.item.severity_label == self.hazard_threshold["severity_label"]
+        )
+
+    def is_impact_eligible(self) -> bool:
+        """Return True if impact passes filters."""
+        key = (self.item.category, self.item.type)
+        threshold = self.impact_threshold.get(key)
+        if threshold is None:
+            return False
+        return self.item.value >= threshold
+
+    def is_eligible(self) -> bool:
+        if self.item.collection == StacItems.CollectionType.IMPACT:
+            return self.is_impact_eligible()
+        elif self.item.collection == StacItems.CollectionType.HAZARD:
+            return self.is_hazard_eligible()
+        return True

alert_system/filters/base_filter.py

@@ -0,0 +1,15 @@
+from .base_filter import BaseFilter
+
+
+class GdacsCycloneFilter(BaseFilter):
+
+    impact_threshold = {
+        ("people", "death"): 1,
+        ("people", "injured"): 10,
+        ("people", "missing"): 1,
+    }
+
+    hazard_threshold = {
+        "severity_value": 5,
+        "severity_label": "Green",
+    }

alert_system/filters/gdacs_cyclone.py

@@ -0,0 +1,14 @@
+from .base_filter import BaseFilter
+
+
+class GdacsFloodFilter(BaseFilter):
+    impact_threshold = {
+        ("people", "death"): 1,
+        ("people", "injured"): 10,
+        ("people", "missing"): 1,
+    }
+
+    hazard_threshold = {
+        "severity_value": 0,
+        "severity_label": "Green",
+    }

alert_system/filters/gdacs_flood.py

@@ -0,0 +1,14 @@
+from .base_filter import BaseFilter
+
+
+class USGSEarthquakeFilter(BaseFilter):
+    impact_threshold = {
+        ("people", "death"): 1,
+        ("people", "injured"): 10,
+        ("people", "missing"): 1,
+    }
+
+    hazard_threshold = {
+        "severity_value": 5,
+        "severity_label": "Green",
+    }