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.

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

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

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:

sites (Any)

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)[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.

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