ALTIMETRY PROCESSING PIPELINE

Monorepo for the independent stages of the altimetry processing pipeline used to generate NASA SSH products.

Pipeline consists of:

Daily along track generation

Daily files -> Crossover -> OER -> Crossover -> Bad pass flagging -> Finalization

Additional product generation

Simple grids -> Indicators -> Imagery for website

                  GSFC Data     S6 Data
                      |            |
                      +------------+
                            |
                            v
                [ Generate Daily Files ]
                            |   
                            v   
                [ Generate Simple Grids ]
                            |  
                            v  
             +--------------+--------------+ 
             |                             | 
             v                             v 
[ Generate ENSO Maps & Imagery]  [ Generate Indicators ]

Description

The directories contained within pipeline/ contain everything for a given stage in the pipeline (note: stage as opposed to step, as the crossover stage is used for two steps). Each stage's directory can be thought of as an independent repository for that stage.

Each stage has been designed to be run on AWS Lambda, and as such there are limitations to how much can be executed locally.

Step Function Architecture

The pipeline is orchestrated by AWS Step Functions using JSONata query language. State machine definitions live in state_machines/ as ASL (Amazon States Language) JSON files. The rendered/ subdirectory contains the same definitions with AWS account/region placeholders resolved.

Hierarchy

A top-level orchestrator (pipeline.asl.json) coordinates three sub-pipelines, each of which delegates per-date work to Distributed Map states that fan out to Lambda functions:

pipeline.asl.json
├── along_track_pipeline.asl.json
│   ├── pipeline_init (Lambda — determines which dates need processing)
│   ├── daily_file.asl.json (Distributed Map → daily_files Lambda)
│   ├── xover.asl.json (Distributed Map → xover Lambda, df_version=p1)
│   ├── oer.asl.json (Distributed Map → oer Lambda)
│   ├── xover.asl.json (Distributed Map → xover Lambda, df_version=p2)
│   ├── bad_pass.asl.json (Distributed Map → bad_pass Lambda)
│   └── finalizer.asl.json (Distributed Map → finalizer Lambda)
├── unifier.asl.json (conditional — runs if source has unify=true)
│   ├── Distributed Map → unifier Lambda
│   └── rewrite_manifest (Lambda — rewrites jobs manifest with NASA-SSH source)
└── simple_grid_pipeline.asl.json
    ├── set_sg_jobs (Lambda — filters manifest to Monday dates)
    ├── simple_grid.asl.json (Distributed Map → simple_grids Lambda)
    ├── enso.asl.json (Distributed Map → enso Lambda)
    └── indicators (Lambda)

Key patterns

• Jobs manifest: pipeline_init writes a jobs.json to S3 listing dates that need processing. All downstream Distributed Maps read from this manifest via ItemReader.
• Distributed Map: Most processing stages use Distributed Map (mode DISTRIBUTED, max concurrency 100) to process each date independently. Results are written to S3 via ResultWriter.
• Input threading: Orchestrator states use Output: "{% $states.input %}" to pass the original input (jobs_key, bucket, source) through to the next state, since child state machine outputs aren't needed upstream.
• Two crossover passes: The xover state machine is invoked twice — once with df_version=p1 (before OER) and once with df_version=p2 (after OER).
• Conditional unification: Sources with unify=true (GSFC, S6) get their finalized daily files copied to a unified NASA-SSH prefix by the unifier. The rewrite_manifest Lambda then produces a new jobs manifest under the NASA-SSH source for downstream simple grid processing.

For a detailed reference of every S3 object written by each stage (success and failure), key patterns, and operator troubleshooting tips, see S3_DATA_FLOW.md.

Source Configuration

Each source has one profile file, utilities/sources/{source}.yaml, holding a shared common: block plus one section per pipeline stage. A stage's config loader (load_source_config) merges common with that stage's section and constructs the stage's config dataclass, so each stage sees only the fields it declares.

Shared block: common:

Source-identity metadata used by multiple stages:

• product_type — reference or high_latitude; determines the daily-file filename family (alt_ref_at_* vs alt_hilat_at_*) and which downstream stages apply
• discovery_type — how pipeline_init finds upstream granules (cmr, thredds, …)
• unify — whether the source participates in NASA-SSH unification
• start_date — first date with available data
• end_date (optional) — last date with available data; omit for ongoing collections
• collections — the upstream collection descriptors discovery reads

Stage sections

Each stage that a source participates in gets its own section (pipeline_init:, daily_files:, finalizer:, xover:, …) carrying only that stage's settings.

The xover: section selects the crossover type (see CONTEXT.md):

• crossover_type: self — a reference product-type source (S6, S6B, GSFC) crossed against its own passes over a forward window. Fields: cycle_length, window_size, window_padding, max_pass_number.
• crossover_type: reference — a high_latitude source (S3B) crossed against the finalized NASA-SSH P3 reference mission over a window centered on the processing day. Adds reference_source and reference_version (see ADR-0006).

Example (utilities/sources/S3B.yaml, abbreviated):

common:
  product_type: high_latitude
  discovery_type: thredds
  start_date: "2018-04-25"

xover:
  crossover_type: reference
  reference_source: NASA-SSH
  reference_version: p3
  cycle_length: 9.9156
  window_size: 12      # interpreted as a ±centered window for reference crossovers
  window_padding: 2
  max_pass_number: 9999

When adding a new source, create utilities/sources/{source}.yaml with a common block and a section for each stage it participates in.

Setup

Dependencies are declared in the root pyproject.toml (the single source of truth) and locked in uv.lock. Each stage is a named optional extra; the dev extra installs the union of them all plus the test/lint tooling. Use uv to create the dev environment:

uv sync --extra dev

This installs the shared utilities package (required to execute containers and unit tests) along with every stage's dependencies.

Running tests:

Tests run one process per stage via scripts/test.sh — each stage is rooted at its own directory so its code resolves as a top-level import, mirroring the Lambda container. See the script header for the rationale.

./scripts/test.sh              # all stages
./scripts/test.sh oer xover    # only the named stages
./scripts/test.sh oer -- -k foo -x   # pass args after `--` through to pytest

Linting:

uv run ruff check .            # config lives in [tool.ruff] in pyproject.toml

Building images:

Images must be built from the root directory context. Each stage Dockerfile exports its extra's locked dependencies with uv and installs the shared utilities package, so no per-stage requirements.txt is needed.

ex:

docker buildx build --platform linux/amd64 --load -t daily_files:latest -f path/to/nasa-ssh-pipeline/pipeline/daily_files/Dockerfile {path/to/nasa-ssh-pipeline/}