pipeline

pipeline ¤

Declarative pipeline orchestrator for ts-shape.

A :class:Pipeline is the single way to chain ts-shape processing steps -- transforms and event detectors -- into one reusable definition. It is linear single-channel:

an optional source step (always first, at most one) calls a ts-shape loader and produces the pipeline's first DataFrame;
a transform step takes the working DataFrame and returns a new one that replaces it (the signal flows on);
a detect step runs against the current working DataFrame, stores its event output under a name, and leaves the working DataFrame unchanged (detectors branch off).

Whether a step is a source, transform or detector is declared by the caller via .source() / .transform() / .detect() -- it is never inferred.

A pipeline with a source step is source-bound: call :meth:run with no argument and the source produces the data. A pipeline without one is DataFrame-driven: pass the input DataFrame to :meth:run, as before.

A step's target may be:

a plain callable df -> df;
a (class, "method") pair -- a classmethod / staticmethod is called on the class, an instance method instantiates the class first. Keyword arguments are routed between the constructor and the method by parameter name automatically.

Keyword-argument values may use two sentinels, resolved at run time:

"$input" -- the DataFrame originally passed to :meth:Pipeline.run;
"$prev" -- the current working DataFrame.

These wire steps that need a second DataFrame (e.g. SegmentProcessor.apply_ranges). If a kwarg names the step's own DataFrame parameter, that value is used instead of the auto-injected one.

Example::

from ts_shape import Pipeline
from ts_shape.transform.calculator.numeric_calc import IntegerCalc
from ts_shape.events.quality.outlier_detection import OutlierDetectionEvents

pipe = (
    Pipeline(name="sensor-quality")
    .transform(IntegerCalc, "scale_column",
               column_name="value_double", factor=0.1)
    .detect(OutlierDetectionEvents, "detect_outliers_zscore",
            name="outliers", value_column="value_double", threshold=3.0)
)
result = pipe.run(dataframe)
result.data                 # final transformed DataFrame
result.events["outliers"]   # detector output
result.to_event_log()       # normalized, combined EventLog

Debugging::

print(pipe.describe())          # preview steps without running
steps = pipe.run_steps(df)      # dict of every intermediate DataFrame

A transform step must preserve the long systime / uuid / value_* schema that downstream detectors expect. Reshaping operations (e.g. DataHarmonizer.pivot_to_wide) are terminal and belong at the end.

PipelineResult `dataclass` ¤

PipelineResult(
    name: str,
    data: DataFrame,
    events: dict[str, DataFrame],
    _detector_ids: dict[str, str | None] = dict(),
)

Outcome of :meth:Pipeline.run.

Attributes:

Name	Type	Description
`name`	`str`	The pipeline's name.
`data`	`DataFrame`	The working DataFrame after the final transform step.
`events`	`dict[str, DataFrame]`	Detector outputs keyed by step name.

to_event_log ¤

to_event_log(*, concat: bool = True) -> Any

Normalize detector outputs into an OCEL event log.

Each detect step added via the (class, method) form is run through :func:ts_shape.eventlog.to_event_log (the pipeline already knows the "ClassName.method" identifier). Plain-callable detect steps have no detector identity and are skipped.

Parameters:

Name	Type	Description	Default
`concat`	`bool`	When True (default) merge all logs into one `EventLog`; when False return a `dict` keyed by step name.	`True`

Raises:

Type	Description
`ValueError`	If no detector step has a known identity to normalize.

Pipeline ¤

Pipeline(name: str = 'pipeline')

A reusable, declarative chain of transform and detector steps.

Build with the fluent :meth:transform and :meth:detect methods, then call :meth:run -- on as many DataFrames as you like.

Initialize an empty pipeline.

Parameters:

Name	Type	Description	Default
`name`	`str`	A label for the pipeline, surfaced in `repr` and results.	`'pipeline'`

steps `property` ¤

steps: list[tuple[str, str]]

The ordered (kind, name) pairs of the configured steps.

source ¤

source(
    target: Any,
    method: str | None = None,
    *,
    name: str | None = None,
    **kwargs: Any
) -> Pipeline

Add a source step -- a loader that produces the pipeline's first frame.

A source step must be the first step, and a pipeline may have at most one. With a source step, :meth:run is called with no DataFrame; without one, :meth:run requires a DataFrame as before.

