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

run_multi_hazard_analysis async

run_multi_hazard_analysis(modules: list[str], aoi_geojson: dict, start_date: str, end_date: str, country: str, gee_project: str = '', extra_params: dict[str, dict] | None = None) -> dict[str, AnalysisOutput]

Run two different risk modules against the same area of interest and date range, for side-by-side comparison of different hazards (as opposed to run_analysis run twice for the same hazard across two different regions).

The two modules run sequentially, sharing one DaskEngine — running them concurrently would double peak memory for no benefit under this package's typical single-worker deployment.

Parameters

modules: Exactly two distinct module names, e.g. ["food_security", "land_degradation"]. extra_params: Per-module extra parameters, e.g. {"food_security": {"model_type": "rf"}, ...}.

Source code in core/runner.py

async def run_multi_hazard_analysis(
    modules: list[str],
    aoi_geojson: dict,
    start_date: str,
    end_date: str,
    country: str,
    gee_project: str = "",
    extra_params: dict[str, dict] | None = None,
) -> dict[str, AnalysisOutput]:
    """
    Run two different risk modules against the same area of interest and
    date range, for side-by-side comparison of different hazards (as
    opposed to ``run_analysis`` run twice for the same hazard across two
    different regions).

    The two modules run sequentially, sharing one ``DaskEngine`` — running
    them concurrently would double peak memory for no benefit under this
    package's typical single-worker deployment.

    Parameters
    ----------
    modules:
        Exactly two distinct module names, e.g. ``["food_security", "land_degradation"]``.
    extra_params:
        Per-module extra parameters, e.g. ``{"food_security": {"model_type": "rf"}, ...}``.
    """
    if len(modules) != 2:
        raise ValueError(
            f"run_multi_hazard_analysis requires exactly 2 modules, got {len(modules)}"
        )
    if modules[0] == modules[1]:
        raise ValueError(
            f"run_multi_hazard_analysis requires 2 distinct modules, got '{modules[0]}' twice"
        )

    for module in modules:
        _ensure_module_registered(module)
        if module not in MODULE_MAP:
            raise ValueError(f"Unknown module: '{module}'. Available: {sorted(MODULE_MAP)}")

    project = gee_project or os.environ.get("GEE_PROJECT", "")
    ensure_gee(project)

    from climate_change.core.dask_engine import DaskEngine

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

    all_params = extra_params or {}
    results: dict[str, AnalysisOutput] = {}
    for module in modules:
        params = dict(all_params.get(module, {}))
        params.setdefault("gee_project", project)
        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)
        results[module] = await use_case.execute(config)

    return results

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 {}

        # A service-account key (GOOGLE_APPLICATION_CREDENTIALS) must take
        # priority over drought_monitoring's authenticate() below, which
        # unconditionally calls ee.Authenticate() (interactive OAuth) first
        # — that fails outright in a headless server/container (no browser,
        # no TTY) even when a valid service-account key is already in place.
        service_account_key_file = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "").strip()

        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
        elif service_account_key_file:
            # ee.Initialize()'s default credentials='persistent' falls back
            # to plain google.auth.default() when no `earthengine
            # authenticate`-issued token is cached — which *does* pick up
            # GOOGLE_APPLICATION_CREDENTIALS, but requests no explicit OAuth
            # scopes, so Earth Engine rejects the resulting token with
            # "invalid_scope: Invalid OAuth scope or ID token audience
            # provided." (confirmed live). ee.ServiceAccountCredentials
            # builds the credentials with Earth Engine's required scopes
            # (oauth.SCOPES) instead — the pattern Earth Engine's own docs
            # recommend for service accounts, and the only one that works.
            try:
                email = ""
                try:
                    with open(service_account_key_file) as f:
                        email = json.load(f).get("client_email", "")
                except (OSError, ValueError):
                    pass  # not a readable JSON key — ServiceAccountCredentials will raise a clear error
                credentials = ee.ServiceAccountCredentials(email, service_account_key_file)
                ee.Initialize(credentials, **kwargs)
            except Exception as exc:
                raise RuntimeError(
                    "GEE initialisation failed using the service-account "
                    "credentials at GOOGLE_APPLICATION_CREDENTIALS.\n"
                    "  • Confirm the file is a valid service-account JSON key.\n"
                    "  • Confirm the Earth Engine API is enabled and the "
                    "project/service account are registered for Earth Engine access.\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 = get_with_retry(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, country: str | None = None) -> xr.Dataset

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

Primary source is Google Earth Engine's WorldPop/GP/100m/pop mirror (see _fetch_population_count_gee) — fast, no full-file download. Falls back to worldpop.org's own data catalog (_fetch_population_count_worldpop_org, which downloads and caches the whole per-country GeoTIFF — much slower, especially for large countries) whenever the GEE path fails outright, or "succeeds" with no valid pixels over the AOI — verified live that this happens for some countries/AOIs (e.g. parts of Uganda) even though the collection has an image for that country/year. The worldpop.org fallback is only attempted when country resolves to an ISO3 code (see _resolve_iso3); if it can't be resolved, the original GEE error/emptiness is raised as-is.

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

Source code in core/population.py

def fetch_population_count(
    aoi: ee.Geometry, scale: int, year: int = 2020, country: str | None = None
) -> xr.Dataset:
    """
    Population COUNT (people per pixel — an additive quantity, not a
    density) for population-exposure reporting.

    Primary source is Google Earth Engine's WorldPop/GP/100m/pop mirror (see
    _fetch_population_count_gee) — fast, no full-file download. Falls back to
    worldpop.org's own data catalog (_fetch_population_count_worldpop_org,
    which downloads and caches the whole per-country GeoTIFF — much slower,
    especially for large countries) whenever the GEE path fails outright, or
    "succeeds" with no valid pixels over the AOI — verified live that this
    happens for some countries/AOIs (e.g. parts of Uganda) even though the
    collection has an image for that country/year. The worldpop.org fallback
    is only attempted when `country` resolves to an ISO3 code (see
    _resolve_iso3); if it can't be resolved, the original GEE error/emptiness
    is raised as-is.

    Deliberately separate from any domain's log-transformed pop_density ML
    feature (e.g. disease/features.py::fetch_pop_density, which now delegates
    here) — this returns raw counts suitable for summing within a risk zone,
    never model input.
    """
    bounds_coords = cast(list, aoi.bounds().coordinates().getInfo())[0]
    lons = [c[0] for c in bounds_coords]
    lats = [c[1] for c in bounds_coords]
    min_lon, min_lat, max_lon, max_lat = min(lons), min(lats), max(lons), max(lats)

    try:
        return _fetch_population_count_gee(aoi, scale, year, min_lon, min_lat, max_lon, max_lat)
    except Exception:
        _log.warning(
            "Earth Engine population fetch failed or had no coverage for this AOI/year; "
            "falling back to worldpop.org",
            exc_info=True,
        )

    iso3 = _resolve_iso3(country)
    if not iso3:
        raise ValueError(
            f"No population data was available for {year}: Earth Engine's WorldPop "
            f"mirror had none, and country {country!r} could not be resolved to an "
            "ISO3 code to try the worldpop.org fallback."
        )
    return _fetch_population_count_worldpop_org(min_lon, min_lat, max_lon, max_lat, iso3, year)

fetch_population_count_safe

fetch_population_count_safe(aoi: Geometry, scale: int, year: int = 2020, country: str | None = None) -> 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, country: str | None = None
) -> 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, country=country)
    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)