Python ETL for Sensor & IoT Data in Emergency Response GIS

import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Any

from pydantic import BaseModel, Field, field_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s",
)
logger = logging.getLogger("etl.contract")


class ReadingType(str, Enum):
    WIND_SPEED = "wind_speed"
    AIR_QUALITY = "air_quality"
    WATER_LEVEL = "water_level"
    TEMPERATURE = "temperature"


class SensorPayload(BaseModel):
    """The ingestion contract. Every field is mandatory and bounded."""

    device_id: str = Field(min_length=1)
    latitude: float = Field(ge=-90.0, le=90.0)
    longitude: float = Field(ge=-180.0, le=180.0)
    reading_type: ReadingType
    reading_value: float
    observed_at: datetime
    crs: str = "EPSG:4326"

    @field_validator("observed_at")
    @classmethod
    def enforce_utc(cls, value: datetime) -> datetime:
        """Reject naive timestamps and normalise everything to UTC."""
        if value.tzinfo is None:
            raise ValueError("observed_at must be timezone-aware")
        return value.astimezone(timezone.utc)


def validate_payload(raw: dict[str, Any]) -> SensorPayload | None:
    """Validate a raw record; return None (and log) instead of raising on failure."""
    try:
        return SensorPayload.model_validate(raw)
    except Exception as exc:  # pydantic.ValidationError and malformed input
        logger.warning("Payload rejected at boundary: %s", exc)
        return None

import asyncio

import aiohttp

logger = logging.getLogger("etl.extract")


async def fetch_with_backoff(
    url: str,
    session: aiohttp.ClientSession,
    sem: asyncio.Semaphore,
    *,
    max_attempts: int = 4,
) -> dict[str, Any] | None:
    """Fetch one endpoint, retrying transient transport failures with backoff."""
    async with sem:  # bound concurrency so surge load cannot exhaust the pool
        for attempt in range(1, max_attempts + 1):
            try:
                timeout = aiohttp.ClientTimeout(total=5)
                async with session.get(url, timeout=timeout) as response:
                    response.raise_for_status()
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                wait = 2 ** (attempt - 1)
                logger.warning(
                    "Transport failure (%s) on %s, attempt %d/%d; retry in %ds",
                    exc, url, attempt, max_attempts, wait,
                )
                await asyncio.sleep(wait)
    logger.error("Endpoint exhausted retries, deferring to dead-letter: %s", url)
    return None


async def extract_batch(urls: list[str], concurrency: int = 32) -> list[dict[str, Any]]:
    """Pull a batch of endpoints concurrently and drop exhausted ones."""
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_backoff(u, session, sem) for u in urls]
        results = await asyncio.gather(*tasks)
    return [r for r in results if r is not None]

import json
import sqlite3

logger = logging.getLogger("etl.deadletter")


