"""Named pipeline presets for common MT processing workflows.
Each preset is a :class:`Preset` object that pairs a description with an
ordered list of :class:`~pycsamt.emtools.pipe._steps.Step` instances. Users
can load a preset via :meth:`~pycsamt.emtools.pipe.Pipeline.from_preset` and
then add, remove, or override steps as needed.
Available presets
-----------------
``"basic_qc"``
Minimal denoising + frequency cleanup. Good for a first-pass inspection.
``"noise_reduction"``
Stacked noise-removal chain for high-EMI environments.
``"full_processing"``
Standard end-to-end workflow: noise → frequency → skew gate → SS → strike.
``"tensor_analysis"``
Tensor-only cleanup (rotation, antisymmetry, sigma-clip, balance).
``"dimensionality_filter"``
Classify, mask, and project to 2-D.
``"publication_ready"``
Full chain suitable for Computers & Geosciences quality output.
``"stratagem_mt"``
emtools-level preset for Stratagem AMT data already loaded as Sites.
Static-shift → AMT band select → powerline notch → Hampel → QC.
Use :mod:`pycsamt.pipeline.stratagem` when you also need coordinate
injection and renaming from raw EDI + CSV inputs.
Method-aware presets
---------------------
The presets above are chosen by processing intent only; the four below are
additionally chosen by EM survey method, since MT/AMT (passive) and
CSAMT/CSUMT (controlled-source) surveys need genuinely different treatment
-- see :doc:`/user_guide/pipeline/presets` for the full rationale.
``"mt_qc"``
Broadband MT quick-look: denoise, frequency cleanup, strike rotation,
and data-driven ("smart") tensor/tipper/strike QC that only fires the
plots the data actually supports.
``"amt_qc"``
Same as ``mt_qc``, trimmed to the standard AMT acquisition band
(10 Hz - 100 kHz).
``"csamt_qc"``
Adds CSAMT-specific near-field/transition-zone correction and its
diagnostic pseudosection on top of ``mt_qc``'s chain.
``"csumt_qc"``
Same as ``csamt_qc``, trimmed to the CSUMT acquisition band and adding
the Bostick depth-section snapshot.
Every method-aware preset opens with a raw-data preview and closes with a
processed-data preview of the same (deterministically) randomly chosen
stations, so a before/after comparison is always available.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from ..emtools.csumt import F_MAX_CSUMT, F_MIN_CSUMT
from ..metadata.survey import _VALID_METHODS as _VALID_SURVEY_METHODS
from ._steps import Step
# ---------------------------------------------------------------------------
# Preset dataclass
# ---------------------------------------------------------------------------
[docs]
@dataclass
class Preset:
"""A named, ordered collection of pipeline steps.
Attributes
----------
name:
Identifier used in :meth:`~pycsamt.emtools.pipe.Pipeline.from_preset`.
description:
One-line description shown when the user prints the preset catalogue.
steps:
Ordered list of ``(label, Step)`` tuples — same format as a
:class:`~pycsamt.emtools.pipe.Pipeline` step list.
"""
name: str
description: str
steps: list[tuple[str, Step]] = field(default_factory=list)
def __repr__(self) -> str:
codes = " → ".join(s.spec.code for _, s in self.steps)
return f"Preset({self.name!r} [{codes}])"
# ---------------------------------------------------------------------------
# Preset definitions
# ---------------------------------------------------------------------------
PRESETS: dict[str, Preset] = {
"basic_qc": Preset(
name="basic_qc",
description=(
"Minimal denoising + frequency cleanup. "
"Good for quick-look inspection."
),
steps=[
("notch", Step("NR001")),
("drop_duplicates", Step("FREQ002")),
("select_band", Step("FREQ001")),
("align_grid", Step("FREQ004")),
("qc_snapshot", Step("QC001")),
],
),
"noise_reduction": Preset(
name="noise_reduction",
description=("Stacked noise-removal chain for high-EMI environments."),
steps=[
("notch", Step("NR001")),
("hampel", Step("NR004")),
("spatial_med", Step("NR005")),
("shrink_trend", Step("NR003")),
("mask_incoher", Step("NR010")),
("qc_snapshot", Step("QC001")),
],
),
"full_processing": Preset(
name="full_processing",
description=(
"Standard end-to-end workflow: "
"noise → frequency → skew gate → static-shift → strike rotation."
),
steps=[
("notch", Step("NR001")),
("drop_dup", Step("FREQ002")),
("select_band", Step("FREQ001")),
("align_grid", Step("FREQ004")),
("mask_skew", Step("SK001")),
("rotate_strike", Step("TZ001")),
("correct_ss", Step("SS001")),
("qc_snapshot", Step("QC001")),
],
),
"tensor_analysis": Preset(
name="tensor_analysis",
description=(
"Tensor-only cleanup: strike rotation, antisymmetry, "
"sigma-clip, and off-diagonal balance."
),
steps=[
("rotate_strike", Step("TZ001")),
("antisymm", Step("TZ002")),
("sigma_clip", Step("TZ003")),
("balance", Step("TZ004")),
("qc_snapshot", Step("QC001")),
],
),
"dimensionality_filter": Preset(
name="dimensionality_filter",
description=(
"Classify 1-D / 2-D / 3-D regions, mask unwanted class, "
"and project to 2-D."
),
steps=[
("classify_dim", Step("DIM001")),
("mask_dim", Step("DIM002")),
("project_2d", Step("DIM003")),
("qc_snapshot", Step("QC001")),
],
),
"publication_ready": Preset(
name="publication_ready",
description=(
"Full chain for publication-quality output "
"(C&G paper standard): noise removal, frequency editing, "
"static-shift correction, strike rotation, and skew gating."
),
steps=[
("notch", Step("NR001")),
("drop_dup", Step("FREQ002")),
("select_band", Step("FREQ001")),
("align_grid", Step("FREQ004")),
("correct_ss", Step("SS001")),
("rotate_strike", Step("TZ001")),
("antisymm", Step("TZ002")),
("mask_skew", Step("SK001")),
("qc_snapshot", Step("QC001")),
],
),
# ── Stratagem-specific preset ─────────────────────────────────────────
"stratagem_mt": Preset(
name="stratagem_mt",
description=(
"emtools-level preset for Stratagem AMT data already loaded "
"as a Sites object. Applies AMA static-shift correction, "
"selects the standard AMT band (10 Hz – 100 kHz), removes "
"powerline harmonics and outliers, and masks incoherent "
"frequency bins. For the full raw EDI + GPS workflow "
"(coordinate injection, hardware SNR masking, renaming) use "
"pycsamt.pipeline.stratagem.StratagemPipeline instead."
),
steps=[
# static shift first — requires complete frequency range
("correct_ss", Step("SS001")),
# trim to AMT acquisition band
("select_band", Step("FREQ001", band_hz=(10.0, 1e5))),
# drop duplicate frequencies from multi-band acquisition
("drop_dup", Step("FREQ002")),
# powerline harmonics (50 Hz grid, 30 harmonics covers AMT range)
("notch", Step("NR001", mains_hz=50.0, n_harm=30, tol_hz=0.08)),
# Hampel outlier filter on impedance tensor
("hampel", Step("NR004")),
# mask frequencies that are incoherent across the profile
("mask_incoher", Step("NR010")),
("qc_snapshot", Step("QC001")),
],
),
# ── Method-aware presets ────────────────────────────────────────────
"mt_qc": Preset(
name="mt_qc",
description=(
"Broadband MT quick-look: denoise, frequency cleanup, strike "
"rotation, and data-driven tensor/tipper/strike QC. "
"Opens and closes with a raw-vs-processed station preview."
),
steps=[
("raw_preview", Step("PRE001")),
("notch", Step("NR001")),
("drop_dup", Step("FREQ002")),
("select_band", Step("FREQ001")),
("align_grid", Step("FREQ004")),
("rotate_strike", Step("TZ001")),
("qc_snapshot", Step("QC001")),
("strike_qc", Step("QC007")),
("tensor_qc_smart", Step("QC005")),
("tipper_qc_smart", Step("QC006")),
("processed_preview", Step("PRE002")),
],
),
"amt_qc": Preset(
name="amt_qc",
description=(
"Same chain as mt_qc, trimmed to the standard AMT acquisition "
"band (10 Hz - 100 kHz)."
),
steps=[
("raw_preview", Step("PRE001")),
("notch", Step("NR001")),
("drop_dup", Step("FREQ002")),
("select_band", Step("FREQ001", band_hz=(10.0, 1e5))),
("align_grid", Step("FREQ004")),
("rotate_strike", Step("TZ001")),
("qc_snapshot", Step("QC001")),
("strike_qc", Step("QC007")),
("tensor_qc_smart", Step("QC005")),
("tipper_qc_smart", Step("QC006")),
("processed_preview", Step("PRE002")),
],
),
"csamt_qc": Preset(
name="csamt_qc",
description=(
"CSAMT quick-look: mt_qc's chain plus near-field/transition-"
"zone correction and its diagnostic pseudosection. "
"source_offset defaults to None (auto-resolved per station from "
"site metadata; stations without a resolvable offset are "
"warned about and left uncorrected, never fail the run)."
),
steps=[
("raw_preview", Step("PRE001")),
("notch", Step("NR001")),
("drop_dup", Step("FREQ002")),
("select_band", Step("FREQ001")),
("align_grid", Step("FREQ004")),
(
"correct_near_field",
Step("SRC001", source_offset=None, verbose=1),
),
("field_zone_snapshot", Step("QC002")),
("normalize_response", Step("SRC002")),
("rotate_strike", Step("TZ001")),
("qc_snapshot", Step("QC001")),
("strike_qc", Step("QC007")),
("tensor_qc_smart", Step("QC005")),
("tipper_qc_smart", Step("QC006")),
("processed_preview", Step("PRE002")),
],
),
"csumt_qc": Preset(
name="csumt_qc",
description=(
"CSUMT quick-look: csamt_qc's chain trimmed to the CSUMT "
"acquisition band, plus a Bostick depth-section snapshot."
),
steps=[
("raw_preview", Step("PRE001")),
("notch", Step("NR001")),
("drop_dup", Step("FREQ002")),
(
"select_band",
Step("FREQ001", band_hz=(F_MIN_CSUMT, F_MAX_CSUMT)),
),
("align_grid", Step("FREQ004")),
(
"correct_near_field",
Step("SRC001", source_offset=None, verbose=1),
),
("field_zone_snapshot", Step("QC002")),
("normalize_response", Step("SRC002")),
("rotate_strike", Step("TZ001")),
("qc_snapshot", Step("QC001")),
("depth_section_snapshot", Step("QC004")),
("strike_qc", Step("QC007")),
("tensor_qc_smart", Step("QC005")),
("tipper_qc_smart", Step("QC006")),
("processed_preview", Step("PRE002")),
],
),
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
[docs]
def list_presets() -> list[Preset]:
"""Return all available :class:`Preset` objects."""
return list(PRESETS.values())
[docs]
def get_preset(name: str) -> Preset:
"""Return the :class:`Preset` for *name*.
Raises
------
KeyError
When *name* is not registered.
"""
if name not in PRESETS:
available = sorted(PRESETS)
raise KeyError(
f"Unknown preset {name!r}. Available presets: {available}"
)
return PRESETS[name]
_METHOD_PRESET_MAP: dict[str, str] = {
"MT": "mt_qc",
"BBMT": "mt_qc",
"LMT": "mt_qc",
"LAMT": "mt_qc",
"AMT": "amt_qc",
"CSAMT": "csamt_qc",
"CSUMT": "csumt_qc",
}
# SurveyMeta._VALID_METHODS predates the CSUMT-specific tooling in
# emtools/csumt.py and does not include "CSUMT" -- accept the union so a
# real, first-class method this feature is about is never rejected by an
# unrelated, older module's vocabulary.
_RECOGNIZED_METHODS = _VALID_SURVEY_METHODS | set(_METHOD_PRESET_MAP)
[docs]
def get_preset_for_method(method: str, level: str = "qc") -> Preset:
"""Return the method-aware :class:`Preset` for an explicit EM method.
Parameters
----------
method:
An EM method string -- either one recognised by
:class:`~pycsamt.metadata.survey.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:
Preset tier. Only ``"qc"`` exists today.
This performs no data inspection or auto-detection -- *method* must be
supplied explicitly (e.g. from a
:class:`~pycsamt.metadata.survey.SurveyMeta.method` the caller already
has), the same way :func:`get_preset` requires an explicit preset name.
Raises
------
ValueError
When *method* is not a recognised EM method, or has no method-aware
preset yet.
Examples
--------
>>> from pycsamt.pipeline import get_preset_for_method
>>> get_preset_for_method("CSAMT").name
'csamt_qc'
"""
m = method.strip().upper()
if m not in _RECOGNIZED_METHODS:
raise ValueError(
f"method {method!r} is not one of {sorted(_RECOGNIZED_METHODS)}."
)
if m not in _METHOD_PRESET_MAP:
raise ValueError(
f"method {method!r} has no method-aware preset yet. "
f"Available: {sorted(_METHOD_PRESET_MAP)}"
)
if level != "qc":
raise ValueError(f"level {level!r} is not available; only 'qc' exists today.")
return get_preset(_METHOD_PRESET_MAP[m])
[docs]
def preset_catalogue() -> str:
"""Return a formatted catalogue of all presets."""
lines = ["Available pipeline presets", "─" * 60]
for p in PRESETS.values():
codes = " → ".join(s.spec.code for _, s in p.steps)
lines.append(f" {p.name:<22s} {p.description}")
lines.append(f" {'':22s} {codes}")
lines.append("")
return "\n".join(lines)
__all__ = [
"Preset",
"PRESETS",
"get_preset",
"get_preset_for_method",
"list_presets",
"preset_catalogue",
]
