Pipeline

class rapidtools.processing.Pipeline(steps: list[Callable] | None = None, cancel_event: Event | None = None)

Bases: object

A smart sequence of processing steps applied to a PhysicalAssetCollection.

This Pipeline is order-agnostic during construction. It automatically inspects the components you add and enforces a strict logical execution sequence before running. This prevents common errors, such as attempting to run AI predictions before image data has been extracted.

The enforced execution order is:

1. Extractors (e.g., AerialImageryExtractor) - Gathers raw data and images.

2. Predictors / Classifiers (e.g., DamagePredictor) - Analyzes the gathered data.

3. Reporters / Exporters (e.g., PDFReporter) - Summarizes and exports the results.

4. Custom / Unknown Steps - Any unrecognized steps default to running last.

Note: The pipeline fully supports standalone execution. If you only provide a Predictor, it will simply run the Predictor without requiring an Extractor.

Parameters:

• steps (list[Callable] | None, optional) – An optional list of initialized processing steps. Each step must be a callable accepting a PhysicalAssetCollection and returning a PhysicalAssetCollection. Defaults to an empty pipeline.

• cancel_event (Event | None, optional) – A cooperative cancellation flag. When set (typically from another thread), the pipeline raises OperationCancelled before starting the next step. The same event can be shared with the individual steps so that they also stop promptly. Defaults to None.

Example

Build a pipeline from an extractor and an analyzer, adding the steps in the “wrong” order. The pipeline reorders them before execution:

>>> import threading
>>> from rapidtools.core import PhysicalAssetCollection
>>> from rapidtools.processing import (
...     AerialImageryExtractor, Gemma4AssetAnalyzer, Pipeline
... )
>>>
>>> stop = threading.Event()
>>> analyzer = Gemma4AssetAnalyzer(
...     prompt='Is this building damaged? Answer yes or no.',
...     cancel_event=stop,
... )
>>> extractor = AerialImageryExtractor(
...     dataset='data/eaton_ortho.tif',
...     save_directory='output/crops',
...     cancel_event=stop,
... )
>>> pipeline = Pipeline(cancel_event=stop)
>>> pipeline.add_step(analyzer).add_step(extractor)
>>> collection = PhysicalAssetCollection.from_geojson('bldgs.geojson')
>>> processed = pipeline.run(collection)

Methods

add_step	Add a new processing step to the pipeline.
run	Execute all steps in the pipeline in the correct logical sequence.

add_step(step: Callable) → Pipeline

Add a new processing step to the pipeline.

Parameters:

step (Callable) – A callable object (like an Extractor or Predictor instance) that accepts and returns a PhysicalAssetCollection.

Returns:

Pipeline – The Pipeline instance itself, allowing for method chaining.

Example

>>> from rapidtools.processing import Pipeline
>>> pipeline = Pipeline()
>>> pipeline.add_step(my_extractor).add_step(my_analyzer)
>>> len(pipeline.steps)
2

run(asset_collection: PhysicalAssetCollection) → PhysicalAssetCollection

Execute all steps in the pipeline in the correct logical sequence.

Parameters:

asset_collection (PhysicalAssetCollection) – The collection of assets to process.

Returns:

PhysicalAssetCollection – The fully processed collection. If the pipeline has no steps, a warning is logged and the input is returned unchanged.

Raises:

OperationCancelled – If cancel_event is set before any step starts.

Example

>>> from rapidtools.processing import Pipeline
>>> pipeline = Pipeline([my_extractor, my_analyzer])
>>> processed = pipeline.run(my_collection)