2.27.2.2. pycsamt.pipeline._steps#

Step and StepResult — user-facing wrappers around StepSpec entries.

A Step binds a registry entry (identified by code or name) with a set of user-supplied parameter overrides. It is the atomic unit the user places into a Pipeline.

A StepResult is the immutable record produced after a step executes. The pipeline collects one per step and bundles them into a PipelineResult.

Functions

call_qc_fn(fn, sites[, before])

Call a registry QC plot function with the arguments it expects.

Classes

Step(code_or_name, **params)

A configured pipeline step.

StepResult(step_idx, step_name, step_code, ...)

Immutable record produced after one pipeline step runs.
class pycsamt.pipeline._steps.Step(code_or_name, **params)[source]

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)
transform(sites)[source]

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

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

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

Return type:

dict

class pycsamt.pipeline._steps.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)[source]

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

True when the step completed without error.

summary_line()[source]

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

Return type:

str