""" load.py Module for loading ensemble models (STAC compatible) and performing optimized inference on large geospatial imagery using dynamic batching and Gaussian blending. """ import math import pathlib import itertools from typing import Literal, Tuple, List import torch import torch.nn import numpy as np import pystac from torch.utils.data import Dataset, DataLoader from tqdm import tqdm # ============================================================================== # 1. HELPER CLASSES & FUNCTIONS # ============================================================================== class EnsembleModel(torch.nn.Module): """ Runtime ensemble model for combining multiple model outputs. Used when loading multiple separate .pt2 files. """ def __init__(self, *models, mode="max"): super(EnsembleModel, self).__init__() self.models = torch.nn.ModuleList(models) self.mode = mode if mode not in ["min", "mean", "median", "max", "none"]: raise ValueError("Mode must be 'none', 'min', 'mean', 'median', or 'max'.") def forward(self, x) -> Tuple[torch.Tensor, torch.Tensor]: """ Returns: - probabilities: (B, 1, H, W) - uncertainty: (B, 1, H, W) (normalized std dev) """ outputs = [model(x) for model in self.models] if not outputs: return None, None # Stack: (B, N, H, W) stacked = torch.stack(outputs, dim=1).squeeze(2) # Aggregation if self.mode == "max": probs = torch.max(stacked, dim=1, keepdim=True)[0] elif self.mode == "mean": probs = torch.mean(stacked, dim=1, keepdim=True) elif self.mode == "median": probs = torch.median(stacked, dim=1, keepdim=True)[0] elif self.mode == "min": probs = torch.min(stacked, dim=1, keepdim=True)[0] elif self.mode == "none": return stacked, None # Uncertainty N = len(outputs) if N > 1: std = torch.std(stacked, dim=1, keepdim=True) std_max = math.sqrt(0.25 * N / (N - 1)) uncertainty = torch.clamp(std / std_max, 0.0, 1.0) else: uncertainty = torch.zeros_like(probs) return probs, uncertainty def get_spline_window(window_size: int, power: int = 2) -> np.ndarray: """Generates a 2D Hann window for smoothing tile edges.""" intersection = np.hanning(window_size) window_2d = np.outer(intersection, intersection) return (window_2d ** power).astype(np.float32) def fix_lastchunk(iterchunks, s2dim, chunk_size): """Adjusts the last chunks to fit within image boundaries.""" itercontainer = [] for index_i, index_j in iterchunks: if index_i + chunk_size > s2dim[0]: index_i = max(s2dim[0] - chunk_size, 0) if index_j + chunk_size > s2dim[1]: index_j = max(s2dim[1] - chunk_size, 0) itercontainer.append((index_i, index_j)) return list(set(itercontainer)) def define_iteration(dimension: tuple, chunk_size: int, overlap: int = 0): """Generates top-left coordinates for sliding window inference.""" dimy, dimx = dimension if chunk_size > max(dimx, dimy): return [(0, 0)] y_step = chunk_size - overlap x_step = chunk_size - overlap iterchunks = list(itertools.product( range(0, dimy, y_step), range(0, dimx, x_step) )) return fix_lastchunk(iterchunks, dimension, chunk_size) # ============================================================================== # 2. DATASET FOR PARALLEL LOADING # ============================================================================== class PatchDataset(Dataset): """ Dataset wrapper to handle image slicing and padding on CPU workers. """ def __init__(self, image: np.ndarray, coords: List[Tuple[int, int]], chunk_size: int, nodata: float = 0): self.image = image self.coords = coords self.chunk_size = chunk_size self.nodata = nodata def __len__(self): return len(self.coords) def __getitem__(self, idx): row_off, col_off = self.coords[idx] # Read patch patch = self.image[:, row_off : row_off + self.chunk_size, col_off : col_off + self.chunk_size] c, h, w = patch.shape patch_tensor = torch.from_numpy(patch).float() # Apply padding if patch is smaller than chunk_size (edges) pad_h = self.chunk_size - h pad_w = self.chunk_size - w if pad_h > 0 or pad_w > 0: patch_tensor = torch.nn.functional.pad(patch_tensor, (0, pad_w, 0, pad_h), "constant", self.nodata) # Identify nodata pixels mask_nodata = (patch_tensor == self.nodata).all(dim=0) return patch_tensor, row_off, col_off, h, w, mask_nodata # ============================================================================== # 3. LOADING & INFERENCE LOGIC # ============================================================================== def compiled_model( path: pathlib.Path, stac_item: pystac.Item, mode: Literal["min", "mean", "median", "max"] = "max", *args, **kwargs ): """ Loads .pt2 model(s). Returns a single model or an EnsembleModel. Automatically unwraps ExportedProgram if possible. """ model_paths = sorted([ asset.href for key, asset in stac_item.assets.items() if asset.href.endswith(".pt2") ]) if not model_paths: raise ValueError("No .pt2 files found in STAC item assets.") # Helper to load and unwrap def load_pt2(p): program = torch.export.load(p) return program.module() if hasattr(program, "module") else program if len(model_paths) == 1: return load_pt2(model_paths[0]) else: models = [load_pt2(p) for p in model_paths] return EnsembleModel(*models, mode=mode) def predict_large( image: np.ndarray, model: torch.nn.Module, chunk_size: int = 512, overlap: int = 128, batch_size: int = 16, num_workers: int = 8, # Recommended: 8-16 device: str = "cuda", nodata: float = 0.0 ) -> Tuple[np.ndarray, np.ndarray] | np.ndarray: """ Optimized inference for large images using Dynamic Batching and Gaussian Blending. """ if image.ndim != 3: raise ValueError(f"Input image must be (C, H, W). Received {image.shape}") # --- 1. Robust Model Unwrapping --- # Fix for torch.export.load() returning an ExportedProgram container if hasattr(model, "module") and callable(model.module): try: unpacked = model.module() if isinstance(unpacked, torch.nn.Module): model = unpacked except Exception: pass # --- 2. Setup Model --- try: model.eval() for p in model.parameters(): p.requires_grad = False except: pass # Only move to device if it's a standard Module (ExportedProgram handles device internally or via input) if isinstance(model, torch.nn.Module): model = model.to(device) bands, height, width = image.shape # --- 3. Check Signature (Ensemble vs Single) --- # Dummy pass (batch=2 to respect dynamic shapes) dummy = torch.randn(2, bands, chunk_size, chunk_size).to(device) with torch.no_grad(): out = model(dummy) is_ensemble = isinstance(out, tuple) and len(out) == 2 # --- 4. Initialize Buffers (Accumulators) --- out_probs = np.zeros((1, height, width), dtype=np.float32) count_map = np.zeros((1, height, width), dtype=np.float32) out_uncert = np.zeros((1, height, width), dtype=np.float32) if is_ensemble else None # --- 5. Prepare Spline Window --- window_spline = get_spline_window(chunk_size, power=2) window_tensor = torch.from_numpy(window_spline).to(device) # --- 6. DataLoader Setup --- coords = define_iteration((height, width), chunk_size, overlap) dataset = PatchDataset(image, coords, chunk_size, nodata) loader = DataLoader( dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers, prefetch_factor=2, pin_memory=True ) # --- 7. Inference Loop --- for batch in tqdm(loader, desc=f"Inference (Batch {batch_size})"): patches, r_offs, c_offs, h_actuals, w_actuals, nodata_masks = batch # Move inputs to GPU patches = patches.to(device, non_blocking=True) nodata_masks = nodata_masks.to(device, non_blocking=True) # (B, H, W) # Forward Pass with torch.no_grad(): if is_ensemble: probs, uncert = model(patches) else: probs = model(patches) uncert = None # Ensure correct dimensions (B, C, H, W) if probs.ndim == 3: probs = probs.unsqueeze(1) if is_ensemble and uncert.ndim == 3: uncert = uncert.unsqueeze(1) # Prepare weights for batch B = patches.size(0) batch_weights = window_tensor.unsqueeze(0).unsqueeze(0).repeat(B, 1, 1, 1) # Zero out weights where input was nodata batch_weights[nodata_masks.unsqueeze(1)] = 0.0 # Apply weights probs_weighted = probs * batch_weights if is_ensemble: uncert_weighted = uncert * batch_weights # Move to CPU probs_cpu = probs_weighted.cpu().numpy() weights_cpu = batch_weights.cpu().numpy() if is_ensemble: uncert_cpu = uncert_weighted.cpu().numpy() # Accumulate in global map for i in range(B): r, c = r_offs[i].item(), c_offs[i].item() h, w = h_actuals[i].item(), w_actuals[i].item() # Slice valid regions valid_probs = probs_cpu[i, :, :h, :w] valid_weights = weights_cpu[i, :, :h, :w] out_probs[:, r:r+h, c:c+w] += valid_probs count_map[:, r:r+h, c:c+w] += valid_weights if is_ensemble: valid_uncert = uncert_cpu[i, :, :h, :w] out_uncert[:, r:r+h, c:c+w] += valid_uncert # --- 8. Normalization --- mask_zero = (count_map == 0) count_map[mask_zero] = 1.0 # Prevent div/0 out_probs /= count_map out_probs[mask_zero] = nodata if is_ensemble: out_uncert /= count_map out_uncert[mask_zero] = nodata return out_probs, out_uncert return out_probs