Parameters:

Name	Type	Description	Default
`target`	`Any`	A callable returning a DataFrame, or a loader class.	required
`method`	`str \| None`	Method name to call on `target` (omit for the callable form). A `classmethod` / `staticmethod` is called on the class; an instance method instantiates the class first, with kwargs routed between constructor and method by name.	`None`
`name`	`str \| None`	Optional step label (defaults to the method/callable name).	`None`
`**kwargs`	`Any`	Forwarded to the loader. The `$input` / `$prev` sentinels are not allowed -- a source has no prior data.	`{}`

Returns:

Type	Description
`Pipeline`	`self`, for chaining.

Raises:

Type	Description
`ValueError`	If the pipeline already has steps -- a source must be the first step, and only one is allowed.

transform ¤

transform(
    target: Any,
    method: str | None = None,
    *,
    name: str | None = None,
    **kwargs: Any
) -> Pipeline

Add a transform step -- its output replaces the working DataFrame.

Parameters:

Name	Type	Description	Default
`target`	`Any`	A callable `df -> df`, or a transform class.	required
`method`	`str \| None`	Method name to call on `target` (omit for the callable form).	`None`
`name`	`str \| None`	Optional step label (defaults to the method/callable name).	`None`
`**kwargs`	`Any`	Forwarded to the transform (routed between constructor and method for stateful classes). Values may be the `$input` / `$prev` sentinels.	`{}`

Returns:

Type	Description
`Pipeline`	`self`, for chaining.

detect ¤

detect(
    target: Any,
    method: str | None = None,
    *,
    name: str | None = None,
    **kwargs: Any
) -> Pipeline

Add a detector step -- its output is stored, the signal is unchanged.

Parameters:

Name	Type	Description	Default
`target`	`Any`	A callable `df -> events_df`, or a detector class.	required
`method`	`str \| None`	Method name to call on `target` (omit for the callable form).	`None`
`name`	`str \| None`	Optional result key (defaults to the method/callable name).	`None`
`**kwargs`	`Any`	Forwarded to the detector (routed between constructor and method). Values may be the `$input` / `$prev` sentinels.	`{}`

Returns:

Type	Description
`Pipeline`	`self`, for chaining.

Raises:

Type	Description
`ValueError`	If the resolved step name collides with an existing detector step.

describe ¤

describe() -> str

Return a human-readable summary of the pipeline without running it.

run ¤

run(dataframe: DataFrame | None = None) -> PipelineResult

Execute every step.

Parameters:

Name	Type	Description	Default
`dataframe`	`DataFrame \| None`	The input timeseries DataFrame. Omit it when the pipeline has a `.source()` step (the source produces it); supply it otherwise.	`None`

Returns:

Name	Type	Description
`A`	`PipelineResult`	class:`PipelineResult` with the final signal and detector
	`PipelineResult`	outputs.

Raises:

Type	Description
`TypeError`	If `dataframe` is passed to a source-bound pipeline, omitted from a sourceless one, is not a DataFrame, or a step returns a non-DataFrame.
`RuntimeError`	If a step raises; the message names the step.

run_steps ¤

run_steps(
    dataframe: DataFrame | None = None,
) -> dict[str, pd.DataFrame]

Execute every step and return all intermediate DataFrames.

Useful for debugging which step changes the data unexpectedly.

Parameters:

Name	Type	Description	Default
`dataframe`	`DataFrame \| None`	The input timeseries DataFrame. Omit it when the pipeline has a `.source()` step; supply it otherwise.	`None`

Returns:

Type	Description
`dict[str, DataFrame]`	A dict keyed by step name. For a DataFrame-driven pipeline, key
`dict[str, DataFrame]`	`"input"` holds the original DataFrame; for a source-bound
`dict[str, DataFrame]`	pipeline the loaded frame is keyed by the source step's name.
`dict[str, DataFrame]`	Transform steps store the transformed signal; detect steps store
`dict[str, DataFrame]`	their event output.

pipeline

pipeline ¤

PipelineResult dataclass ¤

to_event_log ¤

Pipeline ¤

steps property ¤

source ¤

transform ¤

detect ¤

describe ¤

run ¤

run_steps ¤

PipelineResult `dataclass` ¤

steps `property` ¤