Skip to content

README.md

Latest commit

 

History

History
189 lines (148 loc) · 9.73 KB

README.md

File metadata and controls

189 lines (148 loc) · 9.73 KB
Raw

SMOS ARD Pipeline

On-demand SMOS satellite data pipeline that produces Analysis-Ready Data (ARD) cubes and publishes them to the EarthCODE Open Science Catalogue.

The pipeline can be driven in two ways:

  • procodile workflow (workflow.py), runs directly as a Python process; also the source of truth for the ProcessRegistry from which the Airflow DAG can be auto-generated via appligator
  • Airflow DAG (dags/smos_ard_update_dag.py) runs on Kubernetes using KubernetesPodOperator

Status

All core pipeline components are implemented and validated end-to-end.

Component Status Notes
Pipeline steps done fetch → aggregate → publish fully implemented
Typed step I/O done Each step declares exactly what it produces and consumes (FetchResult, AggregateResult, PublishResult). This catches wiring mistakes at definition time rather than at runtime inside a Kubernetes pod, and gives appligator enough information to generate the DAG wiring automatically.
Airflow DAG done KubernetesPodOperator per step; run_step.py entrypoint wired
DAG auto-generation done appligator generates the Airflow DAG from the ProcessRegistry in workflow.py
Docker build done Build context generated automatically from the workflow definition
OSC publication done pipeline stages YAML artefacts to S3; publish_trigger.py runs deep-code separately to open a GitHub PR
Operational run live Daily scheduled updates at 23:00 UTC; global cube appended incrementally each day

Project Structure

smos-ard-pipeline/
├── pixi.toml                           # environment + dependency management
├── pyproject.toml                      # package metadata, ruff/mypy/pytest config
├── build_docker_template.py           # generates docker/ build context via appligator
├── src/
│   └── smos_ard/
│       ├── workflow.py                 # procodile workflow definition + ProcessRegistry
│       ├── fetch.py                    # Step 1: xcube-smos → staging Zarr
│       ├── aggregate.py                # Step 2: temporal mean + QC → ARD Zarr
│       ├── publish.py                  # Step 3: stage YAML artefacts to S3 catalogue-pending/
│       ├── publish_trigger.py          # Separate CLI: deep-code → GitHub PR (runs outside pipeline)
│       └── utils.py                    # Shared S3 helpers
├── dags/
│   ├── smos-ard-service-v2.py          # active production DAG (appligator-generated, daily schedule)
│   ├── main_step.py                    # legacy auto-generated DAG
│   └── smos_ard_update_dag.py          # legacy hand-written DAG
├── tests/                              # pytest unit tests
├── .github/workflows/ci.yml            # CI: lint, typecheck, tests on push/PR
└── docker/                             # generated build context — do not edit or commit

Pipeline Steps

Step Function Description
1 fetch_smos_data Opens xcube-smos store on CREODIAS, writes raw Zarr to s3://{bucket}/staging/{prefix}/raw.zarr
2 aggregate_datacube Temporal resample (mean) + QC masking + variable metadata, writes/appends ARD Zarr to s3://{bucket}/ard/{prefix}/cube.zarr
3 publish_to_catalogue Generates dataset-config.yaml, workflow-config.yaml, and manifest.json; writes them to s3://{bucket}/catalogue-pending/{collection_id}/
— publish_trigger.py Separate CLI: reads staged artefacts from S3, calls deep-code to open a GitHub PR on the OSC metadata repo, archives artefacts to catalogue-published/ on success

Why fetch and aggregate are separate steps

  1. Fault tolerance : if aggregation fails, the raw fetch does not need to repeat
  2. Resource isolation : fetch is network-bound; aggregation is CPU/memory-bound; Kubernetes can size pods differently
  3. Incremental updates : the ARD Zarr supports append-only updates; new data can be fetched and appended without reprocessing the full history

Data Flow

Bulk data (Zarr arrays) flows through S3. Metadata (URIs, extents, variable lists) flows through step outputs / Airflow xcom (≤48KB — never pass array data).

start_date, end_date, bbox, output_prefix, stac_s3_bucket
        │
        ▼
fetch_smos_data          → S3: staging/{prefix}/raw.zarr
        │ FetchResult
        ▼
aggregate_datacube       → S3: ard/{prefix}/cube.zarr
        │ AggregateResult
        ▼
publish_to_catalogue     → S3: catalogue-pending/{collection_id}/
        │                       dataset-config.yaml
        │                       workflow-config.yaml
        │                       manifest.json
        │ PublishResult
        ▼
  catalogue_pending_uri

  ── separate step, run manually or via a trigger ──

