Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
name: CI

on:
push:
pull_request:

jobs:
test:
name: Unit tests + DAG validation
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4

- uses: actions/setup-python@v5
with:
python-version: "3.11"
cache: pip

- name: Install package and test deps
run: |
python -m pip install --upgrade pip
pip install -e .
pip install pytest

# tests/test_workflow.py imports smos_ard.workflow, which imports
# procodile — a local-path development dependency that is not yet on
# PyPI/conda. Exclude it from CI until procodile is publishable.
- name: Run tests + DAG import check
run: pytest tests/ --ignore=tests/test_workflow.py -v
2 changes: 1 addition & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -7,4 +7,4 @@ __pycache__/
notebooks/
.gitaccess
docker/src/
k8s/secret.local.yaml
k8s/
168 changes: 34 additions & 134 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,27 +17,36 @@ All core pipeline components are implemented and validated end-to-end.
| 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 | deep-code creates STAC collection and opens a GitHub PR on the OSC metadata repo |
| Demo run | validated | Global cube produced for a one-week time window |
| 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/
├── build_docker_template.py # generates docker/ build context via appligator
├── pixi.toml # environment + dependency management
├── pyproject.toml # package metadata, ruff/mypy/pytest config
├── run_step.py # container entrypoint (reads STEP_CONFIG/PARAMS/XCOM)
├── 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: STAC item → deep-code → OSC
│ ├── 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_update_dag.py # hand-written Airflow DAG (copy to Airflow DAGs folder)
│ └── main_step.py # auto-generated DAG (produced by appligator)
│ ├── 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
├── k8s/
│ ├── rbac.yaml # Airflow service account role + binding for earthcode namespace
│ └── secret.yaml # template for earthcode-secrets K8s secret
├── notebooks/
│ └── demo_update_service.ipynb # interactive demo of the update service
├── tests/ # pytest unit tests
├── .github/workflows/ci.yml # CI: lint, typecheck, tests on push/PR
└── docker/ # generated build context — do not edit or commit
```

Expand All @@ -47,7 +56,8 @@ smos-ard-pipeline/
|---|---|---|
| 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` | Builds STAC collection + OGC API records via deep-code, opens a GitHub PR on the OSC metadata repo |
| 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

Expand All @@ -61,38 +71,28 @@ smos-ard-pipeline/

```
start_date, end_date, bbox, output_prefix, stac_s3_bucket
fetch_smos_data → S3: staging/{prefix}/raw.zarr
│ FetchResult
│ FetchResult
aggregate_datacube → S3: ard/{prefix}/cube.zarr
│ AggregateResult │
▼ ▼
publish_to_catalogue → GitHub PR on OSC metadata repo
│ → S3: {stac_s3_bucket}/stac/{prefix}/
│ AggregateResult
publish_to_catalogue → S3: catalogue-pending/{collection_id}/
│ dataset-config.yaml
│ workflow-config.yaml
│ manifest.json
│ PublishResult
pull_request_url
```

### Local container test
catalogue_pending_uri

`run_step.py` takes a single JSON argument via `sys.argv[1]`. To test a step locally without Airflow or Kubernetes:
── separate step, run manually or via a trigger ──

```bash
docker run --entrypoint sh -it quay.io/earthcode/smos-ard-pipeline:0.0.5
```

Inside the container, write the payload to a file (avoids shell quoting/newline issues) and invoke `main` directly:

```bash
echo '{"func_module":"smos_ard.workflow","func_qualname":"pipeline_params","inputs":{"start_date":"2025-01-01","end_date":"2025-01-02","bbox":"[-180,-90,180,90]","output_prefix":"test/demo","stac_s3_bucket":null},"output_keys":["start_date","end_date","bbox","output_prefix","stac_s3_bucket"]}' > /tmp/p.json
python -c "from run_step import main; import json; main(**json.load(open('/tmp/p.json')))" && cat /airflow/xcom/return.json
publish_trigger.py → GitHub PR on OSC metadata repo
(reads pending/) → S3: catalogue-published/{collection_id}/{timestamp}/
```

`pipeline_params` is a plain passthrough function with no external dependencies — safe to run without S3 or CREODIAS credentials. Replace `func_qualname` with any other step to test it (credentials required for `fetch_data` onwards).

## procodile Workflow (`workflow.py`)

