Canonical Event Model: GeoSentinel's Unified Schema

from datetime import datetime
from enum import Enum
from typing import Any

from pydantic import BaseModel, Field


class CategoryEnum(str, Enum):
    CONFLICT        = "conflict"
    DISASTER_NATURAL = "disaster_natural"
    WILDFIRE        = "wildfire"
    MOBILITY        = "mobility"
    HUMANITARIAN    = "humanitarian"
    THERMAL_ANOMALY = "thermal_anomaly"
    OTHER           = "other"


class Actor(BaseModel):
    role:       str
    name:       str
    cameo_code: str | None = None


class EventCanonicalCreate(BaseModel):
    # Identity
    event_id_source:      str
    source:               str
    # Temporality — always timezone-aware UTC
    event_time:           datetime
    # Classification
    event_type:           str | None = None
    category:             CategoryEnum
    # Geography (WGS84)
    latitude:             float
    longitude:            float
    location_accuracy_km: float | None = None
    admin1:               str | None   = None
    admin2:               str | None   = None
    country_iso2:         str | None   = None
    geometry:             dict[str, Any] | None = None
    geometry_type:        GeometryTypeEnum | None = None
    # Actors
    actors:               list[Actor] | None = None
    # Metrics
    fatalities:           int | None = None
    severity:             float = Field(..., ge=0.0, le=10.0)
    confidence:           float = Field(..., ge=0.0, le=10.0)
    # Source provenance
    source_url:           str | None        = None
    source_refs:          list[str] | None  = None
    raw_event_id:         int | None        = None
    # Flags
    is_confirmed:         bool = False
    is_rumor:             bool = False
    raw_payload:          dict[str, Any] | None = None

REJECTION_CODES = {
    "INVALID_COORDS":      "Coordinates out of valid range",
    "NULL_COORDS":         "Latitude or longitude is null",
    "FUTURE_DATE":         "event_time is more than 1 hour in the future",
    "NULL_EVENT_TYPE":     "event_type is null or empty",
    "NEGATIVE_FATALITIES": "fatalities value is invalid (< -1)",
    "SCHEMA_ERROR":        "Failed to parse raw payload",
}

def validate_event(event: EventCanonicalCreate) -> ValidationResult:
    now = datetime.now(timezone.utc)
    future_threshold = now + timedelta(hours=1) - timedelta(seconds=1)

    if event.latitude is None or event.longitude is None:
        return ValidationResult(is_valid=False, rejection_code="NULL_COORDS",
            rejection_detail="latitude or longitude is null")

    if event.latitude < -90 or event.latitude > 90:
        return ValidationResult(is_valid=False, rejection_code="INVALID_COORDS",
            rejection_detail=f"latitude {event.latitude} out of range [-90, 90]")

    if event.longitude < -180 or event.longitude > 180:
        return ValidationResult(is_valid=False, rejection_code="INVALID_COORDS",
            rejection_detail=f"longitude {event.longitude} out of range [-180, 180]")

    if event.event_time > future_threshold:
        return ValidationResult(is_valid=False, rejection_code="FUTURE_DATE",
            rejection_detail=f"event_time {event.event_time} is more than 1 hour in the future")

    if event.event_type is None or not str(event.event_type).strip():
        return ValidationResult(is_valid=False, rejection_code="NULL_EVENT_TYPE",
            rejection_detail="event_type is null or empty")

    if event.fatalities is not None and event.fatalities < -1:
        return ValidationResult(is_valid=False, rejection_code="NEGATIVE_FATALITIES",
            rejection_detail=f"fatalities value {event.fatalities} is invalid (minimum allowed is -1)")

    return ValidationResult(is_valid=True, event=event)

import hashlib

def firms_event_id(row: dict) -> str:
    key = f"{row['latitude']}|{row['longitude']}|{row['acq_date']}|{row['acq_time']}|{row['satellite']}"
    return hashlib.sha256(key.encode()).hexdigest()[:32]