land_degradation¶

Vegetation decline and rangeland degradation detection. Entry point: LandDegradationUseCase.

from climate_change import LandDegradationUseCase

Use case¶

use_case ¶

LandDegradationUseCase ¶

LandDegradationUseCase(dask_engine: DaskEngine)

Bases: BaseUseCase

Entry point for the land degradation domain.

Minimal config (all optional — defaults match Northern Burkina Faso 2015–2024):

{ "aoi_geojson": {"type": "Polygon", "coordinates": [...]}, "gee_project": "your-gee-project-id", "start_date": "2015-01-01", "end_date": "2024-12-31", "model_type": "lgbm", # "rf" | "lgbm" | "ensemble" "n_pixels": 3000, "scale": 1000, # metres — uniform for all feature rasters "output_dir": "outputs", "prefix": "land_degradation", }

Source code in land_degradation/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 land_degradation/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 land_degradation/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 degradation COG, and package an AnalysisOutput.

Source code in land_degradation/use_case.py

def run_model(self, features: dict, config: AnalysisConfig) -> AnalysisOutput:
    """Train models, export the degradation COG, and package an `AnalysisOutput`."""
    model_type = config.extra_params.get("model_type", "lgbm")
    dict_config = {"model_type": model_type, **config.extra_params}
    result = self._run_model_dict(features, dict_config)

    raster_paths: dict[str, str] | None = None
    raster_error: str | None = None
    try:
        raster_result = export_degradation_cog(
            rf_model=features["_rf"],
            lgbm_model=features["_lgbm"],
            scaler=features["_scaler"],
            datasets=features["datasets"],
            output_dir=dict_config.get("output_dir", "outputs"),
            prefix=dict_config.get("prefix", "land_degradation"),
            model_type=model_type,
            aoi_geojson=config.aoi_geojson,
            scale=int(dict_config.get("scale", 1000)),
        )
        raster_paths = {"degradation_risk": raster_result["degradation_risk"]}
        _apply_raster_risk_pct(
            result,
            raster_result["risk_pct"],
            raster_result["risk_ha"],
            raster_result.get("population_by_class"),
            raster_result.get("total_population"),
        )
    except Exception as exc:
        raster_error = str(exc)

    geojson_features = []
    if config.aoi_geojson:
        geojson_features.append(
            {
                "type": "Feature",
                "geometry": config.aoi_geojson,
                "properties": {"type": "boundary"},
            }
        )
    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 {},
    }
    if raster_error:
        metadata["raster_error"] = raster_error

    return AnalysisOutput(
        module="land_degradation",
        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 land_degradation/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)

    cog_paths: dict[str, str] | None = None
    cog_error: str | None = None
    try:
        raster_result = export_degradation_cog(
            rf_model=features["_rf"],
            lgbm_model=features["_lgbm"],
            scaler=features["_scaler"],
            datasets=features["datasets"],
            output_dir=config.get("output_dir", "outputs"),
            prefix=config.get("prefix", "land_degradation"),
            model_type=config.get("model_type", "lgbm"),
            aoi_geojson=config.get("aoi_geojson"),
            scale=int(config.get("scale", 1000)),
        )
        cog_paths = {"degradation_risk": raster_result["degradation_risk"]}
        _apply_raster_risk_pct(
            result,
            raster_result["risk_pct"],
            raster_result["risk_ha"],
            raster_result.get("population_by_class"),
            raster_result.get("total_population"),
        )
    except Exception as exc:
        _log.warning("Land degradation COG export failed: %s", exc, exc_info=True)
        cog_error = str(exc)
    result["raster"] = cog_paths or {}
    if cog_error:
        result["raster_error"] = cog_error
    result["stats"].update(
        {
            "bbox": features["bbox"],
            "prefix": config.get("prefix", "land_degradation"),
        }
    )
    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 land_degradation/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 land_degradation/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_ndvi_stack ¶

fetch_ndvi_stack(aoi: Geometry, start: str, end: str, scale: int = 1000) -> xr.Dataset

Download MODIS MOD13A3 pixel-wise NDVI trend at scale metres. Returns Dataset with variables ndvi_slope, ndvi_mean, ndvi_cv.