Pipeline

The pipeline module provides classes for configuring and executing reproducible processing workflows.

Pipeline Class

class Pipeline

Bases: object

Recipe-style pipeline executor.

Executes a sequence of operations defined as StepConfig objects. Each operation receives the PipelineContext, modifies it, and returns it.

Example

>>> steps = [
...     StepConfig("smooth", {"iterations": 100}),
...     StepConfig("remesh", {"n_clusters": 5000}),
...     StepConfig("smooth", {"iterations": 50}),
... ]
>>> pipeline = Pipeline(steps)
>>> result = pipeline.run(mesh)

__init__(steps, registry=None)

Initialize pipeline with steps.

Parameters:

• steps (list[StepConfig]) – List of step configurations

• registry (OperationRegistry | None, default: None) – Operation registry (uses default if not provided)

Return type:

None

classmethod from_yaml(path)

Load pipeline from YAML configuration file.

Parameters:

path (str | Path) – Path to YAML file

Return type:

Pipeline

Returns:

Configured Pipeline instance

Raises:

ConfigurationError – If YAML is invalid or missing required keys

classmethod from_yaml_string(yaml_string)

Load pipeline from YAML string.

Parameters:

yaml_string (str) – YAML configuration as string

Return type:

Pipeline

Returns:

Configured Pipeline instance

run(input_data, verbose=True, resolution=None)

Execute all pipeline steps in sequence.

Parameters:

• input_data (str | Path | ndarray[tuple[Any, ...], dtype[Any]] | PhenoMesh) – Input data - can be: - Path to image file - NumPy array (image or contour) - PhenoMesh object

• verbose (bool, default: True) – Print progress information

• resolution (list[float] | None, default: None) – Spatial resolution [x, y, z] (for image/contour input)

Return type:

PipelineContext

Returns:

PipelineContext with all results

Raises:

PipelineError – If a step fails or input is invalid

validate()

Validate pipeline configuration without running.

Return type:

list[str]

Returns:

List of validation warnings (empty if valid)

Raises:

ConfigurationError – If configuration is invalid

__len__()

Return number of steps.

Return type:

int

__repr__()

Return string representation.

Return type:

str

PipelineContext Class

class PipelineContext

Bases: object

Holds state passed between pipeline steps.

image

Input 3D image array

contour

Binary contour array

mesh

PhenoMesh object

domains

Domain labels array

curvature

Curvature values array

meristem_index

Index of the meristem domain

domain_data

DataFrame with domain measurements

resolution

Spatial resolution [x, y, z]

neighbors

Cached vertex neighbors for performance

image: ndarray[tuple[Any, ...], dtype[Any]] | None = None

contour: ndarray[tuple[Any, ...], dtype[bool]] | None = None

mesh: PhenoMesh | None = None

domains: ndarray[tuple[Any, ...], dtype[integer[Any]]] | None = None

curvature: ndarray[tuple[Any, ...], dtype[floating[Any]]] | None = None

meristem_index: int | None = None

domain_data: DataFrame | None = None

resolution: list[float] | None = None

neighbors: list[ndarray[tuple[Any, ...], dtype[int64]]] | None = None

__init__(image=None, contour=None, mesh=None, domains=None, curvature=None, meristem_index=None, domain_data=None, resolution=None, neighbors=None)

Parameters:

• image (ndarray[tuple[Any, ...], dtype[Any]] | None)

• contour (ndarray[tuple[Any, ...], dtype[bool]] | None)

• mesh (PhenoMesh | None)

• domains (ndarray[tuple[Any, ...], dtype[integer[Any]]] | None)

• curvature (ndarray[tuple[Any, ...], dtype[floating[Any]]] | None)

• meristem_index (int | None)

• domain_data (DataFrame | None)

• resolution (list[float] | None)

• neighbors (list[ndarray[tuple[Any, ...], dtype[int64]]] | None)

Return type:

None

StepConfig Class

class StepConfig

Bases: object

Configuration for a single pipeline step.

name

Name of the operation to execute

parameters

Parameters to pass to the operation

name: str

parameters: dict[str, Any]

__init__(name, parameters=<factory>)

Parameters:

• name (str)

• parameters (dict[str, Any])

Return type:

None

OperationRegistry Class

class OperationRegistry

Bases: object

Registry of available pipeline operations.

Manages the mapping from operation names to callable functions. Supports registering custom operations.

__init__()

Initialize with default operations.

Return type:

None

register(name, function)

Register a custom operation.

Parameters:

• name (str) – Operation name for use in pipeline configs

• function (Callable[..., PipelineContext]) – Callable that takes (PipelineContext, **params) and returns PipelineContext

Return type:

None

get(name)

Get an operation by name.

Parameters:

name (str) – Operation name

Return type:

Callable[..., PipelineContext]

Returns:

The operation callable

Raises:

PipelineError – If operation name is not found

list_operations(category='all')

List available operation names.

Parameters:

category (str, default: 'all') – Filter by category (‘all’, ‘contour’, ‘mesh’, ‘domain’)

Return type:

list[str]

Returns:

List of operation names

__contains__(name)

Check if operation is registered.

Return type:

bool

Parameters:

name (str)

Preset Functions

load_preset(name='default')

Load a preset pipeline configuration.

Parameters:

name (str, default: 'default') – Preset name (defaults to ‘default’)

Return type:

Pipeline

Returns:

Configured Pipeline instance

Raises:

ValueError – If preset name is not found

list_presets()

List available preset names.

Return type:

list[str]

Returns:

List of preset names

get_preset_yaml(name='default')

Get the YAML configuration for a preset.

Parameters:

name (str, default: 'default') – Preset name (defaults to ‘default’)

Return type:

str

Returns:

YAML configuration string

Raises:

ValueError – If preset name is not found

save_pipeline_yaml(pipeline, path)

Save pipeline configuration to YAML file.

Parameters:

• pipeline (Pipeline) – Pipeline to save

• path (str | Path) – Output file path

Return type:

None