terraflow.climate

The climate module provides spatial interpolation and index-based matching for aligning climate observations to raster cells.

Overview

New in v0.2.0: Replaces global mean climate approach with per-cell interpolation using:

• Spatial interpolation: scipy.interpolate.griddata with linear and nearest-neighbor methods
• Index-based matching: Row order or explicit cell ID matching
• Graceful fallbacks: Global mean values for cells outside interpolation range or with sparse data

Quick Example

import pandas as pd
from terraflow.climate import ClimateInterpolator

# Load climate data
climate_df = pd.read_csv("weather_stations.csv")

# Create spatial interpolator
interpolator = ClimateInterpolator(
    climate_data=climate_df,
    strategy="spatial",
    fallback_to_mean=True
)

# Interpolate values for raster cell locations
cell_coords = [(39.14, -100.82), (38.55, -99.20)]
interpolated = interpolator.interpolate(cell_coords)

Use Cases

• Spatial: Weather station networks, satellite gridded data
• Index: Pre-processed per-cell climate datasets

API Reference

climate

Climate data handling with spatial interpolation and matching strategies.

This module provides tools for aligning climate data with raster cells using either index-based lookup or spatial interpolation. It supports graceful fallbacks when climate data is incomplete or sparse.

All climate data validation is performed using pydantic models for type safety, coordinate validation, and consistency checking.

Example: Load and interpolate climate data for raster cells:

```python
import pandas as pd
from terraflow.climate import ClimateInterpolator

# Load climate data with lat/lon coordinates
climate_df = pd.read_csv("climate.csv")

# Create cell coordinates (lat/lon arrays)
cell_lats = [40.5, 40.6, 40.7]
cell_lons = [-74.5, -74.4, -74.3]

# Initialize interpolator with spatial strategy
interpolator = ClimateInterpolator(
    climate_df=climate_df,
    strategy="spatial",
    fallback_to_mean=True
)

# Interpolate climate to cell locations
cell_climate = interpolator.interpolate(cell_lats, cell_lons)
# Returns DataFrame with interpolated temp and rainfall per cell
```

ClimateInterpolator(climate_df, strategy='spatial', cell_id_column=None, fallback_to_mean=True)

Spatially interpolate or index-match climate data to raster cells.

Supports two strategies for aligning climate observations to raster cells: - "spatial": Interpolate climate values using geographic coordinates - "index": Match cells to climate records by row index or cell ID

All configuration uses pydantic models for type safety and validation.

Attributes: climate_df (pd.DataFrame): Climate data with columns like temperature, rainfall. strategy (str): Matching strategy - "spatial" or "index" (validated via pydantic). cell_id_column (str, optional): Column name for cell ID matching (for "index" strategy). fallback_to_mean (bool): If True, use global mean for cells outside interpolation range. _climate_mean (dict): Cached mean values of climate variables. climate_columns (list): List of climate variable columns (non-coordinate columns).

Initialize climate interpolator with pydantic-validated configuration.

Args: climate_df: DataFrame with climate data. Must have 'lat' and 'lon' columns. strategy: "spatial" for interpolation, "index" for direct row matching. cell_id_column: Column name mapping cells to climate records (for "index" strategy). fallback_to_mean: If True, use global mean for missing/out-of-range values.

Raises: ValueError: If strategy is invalid, required columns missing, or invalid coordinates.

Source code in terraflow/climate.py

def __init__(
    self,
    climate_df: pd.DataFrame,
    strategy: Literal["spatial", "index"] = "spatial",
    cell_id_column: Optional[str] = None,
    fallback_to_mean: bool = True,
):
    """Initialize climate interpolator with pydantic-validated configuration.

    Args:
        climate_df: DataFrame with climate data. Must have 'lat' and 'lon' columns.
        strategy: "spatial" for interpolation, "index" for direct row matching.
        cell_id_column: Column name mapping cells to climate records (for "index" strategy).
        fallback_to_mean: If True, use global mean for missing/out-of-range values.

    Raises:
        ValueError: If strategy is invalid, required columns missing, or invalid coordinates.
    """
    # Validate strategy using pydantic-like pattern
    if strategy not in ("spatial", "index"):
        raise ValueError(f"strategy must be 'spatial' or 'index', got '{strategy}'")

    self.climate_df = climate_df.copy()
    self.strategy = strategy
    self.cell_id_column = cell_id_column
    self.fallback_to_mean = fallback_to_mean

    # Validate required columns
    self._validate_columns()

    # Validate and prepare coordinates
    if self.strategy == "spatial":
        self._validate_spatial_data()

    # Cache mean climate values for fallback
    self._climate_mean = self._compute_mean_climate()

    logger.info(
        f"ClimateInterpolator initialized with strategy='{strategy}', "
        f"{len(climate_df)} records, climate_columns={self.climate_columns}"
    )

interpolate(cell_lats, cell_lons)

Interpolate or match climate data to cell locations.

Args: cell_lats: Array of cell latitudes. cell_lons: Array of cell longitudes.

Returns: DataFrame with one row per cell, columns for each climate variable.

Raises: ValueError: If strategy is "index" and cell/climate counts don't match.

Source code in terraflow/climate.py

def interpolate(self, cell_lats: np.ndarray, cell_lons: np.ndarray) -> pd.DataFrame:
    """Interpolate or match climate data to cell locations.

    Args:
        cell_lats: Array of cell latitudes.
        cell_lons: Array of cell longitudes.

    Returns:
        DataFrame with one row per cell, columns for each climate variable.

    Raises:
        ValueError: If strategy is "index" and cell/climate counts don't match.
    """
    cell_lats = np.asarray(cell_lats)
    cell_lons = np.asarray(cell_lons)

    if len(cell_lats) != len(cell_lons):
        raise ValueError(
            f"cell_lats and cell_lons must have same length. "
            f"Got {len(cell_lats)} and {len(cell_lons)}"
        )

    if self.strategy == "spatial":
        return self._interpolate_spatial(cell_lats, cell_lons)
    else:  # "index"
        return self._match_by_index(cell_lats, cell_lons)

CoordinateRange

Bases: BaseModel

Validated geographic coordinate pair.

Ensures latitude is in [-90, 90] and longitude in [-180, 180]. Pydantic provides automatic validation and type coercion.

validate_latitude(v) classmethod

Ensure latitude is in valid range [-90, 90].

Source code in terraflow/climate.py

@field_validator("latitude")
@classmethod
def validate_latitude(cls, v: float) -> float:
    """Ensure latitude is in valid range [-90, 90]."""
    if not isinstance(v, (int, float)):
        raise ValueError(f"Latitude must be numeric, got {type(v).__name__}")
    if v < -90 or v > 90:
        raise ValueError(f"Latitude must be in [-90, 90], got {v}")
    return float(v)

validate_longitude(v) classmethod

Ensure longitude is in valid range [-180, 180].

Source code in terraflow/climate.py

@field_validator("longitude")
@classmethod
def validate_longitude(cls, v: float) -> float:
    """Ensure longitude is in valid range [-180, 180]."""
    if not isinstance(v, (int, float)):
        raise ValueError(f"Longitude must be numeric, got {type(v).__name__}")
    if v < -180 or v > 180:
        raise ValueError(f"Longitude must be in [-180, 180], got {v}")
    return float(v)