food_security¶

Vegetation and climate stress assessment for food-security risk. Entry point: FoodSecurityUseCase.

from climate_change import FoodSecurityUseCase

Use case¶

use_case ¶

FoodSecurityUseCase ¶

FoodSecurityUseCase(dask_engine: DaskEngine)

Bases: BaseUseCase

Entry point for the food security assessment domain.

Minimal config (all optional — defaults match Marsabit County, Kenya 2018–2023):

{ "aoi_geojson": {"type": "Polygon", "coordinates": [...]}, "gee_project": "your-gee-project-id", "start_date": "2018-01-01", "end_date": "2023-12-31", "lt_baseline_start": "2001-01-01", # long-term baseline for VCI/TCI "s2_start": "2020-01-01", # Sentinel-2 MNDWI start "model_type": "rf", # "rf" | "xgboost" | "ensemble" "n_pixels": 3000, "scale": 1000, # metres — MODIS / CHIRPS native "output_dir": "outputs", "prefix": "food_security", }

Source code in food_security/use_case.py

def __init__(self, dask_engine: DaskEngine) -> None:
    super().__init__(dask_engine)

fetch_data ¶

fetch_data(config: AnalysisConfig) -> dict

Authenticate GEE and parse the AOI geometry (see _fetch_from_dict).

Source code in food_security/use_case.py

def fetch_data(self, config: AnalysisConfig) -> dict:
    """Authenticate GEE and parse the AOI geometry (see `_fetch_from_dict`)."""
    flat: dict = {
        "aoi_geojson": config.aoi_geojson,
        "start_date": config.start_date,
        "end_date": config.end_date,
        "country": config.country,
        **config.extra_params,
    }
    return self._fetch_from_dict(flat)

preprocess ¶

preprocess(raw_data: dict, config: AnalysisConfig) -> dict

Build the feature stack and sample training pixels (see _preprocess_raw).

Source code in food_security/use_case.py

def preprocess(self, raw_data: dict, config: AnalysisConfig) -> dict:
    """Build the feature stack and sample training pixels (see `_preprocess_raw`)."""
    return self._preprocess_raw(raw_data)

run_model ¶

run_model(features: dict, config: AnalysisConfig) -> AnalysisOutput

Train models, export the risk COG, and package an AnalysisOutput.

Source code in food_security/use_case.py

def run_model(self, features: dict, config: AnalysisConfig) -> AnalysisOutput:
    """Train models, export the risk COG, and package an `AnalysisOutput`."""
    model_type = config.extra_params.get("model_type", "rf")
    dict_config = {"model_type": model_type, **config.extra_params}
    result = self._run_model_dict(features, dict_config)
    total_area_ha = _aoi_area_ha(config.aoi_geojson)
    spatial_grid = predict_food_security_grid(
        rf_model=features["_rf"],
        xgb_model=features["_xgb"],
        scaler=features["_scaler"],
        datasets=features["datasets"],
        model_type=model_type,
        aoi_geojson=config.aoi_geojson,
    )
    self._apply_spatial_risk_summary(result, spatial_grid, total_area_ha)
    if total_area_ha:
        result["stats"]["total_area_ha"] = round(total_area_ha, 1)
        _attach_risk_area(result.get("charts", {}), total_area_ha)
    raster_paths: dict[str, str] | None = None
    raster_error: str | None = None
    try:
        raster_paths = export_food_security_cog(
            rf_model=features["_rf"],
            xgb_model=features["_xgb"],
            scaler=features["_scaler"],
            datasets=features["datasets"],
            output_dir=dict_config.get("output_dir", "outputs"),
            prefix=dict_config.get("prefix", "food_security"),
            model_type=model_type,
            aoi_geojson=config.aoi_geojson,
        )
    except Exception as exc:
        raster_error = str(exc)

    geojson_features = self._risk_grid_to_geojson(spatial_grid)
    for p in result.pop("_sample_points", []):
        geojson_features.append(
            {
                "type": "Feature",
                "geometry": {"type": "Point", "coordinates": [p["lon"], p["lat"]]},
                "properties": {"risk_class": p["risk_class"]},
            }
        )
    metadata = {
        "model": model_type,
        "country": config.country,
        "start_date": config.start_date,
        "end_date": config.end_date,
        "raster": raster_paths or {},
        "spatial_resolution_m": dict_config.get("scale"),
        "n_pixels_sampled": dict_config.get("n_pixels"),
    }
    if raster_error:
        metadata["raster_error"] = raster_error

    return AnalysisOutput(
        module="food_security",
        geojson={"type": "FeatureCollection", "features": geojson_features},
        raster_path=raster_paths,
        stats={**result["stats"], "country": config.country},
        shap=result.get("charts", {}).get("shap"),
        charts=result.get("charts", {}),
        metadata=metadata,
    )

run ¶

run(config: dict) -> dict

Single AOI analysis — full pipeline in one call.

Source code in food_security/use_case.py

def run(self, config: dict) -> dict:
    """Single AOI analysis — full pipeline in one call."""
    raw_data = self._fetch_from_dict(config)
    features = self._preprocess_raw(raw_data)
    result = self._run_model_dict(features, config)
    total_area_ha = _aoi_area_ha(config.get("aoi_geojson"))
    spatial_grid = predict_food_security_grid(
        rf_model=features["_rf"],
        xgb_model=features["_xgb"],
        scaler=features["_scaler"],
        datasets=features["datasets"],
        model_type=config.get("model_type", "rf"),
        aoi_geojson=config.get("aoi_geojson"),
    )
    self._apply_spatial_risk_summary(result, spatial_grid, total_area_ha)
    if total_area_ha:
        result["stats"]["total_area_ha"] = round(total_area_ha, 1)
        _attach_risk_area(result.get("charts", {}), total_area_ha)

    cog_paths: dict[str, str] | None = None
    try:
        cog_paths = export_food_security_cog(
            rf_model=features["_rf"],
            xgb_model=features["_xgb"],
            scaler=features["_scaler"],
            datasets=features["datasets"],
            output_dir=config.get("output_dir", "outputs"),
            prefix=config.get("prefix", "food_security"),
            model_type=config.get("model_type", "rf"),
            aoi_geojson=config.get("aoi_geojson"),
        )
    except Exception as exc:
        result["raster_error"] = str(exc)
    result["raster"] = cog_paths or {}
    result["geojson"] = {
        "type": "FeatureCollection",
        "features": self._risk_grid_to_geojson(spatial_grid),
    }
    result["stats"].update(
        {
            "bbox": features["bbox"],
            "prefix": config.get("prefix", "food_security"),
        }
    )
    return result

run_date_ranges ¶

run_date_ranges(config: dict, date_ranges: list[dict]) -> list[dict]

Run the same AOI over multiple date-range configurations in parallel.

Source code in food_security/use_case.py

def run_date_ranges(self, config: dict, date_ranges: list[dict]) -> list[dict]:
    """Run the same AOI over multiple date-range configurations in parallel."""
    merged = [{**config, **dr} for dr in date_ranges]
    return self._run_parallel(merged)

run_multi_regions ¶

run_multi_regions(configs: list[dict]) -> list[dict]

Run multiple AOI configs (same timeframe, different regions) in parallel.

Source code in food_security/use_case.py

def run_multi_regions(self, configs: list[dict]) -> list[dict]:
    """Run multiple AOI configs (same timeframe, different regions) in parallel."""
    return self._run_parallel(configs)

Features¶

features ¶

fetch_vci_tci ¶

fetch_vci_tci(aoi: Geometry, lt_start: str, study_start: str, study_end: str, scale: int = 1000) -> xr.Dataset

Compute VCI and TCI from MODIS NDVI and LST relative to the long-term baseline.

VCI = (NDVI_current − NDVI_min) / (NDVI_max − NDVI_min) × 100 TCI = (LST_max − LST_current) / (LST_max − LST_min) × 100

Returns Dataset with variables 'vci' and 'tci'.