osint-dashboard/app/gibs_map.py

"""NASA GIBS basemap catalog + time-domain helper for the OSINT map tab.

GIBS (Global Imagery Browse Services) is what worldview.earthdata.nasa.gov
renders from. Tiles are plain XYZ/WebMercator WMTS rasters, CORS-open
(access-control-allow-origin: *), so the browser can pull them directly —
no tile proxy needed.

Layer metadata is curated here (identifier, display title, tile matrix set,
format, whether it has a Time dimension). Available time windows per layer
come from GIBS' per-layer Domains XML endpoint (tiny), fetched on demand and
cached for a few hours.

Tile URL template (RESTful WMTS):
  https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/{layer}/default/{time}/{tms}/{z}/{y}/{x}.{ext}
The {time} path segment is omitted entirely for static layers (no Time dim).
"""

from __future__ import annotations

import asyncio
import re
import time
from datetime import date, datetime, timedelta

import httpx

GIBS_BASE = "https://gibs.earthdata.nasa.gov/wmts/epsg3857/best"
GIBS_TIMEOUT = 30.0

# ── Curated basemap catalog ────────────────────────────────────────────────
# has_time: True → tile URL includes /default/{time}/; the /api/map/times
# endpoint returns the valid date windows. False → static basemap.
MAP_LAYERS = [
    {
        "id": "BlueMarble_ShadedRelief_Bathymetry",
        "title": "Blue Marble (shaded relief + bathymetry)",
        "subtitle": "Static MODIS composite basemap",
        "tms": "GoogleMapsCompatible_Level8",
        "format": "jpeg",
        "has_time": False,
        "max_zoom": 8,
    },
    {
        "id": "VIIRS_SNPP_CorrectedReflectance_TrueColor",
        "title": "VIIRS S-NPP True Color (daily)",
        "subtitle": "Suomi NPP corrected reflectance",
        "tms": "GoogleMapsCompatible_Level9",
        "format": "jpeg",
        "has_time": True,
        "max_zoom": 9,
    },
    {
        "id": "MODIS_Terra_CorrectedReflectance_TrueColor",
        "title": "MODIS Terra True Color (daily)",
        "subtitle": "Terra corrected reflectance",
        "tms": "GoogleMapsCompatible_Level9",
        "format": "jpeg",
        "has_time": True,
        "max_zoom": 9,
    },
    {
        "id": "MODIS_Aqua_CorrectedReflectance_TrueColor",
        "title": "MODIS Aqua True Color (daily)",
        "subtitle": "Aqua corrected reflectance",
        "tms": "GoogleMapsCompatible_Level9",
        "format": "jpeg",
        "has_time": True,
        "max_zoom": 9,
    },
    {
        "id": "VIIRS_SNPP_DayNightBand_ENCC",
        "title": "VIIRS Night Lights (DNB)",
        "subtitle": "Earth at night, enhanced near-constant contrast",
        "tms": "GoogleMapsCompatible_Level8",
        "format": "png",
        "has_time": True,
        "max_zoom": 8,
    },
]

_LAYER_BY_ID = {l["id"]: l for l in MAP_LAYERS}

# ── Time-domain cache (per layer) ──────────────────────────────────────────
# GIBS Domains XML gives a comma-joined list of ISO ranges:  START/END/PERIOD
_DOMAIN_CACHE: dict[str, tuple[float, dict]] = {}
_DOMAIN_TTL = 6 * 3600  # 6h
_HTTPS_LOCK = asyncio.Lock()

# e.g. 2026-07-16/2026-08-24/P1D
_TIME_RANGE_RE = re.compile(
    r"^(\d{4}-\d{2}-\d{2})/(\d{4}-\d{2}-\d{2})/(P\d+D|P\d+M|PT\d+H)$"
)


def _parse_time_domain(domain: str) -> dict:
    """Expand a Domains XML <Domain> string into usable date metadata.

    Returns {min, max, latest, ranges:[{start,end}]} where ranges keep the
    raw GIBS windows so the frontend can warn when a picked date is outside
    every window (GIBS still serves nearest-time tiles, but this is honest).
    """
    windows = []
    min_d, max_d = None, None
    for part in domain.split(","):
        part = part.strip()
        m = _TIME_RANGE_RE.match(part)
        if not m:
            continue
        start = date.fromisoformat(m.group(1))
        end = date.fromisoformat(m.group(2))
        windows.append({"start": start.isoformat(), "end": end.isoformat()})
        if min_d is None or start < min_d:
            min_d = start
        if max_d is None or end > max_d:
            max_d = end
    return {
        "min": min_d.isoformat() if min_d else None,
        "max": max_d.isoformat() if max_d else None,
        "latest": max_d.isoformat() if max_d else None,
        "ranges": windows,
    }


async def fetch_layer_domain(layer_id: str, tms: str) -> dict:
    """Return parsed time domain for a layer, cached for _DOMAIN_TTL."""
    now = time.time()
    hit = _DOMAIN_CACHE.get(layer_id)
    if hit and now - hit[0] < _DOMAIN_TTL:
        return hit[1]

    url = (
        f"{GIBS_BASE}/1.0.0/{layer_id}/default/{tms}/all/all.xml"
    )
    # One in-flight request per layer; serialize so a burst of layer switches
    # doesn't fan out parallel GIBS hits.
    async with _HTTPS_LOCK:
        hit = _DOMAIN_CACHE.get(layer_id)
        if hit and time.time() - hit[0] < _DOMAIN_TTL:
            return hit[1]
        async with httpx.AsyncClient(timeout=GIBS_TIMEOUT) as client:
            resp = await client.get(url)
            resp.raise_for_status()
        m = re.search(r"<Domain>([^<]+)</Domain>", resp.text)
        if not m:
            raise ValueError(f"No time <Domain> in GIBS response for {layer_id}")
        parsed = _parse_time_domain(m.group(1))
        _DOMAIN_CACHE[layer_id] = (time.time(), parsed)
        return parsed


def default_time_for(layer_id: str) -> str | None:
    """Most recent data date for a time-aware layer (best effort, from cache)."""
    hit = _DOMAIN_CACHE.get(layer_id)
    if not hit:
        return None
    latest = hit[1].get("latest")
    return latest or None