2.27.2.1. pycsamt.pipeline._pipeline#

Core Pipeline class for the pyCSAMT processing engine.

A Pipeline is an ordered sequence of Step objects that is applied to a Sites collection in one call to Pipeline.run().

Design goals#

scikit-learn-style API — steps are stored as (label, Step) tuples; the pipeline prints as a formatted table before running.
Immutable-while-running — once run() is called the step list is frozen until the run completes.
Runs to completion — errors are collected per step; execution continues unless on_step_error="raise" is set in PYCSAMT_PIPE.
Reproducible — the canonical YAML config is saved to the output directory alongside the results.

Examples

Build and run from code:

from pycsamt.emtools.pipe import Pipeline, Step

pipe = Pipeline(
    [
        ("notch", Step("NR001", mains_hz=50)),
        ("band", Step("FREQ001")),
        ("align", Step("FREQ004")),
        ("correct_ss", Step("SS001")),
        ("rotate", Step("TZ001")),
    ]
)
print(pipe)
result = pipe.run(sites, outdir="willy_results/")

Build from preset and customise:

pipe = Pipeline.from_preset("full_processing")
pipe.remove("mask_skew")
pipe.append("skew_band", Step("SK002", threshold=0.25))

Build from a YAML config:

pipe = Pipeline.from_yaml("processing/willy_workflow.yaml")
result = pipe.run(sites)

Classes

`Pipeline`(steps, *[, name, _output_dir])	An ordered, configurable MT processing pipeline.
`PipelineResult`(sites_in, sites_out, ...[, ...])	Return value of `Pipeline.run()`.

class pycsamt.pipeline._pipeline.Pipeline(steps, *, name='pipeline', _output_dir=None)[source]

Bases: PipelineBase

An ordered, configurable MT processing pipeline.

Parameters:

steps (list[tuple[str, Step]] | list[Step]) – Either a list of (label, Step) tuples or a plain list of Step objects (labels are auto-generated from the step name).
name (str)
_output_dir (str | Path | None)

Examples

>>> from pycsamt.emtools.pipe import Pipeline, Step
>>> pipe = Pipeline(
...     [
...         ("notch", Step("NR001")),
...         ("band", Step("FREQ001", band_hz=(0.001, 10000))),
...         ("align", Step("FREQ004")),
...     ]
... )
>>> print(pipe)

append(label, step)[source]

Add step to the end of the pipeline.

Returns self so calls can be chained.

Parameters:

label (str)
step (Step)

Return type:

Pipeline

insert(idx, label, step)[source]

Insert step at position idx (0-based).

Returns self.

Parameters:

idx (int)
label (str)
step (Step)

Return type:

Pipeline

remove(label)[source]

Remove the first step whose label matches label.

Returns self.

Raises:: KeyError – When no step with that label exists.
Parameters:: label (str)
Return type:: Pipeline

replace(label, step)[source]

Replace the step labelled label with step.

Returns self.

Parameters:

label (str)
step (Step)

Return type:

Pipeline

clone()[source]

Return a deep copy of this pipeline.

Return type:: Pipeline

run(sites, *, outdir=<object object>, save_plots=True, save_edis=True, save_report=True, api=None, cache=False, on_step=None, history=False)[source]

Run all steps in order and return a PipelineResult.

Parameters:

sites (Any) – Input Sites collection.
outdir (Any) – Root output directory. Falls back first to the pipeline’s own _output_dir (set by from_yaml), then to output_root.
save_plots (bool) – Generate and save QC figures for each step.
save_edis (bool) – Write processed EDI files after the final step.
save_report (bool) – Write HTML and/or text reports to outdir.
api (Any) – Optional PipelineAPIConfig override. When None the global singleton is used.
cache (bool | str | Path) – False (default) — no caching, identical behavior to every prior release. True — cache each step’s output under cache_root. A path — cache under that directory instead. A rerun of the identical call replays every already-completed step from cache instead of recomputing it — this is also how an interrupted run “resumes.” See Caching And Resume for what is and isn’t safe to cache.
on_step (Callable[[StepResult], None] | None) – Optional callable invoked once per step, immediately after that step’s StepResult is built — for a fresh computation, a cache hit, or a warn/skip error alike (a step that raises under on_step_error="raise" never gets a StepResult and so never fires this either, same as today). An exception raised by on_step itself is caught and warned, never allowed to corrupt or abort the run. This is the hook a notebook, a script, or a future GUI integration would use to observe a run in progress — see Live Observability.
history (bool | str | Path) – False (default) — no history logged. True — append a compact one-line JSON summary of this run to history_path (default ~/.pycsamt/pipeline_history.jsonl). A path — log to that file instead. Read back with load_history().

Return type:

PipelineResult

classmethod from_yaml(path)[source]

Load a pipeline from a YAML config file.

Parameters:: path (str | Path)
Return type:: Pipeline

classmethod from_json(path)[source]

Load a pipeline from a JSON config file.

Parameters:: path (str | Path)
Return type:: Pipeline

classmethod from_py(path)[source]

Load a pipeline from a Python config file.

The file must expose a pipeline_config dict.

Parameters:: path (str | Path)
Return type:: Pipeline

classmethod from_preset(name, pipeline_name=None)[source]

Build a pipeline from a named preset.

Parameters:

name (str) – Preset identifier, e.g. "full_processing".
pipeline_name (str | None) – Optional override for the pipeline label.

Return type:

Pipeline