def init_dead_letter(conn: sqlite3.Connection) -> None:
    """Create the quarantine table if absent."""
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS dead_letter (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT,
            raw_payload TEXT NOT NULL,
            reason TEXT NOT NULL,
            quarantined_at TEXT NOT NULL
        )
        """
    )


def quarantine(conn: sqlite3.Connection, raw: dict[str, Any], reason: str) -> None:
    """Persist a rejected payload for later replay; never lose the bytes."""
    try:
        with conn:  # atomic insert
            conn.execute(
                "INSERT INTO dead_letter (device_id, raw_payload, reason, quarantined_at)"
                " VALUES (?, ?, ?, ?)",
                (
                    str(raw.get("device_id", "unknown")),
                    json.dumps(raw, default=str),
                    reason,
                    datetime.now(timezone.utc).isoformat(),
                ),
            )
        logger.info("Quarantined payload from %s: %s", raw.get("device_id"), reason)
    except sqlite3.Error as exc:
        logger.critical("Dead-letter write failed; payload at risk: %s", exc)
        raise

import geopandas as gpd
from shapely.geometry import Point

logger = logging.getLogger("etl.transform")


def transform_to_incident_layer(
    payloads: list[SensorPayload],
    jurisdictions: gpd.GeoDataFrame,
    *,
    target_epsg: int = 26910,   # UTM Zone 10N — set per operational area
    hazard_radius_m: float = 500.0,
) -> gpd.GeoDataFrame:
    """Project validated readings, attach jurisdiction, and buffer hazard radii in metres."""
    if not payloads:
        return gpd.GeoDataFrame()

    records = [p.model_dump() for p in payloads]
    geometry = [Point(p.longitude, p.latitude) for p in payloads]

    # Build in the storage CRS, then reproject for all metric operations.
    sensors = gpd.GeoDataFrame(records, geometry=geometry, crs="EPSG:4326")
    sensors = sensors.to_crs(epsg=target_epsg)

    if jurisdictions.crs is None or jurisdictions.crs.to_epsg() != target_epsg:
        jurisdictions = jurisdictions.to_crs(epsg=target_epsg)

    joined = gpd.sjoin(sensors, jurisdictions, how="left", predicate="within")

    unmatched = int(joined["index_right"].isna().sum())
    if unmatched:
        logger.warning("%d reading(s) fell outside all jurisdiction polygons", unmatched)

    # Buffer in metres — valid only because we are in a projected CRS.
    joined["hazard_buffer"] = joined.geometry.buffer(hazard_radius_m)
    return joined

logger = logging.getLogger("etl.load")


def init_audit(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS etl_audit (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            device_id TEXT NOT NULL,
            observed_at TEXT NOT NULL,
            target_epsg INTEGER NOT NULL,
            jurisdiction_id TEXT,
            outcome TEXT NOT NULL,
            committed_at TEXT NOT NULL
        )
        """
    )


def load_and_audit(
    conn: sqlite3.Connection,
    feature: dict[str, Any],
    target_epsg: int,
) -> None:
    """Persist one feature and its audit record atomically."""
    try:
        with conn:  # both writes commit or roll back together
            conn.execute(
                "INSERT OR REPLACE INTO incident_features"
                " (device_id, observed_at, geom_wkt) VALUES (?, ?, ?)",
                (feature["device_id"], feature["observed_at"], feature["geom_wkt"]),
            )
            conn.execute(
                "INSERT INTO etl_audit"
                " (device_id, observed_at, target_epsg, jurisdiction_id, outcome, committed_at)"
                " VALUES (?, ?, ?, ?, ?, ?)",
                (
                    feature["device_id"],
                    feature["observed_at"],
                    target_epsg,
                    feature.get("jurisdiction_id"),
                    "loaded",
                    datetime.now(timezone.utc).isoformat(),
                ),
            )
        logger.info("Loaded reading from %s", feature["device_id"])
    except sqlite3.Error as exc:
        logger.critical("Load transaction failed for %s: %s", feature["device_id"], exc)
        raise

from datetime import datetime, timezone


def smoke_test() -> None:
    # 1. The contract rejects an out-of-range coordinate.
    bad = {"device_id": "wx-07", "latitude": 95.0, "longitude": -120.0,
           "reading_type": "wind_speed", "reading_value": 18.0,
           "observed_at": datetime.now(timezone.utc)}
    assert validate_payload(bad) is None, "latitude > 90 must be rejected"

    # 2. The contract rejects a naive timestamp.
    naive = dict(bad, latitude=45.0, observed_at=datetime.now())
    assert validate_payload(naive) is None, "naive observed_at must be rejected"

    # 3. A buffer is only metric in a projected CRS.
    p = validate_payload(dict(bad, latitude=45.0))
    assert p is not None
    import geopandas as gpd
    from shapely.geometry import Point
    g = gpd.GeoDataFrame(geometry=[Point(p.longitude, p.latitude)], crs="EPSG:4326")
    g = g.to_crs(epsg=26910)
    assert g.geometry.buffer(500).area.iloc[0] > 700_000, "buffer area must be ~785k m^2"

    print("smoke test passed")


if __name__ == "__main__":
    smoke_test()