pycsamt.pipeline.stratagem#

pipeline.stratagem#

Stratagem AMT pipeline — extends the standard Pipeline with hardware-survey pre-processing and post-processing hooks.

Two levels of API#

Level 1 — :class:`StratagemPipeline` (Pipeline subclass):

Follows exactly the same run(sites, *, outdir=...) interface as Pipeline. The sites argument accepts every input type that ensure_sites() handles:

Sites object (used as-is)

str or pathlib.Path pointing to a single EDI file

str or pathlib.Path pointing to an EDI directory

A single EDIFile object

A list of EDIFile objects

Optional pre-processing before the emtools steps:

Coordinate injection — when coord_file is set, GPS coordinates from a CSV / XLS / XLSX file are injected into the EDI >HEAD sections.

Hardware SNR masking — when raw_dir is set, zero-stack frequency rows from the Stratagem hardware files are masked.

Optional post-processing after the emtools steps:

Rename — when rename_basename is set, the processed/ EDI files are renamed with the given basename (T2.000.edi, …).

Level 2 — :func:`run_stratagem_preset` / :class:`StratagemPreset`:

Convenience wrappers around StratagemSurvey that use the full Stratagem workflow (coord injection + SS + noise + rename). Useful for quick one-liner scripts that replace the legacy stratagem_edi_process_script.py.

Usage examples#

One-liner (Level 2, backwards-compatible):

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

Pipeline API with a directory of EDIs (Level 1):

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

Pipeline API with a single EDI file:

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

Pipeline API with an already-loaded Sites:

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

Custom step list:

from pycsamt.pipeline import Step
pipe = StratagemPipeline(
    [
        ("correct_ss",  Step("SS001")),
        ("select_band", Step("FREQ001", band_hz=(10.0, 1e5))),
        ("notch",       Step("NR001")),
        ("hampel",      Step("NR004")),
        ("qc",          Step("QC001")),
    ],
    coord_file="2.csv",
    raw_dir="原始数据/2HX",
    epsg=32649,
    rename_basename="T2.",
)
result = pipe.run("2/2EDI", outdir="2/processed")

Functions

`get_stratagem_preset`(name)	Return the `StratagemPreset` for name.
`list_stratagem_presets`()	Return all `StratagemPreset` objects.
`run_stratagem_preset`(preset, edi_dir, ...[, ...])	Execute a named Stratagem preset in one call (convenience wrapper).
`stratagem_preset_catalogue`()	Return a formatted catalogue of all Stratagem presets.

Classes

`StratagemPipeline`(steps, *[, coord_file, ...])	A `Pipeline` extended with Stratagem pre/post-processing.
`StratagemPreset`(name, description[, ...])	Named configuration bundle for `StratagemSurvey`.

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

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.stratagem.StratagemPipeline(steps, *, coord_file=None, raw_dir=None, epsg=32649, utm_zone='49N', order='auto', rename_basename=None, rename_dir=None, name='stratagem_pipeline')[source]#

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)[source]#

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)[source]#

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.stratagem.get_stratagem_preset(name)[source]#

Return the StratagemPreset for name.

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

pycsamt.pipeline.stratagem.list_stratagem_presets()[source]#

Return all StratagemPreset objects.

Return type:: list[StratagemPreset]

pycsamt.pipeline.stratagem.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)[source]#

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.stratagem_preset_catalogue()[source]#

Return a formatted catalogue of all Stratagem presets.

Return type:: str