import os os.environ.setdefault("HF_HOME", "/data/.cache/huggingface") os.environ.setdefault("HF_MODULES_CACHE", "/tmp/hf_modules") os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib") import base64 import hashlib import io import json import time import logging import numpy as np import torch from einops import rearrange from PIL import Image from sklearn.decomposition import PCA import gradio as gr from gradio.routes import HTMLResponse, JSONResponse from olmoearth_pretrain_minimal import load_model_from_id, ModelID, Normalizer from olmoearth_pretrain_minimal.olmoearth_pretrain_v1.utils.constants import Modality from olmoearth_pretrain_minimal.olmoearth_pretrain_v1.utils.datatypes import ( MaskedOlmoEarthSample, MaskValue, ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Model loading (at startup, CPU only) # --------------------------------------------------------------------------- logger.info("Loading OlmoEarth-v1-Base model...") _t0 = time.perf_counter() model = load_model_from_id(ModelID.OLMOEARTH_V1_BASE) model.eval() encoder = model.encoder num_params = sum(p.numel() for p in encoder.parameters()) logger.info(f"Model loaded in {time.perf_counter() - _t0:.1f}s — {num_params:,} params ({num_params / 1e6:.1f}M)") normalizer = Normalizer(std_multiplier=2.0) S2_SPEC = Modality.SENTINEL2_L2A S2_BAND_ORDER = S2_SPEC.band_order # 12 bands NORM_CONFIG = normalizer.norm_config["sentinel2_l2a"] PATCH_SIZE = 4 EMBED_DIM = 768 PAPER_URL = "https://allenai.org/papers/olmoearth" MODEL_URL = "https://huggingface.co/allenai/OlmoEarth-v1-Base" GITHUB_URL = "https://github.com/allenai/olmoearth_pretrain" MAX_EMBED_CACHE = 12 EMBEDDING_CACHE: dict[str, np.ndarray] = {} EMBEDDING_CACHE_ORDER: list[str] = [] # --------------------------------------------------------------------------- # Image processing helpers # --------------------------------------------------------------------------- def normalize_sentinel2(data: np.ndarray) -> np.ndarray: """Normalize data using OlmoEarth's computed Sentinel-2 stats. data shape: (C, H, W) or (T, C, H, W) — last-matched band dim is normalized. """ data = data.astype(np.float32).copy() band_means = np.array([NORM_CONFIG[b]["mean"] for b in S2_BAND_ORDER], dtype=np.float32) band_stds = np.array([NORM_CONFIG[b]["std"] for b in S2_BAND_ORDER], dtype=np.float32) min_vals = band_means - 2.0 * band_stds max_vals = band_means + 2.0 * band_stds ranges = max_vals - min_vals ranges[ranges == 0] = 1.0 # Broadcast: data shape (..., C, H, W), normalize along C axis if data.ndim == 3: data = (data - min_vals[:, None, None]) / ranges[:, None, None] elif data.ndim == 4: data = (data - min_vals[None, :, None, None]) / ranges[None, :, None, None] return np.clip(data, 0, 1) def rgb_to_12band(rgb: np.ndarray) -> np.ndarray: """Expand a 3-channel RGB array to 12-channel Sentinel-2 band order. rgb: (H, W, 3) uint8 or float returns: (12, H, W) float32 normalized """ if rgb.dtype == np.uint8: rgb = rgb.astype(np.float32) h, w, c = rgb.shape # Map RGB -> approximate Sentinel-2 reflectance values # Scale 0-255 to approximate 0-3000 reflectance range scaled = rgb * (3000.0 / 255.0) bands_12 = np.zeros((12, h, w), dtype=np.float32) # B02=blue, B03=green, B04=red, B08=NIR ~ near red bands_12[0] = scaled[:, :, 2] # B02 (blue) <- R channel (reversed for visual) bands_12[1] = scaled[:, :, 1] # B03 (green) bands_12[2] = scaled[:, :, 0] # B04 (red) bands_12[3] = scaled[:, :, 1] * 1.2 # B08 (NIR) - approx from green bands_12[4] = scaled[:, :, 0] * 0.9 # B05 bands_12[5] = scaled[:, :, 1] * 0.9 # B06 bands_12[6] = scaled[:, :, 1] * 0.8 # B07 bands_12[7] = scaled[:, :, 1] * 1.1 # B8A bands_12[8] = scaled[:, :, 2] * 0.7 # B11 (SWIR) bands_12[9] = scaled[:, :, 2] * 0.5 # B12 (SWIR) bands_12[10] = scaled[:, :, 2] * 0.3 # B01 (coastal aerosol) bands_12[11] = scaled[:, :, 0] * 0.4 # B09 (water vapor) return bands_12 def extract_embeddings(img: Image.Image) -> tuple[np.ndarray, int, int]: """Extract per-patch embeddings from a PIL image. Returns (patch_embeddings, n_patch_h, n_patch_w). patch_embeddings shape: (n_patch_h, n_patch_w, embed_dim) """ # Convert to RGB and resize to a multiple of patch_size img = img.convert("RGB") w, h = img.size # Target size: nearest multiple of PATCH_SIZE, capped at 224 max_dim = 224 scale = min(max_dim / w, max_dim / h, 1.0) new_w = int(round(w * scale / PATCH_SIZE)) * PATCH_SIZE new_h = int(round(h * scale / PATCH_SIZE)) * PATCH_SIZE new_w = max(new_w, PATCH_SIZE) new_h = max(new_h, PATCH_SIZE) img = img.resize((new_w, new_h), Image.BILINEAR) rgb = np.array(img, dtype=np.float32) # (H, W, 3) bands_12 = rgb_to_12band(rgb) # (12, H, W) bands_12 = normalize_sentinel2(bands_12) # normalize per band # Build model input: (B, H, W, T=1, BandSets=3, Bands) # Sentinel-2 bandsets: [B02,B03,B04,B08], [B05,B06,B07,B8A,B11,B12], [B01,B09] bandset_indices = S2_SPEC.bandsets_as_indices() # [[0,1,2,3], [4,5,6,7,8,9], [10,11]] bandsets_data = [] for indices in bandset_indices: bandsets_data.append(bands_12[indices]) # (n_bands, H, W) # Stack: (BandSets, Bands, H, W) -> (H, W, T=1, BandSets, Bands) # But the model expects (B, H, W, T, BandSets, Bands) via the sample # Actually the sample format is (B, H, W, T, BandSets, Bands) # Wait - looking at the tutorial, the data is (B, H, W, T, BandSets) where each entry has D dim after embedding # Let me re-check: the input tensor is (B, H, W, T, BandSets) where each entry is the raw data # No - looking more carefully at the tutorial code: # s2_tensor = rearrange(s2_tensor, 't c h w -> 1 h w t c') # This gives (1, H, W, T, C=12) - all 12 bands in one dimension # And mask is (1, H, W, T, 3) - 3 bandsets # So the input is (B, H, W, T, Bands=12) and the encoder internally splits into bandsets s2_data = np.stack(bands_12, axis=0) # (12, H, W) s2_tensor = torch.from_numpy(s2_data.astype(np.float32)) # (12, H, W) s2_tensor = rearrange(s2_tensor, 'c h w -> 1 h w 1 c') # (1, H, W, T=1, 12) h, w = new_h, new_w mask = torch.full((1, h, w, 1, 3), MaskValue.ONLINE_ENCODER.value, dtype=torch.int32) timestamps = torch.zeros((1, 1, 3), dtype=torch.int32) timestamps[0, :, 1] = 0 # month = 0 (January) sample = MaskedOlmoEarthSample( sentinel2_l2a=s2_tensor, sentinel2_l2a_mask=mask, timestamps=timestamps, ) with torch.no_grad(): output = encoder(sample, fast_pass=True, patch_size=PATCH_SIZE) tokens = output["tokens_and_masks"].sentinel2_l2a # Shape: (B, P_H, P_W, T, BandSets, D) # Pool across T and BandSets to get (B, P_H, P_W, D) patch_emb = tokens.mean(dim=[3, 4]) # (1, P_H, P_W, D) patch_emb = patch_emb[0].float().cpu().numpy() # (P_H, P_W, D) n_ph, n_pw, d = patch_emb.shape return patch_emb, n_ph, n_pw def embeddings_to_rgb_pca(patch_emb: np.ndarray) -> np.ndarray: """PCA project embeddings to 3 dims, normalize to 0-255 RGB. patch_emb: (P_H, P_W, D) returns: (P_H, P_W, 3) uint8 """ ph, pw, d = patch_emb.shape flat = patch_emb.reshape(-1, d) # (P_H*P_W, D) pca = PCA(n_components=3) pca.fit(flat) projected = pca.transform(flat) # (P_H*P_W, 3) # Normalize each component independently to [0, 1] for i in range(3): col = projected[:, i] cmin, cmax = col.min(), col.max() if cmax - cmin > 1e-8: projected[:, i] = (col - cmin) / (cmax - cmin) else: projected[:, i] = 0.5 rgb = (projected * 255).clip(0, 255).astype(np.uint8) rgb = rgb.reshape(ph, pw, 3) return rgb def embeddings_to_heatmap(patch_emb: np.ndarray) -> np.ndarray: """Create a similarity heatmap from embeddings. Shows cosine similarity of each patch to the mean embedding. returns: (P_H, P_W, 3) uint8 """ ph, pw, d = patch_emb.shape flat = patch_emb.reshape(-1, d) mean_emb = flat.mean(axis=0) # Cosine similarity norms = np.linalg.norm(flat, axis=1) * np.linalg.norm(mean_emb) norms[norms == 0] = 1.0 sim = (flat @ mean_emb) / norms sim = sim.reshape(ph, pw) # Map to color: low sim = dark blue, high sim = cyan/green sim_norm = (sim - sim.min()) / (sim.max() - sim.min() + 1e-8) r = (sim_norm * 50).astype(np.uint8) g = (sim_norm * 200 + 55).astype(np.uint8) b = (sim_norm * 200 + 55).astype(np.uint8) return np.stack([r, g, b], axis=-1) def embeddings_to_query_heatmap( patch_emb: np.ndarray, x_norm: float, y_norm: float, ) -> tuple[np.ndarray, dict]: """Create a brightness heatmap from cosine similarity to one clicked patch.""" ph, pw, d = patch_emb.shape q_col = int(np.clip(np.floor(float(x_norm) * pw), 0, pw - 1)) q_row = int(np.clip(np.floor(float(y_norm) * ph), 0, ph - 1)) flat = patch_emb.reshape(-1, d).astype(np.float32) flat_norms = np.linalg.norm(flat, axis=1, keepdims=True) flat_norms[flat_norms == 0] = 1.0 flat_unit = flat / flat_norms query_idx = q_row * pw + q_col query = flat_unit[query_idx] sim = (flat_unit @ query).reshape(ph, pw) lo, hi = np.percentile(sim, [2, 98]) if hi - lo < 1e-8: lo, hi = float(sim.min()), float(sim.max()) sim_norm = np.clip((sim - lo) / (hi - lo + 1e-8), 0, 1) sim_norm = sim_norm ** 0.85 # Brightness encodes similarity; color keeps the map legible on a dark UI. r = (12 + sim_norm * 243).astype(np.uint8) g = (18 + sim_norm * 226).astype(np.uint8) b = (34 + sim_norm * 166).astype(np.uint8) rgb = np.stack([r, g, b], axis=-1) # Mark the clicked query patch. for rr in range(max(0, q_row - 1), min(ph, q_row + 2)): rgb[rr, q_col] = [255, 205, 64] for cc in range(max(0, q_col - 1), min(pw, q_col + 2)): rgb[q_row, cc] = [255, 205, 64] return rgb, { "row": int(q_row), "col": int(q_col), "grid_h": int(ph), "grid_w": int(pw), "x_norm": float(np.clip(x_norm, 0, 1)), "y_norm": float(np.clip(y_norm, 0, 1)), "max_similarity": float(sim.max()), "mean_similarity": float(sim.mean()), } def upsample_to_image(small: np.ndarray, target_h: int, target_w: int) -> np.ndarray: """Nearest-neighbor upsample a small array to target size.""" pil = Image.fromarray(small) pil = pil.resize((target_w, target_h), Image.NEAREST) return np.array(pil) def resize_for_display(img: Image.Image, max_display: int = 512) -> Image.Image: """Resize an input image to the size used by the frontend panels.""" img_rgb = img.convert("RGB") w, h = img_rgb.size scale = min(max_display / w, max_display / h, 1.0) disp_w, disp_h = int(w * scale), int(h * scale) return img_rgb.resize((disp_w, disp_h), Image.BILINEAR) def decode_image_b64(image_b64: str) -> tuple[bytes, Image.Image]: """Decode either a raw base64 payload or a data URL.""" if "," in image_b64 and image_b64.strip().startswith("data:"): image_b64 = image_b64.split(",", 1)[1] img_data = base64.b64decode(image_b64) return img_data, Image.open(io.BytesIO(img_data)) def image_cache_key(img_data: bytes) -> str: return hashlib.sha256(img_data).hexdigest()[:20] def remember_embeddings(cache_key: str, patch_emb: np.ndarray) -> None: EMBEDDING_CACHE[cache_key] = patch_emb if cache_key in EMBEDDING_CACHE_ORDER: EMBEDDING_CACHE_ORDER.remove(cache_key) EMBEDDING_CACHE_ORDER.append(cache_key) while len(EMBEDDING_CACHE_ORDER) > MAX_EMBED_CACHE: stale_key = EMBEDDING_CACHE_ORDER.pop(0) EMBEDDING_CACHE.pop(stale_key, None) def img_to_b64(img: Image.Image) -> str: buf = io.BytesIO() img.save(buf, format="PNG") return base64.b64encode(buf.getvalue()).decode() def numpy_to_b64(arr: np.ndarray) -> str: img = Image.fromarray(arr) return img_to_b64(img) # --------------------------------------------------------------------------- # Real satellite imagery examples (pre-computed, cached) # --------------------------------------------------------------------------- EXAMPLES_DIR = os.path.join(os.path.dirname(__file__), "examples") def _load_examples(): """Load real satellite examples with pre-computed model outputs from examples/ dir.""" manifest_path = os.path.join(EXAMPLES_DIR, "manifest.json") with open(manifest_path) as f: manifest = json.load(f) examples = [] for entry in manifest: img_path = os.path.join(EXAMPLES_DIR, os.path.basename(entry["filename"])) pca_path = os.path.join(EXAMPLES_DIR, os.path.basename(entry["pca_file"])) heat_path = os.path.join(EXAMPLES_DIR, os.path.basename(entry["heatmap_file"])) img = Image.open(img_path) pca_img = Image.open(pca_path) heat_img = Image.open(heat_path) examples.append({ "id": entry["id"], "label": entry["label"], "emoji": entry["emoji"], "description": entry["description"], "source": entry["source"], "image": img_to_b64(img), "pca_rgb": img_to_b64(pca_img), "heatmap": img_to_b64(heat_img), "stats": { "n_patches": entry["n_patches"], "patch_grid": entry["patch_grid"], "embed_dim": EMBED_DIM, "mean_norm": entry["mean_norm"], "inference_time": 0.0, }, "patch_emb_shape": [56, 56, EMBED_DIM], }) logger.info(f"Loaded cached example: {entry['id']} - {entry['label']}") return examples CACHED_EXAMPLES = _load_examples() # --------------------------------------------------------------------------- # API endpoints # --------------------------------------------------------------------------- app = gr.Server() @app.api(name="process_image", concurrency_limit=2, time_limit=120) def process_image(image_b64: str) -> str: """Process a base64-encoded image and return visualization data as JSON string. Returns JSON with: input_image, pca_rgb, heatmap, embedding_stats, n_patches. All images are base64-encoded PNGs. """ try: img_data, img = decode_image_b64(image_b64) except Exception as e: return json.dumps({"error": f"Failed to decode image: {e}"}) t0 = time.perf_counter() patch_emb, ph, pw = extract_embeddings(img) inference_time = time.perf_counter() - t0 cache_key = image_cache_key(img_data) remember_embeddings(cache_key, patch_emb) # Generate visualizations pca_rgb = embeddings_to_rgb_pca(patch_emb) heatmap = embeddings_to_heatmap(patch_emb) # Resize model outputs to match input image size img_rgb = resize_for_display(img) disp_w, disp_h = img_rgb.size pca_up = upsample_to_image(pca_rgb, disp_h, disp_w) heat_up = upsample_to_image(heatmap, disp_h, disp_w) # Compute embedding stats flat = patch_emb.reshape(-1, EMBED_DIM) stats = { "n_patches": int(ph * pw), "patch_grid": f"{ph} × {pw}", "embed_dim": EMBED_DIM, "mean_norm": float(np.linalg.norm(flat, axis=1).mean()), "inference_time": round(inference_time, 3), } result = { "input_image": numpy_to_b64(np.array(img_rgb)), "pca_rgb": numpy_to_b64(pca_up), "heatmap": numpy_to_b64(heat_up), "stats": stats, "patch_emb_shape": [ph, pw, EMBED_DIM], "cache_key": cache_key, } return json.dumps(result) @app.api(name="query_similarity", concurrency_limit=2, time_limit=120) def query_similarity(image_b64: str, x_norm: float, y_norm: float, cache_key: str = "") -> str: """Return a similarity heatmap using the clicked patch as the query.""" try: img_data, img = decode_image_b64(image_b64) except Exception as e: return json.dumps({"error": f"Failed to decode image: {e}"}) cache_key = cache_key or image_cache_key(img_data) t0 = time.perf_counter() patch_emb = EMBEDDING_CACHE.get(cache_key) cache_hit = patch_emb is not None if patch_emb is None: patch_emb, _, _ = extract_embeddings(img) remember_embeddings(cache_key, patch_emb) elapsed = time.perf_counter() - t0 heatmap, query = embeddings_to_query_heatmap(patch_emb, x_norm, y_norm) img_rgb = resize_for_display(img) disp_w, disp_h = img_rgb.size heat_up = upsample_to_image(heatmap, disp_h, disp_w) flat = patch_emb.reshape(-1, EMBED_DIM) stats = { "n_patches": int(patch_emb.shape[0] * patch_emb.shape[1]), "patch_grid": f"{patch_emb.shape[0]} × {patch_emb.shape[1]}", "embed_dim": EMBED_DIM, "mean_norm": float(np.linalg.norm(flat, axis=1).mean()), "inference_time": round(elapsed, 3), } return json.dumps({ "heatmap": numpy_to_b64(heat_up), "stats": stats, "query": query, "cache_key": cache_key, "cache_hit": cache_hit, }) @app.api(name="get_samples", concurrency_limit=1, time_limit=30) def get_samples() -> str: """Return all cached sample images with pre-computed model outputs.""" return json.dumps(CACHED_EXAMPLES) @app.api(name="get_model_info", concurrency_limit=1, time_limit=10) def get_model_info() -> str: """Return model information.""" info = { "model_name": "OlmoEarth-v1-Base", "model_id": "allenai/OlmoEarth-v1-Base", "architecture": "ViT-Base (Flexible ViT)", "parameters": f"{num_params / 1e6:.1f}M", "embedding_dim": EMBED_DIM, "encoder_depth": 12, "encoder_heads": 12, "patch_size": PATCH_SIZE, "modalities": [ "Sentinel-1", "Sentinel-2", "Landsat-8", "WorldCover", "OpenStreetMap", "SRTM", "Canopy Height", ], "description": "A spatio-temporal, multimodal Earth observation foundation model that turns aligned satellite observations into reusable patch embeddings.", "paper_url": PAPER_URL, "huggingface_url": MODEL_URL, "github_url": GITHUB_URL, "paper_context": { "earth_observation_foundation_models": ( "Earth observation foundation models are reusable encoders for satellite and geospatial data. " "They learn patch-level representations that can transfer to land-cover mapping, crop and ecosystem classification, segmentation, and detection with limited labels." ), "why_special": [ "Multimodal inputs: Sentinel-1 radar, Sentinel-2 optical, Landsat-8, and map layers including WorldCover, OpenStreetMap, SRTM elevation, and canopy height.", "Multi-temporal context: each pretraining sample covers a 2.56 km by 2.56 km region with up to 12 monthly timesteps over one year.", "Latent MIM Lite: a stable latent masked-image-modeling objective that predicts frozen random-projection targets instead of pixels.", "Modality-aware masking: the model learns to reconstruct missing bandsets or modalities rather than relying on easy nearby tokens.", "Strong transfer: the paper reports best performance on 15 of 24 frozen embedding tasks and 19 of 29 fine-tuning tasks against 12 other EO foundation models.", ], "pretraining_dataset": "285,288 globally sampled locations selected from OpenStreetMap categories, with observations resampled to 10 m per pixel.", }, } return json.dumps(info) # --------------------------------------------------------------------------- # Custom HTML frontend # --------------------------------------------------------------------------- FRONTEND_HTML = r"""
Satellite Vision Explorer