core

Orchestration, GEE authentication, caching, and distributed execution shared by every domain module. See Concepts for how these fit together.

Analysis contract & base use case

base_use_case

AnalysisConfig dataclass

AnalysisConfig(module: str, aoi_geojson: dict, start_date: str, end_date: str, country: str, extra_params: dict = dict())

Input contract accepted by every BaseUseCase.execute().

Built by core.runner.run_analysis from the arguments passed to the top-level run_analysis() function; extra_params carries module- and model-specific overrides (e.g. model_type, scale, n_pixels).

AnalysisOutput dataclass

AnalysisOutput(module: str, geojson: dict, raster_path: str | dict[str, str] | None, stats: dict, shap: dict | None, charts: dict, metadata: dict)

Output contract returned by every domain use case.

Fields are intentionally module-agnostic dicts/paths so all five domain modules (drought, flood, food_security, disease, land_degradation) can share one result shape; treat stats and charts as module-specific payloads (see each module's model.py build_*_charts for their shape).

BaseUseCase

BaseUseCase(dask_engine: DaskEngine)

Bases: ABC

Abstract base for all five climate risk use-case modules. Subclasses must implement fetch_data, preprocess, and run_model. execute() orchestrates the full pipeline with caching.

Source code in core/base_use_case.py

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

fetch_data abstractmethod

fetch_data(config: AnalysisConfig) -> dict

Fetch raw EO data from GEE / CHIRPS / ERA5 as xr.Dataset.

Source code in core/base_use_case.py

@abstractmethod
def fetch_data(self, config: AnalysisConfig) -> dict:
    """Fetch raw EO data from GEE / CHIRPS / ERA5 as xr.Dataset."""

preprocess abstractmethod

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

Clip to AOI, compute indices, build Dask feature arrays. Must be decorated with @feature_cache.cache in subclasses. Must return in-memory feature dict — no disk writes.

Source code in core/base_use_case.py

@abstractmethod
def preprocess(self, raw_data: dict, config: AnalysisConfig) -> dict:
    """
    Clip to AOI, compute indices, build Dask feature arrays.
    Must be decorated with @feature_cache.cache in subclasses.
    Must return in-memory feature dict — no disk writes.
    """

run_model abstractmethod

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

Run ML model and return AnalysisOutput with GeoJSON + stats + charts.

Source code in core/base_use_case.py

@abstractmethod
def run_model(self, features: dict, config: AnalysisConfig) -> AnalysisOutput:
    """Run ML model and return AnalysisOutput with GeoJSON + stats + charts."""

explain

explain(features: dict) -> dict

Override in subclasses that support SHAP explanations.

Source code in core/base_use_case.py

def explain(self, features: dict) -> dict:
    """Override in subclasses that support SHAP explanations."""
    return {}

execute async

execute(config: AnalysisConfig) -> AnalysisOutput

Main entry point. Called by core.runner.run_analysis(). Checks in-process cache first; runs full pipeline on miss. Each sync stage runs in a thread pool so the event loop stays free.

Source code in core/base_use_case.py

async def execute(self, config: AnalysisConfig) -> AnalysisOutput:
    """
    Main entry point. Called by core.runner.run_analysis().
    Checks in-process cache first; runs full pipeline on miss.
    Each sync stage runs in a thread pool so the event loop stays free.
    """
    import asyncio

    cache_key = self._cache_key(config)
    cached = analysis_cache.get(cache_key)
    if cached:
        return cached

    raw = await asyncio.to_thread(self.fetch_data, config)
    features = await asyncio.to_thread(self.preprocess, raw, config)
    output = await asyncio.to_thread(self.run_model, features, config)
    explain_payload = await asyncio.to_thread(self.explain, features)
    output.shap = explain_payload or output.shap or output.charts.get("shap")

    analysis_cache.set(cache_key, output, expire=3600)
    return output

Orchestrator

runner

register_module

register_module(name: str, cls: type[BaseUseCase]) -> None

Register a use-case class under the given module name.

Source code in core/runner.py

def register_module(name: str, cls: type[BaseUseCase]) -> None:
    """Register a use-case class under the given module name."""
    MODULE_MAP[name] = cls

run_analysis async

run_analysis(module: str, aoi_geojson: dict, start_date: str, end_date: str, country: str, gee_project: str = '', extra_params: dict | None = None, openai_api_key: str | None = None, report_output_dir: str | None = None, map_png_bytes: bytes | None = None) -> AnalysisOutput

Main public API of the climate_change package. The Celery task calls only this function — no other file in the infrastructure layer imports climate_change directly.

Parameters

gee_project: Google Earth Engine Cloud project ID. Falls back to the GEE_PROJECT environment variable when not supplied. Authentication runs via drought_monitoring.gee.authenticate before any use-case logic executes.

Optional

openai_api_key: if provided, GPT-4o interprets the results and the text is stored in output.metadata["ai_interpretation"]. report_output_dir: if provided, a PDF report is written to this directory and the path stored in output.metadata["report_path"]. map_png_bytes: optional PNG screenshot embedded in the PDF report.

Source code in core/runner.py

async def run_analysis(
    module: str,
    aoi_geojson: dict,
    start_date: str,
    end_date: str,
    country: str,
    gee_project: str = "",
    extra_params: dict | None = None,
    openai_api_key: str | None = None,
    report_output_dir: str | None = None,
    map_png_bytes: bytes | None = None,
) -> AnalysisOutput:
    """
    Main public API of the climate_change package.
    The Celery task calls only this function — no other file in the
    infrastructure layer imports climate_change directly.

    Parameters
    ----------
    gee_project:
        Google Earth Engine Cloud project ID.  Falls back to the
        ``GEE_PROJECT`` environment variable when not supplied.
        Authentication runs via ``drought_monitoring.gee.authenticate``
        before any use-case logic executes.

    Optional:
        openai_api_key: if provided, GPT-4o interprets the results and the
            text is stored in output.metadata["ai_interpretation"].
        report_output_dir: if provided, a PDF report is written to this
            directory and the path stored in output.metadata["report_path"].
        map_png_bytes: optional PNG screenshot embedded in the PDF report.
    """
    _ensure_module_registered(module)
    if module not in MODULE_MAP:
        raise ValueError(f"Unknown module: '{module}'. Available: {sorted(MODULE_MAP)}")

    # ── GEE authentication — happens once per process before any use-case ──────
    project = gee_project or os.environ.get("GEE_PROJECT", "")
    ensure_gee(project)

    # Inject resolved project into extra_params so each use-case run() can
    # access it without re-deriving from the environment.
    params = dict(extra_params or {})
    params.setdefault("gee_project", project)

    from climate_change.core.dask_engine import DaskEngine

    dask_engine = DaskEngine()
    dask_engine.get_client()  # start cluster (idempotent)

    config = AnalysisConfig(
        module=module,
        aoi_geojson=aoi_geojson,
        start_date=start_date,
        end_date=end_date,
        country=country,
        extra_params=params,
    )

    use_case: BaseUseCase = MODULE_MAP[module](dask_engine)
    output = await use_case.execute(config)

    ai_text: str | None = None
    if openai_api_key:
        from climate_change.ai_interpreter.interpreter import AIInterpreter

        try:
            ai_text = AIInterpreter(openai_api_key).interpret(output)
            output.metadata["ai_interpretation"] = ai_text
        except Exception as exc:
            output.metadata["ai_interpretation_error"] = str(exc)

    if report_output_dir:
        from datetime import datetime, timezone
        from pathlib import Path

        from climate_change.reporting.report_builder import ReportBuilder

        ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        report_path = Path(report_output_dir) / f"{module}_{ts}.pdf"
        ReportBuilder(report_path).build(output, ai_text, map_png_bytes)
        output.metadata["report_path"] = str(report_path)

    return output

Google Earth Engine authentication

gee_auth

GEE authentication helpers for climate_change.

Usage contexts

Local / notebook Leave gee_project empty — the project is resolved from the .env file, the GEE_PROJECT environment variable, or an interactive prompt.

Backend / API (recommended pattern) 1. At user registration, call validate_gee_project(project_id) to verify the project is reachable before storing it in the database. 2. On every analysis request, pass the stored project ID through run_analysis(..., gee_project=user.gee_project). 3. Users are never prompted again — authentication is automatic from their stored project ID.

Example FastAPI handler::

    @router.post("/analyse")
    async def analyse(body: AnalysisRequest, user: User = Depends(get_current_user)):
        # gee_project was verified at registration and stored in the DB.
        # Authentication happens automatically — the user is never asked again.
        return await run_analysis(
            module=body.module,
            aoi_geojson=body.aoi,
            start_date=body.start_date,
            end_date=body.end_date,
            country=body.country,
            gee_project=user.gee_project,
        )

Database security guidance

The GEE project ID stored in the database is not a secret key — it is a Google Cloud project identifier. However, it should be protected because:

• If leaked it could be used by someone with valid GEE credentials to bill charges against the project owner's quota.
• Users must never be able to read or impersonate another user's project ID.

Recommended storage practices:

1. Encrypt the column at rest — use your database's column-level encryption (PostgreSQL pgcrypto, MySQL AES_ENCRYPT) or a secrets manager (AWS KMS, GCP Cloud KMS, HashiCorp Vault) to encrypt the value before writing it. The encryption key lives outside the database.

2. Row-level security — ensure the API never returns another user's gee_project field. Add a database-level policy (e.g. PostgreSQL RLS) so queries are automatically scoped to the authenticated user.

3. Encrypt the full database at rest — all major cloud databases (RDS, Cloud SQL, Supabase) support transparent disk-level encryption; enable it.

4. Audit access — log every read of the gee_project column so suspicious bulk-reads are detectable.

What is NOT stored here:

• OAuth tokens and service-account keys stay on the server filesystem after earthengine authenticate. They are never stored in the user DB. Only the project ID (a non-secret identifier) is stored.

ensure_gee

ensure_gee(project: str = '', *, allow_prompt: bool = False) -> None

Authenticate and initialise GEE for the given project, once per project per process.

Multi-user API behaviour

Each unique project ID is initialised independently. Requests from User A and User B (different GEE projects) each trigger their own initialisation without interfering with each other. Subsequent requests from the same user skip re-initialisation (idempotent).

Project resolution order

1. project argument (always used in API context)
2. GEE_PROJECT environment variable / .env file (local use)
3. Interactive prompt (when allow_prompt=True, local / notebook only)

Parameters

project : str GEE Cloud project ID. In API context this is always the value stored for the authenticated user — never empty. allow_prompt : bool False in API/server contexts; raises immediately on missing project.

Source code in core/gee_auth.py

def ensure_gee(project: str = "", *, allow_prompt: bool = False) -> None:
    """
    Authenticate and initialise GEE for the given project, once per project
    per process.

    Multi-user API behaviour
    ------------------------
    Each unique project ID is initialised independently.  Requests from
    User A and User B (different GEE projects) each trigger their own
    initialisation without interfering with each other.  Subsequent requests
    from the same user skip re-initialisation (idempotent).

    Project resolution order
    ------------------------
    1. ``project`` argument  (always used in API context)
    2. ``GEE_PROJECT`` environment variable / ``.env`` file  (local use)
    3. Interactive prompt (when ``allow_prompt=True``, local / notebook only)

    Parameters
    ----------
    project : str
        GEE Cloud project ID.  In API context this is always the value stored
        for the authenticated user — never empty.
    allow_prompt : bool
        ``False`` in API/server contexts; raises immediately on missing project.
    """
    import os

    with _lock:
        resolved = _resolve_project(project, allow_prompt=allow_prompt)
        cache_key = resolved or _ANONYMOUS_KEY

        if cache_key in _initialised_projects:
            return

        # Detect Dask worker context — workers cannot run interactive auth.
        in_worker = False
        try:
            from dask.distributed import get_worker

            get_worker()
            in_worker = True
        except (ImportError, ValueError):
            pass

        import ee

        kwargs: dict = {"project": resolved} if resolved else {}

        if in_worker:
            try:
                ee.Initialize(**kwargs)
            except Exception as exc:
                raise RuntimeError(
                    "GEE initialisation failed inside a Dask worker.\n"
                    "  Run  earthengine authenticate  in a terminal before\n"
                    "  starting the Dask cluster so credentials are on disk.\n"
                    f"Original error: {exc}"
                ) from exc
        else:
            try:
                from drought_monitoring.gee import authenticate

                authenticate(project=resolved or None, quiet=True)
            except ImportError:
                # drought_monitoring (ml extras) not installed — init directly
                ee.Initialize(**kwargs)

        _initialised_projects.add(cache_key)
        # Keep env in sync so sub-processes and libraries can read it.
        if resolved:
            os.environ["GEE_PROJECT"] = resolved

validate_gee_project

validate_gee_project(project: str) -> None

Verify that a GEE project ID is valid and reachable.

Call this once at user registration before storing the project ID in the database. If this succeeds, ensure_gee will authenticate automatically on every subsequent request — the user is never asked again.

Parameters

project : str The GEE Cloud project ID the user supplied during registration.

Raises

ValueError The project ID string is empty. RuntimeError GEE authentication or initialisation failed for this project.

Example (FastAPI registration endpoint)::

@router.post("/register")
async def register(body: RegisterRequest):
    try:
        validate_gee_project(body.gee_project)
    except (ValueError, RuntimeError) as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    # Store encrypted in DB — user will never be prompted again.
    user = await create_user(body, gee_project=encrypt(body.gee_project))
    return {"id": user.id}

Source code in core/gee_auth.py

def validate_gee_project(project: str) -> None:
    """
    Verify that a GEE project ID is valid and reachable.

    Call this **once at user registration** before storing the project ID in
    the database.  If this succeeds, ``ensure_gee`` will authenticate
    automatically on every subsequent request — the user is never asked again.

    Parameters
    ----------
    project : str
        The GEE Cloud project ID the user supplied during registration.

    Raises
    ------
    ValueError
        The project ID string is empty.
    RuntimeError
        GEE authentication or initialisation failed for this project.

    Example (FastAPI registration endpoint)::

        @router.post("/register")
        async def register(body: RegisterRequest):
            try:
                validate_gee_project(body.gee_project)
            except (ValueError, RuntimeError) as exc:
                raise HTTPException(status_code=422, detail=str(exc))
            # Store encrypted in DB — user will never be prompted again.
            user = await create_user(body, gee_project=encrypt(body.gee_project))
            return {"id": user.id}
    """
    if not project or not project.strip():
        raise ValueError(
            "GEE project ID cannot be empty. "
            "Provide your Google Cloud project ID that has the Earth Engine API enabled."
        )
    try:
        ensure_gee(project.strip(), allow_prompt=False)
    except Exception as exc:
        raise RuntimeError(
            f"Could not authenticate with Google Earth Engine "
            f"using project '{project}'.\n"
            "Make sure:\n"
            "  1. The project exists in Google Cloud Console.\n"
            "  2. The Earth Engine API is enabled for this project.\n"
            "  3. Your service-account key or OAuth credentials are valid.\n"
            f"Details: {exc}"
        ) from exc

startup_init_gee

startup_init_gee() -> None

Authenticate and initialise GEE once at server startup using the organisation's project configured in GEE_PROJECT (env / .env file).

Call this from the FastAPI lifespan inside asyncio.to_thread because ee.Initialize is synchronous.

Logs a warning rather than raising if GEE_PROJECT is not set, so the server still starts in environments where GEE is not yet configured.

Source code in core/gee_auth.py

def startup_init_gee() -> None:
    """
    Authenticate and initialise GEE once at server startup using the
    organisation's project configured in ``GEE_PROJECT`` (env / .env file).

    Call this from the FastAPI lifespan inside ``asyncio.to_thread`` because
    ``ee.Initialize`` is synchronous.

    Logs a warning rather than raising if ``GEE_PROJECT`` is not set, so the
    server still starts in environments where GEE is not yet configured.
    """
    import logging

    logger = logging.getLogger(__name__)

    project = os.environ.get("GEE_PROJECT", "").strip()
    if not project:
        logger.warning(
            "GEE_PROJECT is not set — Earth Engine analyses will fail until "
            "GEE_PROJECT is configured and the server is restarted."
        )
        return

    try:
        ensure_gee(project, allow_prompt=False)
        logger.info("Google Earth Engine initialised with project '%s'.", project)
    except Exception as exc:
        logger.error(
            "GEE initialisation failed at startup for project '%s': %s. "
            "Earth Engine analyses will be unavailable.",
            project,
            exc,
        )

Caching

cache

SimpleCache

SimpleCache()

In-process result cache. Replaced by Redis at the API layer.

Source code in core/cache.py

def __init__(self) -> None:
    self._store: dict[str, tuple] = {}

Distributed execution

dask_engine

DaskEngine

Manages a Dask LocalCluster for distributed in-memory computation. Singleton pattern — one cluster per process.

Two tiers of parallelism: - Dask Futures (submit / gather) : CPU-bound tasks (ML training, xarray ops). - I/O thread pool (run_io_parallel): GEE HTTP downloads that share a session.

get_client classmethod

get_client() -> Client

Return the singleton Dask Client, starting a LocalCluster if needed.

Source code in core/dask_engine.py

@classmethod
def get_client(cls) -> Client:
    """Return the singleton Dask Client, starting a LocalCluster if needed."""
    if cls._client is None or cls._client.status == "closed":
        cluster = LocalCluster(
            n_workers=int(os.environ.get("DASK_WORKERS", 4)),
            threads_per_worker=int(os.environ.get("DASK_THREADS_PER_WORKER", 2)),
            memory_limit=os.environ.get("DASK_MEMORY_LIMIT", "4GB"),
            silence_logs=True,
        )
        cls._client = Client(cluster)
    return cls._client

get_client_if_running classmethod

get_client_if_running() -> Client | None

Return the active Dask Client, or None if the cluster has not been started.

Source code in core/dask_engine.py

@classmethod
def get_client_if_running(cls) -> Client | None:
    """Return the active Dask Client, or None if the cluster has not been started."""
    if cls._client is not None and cls._client.status == "closed":
        cls._client = None
    return cls._client

shutdown classmethod

shutdown() -> None

Shut down the cluster and release resources.

Source code in core/dask_engine.py

@classmethod
def shutdown(cls) -> None:
    """Shut down the cluster and release resources."""
    if cls._client:
        cls._client.close()
        cls._client = None

submit classmethod

submit(fn: Callable, *args: Any, **kwargs: Any) -> Future

Submit a callable to the Dask cluster. Returns a Future.

Source code in core/dask_engine.py

@classmethod
def submit(cls, fn: Callable, *args: Any, **kwargs: Any) -> Future:
    """Submit a callable to the Dask cluster. Returns a Future."""
    return cls.get_client().submit(fn, *args, pure=False, **kwargs)

gather classmethod

gather(futures: list) -> list

Block until all futures complete and return results in submission order.

Source code in core/dask_engine.py

@classmethod
def gather(cls, futures: list) -> list:
    """Block until all futures complete and return results in submission order."""
    return cast(list, cls.get_client().gather(futures))

run_io_parallel staticmethod

run_io_parallel(tasks: dict[str, Callable[[], Any]], max_workers: int | None = None) -> dict[str, Any]

Execute zero-arg callables concurrently via threads.

Ideal for GEE HTTP downloads because threads share the authenticated Earth Engine session within a process, avoiding per-process re-auth.

Parameters

tasks : {key: zero-arg callable} max_workers: defaults to len(tasks).

Returns

dict with the same keys mapping to each callable's return value.

Source code in core/dask_engine.py

@staticmethod
def run_io_parallel(
    tasks: dict[str, Callable[[], Any]],
    max_workers: int | None = None,
) -> dict[str, Any]:
    """
    Execute zero-arg callables concurrently via threads.

    Ideal for GEE HTTP downloads because threads share the authenticated
    Earth Engine session within a process, avoiding per-process re-auth.

    Parameters
    ----------
    tasks      : {key: zero-arg callable}
    max_workers: defaults to len(tasks).

    Returns
    -------
    dict with the same keys mapping to each callable's return value.
    """
    n = max_workers or len(tasks)
    if not n:
        return {}
    with ThreadPoolExecutor(max_workers=n) as pool:
        futures = {k: pool.submit(fn) for k, fn in tasks.items()}
    return {k: f.result() for k, f in futures.items()}

clip_raster_to_aoi staticmethod

clip_raster_to_aoi(da_array, aoi_geojson: dict)

Clip a dask-backed xarray DataArray to an AOI polygon — lazy.

Source code in core/dask_engine.py

@staticmethod
def clip_raster_to_aoi(da_array, aoi_geojson: dict):
    """Clip a dask-backed xarray DataArray to an AOI polygon — lazy."""
    import rioxarray  # noqa: F401 — registers .rio accessor
    from shapely.geometry import shape

    geo_type = aoi_geojson.get("type")
    if geo_type == "Feature":
        geom_dict = aoi_geojson["geometry"]
    elif geo_type == "FeatureCollection":
        features = aoi_geojson.get("features", [])
        geom_dict = features[0]["geometry"] if features else aoi_geojson
    else:
        geom_dict = aoi_geojson
    geom = shape(geom_dict)
    return da_array.rio.clip([geom], crs="EPSG:4326", drop=True)

compute_with_progress staticmethod

compute_with_progress(lazy_result)

Trigger Dask computation and return the materialised result.

Source code in core/dask_engine.py

@staticmethod
def compute_with_progress(lazy_result):
    """Trigger Dask computation and return the materialised result."""
    return lazy_result.compute()

Land-cover masking

landcover_mask

worldcover_map

worldcover_map(aoi: Geometry) -> ee.Image

Raw ESA WorldCover v200 'Map' band (class codes 10-100), clipped to aoi.

Source code in core/landcover_mask.py

def worldcover_map(aoi: ee.Geometry) -> ee.Image:
    """Raw ESA WorldCover v200 'Map' band (class codes 10-100), clipped to aoi."""
    import ee  # local import — ee must be initialised before calling this

    return ee.ImageCollection(WORLDCOVER_COLLECTION).first().select("Map").clip(aoi)

exclusion_mask

exclusion_mask(aoi: Geometry, exclude_classes: list[int]) -> ee.Image

Boolean ee.Image (1 = keep, 0 = excluded) derived from ESA WorldCover landcover classes. Used to mask specific landcover classes (e.g. water bodies, built-up areas) out of an AOI before feature datasets are built or sampled.

Source code in core/landcover_mask.py

def exclusion_mask(aoi: ee.Geometry, exclude_classes: list[int]) -> ee.Image:
    """
    Boolean ee.Image (1 = keep, 0 = excluded) derived from ESA WorldCover landcover
    classes. Used to mask specific landcover classes (e.g. water bodies, built-up
    areas) out of an AOI before feature datasets are built or sampled.
    """
    lc = worldcover_map(aoi)
    mask = lc.neq(exclude_classes[0])
    for class_value in exclude_classes[1:]:
        mask = mask.And(lc.neq(class_value))
    return mask

fetch_landcover_class_array

fetch_landcover_class_array(aoi_geojson: dict, scale: int = 100) -> xr.DataArray

Download raw ESA WorldCover class codes over aoi_geojson as an xr.DataArray with (lon, lat) dims. Used where the AOI is only available as GeoJSON (no live ee.Geometry), e.g. to mask an already-computed spatial dataset.

Source code in core/landcover_mask.py

def fetch_landcover_class_array(aoi_geojson: dict, scale: int = 100) -> xr.DataArray:
    """
    Download raw ESA WorldCover class codes over aoi_geojson as an xr.DataArray
    with (lon, lat) dims. Used where the AOI is only available as GeoJSON (no
    live ee.Geometry), e.g. to mask an already-computed spatial dataset.
    """
    import ee
    import rioxarray as rxr

    from climate_change.core.base_use_case import _ee_geometry_from_geojson

    aoi = _ee_geometry_from_geojson(aoi_geojson)
    lc = ee.ImageCollection(WORLDCOVER_COLLECTION).first().select("Map").clip(aoi)
    url = lc.getDownloadURL(
        {"region": aoi, "scale": scale, "crs": "EPSG:4326", "format": "GEO_TIFF"}
    )

    resp = requests.get(url, timeout=600)
    try:
        resp.raise_for_status()
    except HTTPError as exc:
        body = resp.text[:1000] if resp.text else ""
        raise HTTPError(f"{exc}. Earth Engine response: {body}", response=resp) from exc

    da = cast("xr.DataArray", rxr.open_rasterio(io.BytesIO(resp.content))).squeeze()
    return da.rename({"x": "lon", "y": "lat"})

Population exposure

population

fetch_population_count

fetch_population_count(aoi: Geometry, scale: int, year: int = 2020) -> xr.Dataset

WorldPop GP 100 m population COUNT (people per pixel — an additive quantity, not a density) for population-exposure reporting.

Deliberately separate from any domain's log-transformed pop_density ML feature (e.g. disease/features.py::fetch_pop_density) — this returns raw counts suitable for summing within a risk zone, never model input.

Always fetches at (up to) the native ~100 m resolution regardless of the requested scale — i.e. min(scale, 100). An earlier version of this function used ee.Image.reduceResolution(Reducer.sum()) to downsample to a domain's coarser working scale before download, on the assumption that would sum native pixels into each output pixel. Verified against live GEE data this was wrong: reduceResolution's default area-weighting turns sum() into something closer to a coverage fraction (single-pixel test: output value was ~1 instead of the true ~25/~100 contributing pixels), and even with .unweighted() the boundary handling still over-counted by 20-60%+ against a server-side reduceRegion ground truth, including at a realistic ~100km county-sized AOI (not just a small edge-effect-prone test box). There is no reliable way found to make GEE aggregate this correctly server-side, so this function never downsamples population at all — see population_exposure() below for how the resolution mismatch with a domain's coarser risk-classification grid is actually handled (by upsampling the classification, not downsampling the population).

Returns Dataset with variable 'population', values in people-per-pixel, already masked (0.0) where WorldPop's -99999 nodata sentinel appears — verified live that the GeoTIFF export does not set a GDAL/rasterio nodata tag (da.rio.nodata comes back None), so summing the raw array would otherwise corrupt totals by many orders of magnitude.

Source code in core/population.py

def fetch_population_count(aoi: ee.Geometry, scale: int, year: int = 2020) -> xr.Dataset:
    """
    WorldPop GP 100 m population COUNT (people per pixel — an additive
    quantity, not a density) for population-exposure reporting.

    Deliberately separate from any domain's log-transformed pop_density ML
    feature (e.g. disease/features.py::fetch_pop_density) — this returns raw
    counts suitable for summing within a risk zone, never model input.

    Always fetches at (up to) the native ~100 m resolution regardless of the
    requested `scale` — i.e. min(scale, 100). An earlier version of this
    function used ee.Image.reduceResolution(Reducer.sum()) to downsample to a
    domain's coarser working scale before download, on the assumption that
    would sum native pixels into each output pixel. Verified against live GEE
    data this was wrong: reduceResolution's default area-weighting turns
    sum() into something closer to a coverage fraction (single-pixel test:
    output value was ~1 instead of the true ~25/~100 contributing pixels),
    and even with `.unweighted()` the boundary handling still over-counted by
    20-60%+ against a server-side reduceRegion ground truth, including at a
    realistic ~100km county-sized AOI (not just a small edge-effect-prone
    test box). There is no reliable way found to make GEE aggregate this
    correctly server-side, so this function never downsamples population at
    all — see population_exposure() below for how the resolution mismatch
    with a domain's coarser risk-classification grid is actually handled (by
    upsampling the *classification*, not downsampling the population).

    Returns Dataset with variable 'population', values in people-per-pixel,
    already masked (0.0) where WorldPop's -99999 nodata sentinel appears —
    verified live that the GeoTIFF export does not set a GDAL/rasterio
    nodata tag (da.rio.nodata comes back None), so summing the raw array
    would otherwise corrupt totals by many orders of magnitude.
    """
    import ee
    import xarray as xr

    pop_collection = (
        ee.ImageCollection(POPULATION_COLLECTION)
        .filterBounds(aoi)
        .filter(ee.Filter.eq("year", year))
    )
    pop_img = _require_population_image(pop_collection, year).select("population").clip(aoi)
    fetch_scale = min(scale, _NATIVE_SCALE)

    url = pop_img.getDownloadURL(
        {"region": aoi, "scale": fetch_scale, "crs": "EPSG:4326", "format": "GEO_TIFF"}
    )
    da = _download_band(url).squeeze()
    # Population can't be negative — clamp WorldPop's unmasked -99999
    # sentinel (and any other negative noise) to 0 rather than relying on a
    # nodata tag that isn't actually present in the export.
    da = da.copy(data=np.where(da.values < 0, 0.0, da.values))
    return xr.Dataset({"population": da.rename({"x": "lon", "y": "lat"})})

fetch_population_count_safe

fetch_population_count_safe(aoi: Geometry, scale: int, year: int = 2020) -> xr.Dataset | None

Best-effort wrapper around fetch_population_count.

Population exposure is an enrichment on top of the core risk analysis, not a required input — WorldPop being unavailable for a given AOI/year (or a transient network failure) must not crash the whole analysis the way a missing rainfall/temperature/NDVI band legitimately should. Returns None on any failure; callers should skip exposure reporting in that case.

Source code in core/population.py

def fetch_population_count_safe(
    aoi: ee.Geometry, scale: int, year: int = 2020
) -> xr.Dataset | None:
    """
    Best-effort wrapper around fetch_population_count.

    Population exposure is an enrichment on top of the core risk analysis,
    not a required input — WorldPop being unavailable for a given AOI/year
    (or a transient network failure) must not crash the whole analysis the
    way a missing rainfall/temperature/NDVI band legitimately should. Returns
    None on any failure; callers should skip exposure reporting in that case.
    """
    try:
        return fetch_population_count(aoi, scale=scale, year=year)
    except Exception:
        _log.warning(
            "Population fetch failed; population exposure stats will be omitted",
            exc_info=True,
        )
        return None

population_by_class

population_by_class(class_grid: ndarray, population_grid: ndarray, classes: list) -> dict[str, float]

Zonal sum of population per value present in class_grid.

class_grid and population_grid must already be the same shape and cover the same grid. In practice, prefer population_exposure() below, which handles the (typical) case where the risk-classification grid is coarser than population's native resolution — calling this directly requires the caller to have already resolved that mismatch. Works with int-coded risk grids (disease/flood/food_security/land_degradation) or string-labelled class grids (drought's CDI severity classes) alike — classes is whatever set of class tokens is meaningful for that grid's dtype. Returned dict is keyed by str(class token).

