2.27. 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.

2.27.1. 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 built-in 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)

Cache step outputs so an interrupted run resumes instead of recomputing:

result = pipe.run(sites, outdir="results/", cache=True)

Observe a run live, or log it for later comparison:

result = pipe.run(sites, on_step=lambda sr: print(sr.summary_line()))
result = pipe.run(sites, history=True)  # -> load_history() later

A finished PipelineResult renders inline in Jupyter (_repr_html_), and configure_pipe(progress_style="rich") (or CLI --live) renders a live-updating status table in the terminal while a run is in progress.

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:
Return type:

Pipeline

insert(idx, label, step)#

Insert step at position idx (0-based).

Returns self.

Parameters:
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:
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, cache=False, on_step=None, history=False)#

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)#

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 = '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 (dict) – 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)
spec: StepSpec#
params: dict#
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, before=None)#

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

before is the step’s input; comparison plots (before/after) need it and are skipped when it is not given (see call_qc_fn()).

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:
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, cached=False)#

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.

  • cached (bool) – True when this step’s result was replayed from the pipeline’s step cache (Pipeline.run(..., cache=...)) instead of actually being recomputed. Only ever True when caching was enabled for this run.

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 = 0#
n_sites_out: int = 0#
error: Exception | None = None#
cached: bool = False#
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, origin='builtin')#

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.

  • origin (str) – "builtin" for the 47 steps shipped with pyCSAMT (default), or "plugin" for anything added via register_step(). Stamped automatically by register_step() — callers never need to set it.

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#
origin: str = 'builtin'#
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]

pycsamt.pipeline.register_step(spec, *, replace_existing=False, validate=True)#

Register a third-party StepSpec into the pipeline registry.

This is the extension point for anything outside the 47 steps shipped with pyCSAMT — a plugin package (see pycsamt.pipeline.discover_plugins()) or a one-off custom step defined in a user script. Once registered, the step is usable everywhere a built-in step is: Step(code), the CLI (pycsamt pipe steps, pipe run --steps ...), presets, etc.

Parameters:

  • spec (StepSpec) – The step to register. Its origin is always overwritten to "plugin" regardless of what the caller passed.

  • replace_existing (bool) – If False (default), raises ValueError when spec.code or spec.name already exists in the registry. Set True to overwrite an existing entry (built-in or previously-registered plugin) — e.g. to patch a built-in step’s implementation.

  • validate (bool) – If True (default), resolves spec.get_fn() once before inserting, so a typo’d mod/fn_name or a broken override_fn is caught at registration time rather than at the first pipeline run. On failure the registry is left untouched.

Returns:

The registered spec (with origin="plugin" stamped on it).

Return type:

StepSpec

Raises:
pycsamt.pipeline.unregister_step(code_or_name, *, missing_ok=False)#

Remove a previously-registered step from the pipeline registry.

Parameters:

  • code_or_name (str) – The step’s code or name (see lookup_step()).

  • missing_ok (bool) – If False (default), raises KeyError when no such step is registered. Set True to no-op instead.

Return type:

None

pycsamt.pipeline.discover_plugins(*, group='pycsamt.pipeline.steps', on_error='warn')#

Load every pycsamt.pipeline.steps entry point and run it.

Each entry point must resolve to a zero-argument callable; it is expected to call pycsamt.pipeline.register_step() itself for whatever steps it contributes. Never called automatically — see the module docstring.

Parameters:

  • group (str) – Entry-point group to scan. Defaults to ENTRY_POINT_GROUP.

  • on_error (str) – "warn" (default): a plugin that fails to load or raises is reported as a failed PluginLoadResult and a UserWarning; discovery continues with the remaining plugins. "raise": the first failure propagates immediately.

Returns:

One entry per discovered entry point, in discovery order.

Return type:

list[PluginLoadResult]

class pycsamt.pipeline.PluginLoadResult(name, ok, error=None)#

Bases: object

Outcome of loading one pycsamt.pipeline.steps entry point.

Variables:

  • name (str) – The entry point’s registered name (left-hand side in [project.entry-points."pycsamt.pipeline.steps"]).

  • ok (bool) – Whether the entry point loaded and ran without raising.

  • error (str | None) – str(exception) when ok is False, else None.

Parameters:
name: str#
ok: bool#
error: str | None = None#
pycsamt.pipeline.register_ai_steps(*, replace_existing=False)#

Register the opt-in AI step(s) into the pipeline step registry.

