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.

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.