Spaces:
Sleeping
Sleeping
File size: 4,056 Bytes
6dad9a7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Curated Sentinel-2 (Track-B) AOI registry for the ``/sentinel2`` demo tab (M5, PRD §6.3/§10).
A curated Sentinel-2 *AOI* is a real-world location with large, obvious change visible even at 10 m
(reclamation, a filling reservoir, an airport built from farmland, a solar park, a desert city). Its
before/after imagery and change prediction are **baked offline** by ``build_sentinel2.py`` and
served here straight from the cache — **no runtime inference, no runtime STAC, no GPU**, exactly
like the aerial curated mode. The runtime image stays STAC-free (no ``pystac``/``rasterio`` deps).
The registry reads ``<data_dir>/manifest.json`` (per-AOI metadata: title, MGRS tile, centre, the
acquisition dates + cloud cover) and the baked ``<data_dir>/_predictions.json`` (the same cache
schema ``inference.py.predict`` returns), and exposes both to the API.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from PIL import Image
_IMAGE_KINDS = ("before", "after", "overlay")
class Sentinel2Registry:
"""Reads the baked Sentinel-2 manifest + prediction cache and serves them (cache only)."""
def __init__(self, data_dir: str | Path) -> None:
self.data_dir = Path(data_dir)
self.aois: dict[str, dict[str, Any]] = {}
self.predictions: dict[str, dict[str, Any]] = {}
self.reload()
def reload(self) -> None:
self.aois.clear()
self.predictions.clear()
manifest = self.data_dir / "manifest.json"
if manifest.exists():
for entry in json.loads(manifest.read_text()).get("pairs", []):
aid = str(entry["id"])
if (self.data_dir / aid / "before.png").exists():
self.aois[aid] = entry
cache = self.data_dir / "_predictions.json"
if cache.exists():
try:
data = json.loads(cache.read_text())
except json.JSONDecodeError:
data = {}
for aid, pred in data.items():
if aid in self.aois:
self.predictions[aid] = pred
def list(self) -> list[dict[str, Any]]:
"""Per-AOI metadata + the baked prediction summary (stats/threshold/tiles), minus the heavy
``overlay_png`` data URL — the overlay is served as a PNG file via :meth:`image_path`."""
out = []
for aid, entry in self.aois.items():
pred = self.predictions.get(aid, {})
out.append(
{
"id": aid,
"title": entry.get("title", aid),
"description": entry.get("description", ""),
"source": entry.get("source", "Sentinel-2 L2A · 10 m"),
"tile": entry.get("tile", ""),
"center": entry.get("center"),
"width": entry.get("width"),
"height": entry.get("height"),
"date_before": entry.get("date_before"),
"date_after": entry.get("date_after"),
"cloud_before": entry.get("cloud_before"),
"cloud_after": entry.get("cloud_after"),
"model_id": pred.get("model_id", ""),
"threshold": pred.get("threshold"),
"is_placeholder": pred.get("is_placeholder", False),
"n_tiles": pred.get("n_tiles"),
"input_size": pred.get("input_size"),
"elapsed_ms": pred.get("elapsed_ms"),
"stats": pred.get("stats", {}),
}
)
return out
def image_path(self, aoi_id: str, which: str) -> Path:
if aoi_id not in self.aois:
raise KeyError(aoi_id)
if which not in _IMAGE_KINDS:
raise ValueError(which)
return self.data_dir / aoi_id / f"{which}.png"
def dimensions(self, aoi_id: str) -> tuple[int, int]:
with Image.open(self.image_path(aoi_id, "before")) as im:
return im.size
|