pycsamt.pipeline#

Declarative processing pipelines, step registries, presets, output handling, and report generation.

pyCSAMT processing pipeline.

A declarative, automated MT processing engine that chains emtools operations from raw Sites to publication-ready output.

This package is the single entry point for all pipeline-related symbols. The processing engine lives in the private sub-modules of this package (_registry, _steps, _pipeline, …). Runtime configuration is provided by pycsamt.api.pipe and re-exported here so users only need one import location.

Typical usage#

Build and run from code:

from pycsamt.pipeline 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/")

# shortest path — Pipeline and Step are also on pycsamt directly
from pycsamt import Pipeline, Step

Load from a config file:

pipe = Pipeline.from_yaml("config/workflow.yaml")
pipe = Pipeline.from_preset("publication_ready")

Discover available steps and presets:

from pycsamt.pipeline import list_steps, preset_catalogue
list_steps()                  # all 33 StepSpec objects
list_steps("noise_removal")   # by category
print(preset_catalogue())     # named presets

Configure pipeline output globally:

from pycsamt.pipeline import configure_pipe
configure_pipe(plot_dpi=300, plot_fmt="pdf", output_root="results/")

# or temporarily with a context manager
from pycsamt.pipeline import PYCSAMT_PIPE
with PYCSAMT_PIPE.context(show_progress=False):
    result = pipe.run(sites)

class pycsamt.pipeline.PipelineBase#

Bases: object

Base class for pyCSAMT pipeline objects.

Inherit from this to gain registry introspection, to_py export, and scaffold template generation. Pipeline inherits this.

static available_steps(category=None)#

Return all registered StepSpec objects.

Parameters:: category (str | None) – When supplied, restrict to steps in that category (e.g. "noise_removal", "frequency").
Return type:: list

Examples

>>> Pipeline.available_steps()
>>> Pipeline.available_steps("static_shift")

static available_categories()#

Return a sorted list of step category names.

Examples

>>> Pipeline.available_categories()
['dimensionality', 'frequency', 'noise_removal', ...]

Return type:: list[str]

static step_info(code_or_name)#

Return a formatted info block for a single step.

Parameters:: code_or_name (str) – Registry code ("NR001") or snake-case name ("notch_powerline").
Return type:: str

Examples

>>> print(Pipeline.step_info("NR001"))
>>> print(Pipeline.step_info("correct_ss_ama"))

classmethod catalogue(category=None)#

Return a full formatted catalogue of all available steps.

Parameters:: category (str | None) – Restrict output to one category, or None for all.
Return type:: str

Examples

>>> print(Pipeline.catalogue())
>>> print(Pipeline.catalogue("tensor"))

to_py(path=None)#

Serialize this pipeline to a Python config script.

Produces a pipeline_config dict in a clean, readable .py file that can be edited and reloaded with from_py(). Steps are grouped by category and commented with their human-readable labels.

Parameters:: path (str | Path | None) – If given, write the output to this file path.
Returns:: The generated Python source as a string (always returned, whether or not path is provided).
Return type:: str

Examples

>>> src = pipe.to_py()
>>> print(src)
>>> pipe.to_py("config/my_workflow.py")

classmethod scaffold(path=None, *, fmt='yaml', preset=None, name='my_workflow', outdir='pipe_results')#

Generate a ready-to-edit starter pipeline config.

The scaffold contains: - Active steps from preset (or a sensible default set) - All other available steps commented out with descriptions

Parameters:

path (str | Path | None) – Write the output to this file. Extension is inferred from fmt if the path has none.
fmt (str) – Output format: "yaml" (default), "json", or "py".
preset (str | None) – Name of a built-in preset to use as the active step set (e.g. "basic_qc", "full_processing"). When None a minimal basic_qc-style set is used.
name (str) – Pipeline name written into the config.
outdir (str) – Default output directory written into the config.

Returns:

The generated config content as a string.

Return type:

str

Examples

>>> print(Pipeline.scaffold())
>>> Pipeline.scaffold("config/starter.yaml", preset="full_processing")
>>> Pipeline.scaffold("config/starter.py",   fmt="py")