`workflow.py` is the single source of truth for the pipeline. It uses the [procodile](https://github.com/deepesdl/procodile) framework to declare the pipeline as a typed, inspectable workflow rather than a plain Python script. Three procodile constructs do the work:
Expand All @@ -117,7 +117,7 @@ PublishResult — output of publish_data step
| `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` | `null` | S3 bucket for the STAC catalog and deep-code user storage. Defaults to the ARD cube bucket when not set |
| `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

Expand Down Expand Up @@ -157,112 +157,12 @@ pixi run appligator smos_ard.workflow:registry \

## Airflow DAG

The hand-written DAG in `dags/smos_ard_update_dag.py` uses `KubernetesPodOperator`. Each step runs `run_step.py` as the container entrypoint, which:
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`

### Trigger via Airflow CLI

```bash
# Production global run
airflow dags trigger smos_ard_update_service --conf '{
"start_date": "2024-01-01",
"end_date": "2024-01-07"
}'

# Demo run with small bbox and isolated prefix
airflow dags trigger smos_ard_update_service --conf '{
"start_date": "2024-01-01",
"end_date": "2024-01-02",
"bbox": [10, 50, 12, 52],
"output_prefix": "smos-sm/demo"
}'

# Run with STAC catalog written to a separate bucket
airflow dags trigger smos_ard_update_service --conf '{
"start_date": "2024-01-01",
"end_date": "2024-01-07",
"stac_s3_bucket": "my-stac-catalog-bucket"
}'
```

### Trigger via REST API

```bash
curl -X POST http://localhost:8080/api/v1/dags/smos_ard_update_service/dagRuns \
-H "Content-Type: application/json" \
-d '{"conf": {"start_date": "2024-01-01", "end_date": "2024-01-07"}}'
```

## Setup

### 1. Build and push the image

The `docker/` directory is the build context, generated by appligator. It contains a two-stage Dockerfile (pixi build stage → lean Debian runtime), a `pixi.toml`, a `pyproject.toml`, and a locked `pixi.lock`.

**Why `pixi.toml` and `pyproject.toml` are needed alongside `pixi.lock`:**
- `pixi.toml` — pixi unconditionally requires a manifest to run; `pixi install --locked` won't start without one. The lock file has all resolved packages, but pixi still needs the manifest to know the project name, channels, and which environments exist. The `docker/pixi.toml` is a minimal stub that satisfies this requirement.
- `pyproject.toml` — needed by the `pip install --no-deps -e .` step that installs the `smos_ard` package from the copied `src/`. `smos-ard-pipeline` cannot be declared in `pixi.toml` as a `path = "."` dependency and use `--locked`, because pixi hashes the local source when generating the lock file on the host; in Docker the path content differs, so pixi reports the lock as out-of-date and refuses to install. Installing it separately with pip after `pixi install --locked` sidesteps this.

`procodile` and `gavicore` exist on PyPI but the latest changes needed by this pipeline are only on the `main` branch and not yet in a published release, so the Dockerfile installs them directly from the [`eo-tools/eozilla`](https://github.com/eo-tools/eozilla) git repo during the build stage. The `smos_ard` package is installed from `src/` which is copied in from the project root.

The build context is `docker/`. Since `src/` lives at the project root it must be copied in before building (`docker/src/` is gitignored):

```bash
# Copy smos_ard source into the build context
cp -r src docker/src

# Build and push
docker build -t earthcode/smos-ard-pipeline:0.0.1 docker/
docker push earthcode/smos-ard-pipeline:latest
```

### 2. Apply Kubernetes RBAC

The Airflow service account (`airflow-sa` in the `airflow` namespace) needs permission to create and manage pods in the `earthcode` namespace. Apply the role and binding once:

```bash
kubectl apply -f k8s/rbac.yaml
```

### 3. Create Kubernetes secret

Every pod receives these environment variables from the `earthcode-secrets` secret:

| Variable | Required | Used by | Description |
|---|---|---|---|
| `AWS_S3_ACCESS_KEY_ID` | yes | fetch, aggregate, publish | AWS credentials for reading/writing S3 |
| `AWS_S3_SECRET_ACCESS_KEY` | yes | fetch, aggregate, publish | AWS credentials for reading/writing S3 |
| `AWS_DEFAULT_REGION` | no | fetch, aggregate, publish | AWS region; defaults to `eu-central-1` |
| `CREODIAS_S3_KEY` | yes | fetch | CREODIAS S3 key for the xcube-smos data store |
| `CREODIAS_S3_SECRET` | yes | fetch | CREODIAS S3 secret for the xcube-smos data store |
| `GITHUB_USERNAME` | yes | publish | GitHub username for opening the OSC catalogue PR |
| `GITHUB_TOKEN` | yes | publish | GitHub PAT with `repo` scope for opening the OSC catalogue PR |

> `publish` also needs `S3_USER_STORAGE_{BUCKET,KEY,SECRET}` (expected by deep-code). `KEY` and `SECRET` are derived at runtime from `AWS_S3_ACCESS_KEY_ID` / `AWS_S3_SECRET_ACCESS_KEY`. `BUCKET` defaults to the ARD cube bucket but can be overridden via the `stac_s3_bucket` DAG parameter when the STAC catalog should live in a separate bucket.

```bash
kubectl create secret generic earthcode-secrets \
--namespace=earthcode \
--from-literal=AWS_S3_ACCESS_KEY_ID=<key> \
--from-literal=AWS_S3_SECRET_ACCESS_KEY=<secret> \
--from-literal=AWS_DEFAULT_REGION=eu-central-1 \
--from-literal=CREODIAS_S3_KEY=<key> \
--from-literal=CREODIAS_S3_SECRET=<secret> \
--from-literal=GITHUB_USERNAME=<user> \
--from-literal=GITHUB_TOKEN=<token>
```

### 4. Deploy the DAG

Airflow picks up DAGs from the S3 bucket `s3://airflow-dags-bc/earthcode/`. Copy the generated DAG there to make it visible in the Airflow UI:

```bash
aws s3 cp dags/main_step.py s3://airflow-dags-bc/earthcode/
```

Airflow polls the bucket periodically, so the DAG will appear within a minute or two.
For deployment instructions, secret setup, trigger commands, and scheduled run behaviour see [deployment_guide.md](deployment_guide.md).

## Development

Expand Down
56 changes: 56 additions & 0 deletions changes.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
# Changelog

## v0.1 — Initial Release

### Data Pipeline

- **Fetch** (`smos_ard.fetch`): opens SMOS L2C soil moisture (`SMOS-L2C-SM`) and ocean
salinity (`SMOS-L2C-OS`) data from the CREODIAS EODATA archive via xcube-smos; writes
raw orbital-pass data to a staging Zarr store on S3
(`s3://{bucket}/staging/{prefix}/raw.zarr`)

- **Aggregate** (`smos_ard.aggregate`): reads staging Zarr, deduplicates and sorts the
up-to-29 orbital passes per day, resamples to configurable temporal periods (default
daily mean), applies physical-range QC masks (`Soil_Moisture ∈ [0, 1]` m³/m³;
`SSS_corr ∈ [0, 45]` PSU), enriches variable metadata with long names and GCMD keyword
URLs, and writes/appends the ARD cube to S3 (`s3://{bucket}/ard/{prefix}/cube.zarr`)
using zappend for atomic incremental updates

- **Publish** (`smos_ard.publish`): generates `dataset-config.yaml`,
`workflow-config.yaml`, and `manifest.json` from pipeline metadata and stages them to
`s3://{bucket}/catalogue-pending/{collection_id}/`; no deep-code dependency in the
pipeline image

- **Publish trigger** (`smos_ard.publish_trigger`): standalone CLI that reads staged
artefacts from S3, calls `deep_code.tools.publish.Publisher` to fork the OSC catalogue
repo and open a GitHub PR, then archives the artefacts to
`catalogue-published/{collection_id}/{timestamp}/` on success

### Orchestration

- Airflow DAG (`dags/smos-ard-service-v2.py`) generated by appligator from the procodile
`ProcessRegistry` in `workflow.py`; runs daily at 23:00 UTC via
`KubernetesPodOperator` on the `earthcode` namespace
- `run_step.py` container entrypoint: merges `STEP_CONFIG`, `STEP_PARAMS`, and
`STEP_XCOM` environment variables, dynamically dispatches to the target function, and
writes the return dict to `/airflow/xcom/return.json`
- `start_date` and `end_date` derived automatically from Airflow's `{{ ds }}` logical
date on scheduled runs; manual overrides supported via DAG params

### ARD Cube

- Format: Zarr v2, consolidated metadata
- Default spatial coverage: global (`[-180, -90, 180, 90]`)
- Default temporal aggregation: daily mean (`1D`)
- Chunk layout: `{time: 1, lat: 2048, lon: 2048}`
- Write modes: 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`

### Bug Fixes

- Fixed alternate-day append failures caused by `_validate_append_compat` running before
`_clip_to_new_times`: the boundary day (`new_min == existing_max`) was incorrectly
rejected before the overlap could be stripped
Loading
Loading