osint-dashboard/app/fire_sources.py

"""NASA FIRMS active fire / hotspot ingestor.

Fetches the VIIRS active-fire detection CSV for an area from NASA FIRMS and
publishes each hotspot to NATS JetStream on the ``events.fire`` subject. The
NATS consumer (app/ingestor.py) routes those messages to the ``fires``
TimescaleDB hypertable, deduped by the natural key (lat, lon, acq_time,
satellite).

Zero-cost data: the FIRMS MAP_KEY is free (https://firms.modaps.eosdis.nasa.gov/api/map_key_info/)
and the CSV area endpoint is unlimited for personal/research use.

Endpoint (documented at https://firms.modaps.eosdis.nasa.gov/api/area/csv/):
    https://firms.modaps.eosdis.nasa.gov/api/area/csv/{MAP_KEY}/{DATASET}/{bbox}
    bbox = "minlon,minlat,maxlon,maxlat"  (e.g. "-125,24,-66,50")

CSV columns (VIIRS): latitude, longitude, bright_ti4, scan, track, acq_date,
acq_time, satellite, instrument, confidence, version, bright_ti5, frp, daynight.
``acq_time`` is an integer HHMM in UTC (seconds are 0); ``acq_date`` is the
UTC date (YYYY-MM-DD).
"""

from __future__ import annotations

import csv
import io
import json
import logging
import os
from datetime import datetime, timezone

import httpx
import nats

from config import FIRMS_DATASET, FIRMS_BBOX, FIRMS_TIMEOUT, NATS_URL
from keystore import get_api_key

logger = logging.getLogger("osint.firms")

# ── FIRMS API ─────────────────────────────────────────────────────────────

FIRMS_AREA_CSV = (
    "https://firms.modaps.eosdis.nasa.gov/api/area/csv/{key}/{dataset}/{bbox}"
)

# The canonical VIIRS CSV header FIRMS returns. Used to (a) locate the real
# header row if FIRMS ever prepends a legend line and (b) validate a download.
FIRMS_CSV_COLUMNS = (
    "latitude", "longitude", "bright_ti4", "scan", "track", "acq_date",
    "acq_time", "satellite", "instrument", "confidence", "version",
    "bright_ti5", "frp", "daynight",
)


def _to_float(value: object) -> float | None:
    """Best-effort float conversion; returns None on empty/garbage values."""
    if value is None:
        return None
    try:
        return float(value)
    except (ValueError, TypeError):
        return None


def normalize_acq_time(acq_date: object, acq_time: object) -> datetime | None:
    """Combine FIRMS acq_date + acq_time into a timezone-aware UTC datetime.

    acq_date is 'YYYY-MM-DD'; acq_time is an integer HHMM in UTC with seconds
    truncated. Returns None when the values can't be parsed (dropped).
    """
    if not acq_date or acq_time is None or acq_time == "":
        return None
    try:
        hhmm = str(int(acq_time)).zfill(4)
        return datetime.strptime(
            f"{acq_date} {hhmm}", "%Y-%m-%d %H%M"
        ).replace(tzinfo=timezone.utc)
    except (ValueError, TypeError):
        return None


