feat: extensible transforms pipeline for zarr build

.env.example

@@ -1,5 +1,5 @@
 # ── Instance configuration ────────────────────────────────────────────────────
-# Path to the instance config file (extent, optional datasets_dir).
+# Path to the instance config file (extent, optional templates_dir).
 # Copy the example before editing: cp climate-api.yaml.example climate-api.yaml
 # climate-api.yaml is gitignored so your local extent stays out of version control.
 # When running via `make run` from the repo root, the relative path below works.
@@ -17,9 +17,6 @@ CLIMATE_API_CONFIG=./climate-api.yaml
 # See docs/setup_guide.md for registration and .netrc setup instructions.
 
 # ── Download and ingestion ────────────────────────────────────────────────────
-# Override the download cache directory (default: data/downloads).
-# CACHE_OVERRIDE=/path/to/cache
-
 # Fallback bounding box used when a request does not include an explicit bbox.
 # Format: xmin,ymin,xmax,ymax
 # DOWNLOAD_BBOX=-13.5,6.9,-10.1,10.0

Makefile

@@ -7,7 +7,7 @@ sync: ## Install dependencies with uv
 	uv sync
 
 run: openapi ## Start the app with uvicorn
-	uv run uvicorn climate_api.main:app --reload
+	uv run uvicorn climate_api.main:app --reload --reload-include "*.html" --reload-include "*.yaml" --reload-include "*.yml"
 
 lint: ## Check linting, formatting, and types (no autofix)
 	uv run ruff check .

climate-api.yaml.example

@@ -8,4 +8,6 @@ extent:
   bbox: [-13.5, 6.9, -10.1, 10.0]
   country_code: SLE
 
-# datasets_dir: ./datasets/   # optional — custom templates merged with built-ins
+data_dir: ./data   # required — directory for downloaded NetCDF files and Zarr stores
+
+# templates_dir: ./templates/   # optional — root for custom templates; datasets go in templates/datasets/

climate_api/config.py

@@ -7,6 +7,8 @@
 
 import yaml
 
+_MISSING = object()
+
 
 def _substitute_env_vars(text: str) -> str:
     """Replace ${VAR:-default} patterns with values from the environment."""
@@ -54,3 +56,31 @@ def _load_config() -> dict[str, Any]:
         raise ValueError(f"CLIMATE_API_CONFIG must be a YAML mapping at the top level: {path}")
     _cache = dict(loaded or {})
     return _cache
+
+
+def get_data_dir() -> Path | None:
+    """Return the data directory declared in CLIMATE_API_CONFIG.
+
+    Returns None when CLIMATE_API_CONFIG is unset or points to a file that does
+    not exist (e.g. CI environments where the config is gitignored).
+
+    Raises ValueError if the config file exists but data_dir is not set, so
+    misconfigured instances fail fast at startup rather than silently sharing
+    a default directory with other instances.
+
+    """
+    config_path = get_config_path()
+    if config_path is None or not config_path.exists():
+        return None
+
+    config = get_config()
+    raw = config.get("data_dir", _MISSING)
+    if raw is _MISSING:
+        raise ValueError(
+            "data_dir is required in CLIMATE_API_CONFIG when a config file is present. "
+            "Set it to the directory where downloaded data should be stored, "
+            "e.g. data_dir: ./data"
+        )
+    if not isinstance(raw, (str, Path)):
+        raise ValueError(f"data_dir in CLIMATE_API_CONFIG must be a path string, got {type(raw).__name__}")
+    return (config_path.parent / raw).resolve()

climate_api/data/datasets/chirps3.yaml

@@ -7,9 +7,21 @@
   sync_execution: append
   sync_availability:
     latest_available_function: climate_api.providers.availability.chirps3_daily_latest_available
-  ingestion: 
+  extents:
+    spatial:
+      bbox: [-180, -50, 180, 50]
+      crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84
+    temporal:
+      begin: "1981-01-01"
+      trs: http://www.opengis.net/def/uom/ISO-8601/0/Gregorian
+      resolution: P1D
+  ingestion:
     function: dhis2eo.data.chc.chirps3.daily.download
   units: mm
   resolution: 5 km x 5 km
   source: CHIRPS v3
   source_url: https://www.chc.ucsb.edu/data/chirps3
+  display:
+    colormap: blues
+    range: [0.0, 20.0]
+    nodata: -9999.0

climate_api/data/datasets/era5_land.yaml

@@ -8,15 +8,28 @@
   sync_availability:
     latest_available_function: climate_api.providers.availability.lagged_latest_available
     lag_hours: 120
-  ingestion: 
+  extents:
+    spatial:
+      bbox: [-180, -90, 180, 90]
+      crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84
+    temporal:
+      begin: "1950-01-01"
+      trs: http://www.opengis.net/def/uom/ISO-8601/0/Gregorian
+      resolution: PT1H
+  ingestion:
     function: dhis2eo.data.destine.era5_land.hourly.download
     default_params:
       variables: ['t2m']
