170 lines
6.3 KiB
Python
170 lines
6.3 KiB
Python
"""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, timezone
|
||
|
||
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).
|
||
|
||
GIBS' published windows include the current UTC day as soon as it starts,
|
||
but daily mosaics (MODIS/VIIRS true color, night lights) aren't actually
|
||
ingested until hours later — tiles for "today" 404 until then. So we clamp
|
||
daily-layer window ends to *yesterday* UTC: 'latest' then always points at
|
||
imagery that really exists.
|
||
"""
|
||
windows = []
|
||
min_d, max_d = None, None
|
||
# 00:00–~03:00+ UTC the newest full day is always yesterday; use a small
|
||
# margin so we never advertise a same-day mosaic that isn't ingested yet.
|
||
cutoff = datetime.now(timezone.utc).date() - timedelta(days=1)
|
||
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))
|
||
if m.group(3) == "P1D" and end > cutoff:
|
||
end = cutoff
|
||
if start > end:
|
||
continue # whole window was "not yet ingested" days
|
||
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
|