def parse_firms_csv(text: str) -> list[dict]:
    """Parse a FIRMS area CSV payload into normalized fire messages.

    Returns one dict per hotspot with the fields stored in the ``fires`` table
    (acq_time already combined into a UTC ISO timestamp). Rows that don't look
    like valid VIIRS detections are skipped rather than failing the whole poll.
    """
    rows = list(csv.reader(io.StringIO(text)))
    if not rows:
        return []

    # Locate the real header row. FIRMS normally returns the CSV header first,
    # but occasionally prepends a legend/info line, so scan until we see the
    # canonical header.
    header_idx = 0
    for i, row in enumerate(rows):
        if row and row[0].strip().lower() == "latitude" and len(row) >= 4:
            header_idx = i
            break
    header = [c.strip().lower() for c in rows[header_idx]]
    # Guard against a header that isn't actually the FIRMS one.
    if "latitude" not in header or "longitude" not in header:
        logger.warning("FIRMS payload does not look like a hotspot CSV (first row: %r)", header[:6])
        return []

    points: list[dict] = []
    for row in rows[header_idx + 1:]:
        if len(row) < len(header):
            continue
        rec = dict(zip(header, row))
        lat = _to_float(rec.get("latitude"))
        lon = _to_float(rec.get("longitude"))
        if lat is None or lon is None:
            continue
        acq_time = normalize_acq_time(rec.get("acq_date"), rec.get("acq_time"))
        if acq_time is None:
            continue
        points.append({
            "latitude": lat,
            "longitude": lon,
            "brightness": _to_float(rec.get("bright_ti4")),
            "confidence": str(rec.get("confidence") or "").strip(),
            "acq_time": acq_time.isoformat(),
            "satellite": str(rec.get("satellite") or "").strip(),
            "instrument": str(rec.get("instrument") or "").strip(),
            "bright_ti5": _to_float(rec.get("bright_ti5")),
            "frp": _to_float(rec.get("frp")),
            "daynight": str(rec.get("daynight") or "").strip(),
            "scan": _to_float(rec.get("scan")),
            "track": _to_float(rec.get("track")),
            "version": str(rec.get("version") or "").strip(),
        })
    return points


async def publish_fire_batch(points: list[dict]) -> int:
    """Publish a batch of hotspot messages to NATS JetStream in one connection.

    Fires are bulk data (thousands of detections per poll), so opening a NATS
    connection per message — as the RSS/GDELT path does — would be wasteful.
    One connection, one JetStream context, one close, with the batch flushed
    before returning so a crash mid-write can't silently drop half a poll.
    """
    if not points:
        return 0
    nc = await nats.connect(NATS_URL)
    js = nc.jetstream()
    try:
        for pt in points:
            msg = {"source_type": "fire", **pt}
            await js.publish("events.fire", json.dumps(msg).encode())
    finally:
        await nc.close()
    return len(points)


async def ingest_fires(bbox: str | None = None) -> int:
    """Fetch the FIRMS hotspot CSV for an area and publish it to NATS.

    Returns the number of hotspot messages published. Idle (0) and a logged
    warning when FIRMS_MAP_KEY is not set, so the ingester keeps running for
    the other sources.
    """
    # Key resolution: env var first (12-factor), then the shared Postgres
    # api_keys table so a key saved via the dashboard Keys page is picked up
    # on the next poll without a restart.
    map_key = os.getenv("FIRMS_MAP_KEY", "") or await get_api_key("FIRMS_MAP_KEY")
    if not map_key:
        logger.warning(
            "FIRMS_MAP_KEY not set — fire ingest disabled. "
            "Set it in the dashboard Keys page or .env. "
            "Free key: https://firms.modaps.eosdis.nasa.gov/api/map_key_info/"
        )
        return 0

    area = bbox or FIRMS_BBOX
    url = FIRMS_AREA_CSV.format(
        key=map_key, dataset=FIRMS_DATASET, bbox=area
    )

    async with httpx.AsyncClient(timeout=FIRMS_TIMEOUT) as client:
        resp = await client.get(url)
        resp.raise_for_status()
        text = resp.text

    # FIRMS returns HTTP 200 with a plain-text error for some failure modes
    # (bad key, invalid bbox); surface the first line for debuggability.
    if "latitude" not in text.lower()[:4096]:
        first_line = text.strip().splitlines()[0][:200] if text.strip() else "(empty)"
        logger.warning("FIRMS CSV download returned no hotspot data (%s)", first_line)
        return 0

    points = parse_firms_csv(text)
    published = await publish_fire_batch(points)
    logger.info(
        "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d",
        len(points), area, FIRMS_DATASET, published,
    )
    return published