+  transforms:
+    - climate_api.transforms.convert_units
   units: kelvin
   convert_units: degC
   resolution: 9 km x 9 km
   source: ERA5-Land Reanalysis
   source_url: https://earthdatahub.destine.eu/collections/era5/datasets/reanalysis-era5-land
+  display:
+    colormap: rdbu_r
+    range: [15.0, 40.0]
 
 - id: era5land_precipitation_hourly
   name: Total precipitation (ERA5-Land)
@@ -28,13 +41,27 @@
   sync_availability:
     latest_available_function: climate_api.providers.availability.lagged_latest_available
     lag_hours: 120
-  ingestion: 
+  extents:
+    spatial:
+      bbox: [-180, -90, 180, 90]
+      crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84
+    temporal:
+      begin: "1950-01-01"
+      trs: http://www.opengis.net/def/uom/ISO-8601/0/Gregorian
+      resolution: PT1H
+  ingestion:
     function: dhis2eo.data.destine.era5_land.hourly.download
     default_params:
       variables: ['tp']
-  pre_process: ['deaccumulate_era5']
+  transforms:
+    - climate_api.transforms.deaccumulate_era5
+    - climate_api.transforms.convert_units
   units: m
   convert_units: mm
   resolution: 9 km x 9 km
   source: ERA5-Land Reanalysis
   source_url: https://earthdatahub.destine.eu/collections/era5/datasets/reanalysis-era5-land
+  display:
+    colormap: blues
+    range: [0.0, 5.0]
+    nodata: 0.0

climate_api/data/datasets/worldpop.yaml

@@ -8,6 +8,15 @@
     latest_available_function: climate_api.providers.availability.worldpop_release_latest_available
     # WorldPop projections are intentionally request-driven for future years.
     allow_future: true
+  extents:
+    spatial:
+      bbox: [-180, -90, 180, 90]
+      crs: http://www.opengis.net/def/crs/OGC/1.3/CRS84
+    temporal:
+      begin: "2015"
+      end: "2030"
+      trs: http://www.opengis.net/def/uom/ISO-8601/0/Gregorian
+      resolution: P1Y
   ingestion:
     function: dhis2eo.data.worldpop.pop_total.yearly.download
     multiscales:
@@ -18,3 +27,7 @@
   resolution: 100m x 100m
   source: WorldPop Global2
   source_url: https://hub.worldpop.org/project/categories?id=3
+  display:
+    colormap: reds
+    range: [0.0, 25.0]
+    nodata: 0.0

climate_api/data_manager/services/downloader.py

@@ -16,19 +16,17 @@
 from geozarr_toolkit import MultiscalesConventionMetadata, create_geozarr_attrs
 from topozarr.coarsen import create_pyramid
 
+from climate_api import config as api_config
+
 from .utils import get_lon_lat_dims, get_time_dim
 
 logger = logging.getLogger(__name__)
 
 
 def _resolve_download_dir() -> Path:
-    # CACHE_OVERRIDE keeps existing Docker/dev deployments working unchanged.
-    override = os.getenv("CACHE_OVERRIDE")
-    if override:
-        return Path(override)
-    # Default to an XDG-compliant user-writable location so the package works
-    # when installed with pip (where a package-relative path would land inside
-    # site-packages and typically be non-writable).
+    data_dir = api_config.get_data_dir()
+    if data_dir is not None:
+        return data_dir / "downloads"
     xdg_data = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local" / "share"))
     return xdg_data / "climate-api" / "downloads"
 