Currently just AI001 / audit_survey. Never called automatically — see the module docstring for why.

Parameters:

replace_existing (bool) – Forwarded to register_step().

Returns:

The registered spec(s), each stamped origin="plugin" by register_step.

Return type:

list[StepSpec]

class pycsamt.pipeline.StepCache(root=None)#

Bases: object

Sharded, content-addressed disk cache for pipeline step outputs.

Layout: root/<key[:2]>/<key>.joblib. Writes are atomic (temp file in the same directory, then os.replace), so a killed process never leaves a half-written entry visible. A corrupt or unreadable entry is treated as a miss — warns, never crashes the run — the same “one bad entry must not break everything else” principle pycsamt.pipeline.discover_plugins() already established.

Parameters:

root (str | Path | None)

get(key)#

Return the cached value for key, or the _MISS sentinel.

Parameters:

key (str)

Return type:

Any

put(key, value)#

Atomically store value under key.

Parameters:
Return type:

None

clear()#

Remove every entry (and the root directory itself) from disk.

Return type:

None

pycsamt.pipeline.fingerprint_sites(sites)#

Deterministic content hash of a Sites-like collection.

Tries the real Sites shape first — iterates stations, hashing each one’s name plus its freq/z arrays, in iteration order. Order is part of the fingerprint on purpose: under-caching on a harmless reorder is a safer failure mode than treating two differently-ordered (and potentially differently-behaving) inputs as identical.

Falls back to hashing the pickled object for anything that doesn’t match that shape (plain test doubles, custom plugin-defined sites-like objects) — this keeps the cache usable without special-casing every possible “sites” representation.

Parameters:

sites (Any)

Return type:

str

pycsamt.pipeline.load_history(path=None, *, last=None)#

Read back previously logged run summaries, oldest first.

Parameters:

  • path (str | Path | None) – Defaults to default_history_path().

  • last (int | None) – When given, return only the most recent last entries.

  • skipped (A malformed line (e.g. from a write interrupted mid-flush) is)

  • non-event (rather than raising — the same "one bad entry is a)

  • a (not)

Return type:

list[dict]

:param crash” posture StepCache already uses.:

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]]#
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.get_preset_for_method(method, level='qc')#

Return the method-aware Preset for an explicit EM method.

Parameters:

  • method (str) – An EM method string – either one recognised by SurveyMeta ("MT", "AMT", "CSAMT", "CSEM", "TEM", "BBMT", "LAMT", "LMT") or "CSUMT" (recognised here even though SurveyMeta’s vocabulary predates it). "MT", "BBMT", "LAMT", "LMT", "AMT", "CSAMT", and "CSUMT" are mapped to a preset today; "CSEM"/"TEM" are recognised but have no method-aware preset yet. Case-insensitive.

  • level (str) – Preset tier. Only "qc" exists today.

  • be (This performs no data inspection or auto-detection -- method must)

  • a (supplied explicitly (e.g. from)

Return type:

Preset

:param method the caller already: :param has): :param the same way get_preset() requires an explicit preset name.:

Raises:

ValueError – When method is not a recognised EM method, or has no method-aware preset yet.

Parameters:
Return type:

Preset

Examples

>>> from pycsamt.pipeline import get_preset_for_method
>>> get_preset_for_method("CSAMT").name
'csamt_qc'
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:
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:
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:
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:
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]]#
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:

2.27. 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.

2.27. 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:
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

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

2.27.2. 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._plugins

Entry-point discovery for third-party pipeline steps.

pycsamt.pipeline._cache

Content-addressed disk cache for pipeline step outputs.

pycsamt.pipeline._history

Append-only run history log for Pipeline.run(..., history=...).

pycsamt.pipeline._richview

Live rich.table.Table rendering for progress_style="rich".

pycsamt.pipeline._presets

Named pipeline presets for common MT processing workflows.

pycsamt.pipeline._smart_qc

Data-driven ("smart") QC plot gating for method-aware presets.

pycsamt.pipeline._preview

Random-station raw-vs-processed preview for method-aware presets.

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._dashboard

The "dashboard" report tier — a branded, chart-carrying HTML report.

pycsamt.pipeline._base

Base class for pyCSAMT pipeline objects.

pycsamt.pipeline.ai_steps

Opt-in AI-backed pipeline step: the domain-gap survey audit.

pycsamt.pipeline.stratagem

pipeline.stratagem

pycsamt.pipeline.plot

Plot helpers for pycsamt.pipeline results.
On this page