class pycsamt.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.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

class pycsamt.pipeline.Step(code_or_name, **params)#

Bases: object

A configured pipeline step.

Parameters:

code_or_name (str) – Registry code (e.g. "NR001") or snake-case name (e.g. "notch_powerline"). Both forms are accepted.
**params (Any) – Keyword arguments forwarded to the underlying transform function. Values here are merged on top of the registry defaults.

Examples

>>> from pycsamt.emtools.pipe import Step
>>> Step("NR001", mains_hz=50)
Step [NR001] Power-line Harmonic Notch  (mains_hz=50, n_harm=30, tol_hz=0.08)
>>> Step("notch_powerline")
Step [NR001] Power-line Harmonic Notch  (mains_hz=50, n_harm=30, tol_hz=0.08)

transform(sites)#

Apply the step’s transform function to sites.

If the step has returns_sites=False (diagnostic-only), the input sites is returned unchanged.

Parameters:: sites (Any)
Return type:: Any

generate_qc_plots(sites)#

Call QC plot functions on sites and return (name, Figure) pairs.

Return values are normalised to matplotlib.figure.Figure so the pipeline can call savefig regardless of whether the underlying emtools function returned a Figure or an Axes object. Failures are silently skipped so that a broken QC plot never kills a successful processing run.

Parameters:: sites (Any)
Return type:: list

to_dict()#

Serialise step to a plain dict (used by to_yaml / to_json).

Return type:: dict

class pycsamt.pipeline.StepResult(step_idx, step_name, step_code, step_label, params, elapsed_sec, plots=<factory>, n_sites_in=0, n_sites_out=0, error=None)#

Bases: object

Immutable record produced after one pipeline step runs.

Variables:

step_idx (int) – 1-based position in the pipeline.
step_name (str) – User-supplied label for this step in the pipeline.
step_code (str) – Registry code (e.g. "NR001").
step_label (str) – Human-readable label from the registry.
params (dict) – Parameters that were passed to the transform function.
elapsed_sec (float) – Wall-clock time in seconds for this step (transform + QC plots).
plots (list[pathlib.Path]) – Paths of figures saved to disk during this step.
n_sites_in (int) – Number of sites fed into this step.
n_sites_out (int) – Number of sites coming out of this step.
error (Exception | None) – If the step raised an exception and execution continued (on_step_error != "raise"), the exception is stored here. None means the step completed without error.

Parameters:

step_idx (int)
step_name (str)
step_code (str)
step_label (str)
params (dict)
elapsed_sec (float)
plots (list[Path])
n_sites_in (int)
n_sites_out (int)
error (Exception | None)

step_idx: int#

step_name: str#

step_code: str#

step_label: str#

params: dict#

elapsed_sec: float#

plots: list[Path]#

n_sites_in: int = 0#

n_sites_out: int = 0#

error: Exception | None = None#

property ok: bool#: True when the step completed without error.

summary_line()#

Return a one-line text summary of this step result.

Return type:: str

class pycsamt.pipeline.StepSpec(code, name, label, category, defaults=<factory>, returns_sites=True, mod=None, fn_name=None, qc_defs=<factory>, override_fn=None)#

Bases: object

Immutable descriptor for one pipeline step.

Parameters:

code (str) – Short uppercase identifier, e.g. "NR001".
name (str) – Snake-case name, e.g. "notch_powerline".
label (str) – Human-readable label shown in pipeline __repr__.
category (str) – Logical group ("frequency", "noise_removal", "static_shift", "tensor", "dimensionality", "skew", "source_effects", "qc").
defaults (dict) – Keyword arguments passed to the transform function when not overridden by the user.
returns_sites (bool) – True – step transforms the Sites object (default). False – diagnostic-only step; the Sites pass through unchanged.
mod (str | None) – Dotted module path for the primary transform function. None when override_fn is provided.
fn_name (str | None) – Name of the transform function inside mod. None when override_fn is provided.
qc_defs (list[tuple[str, str]]) – List of (module_path, function_name) pairs. The pipeline calls these after each step to generate QC figures.
override_fn (Callable | None) – Direct callable override. When set, mod / fn_name are ignored.