Source code in core/population.py

def population_by_class(
    class_grid: np.ndarray,
    population_grid: np.ndarray,
    classes: list,
) -> dict[str, float]:
    """
    Zonal sum of population per value present in class_grid.

    class_grid and population_grid must already be the same shape and cover
    the same grid. In practice, prefer population_exposure() below, which
    handles the (typical) case where the risk-classification grid is coarser
    than population's native resolution — calling this directly requires the
    caller to have already resolved that mismatch. Works with int-coded risk
    grids (disease/flood/food_security/land_degradation) or string-labelled
    class grids (drought's CDI severity classes) alike — `classes` is
    whatever set of class tokens is meaningful for that grid's dtype.
    Returned dict is keyed by str(class token).
    """
    if class_grid.shape != population_grid.shape:
        raise ValueError(
            f"class_grid shape {class_grid.shape} != population_grid shape "
            f"{population_grid.shape} — must be aligned to the same grid. "
            "Use population_exposure() if the two grids have different resolutions."
        )
    pop_valid = np.nan_to_num(population_grid.astype(np.float64), nan=0.0)
    return {str(cls): round(float(pop_valid[class_grid == cls].sum()), 1) for cls in classes}

upsample_class_grid

upsample_class_grid(class_grid: ndarray, class_transform, target_shape: tuple[int, int], target_transform, crs: str = 'EPSG:4326') -> np.ndarray