@@ -54,6 +52,7 @@ def download_dataset(
     When running in the background-task path, the download is deferred and this function
     returns an empty list because no files have been created yet.
     """
+    _validate_spatial_coverage(dataset, bbox if bbox is not None else _bbox_from_env())
     ingestion = dataset["ingestion"]
     eo_download_func_path = ingestion["function"]
     eo_download_func = _get_dynamic_function(eo_download_func_path)
@@ -143,6 +142,7 @@ def build_dataset_zarr(dataset: dict[str, Any], *, start: str | None = None, end
     dims = [lon_dim, lat_dim]
 
     ds = _select_time_range(ds, dataset=dataset, start=start, end=end)
+    ds = _run_transforms(ds, dataset)
 
     xmin = ds[lon_dim].min().item()
     xmax = ds[lon_dim].max().item()
@@ -243,6 +243,16 @@ def _select_time_range(
     return selected
 
 
+def _run_transforms(ds: xr.Dataset, dataset: dict[str, Any]) -> xr.Dataset:
+    for entry in dataset.get("transforms", []):
+        func_path = entry if isinstance(entry, str) else entry["function"]
+        params = {} if isinstance(entry, str) else entry.get("params", {})
+        func = _get_dynamic_function(func_path)
+        logger.info("Applying transform %s to dataset %s", func_path, dataset.get("id", "?"))
+        ds = func(ds, dataset, **params)
+    return ds
+
+
 def _compute_time_space_chunks(
     ds: xr.Dataset,
     dataset: dict[str, Any],
@@ -289,6 +299,39 @@ def get_zarr_path(dataset: dict[str, Any]) -> Path | None:
     return None
 
 
+def _validate_spatial_coverage(dataset: dict[str, Any], bbox: list[float] | None) -> None:
+    """Raise HTTP 400 if the request bbox falls outside the dataset's declared extents."""
+    extents = dataset.get("extents")
+    if not extents or bbox is None:
+        return
+    spatial = extents.get("spatial")
+    if not spatial:
+        return
+    cov_bbox = spatial.get("bbox")
+    if not isinstance(cov_bbox, (list, tuple)) or len(cov_bbox) != 4:
+        return
+    cov_xmin, cov_ymin, cov_xmax, cov_ymax = cov_bbox
+    xmin, ymin, xmax, ymax = bbox
+    if ymin > cov_ymax or ymax < cov_ymin:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Dataset '{dataset['id']}' does not cover this extent. "
+                f"Latitude coverage: {cov_ymin}°–{cov_ymax}°, "
+                f"requested: {ymin}°–{ymax}°."
+            ),
+        )
+    if xmin > cov_xmax or xmax < cov_xmin:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Dataset '{dataset['id']}' does not cover this extent. "
+                f"Longitude coverage: {cov_xmin}°–{cov_xmax}°, "
+                f"requested: {xmin}°–{xmax}°."
+            ),
+        )
+
+
 def _get_dynamic_function(full_path: str) -> Callable[..., Any]:
     """Import and return a function given its dotted module path."""
     parts = full_path.split(".")

climate_api/data_registry/services/datasets.py

@@ -23,7 +23,7 @@ def list_datasets() -> list[dict[str, Any]]:
     """Load all dataset templates and return a flat list.
 
     Built-in templates from climate_api/data/datasets/ are always loaded. When
-    datasets_dir is set in CLIMATE_API_CONFIG, templates from that directory are
+    templates_dir is set in CLIMATE_API_CONFIG, templates from that directory are
     merged on top — a custom template with the same id overrides the built-in one.
 
     CONFIGS_DIR (test override via monkeypatch) bypasses this and loads only
@@ -34,16 +34,18 @@ def list_datasets() -> list[dict[str, Any]]:
 
     merged: dict[str, dict[str, Any]] = {d["id"]: d for d in _load_builtin_datasets()}
 
-    config_datasets_dir = api_config.get_config().get("datasets_dir")
-    if config_datasets_dir:
-        if not isinstance(config_datasets_dir, (str, Path)):
+    config_templates_dir = api_config.get_config().get("templates_dir")
+    if config_templates_dir:
+        if not isinstance(config_templates_dir, (str, Path)):
             raise ValueError(
-                f"datasets_dir in CLIMATE_API_CONFIG must be a path string, got {type(config_datasets_dir).__name__}"
+                f"templates_dir in CLIMATE_API_CONFIG must be a path string, got {type(config_templates_dir).__name__}"
             )
         config_path = api_config.get_config_path()
-        resolved = (config_path.parent / config_datasets_dir).resolve() if config_path else Path(config_datasets_dir)
-        for dataset in _load_from_dir(resolved):
-            merged[dataset["id"]] = dataset
+        root = (config_path.parent / config_templates_dir).resolve() if config_path else Path(config_templates_dir)
+        datasets_subdir = root / "datasets"
+        if datasets_subdir.is_dir():
+            for dataset in _load_from_dir(datasets_subdir):
+                merged[dataset["id"]] = dataset
 
     return list(merged.values())
 

climate_api/ingestions/services.py

@@ -49,10 +49,11 @@
 
 
 def _resolve_artifacts_dir() -> Path:
-    # CACHE_OVERRIDE keeps existing Docker/dev deployments working unchanged.
-    override = os.getenv("CACHE_OVERRIDE")
-    if override:
-        return Path(override) / "artifacts"
+    from climate_api import config as api_config
+
+    data_dir = api_config.get_data_dir()
+    if data_dir is not None:
+        return data_dir / "artifacts"
     xdg_data = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local" / "share"))
     return xdg_data / "climate-api" / "artifacts"
 

climate_api/publications/services.py

@@ -19,9 +19,11 @@
 
 
 def _resolve_pygeoapi_dir() -> Path:
-    override = os.getenv("CACHE_OVERRIDE")
-    if override:
-        return Path(override) / "pygeoapi"
+    from climate_api import config as api_config
+
+    data_dir = api_config.get_data_dir()
+    if data_dir is not None:
+        return data_dir / "pygeoapi"
     xdg_data = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local" / "share"))
     return xdg_data / "climate-api" / "pygeoapi"