Merge pull request #2 from deepesdl/tejas-xxx-decouple-publishing-to-osc

decouple publishing to osc

Author: TejasMorbagal

.github/workflows/ci.yml

@@ -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

.gitignore

@@ -7,4 +7,4 @@ __pycache__/
 notebooks/
 .gitaccess
 docker/src/
-k8s/secret.local.yaml
+k8s/

README.md

@@ -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
 ```
 
@@ -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
 
@@ -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:
@@ -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
 
@@ -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
 

changes.md

@@ -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