publish_trigger.py       → GitHub PR on OSC metadata repo
  (reads pending/)       → S3: catalogue-published/{collection_id}/{timestamp}/

procodile Workflow (workflow.py)

workflow.py is the single source of truth for the pipeline. It uses the procodile framework to declare the pipeline as a typed, inspectable workflow rather than a plain Python script. Three procodile constructs do the work:

  • ProcessRegistry : a top-level container that holds all workflows. appligator reads this registry at code-generation time to produce the Airflow DAG.
  • @registry.main(...) : marks the pipeline entry point (process_pipeline). Declares the user-facing inputs (start_date, end_date, bbox, output_prefix, stac_s3_bucket) with types, titles, and defaults. These become the Airflow DAG parameters. Important: calling process_pipeline(...) directly runs the full pipeline via procodile — it does not just execute the decorated function body. The DAG's main_step KPO therefore calls pipeline_params (a plain, undecorated passthrough) instead, so the pod only validates and forwards the inputs without triggering downstream steps.
  • @process_pipeline.step(...) : registers each pipeline step and declares where its inputs come from, either FromMain(output=...) (a value passed in by the user) or FromStep(step_id=..., output=...) (a value produced by an upstream step). This wiring is what appligator turns into xcom_pull calls and task dependencies in the generated DAG.

Each step also has an explicit Pydantic output model, no untyped dicts passed between steps:

FetchResult        — output of fetch_data step
AggregateResult    — output of aggregate_data step (only fields needed by publish)
PublishResult      — output of publish_data step

User inputs

Parameter Format Default Description
start_date date — Start of the time range to process
end_date date — End of the time range to process
bbox bbox [-180, -90, 180, 90] Spatial extent. Defaults to global; use a smaller bbox for demos
output_prefix string smos-sm/global S3 key prefix for staging and ARD stores. Use a unique prefix for demo/test runs to avoid writing to the production cube
stac_s3_bucket string s3://deep-esdl-public/stac/smos-sm/global/ S3 bucket for the STAC catalog and deep-code user storage. Defaults to the ARD cube bucket when not set

Running the workflow

# Validate wiring only (no credentials needed)
python -c "import workflow"

# Full global production run
python workflow.py

# Quick demo run (small bbox, isolated prefix)
python -c "
from workflow import process_pipeline
process_pipeline.run(
    start_date='2026-01-06',
    end_date='2026-01-07',
    bbox=[10, 50, 12, 52],
    output_prefix='smos-sm/demo',
)
"

The __main__ block in workflow.py is pre-configured with a small demo bbox and the smos-sm/demo prefix for convenience.

Generating the Airflow DAG from the registry

appligator reads the ProcessRegistry from smos_ard.workflow, converts each workflow to an intermediate representation, and renders Python DAG files for Airflow 3+.

pixi run appligator smos_ard.workflow:registry \
    --dags-folder /opt/airflow/dags \
    --image-name earthcode/smos-ard-pipeline:latest \
    --secret-name earthcode-credentials

--secret-name (repeatable) injects a Kubernetes secret as environment variables into every pod via env_from. Pass it multiple times to inject more than one secret. Omit it to generate a DAG without secret injection (useful for local testing with --skip-build).

Airflow DAG

The active DAG (dags/smos-ard-service-v2.py) uses KubernetesPodOperator. Each step runs run_step.py as the container entrypoint, which:

  1. Reads STEP_CONFIG (static), STEP_PARAMS (DAG params), STEP_XCOM (upstream xcom)
  2. Merges inputs and dynamically calls the target function
  3. Writes the return dict to /airflow/xcom/return.json

For deployment instructions, secret setup, trigger commands, and scheduled run behaviour see deployment_guide.md.

Development

# Install dev environment (requires pixi: https://pixi.sh)
pixi install -e dev

# Run all checks (lint + typecheck + tests)
pixi run -e dev check

# Run tests only
pixi run -e dev test

ARD Cube Details

  • Format: Zarr v2, consolidated metadata
  • Default coverage: Global ([-180, -90, 180, 90])
  • Default aggregation: Daily mean (1D)
  • Chunk sizes: {time: 1, lat: 2048, lon: 2048}
  • Write mode: Initial write on first run; atomic append via zappend on subsequent runs; skipped if all timesteps already present
  • Variables: Soil_Moisture, Chi_2, Chi_2_P, N_RFI_X, N_RFI_Y, RFI_Prob, Soil_Moisture_DQX
  • Production location: s3://earthcode-ard-cubes/ard/smos-sm/global/cube.zarr