Skip to content

Pipelines

A single algorithm answers one question. A real analysis is a sequence - read, normalise, measure, fit, plot. A Pipeline is that sequence made explicit, repeatable and shareable.

Building a pipeline

The fluent PipelineBuilder chains steps:

from spectro_kernel import PipelineBuilder, WorkContext

pipeline = (
    PipelineBuilder()
    .named("balmer analysis")
    .add("normalize_polynomial", order=3)
    .add("snr_der")
    .add("fit_gaussian_line", line_center_angstrom=6562.8, window_angstrom=30)
    .build()
)

result = pipeline.execute(WorkContext(spectrum=spec))

build() validates every referenced algorithm against the registry, so a typo fails immediately - not halfway through a run.

Executing it

execute threads one WorkContext through every step:

flowchart LR
    CTX["WorkContext"] --> S1["step 1<br/>normalize_polynomial"]
    S1 --> S2["step 2<br/>snr_der"]
    S2 --> S3["step 3<br/>fit_gaussian_line"]
    S3 --> RES["PipelineResult"]
    S1 -.records.-> H["ctx.history"]
    S2 -.records.-> H
    S3 -.records.-> H

It returns a PipelineResult:

result.success        # did every step succeed?
result.failed_at      # index of the first failing step, or None
result.outputs        # the AlgorithmOutput of each step
result.context        # the final WorkContext, with all results
result.history        # the full audit trail
result.to_dict()      # a JSON-safe summary of the whole run

By default execution stops at the first failure; pass stop_on_error=False to run every step regardless.

Presets - pipelines as YAML

A preset is a pipeline written as a YAML file: a named, versioned, shareable recipe.

# balmer_quick.yaml
name: balmer_quick
description: Quick analysis of the Balmer lines on a loaded spectrum.
version: 1.0.0
steps:
  - name: Normalise continuum
    algorithm: normalize_polynomial
    params: { order: 3, sigma_clip: 3.0 }
  - name: Compute SNR
    algorithm: snr_der
  - name: Fit H-alpha
    algorithm: fit_gaussian_line
    params: { line_center_angstrom: 6562.79, window_angstrom: 30.0, label: H-alpha }

Load and run it:

pipeline = PipelineBuilder().from_preset("balmer_quick").build()
result = pipeline.execute(WorkContext(spectrum=spec))

…or from the command line:

spectro presets                                  # list bundled presets
spectro pipeline balmer_quick --input obs.fits    # run one

Presets are why an analysis can be cited and reproduced: "this measurement used the balmer_quick v1.0.0 preset" is a precise, checkable statement. Since v0.7.0 a preset can declare variables (the instrument- or observer-dependent values) that a profile supplies at run time, and every run closes with a pipeline:<name> record in the history naming the recipe, its version and the bound values. The Recipes section explains the idea and the format page is the reference.

Reproducibility

Reproducibility is built in, not bolted on. Three mechanisms combine:

  1. Per-algorithm versions. Every algorithm carries a version. A behaviour change means a version bump, so an old result is never silently re-interpreted.
  2. The audit trail. Every step records its algorithm, version, parameters, input/output hashes and duration into ctx.history.
  3. State hashing. WorkContext.hash_state() and Spectrum1D.content_hash() give stable digests of the scientific content, excluding history - so an identical input run with identical parameters yields an identical hash.

Together: given a result, you can read exactly which algorithms and versions produced it, with which parameters, and replay the chain.