Nearest-neighbour upsample a (coarse) classified grid onto a finer target grid — typically population's native ~100 m resolution. Nearest-neighbour is the semantically correct choice for a categorical class grid (each finer output pixel just inherits its parent coarse cell's class), unlike downsampling an additive quantity like population counts, which is why the resolution mismatch is resolved this direction and not the other.

Source code in core/population.py

def upsample_class_grid(
    class_grid: np.ndarray,
    class_transform,
    target_shape: tuple[int, int],
    target_transform,
    crs: str = "EPSG:4326",
) -> np.ndarray:
    """
    Nearest-neighbour upsample a (coarse) classified grid onto a finer target
    grid — typically population's native ~100 m resolution. Nearest-neighbour
    is the semantically correct choice for a categorical class grid (each
    finer output pixel just inherits its parent coarse cell's class), unlike
    downsampling an additive quantity like population counts, which is why
    the resolution mismatch is resolved this direction and not the other.
    """
    import rasterio.warp

    dst = np.zeros(target_shape, dtype=class_grid.dtype)
    rasterio.warp.reproject(
        source=class_grid,
        destination=dst,
        src_transform=class_transform,
        src_crs=crs,
        dst_transform=target_transform,
        dst_crs=crs,
        resampling=rasterio.warp.Resampling.nearest,
    )
    return dst

