2.1. pycsamt#

pycsamt v2 — Python toolbox for audio-magnetotellurics (AMT & CSAMT).

Documentation: https://pycsamt.org/ Source: earthai-tech/pycsamt

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

pycsamt.configure_pipe(**kw)#

Configure the global PYCSAMT_PIPE singleton.

Parameters:

kw (Any)

Return type:

None

pycsamt.reset_pipe()#

Reset PYCSAMT_PIPE to package defaults.

Return type:

None