code: str#

name: str#

label: str#

category: str#

defaults: dict#

returns_sites: bool = True#

mod: str | None = None#

fn_name: str | None = None#

qc_defs: list[tuple[str, str]]#

override_fn: Callable | None = None#

get_fn()#

Return (lazily imported) transform function.

Return type:: Callable

get_qc_fns()#

Return list of (fn_name, callable) QC plot functions.

Return type:: list[tuple[str, Callable]]

pycsamt.pipeline.lookup_step(code_or_name)#

Return the StepSpec for code_or_name.

Parameters:: code_or_name (str) – Either the uppercase code ("NR001") or the snake-case name ("notch_powerline").
Raises:: KeyError – When neither a code nor a name matches.
Return type:: StepSpec

pycsamt.pipeline.list_steps(category=None)#

Return all registered StepSpec objects, optionally filtered.

Parameters:: category (str | None) – When supplied, only steps whose category matches this string are returned.
Return type:: list[StepSpec]

pycsamt.pipeline.step_codes()#

Return a sorted list of all step codes.

Return type:: list[str]

pycsamt.pipeline.step_names()#

Return a sorted list of all step names.

Return type:: list[str]

pycsamt.pipeline.categories()#

Return a sorted list of distinct step categories.

Return type:: list[str]

class pycsamt.pipeline.Preset(name, description, steps=<factory>)#

Bases: object

A named, ordered collection of pipeline steps.

Variables:

name (str) – Identifier used in from_preset().
description (str) – One-line description shown when the user prints the preset catalogue.
steps (list[tuple[str, pycsamt.pipeline._steps.Step]]) – Ordered list of (label, Step) tuples — same format as a Pipeline step list.

Parameters:

name (str)
description (str)
steps (list[tuple[str, Step]])

name: str#

description: str#

steps: list[tuple[str, Step]]#

pycsamt.pipeline.get_preset(name)#

Return the Preset for name.

Raises:: KeyError – When name is not registered.
Parameters:: name (str)
Return type:: Preset

pycsamt.pipeline.list_presets()#

Return all available Preset objects.

Return type:: list[Preset]

pycsamt.pipeline.preset_catalogue()#

Return a formatted catalogue of all presets.

Return type:: str

pycsamt.pipeline.load_yaml(path)#

Parse a YAML pipeline config file and return the raw dict.

Parameters:: path (str | Path)
Return type:: dict[str, Any]

pycsamt.pipeline.load_json(path)#

Parse a JSON pipeline config file and return the raw dict.

Parameters:: path (str | Path)
Return type:: dict[str, Any]

pycsamt.pipeline.load_py(path)#

Import a Python config file and return its pipeline_config dict.

Parameters:: path (str | Path)
Return type:: dict[str, Any]

pycsamt.pipeline.plot_pipeline_dashboard(result, *, figsize=(12.0, 8.0))#

Create a compact dashboard for a completed pipeline run.

The dashboard combines run-level cards, per-step status, timing, station-count flow, and generated-figure counts. It is the most useful single figure to place in a processing report or notebook.

Returns:

Figure containing the dashboard.

Return type:

matplotlib.figure.Figure

Parameters:

result (PipelineResult)
figsize (tuple[float, float])

pycsamt.pipeline.plot_pipeline_status(result, *, ax=None, ok_color='#2f9e44', error_color='#c92a2a', annotate_errors=True)#

Plot per-step success status for a pipeline run.

Failed steps are coloured red and, when available, the exception message is written below the corresponding bar. Empty results draw a neutral placeholder instead of failing.

Parameters:

result (PipelineResult)
ax (Any | None)
ok_color (str)
error_color (str)
annotate_errors (bool)

Return type:

Any

pycsamt.pipeline.plot_pipeline_timing(result, *, ax=None, color='#2f6f8f', slow_color='#f08c00', slow_quantile=0.8, annotate=True)#

Plot elapsed time per pipeline step.

Steps above slow_quantile are highlighted, helping users quickly spot expensive processing or plotting stages.

Parameters:

result (PipelineResult)
ax (Any | None)
color (str)
slow_color (str)
slow_quantile (float)
annotate (bool)

Return type:

Any

pycsamt.pipeline.plot_site_count_flow(result, *, ax=None, color_in='#687582', color_out='#7c4d79', drop_color='#c92a2a', annotate=True)#

Plot station counts entering and leaving each pipeline step.

A faint red band marks steps where the output site count is lower than the input site count, which is useful for QC workflows that reject stations.

Parameters:

result (PipelineResult)
ax (Any | None)
color_in (str)
color_out (str)
drop_color (str)
annotate (bool)

Return type:

Any

class pycsamt.pipeline.StratagemPreset(name, description, survey_defaults=<factory>, steps=<factory>)#

Bases: object

Named configuration bundle for StratagemSurvey.

Used by run_stratagem_preset(). Each preset describes an ordered sequence of StratagemSurvey method calls with default keyword arguments.

Variables:

name (str)
description (str)
survey_defaults (dict) – Default keyword arguments for __init__.
steps (list of (str, dict)) – Ordered (method_name, kwargs) pairs executed on the survey.

Parameters:

name (str)
description (str)
survey_defaults (dict)
steps (list[tuple[str, dict]])

name: str#

description: str#

survey_defaults: dict#

steps: list[tuple[str, dict]]#

class pycsamt.pipeline.StratagemPipeline(steps, *, coord_file=None, raw_dir=None, epsg=32649, utm_zone='49N', order='auto', rename_basename=None, rename_dir=None, name='stratagem_pipeline')#

Bases: Pipeline

A Pipeline extended with Stratagem pre/post-processing.

Follows exactly the same run(sites, *, outdir=...) interface as the standard pipeline. The sites argument is normalised via ensure_sites(), so it accepts:

Sites
str / pathlib.Path pointing to a directory of EDIs
str / pathlib.Path pointing to a single EDI file
A single EDIFile object
A list of EDIFile objects

Optional pre-processing (applied before emtools steps)#

Coordinate injection — when coord_file is given, GPS coordinates from a CSV / XLS / XLSX table are written into each EDI >HEAD section via CoordinateInjector.
Hardware SNR mask — when raw_dir is given, zero-stack frequency rows in the Stratagem raw files are masked in the impedance tensor via FrequencyFilter.

Optional post-processing (applied after emtools steps)#

Rename — when rename_basename is given, the output EDI files in <outdir>/processed/ are copied to <outdir>/renamed/ with a standardised naming convention using EDIRenamer.

param steps:: emtools processing steps (same format as Pipeline).
type steps:: list of (str, Step) or list of Step
param coord_file:: GPS coordinate table (CSV / XLS / XLSX).
type coord_file:: path-like, optional
param raw_dir:: Directory of raw Stratagem hardware files for hardware SNR masking.
type raw_dir:: path-like, optional
param epsg:: EPSG code of the projected CRS in coord_file.
type epsg:: int, default 32649
param utm_zone:
type utm_zone:: str, default '49N'
param order:: Station ordering for StationLocator.
type order:: str, default 'auto'
param rename_basename:: When given, output EDIs are renamed {basename}000.edi, …
type rename_basename:: str, optional
param rename_dir:: Destination for renamed files. Defaults to <outdir>/renamed/.
type rename_dir:: path-like, optional
param name:
type name:: str, default 'stratagem_pipeline'

Examples

Load a directory and inject coordinates:

from pycsamt.pipeline.stratagem import StratagemPipeline
pipe = StratagemPipeline.from_preset(
    "stratagem_mt",
    coord_file="2.csv",
    epsg=32649,
    rename_basename="T2.",
)
result = pipe.run("2/2EDI", outdir="2/processed")
print(result.summary())

Single EDI file:

result = pipe.run("2/2EDI/Z2HX002.edi", outdir="tmp/single/")

Already-loaded Sites:

from pycsamt.emtools._core import ensure_sites
S = ensure_sites("2/2EDI")
result = pipe.run(S, outdir="2/processed")

Custom steps with hardware mask:

from pycsamt.pipeline import Step
pipe = StratagemPipeline(
    [("ss", Step("SS001")), ("band", Step("FREQ001", band_hz=(10, 1e5)))],
    coord_file="2.csv",
    raw_dir="原始数据/2HX",
    epsg=32649,
)
result = pipe.run("2/2EDI", outdir="out/")

run(sites, *, outdir=None, save_plots=True, save_edis=True, save_report=True, api=None, rename_basename=None, rename_dir=None, overwrite=False)#

Run the pipeline on sites.

Parameters:

sites (Sites | str | Path | EDIFile | list[EDIFile]) – Input data. Any form accepted by ensure_sites().
outdir (path-like, optional) – Root output directory (same as Pipeline.run()).
save_plots (bool, default True)
save_edis (bool, default True)
save_report (bool, default True)
api (PipelineAPIConfig, optional)
rename_basename (str, optional) – Override the rename_basename set in __init__.
rename_dir (path-like, optional) – Override the rename_dir set in __init__.
overwrite (bool, default False) – Overwrite existing renamed files.

Return type:

PipelineResult

classmethod from_preset(name='stratagem_mt', *, coord_file=None, raw_dir=None, epsg=32649, utm_zone='49N', order='auto', rename_basename=None, rename_dir=None, pipeline_name=None)#

Build a StratagemPipeline from a named emtools preset.

Parameters:

name (str, default 'stratagem_mt') – Any preset in PRESETS (e.g. 'stratagem_mt', 'full_processing', 'publication_ready').
coord_file (str | Path | None) – Forwarded to the constructor.
raw_dir (str | Path | None) – Forwarded to the constructor.
epsg (int) – Forwarded to the constructor.
utm_zone (str) – Forwarded to the constructor.
order (str) – Forwarded to the constructor.
rename_basename (str | None) – Forwarded to the constructor.
rename_dir (str | Path | None) – Forwarded to the constructor.
pipeline_name (str, optional) – Override the pipeline label.

Return type:

StratagemPipeline

Examples

>>> pipe = StratagemPipeline.from_preset(
...     "stratagem_mt",
...     coord_file="2.csv",
...     epsg=32649,
...     rename_basename="T2.",
... )
>>> result = pipe.run("2/2EDI", outdir="2/processed")

Parameters:

steps (list[tuple[str, Step]] | list[Step])
coord_file (str | Path | None)
raw_dir (str | Path | None)
epsg (int)
utm_zone (str)
order (str)
rename_basename (str | None)
rename_dir (str | Path | None)
name (str)

pycsamt.pipeline.get_stratagem_preset(name)#

Return the StratagemPreset for name.

Parameters:: name (str)
Return type:: StratagemPreset

pycsamt.pipeline.list_stratagem_presets()#

Return all StratagemPreset objects.

Return type:: list[StratagemPreset]

pycsamt.pipeline.run_stratagem_preset(preset, edi_dir, coord_file, outdir, *, raw_dir=None, epsg=32649, utm_zone='49N', rename_basename='S', rename_dir=None, step_overrides=None, overwrite=False, verbose=0)#

Execute a named Stratagem preset in one call (convenience wrapper).

Uses StratagemSurvey internally for the full EDI-directory + GPS-CSV workflow. For the pipeline-style API (run(sites, outdir=...)) use StratagemPipeline directly.

Parameters:

preset ({'basic', 'full_processing', 'publication_ready'})
edi_dir (path-like)
coord_file (path-like)
outdir (path-like)
raw_dir (path-like, optional)
epsg (int, default 32649)
utm_zone (str, default '49N')
rename_basename (str, default 'S')
rename_dir (path-like, optional)
step_overrides (dict, optional) – Per-step parameter overrides (see StratagemPipeline).
overwrite (bool, default False)
verbose (int, default 0)

Returns:

The completed survey object.

Return type:

StratagemSurvey

Examples

Replicate the legacy script:

sv = run_stratagem_preset(
    "basic",
    edi_dir="2/2EDI",
    coord_file="2.csv",
    outdir="2/processed",
    epsg=32649,
    rename_basename="T2.",
)

With hardware files:

sv = run_stratagem_preset(
    "full_processing",
    edi_dir="2/2EDI",
    coord_file="2.csv",
    outdir="2/processed",
    raw_dir="原始数据/2HX",
    epsg=32649,
    rename_basename="T2.",
)

pycsamt.pipeline.stratagem_preset_catalogue()#

Return a formatted catalogue of all Stratagem presets.

Return type:: str

class pycsamt.pipeline.PipelineAPIConfig(output_root='pipe_results', processed_subdir='processed', plots_subdir='plots', on_step_error='warn', save_intermediate=False, show_progress=True, progress_style='bar', repr_width=80, plot_dpi=150, plot_fmt='png', report_formats=<factory>)#

Bases: object

Runtime configuration for the pyCSAMT pipeline engine.

Variables:

output_root (str) – Default root directory when no outdir is passed to run().
processed_subdir (str) – Sub-directory inside output_root that receives the processed EDI files.
plots_subdir (str) – Sub-directory inside output_root that receives all QC figures, organised into one sub-folder per pipeline step.
on_step_error (str) – Behaviour when a step raises an exception. "raise" re-raises immediately; "warn" logs a warning and continues with the previous sites; "skip" is silent.
save_intermediate (bool) – If True, EDI files are also written after each step (into plots_subdir/<step>/edi_snapshot/).
show_progress (bool) – Enable console progress output while the pipeline runs.
progress_style (str) – "bar" – tqdm-style progress bar (requires tqdm; falls back to "log" when unavailable). "log" – one line per step. "silent" – no output.
repr_width (int) – Character width used when printing the pipeline summary.
plot_dpi (int) – DPI for saved figures.
plot_fmt (str) – File extension / format for saved figures ("png", "pdf", "svg").
report_formats (tuple) – Sequence of report types to write. Each item must be "html" or "txt".

Parameters:

output_root (str)
processed_subdir (str)
plots_subdir (str)
on_step_error (str)
save_intermediate (bool)
show_progress (bool)
progress_style (str)
repr_width (int)
plot_dpi (int)
plot_fmt (str)
report_formats (tuple)

output_root: str = 'pipe_results'#

processed_subdir: str = 'processed'#

plots_subdir: str = 'plots'#

on_step_error: str = 'warn'#

save_intermediate: bool = False#

show_progress: bool = True#

progress_style: str = 'bar'#

repr_width: int = 80#

plot_dpi: int = 150#

plot_fmt: str = 'png'#

report_formats: tuple#

configure(**kw)#

Set one or more configuration attributes by keyword.

Parameters:: kw (Any)
Return type:: None

context(**kw)#

Temporarily override config values, then restore them.

Parameters:: kw (Any)
Return type:: Generator[PipelineAPIConfig, None, None]

reset()#

Restore all attributes to package defaults.

Return type:: None

clone()#

Return a deep copy of this config.

Return type:: PipelineAPIConfig

pycsamt.pipeline.configure_pipe(**kw)#

Configure the global PYCSAMT_PIPE singleton.

Parameters:: kw (Any)
Return type:: None

pycsamt.pipeline.reset_pipe()#

Reset PYCSAMT_PIPE to package defaults.

Return type:: None

Pipeline Modules#

`pycsamt.pipeline._pipeline`	Core Pipeline class for the pyCSAMT processing engine.
`pycsamt.pipeline._steps`	Step and StepResult — user-facing wrappers around StepSpec entries.
`pycsamt.pipeline._registry`	Step registry for the pyCSAMT processing pipeline.
`pycsamt.pipeline._presets`	Named pipeline presets for common MT processing workflows.
`pycsamt.pipeline._config`	Config-file loaders for the pyCSAMT pipeline.
`pycsamt.pipeline._output`	Output-directory management for the pyCSAMT pipeline.
`pycsamt.pipeline._report`	Pipeline report generators.
`pycsamt.pipeline._base`	Base class for pyCSAMT pipeline objects.
`pycsamt.pipeline.stratagem`	pipeline.stratagem
`pycsamt.pipeline.plot`	Plot helpers for `pycsamt.pipeline` results.