population_exposure

population_exposure(class_grid: ndarray, class_transform, population_ds: Dataset, classes: list) -> dict[str, float]

Zonal sum of population per class, handling the resolution mismatch between a domain's (coarser) risk-classification grid and population's native ~100 m grid: upsamples class_grid onto population's grid via nearest-neighbour, then delegates to population_by_class. This is the primary entry point domains should use — see population_by_class's docstring for why population itself is never downsampled instead.

class_grid may hold int-coded classes (disease/flood/food_security/ land_degradation's risk grids) or string labels (drought's CDI severity classes) — GDAL/rasterio's reprojection (used internally to upsample class_grid) only supports numeric dtypes, so a non-numeric class_grid is transparently encoded to small integers before upsampling and decoded back to the original classes labels in the returned dict.

Source code in core/population.py

def population_exposure(
    class_grid: np.ndarray,
    class_transform,
    population_ds: xr.Dataset,
    classes: list,
) -> dict[str, float]:
    """
    Zonal sum of population per class, handling the resolution mismatch
    between a domain's (coarser) risk-classification grid and population's
    native ~100 m grid: upsamples class_grid onto population's grid via
    nearest-neighbour, then delegates to population_by_class. This is the
    primary entry point domains should use — see population_by_class's
    docstring for why population itself is never downsampled instead.

    class_grid may hold int-coded classes (disease/flood/food_security/
    land_degradation's risk grids) or string labels (drought's CDI severity
    classes) — GDAL/rasterio's reprojection (used internally to upsample
    class_grid) only supports numeric dtypes, so a non-numeric class_grid is
    transparently encoded to small integers before upsampling and decoded
    back to the original `classes` labels in the returned dict.
    """
    from rasterio.transform import from_bounds

    pop_da = population_ds["population"]
    lons = np.asarray(pop_da["lon"].values)
    lats = np.asarray(pop_da["lat"].values)
    pop_transform = from_bounds(
        float(lons.min()),
        float(lats.min()),
        float(lons.max()),
        float(lats.max()),
        len(lons),
        len(lats),
    )
    pop_grid = np.asarray(pop_da.values).astype(np.float64)

    if np.issubdtype(class_grid.dtype, np.number):
        upsampled_class = upsample_class_grid(
            class_grid, class_transform, pop_grid.shape, pop_transform
        )
        return population_by_class(upsampled_class, pop_grid, classes=classes)

    label_to_idx = {label: i for i, label in enumerate(classes)}
    encoded = np.full(class_grid.shape, -1, dtype=np.int32)
    for label, idx in label_to_idx.items():
        encoded[class_grid == label] = idx
    upsampled_encoded = upsample_class_grid(encoded, class_transform, pop_grid.shape, pop_transform)
    by_idx = population_by_class(upsampled_encoded, pop_grid, classes=list(label_to_idx.values()))
    return {label: by_idx[str(idx)] for label, idx in label_to_idx.items()}

at_risk_population

at_risk_population(pop_by_class: dict[str, float], at_risk_labels: list[str]) -> float

Sum population across the given at-risk class labels (keys of pop_by_class).

Source code in core/population.py

def at_risk_population(pop_by_class: dict[str, float], at_risk_labels: list[str]) -> float:
    """Sum population across the given at-risk class labels (keys of pop_by_class)."""
    return round(sum(pop_by_class.get(label, 0.0) for label in at_risk_labels), 1)