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)

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)

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)

Insert step at position idx (0-based).

Returns self.

Parameters:

• idx (int)

• label (str)

• step (Step)

Return type:

Pipeline

remove(label)

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)

Replace the step labelled label with step.

Returns self.

Parameters:

• label (str)

• step (Step)

Return type:

Pipeline

clone()

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)

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.

Return type:

PipelineResult

classmethod from_yaml(path)

Load a pipeline from a YAML config file.

Parameters:

path (str | Path)

Return type:

Pipeline

classmethod from_json(path)

Load a pipeline from a JSON config file.

Parameters:

path (str | Path)

Return type:

Pipeline

classmethod from_py(path)

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)

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

See also

pycsamt.emtools.pipe.preset_catalogue

to_yaml_string()

Serialise this pipeline to a YAML string.

Return type:

str

to_yaml(path)

Write this pipeline config to path as YAML.

Parameters:

path (str | Path)

Return type:

None

to_json(path)

Write this pipeline config to path as JSON.

Parameters:

path (str | Path)

Return type:

None

describe()

Return a pandas.DataFrame describing the pipeline steps.

Return type:

Any

steps_in_category(category)

Return steps belonging to category.

Parameters:

category (str)

Return type:

list[tuple[str, Step]]

class pycsamt.pipeline._pipeline.PipelineResult(sites_in, sites_out, step_results, outdir, elapsed_sec, processed_paths=<factory>, pipeline_name='pipeline')

Bases: object

Return value of Pipeline.run().

Variables:

• sites_in (Any) – Original Sites passed to Pipeline.run().

• sites_out (Any) – Fully processed Sites after all steps.

• step_results (list[pycsamt.pipeline._steps.StepResult]) – One StepResult per pipeline step, in order.

• outdir (pathlib.Path | None) – Root output directory (None when no output was requested).

• elapsed_sec (float) – Total wall-clock time for the run.

• processed_paths (list[pathlib.Path]) – Paths of EDI files written to <outdir>/processed/.

• pipeline_name (str) – Name label of the pipeline.

Parameters:

• sites_in (Any)

• sites_out (Any)

• step_results (list[StepResult])

• outdir (Path | None)

• elapsed_sec (float)

• processed_paths (list[Path])

• pipeline_name (str)

sites_in: Any

sites_out: Any

step_results: list[StepResult]

outdir: Path | None

elapsed_sec: float

processed_paths: list[Path]

pipeline_name: str = 'pipeline'

property plots: list[Path]

All figure paths generated across every step.

property n_errors: int

Number of steps that raised an error.

property ok: bool

True when every step completed without error.

summary()

Return a compact text summary of the run.

Return type:

str