disease¶

Climate suitability for disease outbreaks (e.g. malaria). Entry point: DiseaseRiskUseCase.

from climate_change import DiseaseRiskUseCase

Use case¶

use_case ¶

DiseaseRiskUseCase ¶

DiseaseRiskUseCase(dask_engine: DaskEngine)

Bases: BaseUseCase

Entry point for the climate-driven disease surveillance domain.

Minimal config (all optional — defaults match Kisumu County, Kenya 2021–2023):

{ "aoi_geojson": {"type": "Polygon", "coordinates": [...]}, "gee_project": "your-gee-project-id", "start_date": "2021-01-01", "end_date": "2023-12-31", "model_type": "gbm", # "gbm" | "xgboost" | "ensemble" "n_pixels": 3000, "scale": 1000, # metres — MODIS / CHIRPS native "output_dir": "outputs", "prefix": "disease", }

Source code in disease/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 disease/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 disease/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 disease/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", "gbm")
    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
    raster_result: DiseaseCogResult | None = None
    try:
        raster_result = export_disease_cog(
            gbm_model=features["_gbm"],
            xgb_model=features["_xgb"],
            scaler=features["_scaler"],
            datasets=features["datasets"],
            output_dir=dict_config.get("output_dir", "outputs"),
            prefix=dict_config.get("prefix", "disease"),
            model_type=model_type,
            aoi_geojson=config.aoi_geojson,
        )
        raster_paths = {"disease_risk": raster_result["disease_risk"]}
        _apply_raster_risk_pct(result, raster_result["risk_pct"])
    except Exception as exc:
        raster_error = str(exc)

    total_area_ha = _aoi_area_ha(config.aoi_geojson)
    if total_area_ha:
        result["stats"]["total_area_ha"] = round(total_area_ha, 1)
        _attach_risk_area(result.get("charts", {}).get("riskDist"), total_area_ha)
    _attach_population_stats(result, raster_result)

    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="disease",
        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 disease/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
    raster_result: DiseaseCogResult | None = None
    try:
        raster_result = export_disease_cog(
            gbm_model=features["_gbm"],
            xgb_model=features["_xgb"],
            scaler=features["_scaler"],
            datasets=features["datasets"],
            output_dir=config.get("output_dir", "outputs"),
            prefix=config.get("prefix", "disease"),
            model_type=config.get("model_type", "gbm"),
            aoi_geojson=config.get("aoi_geojson"),
        )
        cog_paths = {"disease_risk": raster_result["disease_risk"]}
        _apply_raster_risk_pct(result, raster_result["risk_pct"])
    except Exception as exc:
        _log.warning("Disease 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

    total_area_ha = _aoi_area_ha(config.get("aoi_geojson"))
    if total_area_ha:
        result["stats"]["total_area_ha"] = round(total_area_ha, 1)
        _attach_risk_area(result.get("charts", {}).get("riskDist"), total_area_ha)
    _attach_population_stats(result, raster_result)

    result["stats"].update(
        {
            "bbox": features["bbox"],
            "prefix": config.get("prefix", "disease"),
        }
    )
    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 disease/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 disease/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¶