ImagePipeline Class Methods#

Overview#

The ImagePipeline class provides a comprehensive interface for sequential execution of image processing operations and measurements on images. It manages two queues: a processing queue containing operations applied sequentially to each image, and a measurement queue containing feature extractors that analyze images and produce results as pandas DataFrames.

The pipeline supports:

Sequential operations: Apply detection, enhancement, refinement, and correction operations in order
Measurement extraction: Extract shape, intensity, texture, and custom features
Performance monitoring: Track execution time with benchmarking
Serialization: Save/load pipeline configurations as JSON
Batch processing: Process multiple images efficiently
Verbose logging: Progress reporting during execution

Pipelines are the recommended approach for reproducible image analysis workflows, ensuring consistent processing across image sets and facilitating parameter optimization.

Initialization & Configuration#

ImagePipeline.__init__(**data: Any) → None#

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)
Return type:: None

Create a new pipeline with specified operations and measurements. Operations are applied sequentially in the order provided. Measurements are extracted after all operations complete.

Parameters:

ops: List or dict of ImageOperation instances (detectors, enhancers, refiners)
meas: List or dict of MeasureFeatures instances (shape, intensity, custom features)
benchmark: Enable execution time tracking (default: False)
verbose: Enable progress reporting and diagnostic output (default: False)

Example:

from phenotypic import ImagePipeline
from phenotypic.detect import OtsuDetector
from phenotypic.enhance import GaussianBlur
from phenotypic.refine import RemoveSmallObjects
from phenotypic.measure import MeasureShape, MeasureIntensity

# Create pipeline with operations and measurements
pipeline = ImagePipeline(
    ops=[
        GaussianBlur(sigma=1.5),
        OtsuDetector(),
        RemoveSmallObjects(min_size=100)
    ],
    meas=[
        MeasureShape(),
        MeasureIntensity()
    ],
    benchmark=True,
    verbose=True
)

# Apply to image
img = Image.imread('plate.jpg')
results = pipeline.apply_and_measure(img)

ImagePipeline.set_ops(ops: List[ImageOperation | ImagePipeline] | Dict[str, ImageOperation | ImagePipeline])#

Sets the operations to be performed. The operations can be passed as either a list of ImageOperation or ImagePipeline instances or a dictionary mapping operation names to ImageOperation or ImagePipeline instances. This method ensures that each operation in the list has a unique name. Raises a TypeError if the input is neither a list nor a dictionary.

Parameters:: ops (List[ImageOperation | ImagePipeline] | Dict[str, ImageOperation | ImagePipeline]) – A list of ImageOperation or ImagePipeline objects, or a dictionary where keys are operation names and values are ImageOperation or ImagePipeline objects.
Raises:: TypeError – If the input is not a list or a dictionary.

Register or update the operations queue. Operations can be provided as a list (auto-named by class) or a dictionary (explicit names). If operations with duplicate class names are provided, automatic suffix numbering ensures uniqueness.

Example:

pipeline = ImagePipeline()

# Set operations as list
pipeline.set_ops([
    GaussianBlur(sigma=1.5),
    OtsuDetector(),
    RemoveSmallObjects(min_size=100)
])
# Names: 'GaussianBlur', 'OtsuDetector', 'RemoveSmallObjects'

# Set operations as dict with custom names
pipeline.set_ops({
    'blur': GaussianBlur(sigma=1.5),
    'detect': OtsuDetector(),
    'refine': RemoveSmallObjects(min_size=100)
})

ImagePipeline.set_meas(measurements: List[MeasureFeatures] | Dict[str, MeasureFeatures])#

Sets the measurements to be used for further computation. The input can be either a list of MeasureFeatures objects or a dictionary with string keys and MeasureFeatures objects as values.

The method processes the given input to construct a dictionary mapping measurement names to MeasureFeatures instances. If a list is passed, unique class names of the MeasureFeatures instances in the list are used as keys.

Parameters:: measurements (List[MeasureFeatures] | Dict[str, MeasureFeatures]) – A collection of measurement features either as a list of MeasureFeatures objects, where class names are used as keys for dictionary creation, or as a dictionary where keys are predefined strings and values are MeasureFeatures objects.
Raises:: TypeError – If the measurements argument is neither a list nor a dictionary.

Register or update the measurements queue. Measurements can be provided as a list (auto-named by class) or a dictionary (explicit names).

Example:

pipeline = ImagePipeline()

# Set measurements as list
pipeline.set_meas([
    MeasureShape(),
    MeasureIntensity(),
    MeasureTexture()
])

# Set measurements as dict with custom names
pipeline.set_meas({
    'shape': MeasureShape(),
    'intensity': MeasureIntensity(),
    'texture': MeasureTexture()
})

Pipeline Execution#

ImagePipeline.apply(image: Image, inplace: bool = False, reset: bool | None = None) → GridImage | Image#

The class provides an interface to process and apply a series of operations on an Image. The operations are maintained in a queue and executed sequentially when applied to the given Image.

Parameters:

image (Image) – The arr Image to be processed. The type Image refers to an instance of the Image object to which transformations are applied.
inplace (bool, optional) – A flag indicating whether to apply the transformations directly on the provided Image (True) or create a copy of the Image before performing transformations (False). Defaults to False.
reset (bool, optional) – Whether to reset the image before applying the pipeline. If None (default), uses the pipeline’s reset setting from __init__. If explicitly set to True or False, overrides the pipeline setting.

Return type:

Union[GridImage, Image]

Apply all registered operations to an image sequentially. Operations are executed in the order they were registered. Each operation receives the output of the previous operation.

Parameters:

image: Image instance to process
inplace: If True, modify image directly; if False, create a copy (default: False)
reset: If True, reset image state before applying pipeline (default: True)

Returns: Processed Image (or GridImage if input was GridImage)

Example:

# Apply pipeline to single image
img = Image.imread('plate.jpg')
processed = pipeline.apply(img, inplace=False)

# Show before and after
import matplotlib.pyplot as plt
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 6))
img.show(ax=ax1)
ax1.set_title('Original')
processed.plot.overlay(ax=ax2)
ax2.set_title('Processed')

ImagePipeline.measure(image: Image, include_metadata: bool = True, apply_post: bool = True) → pd.DataFrame#

Measures properties of a given image and optionally includes metadata. The method performs measurements using a set of predefined measurement operations. If benchmarking is enabled, the execution time of each measurement is recorded. When verbose mode is active, detailed logging of the measurement process is displayed. A progress bar is used to track progress if the tqdm library is available.

Parameters:

image (Image) – The image object for which measurements are performed. It must support the info method and optionally a grid or objects attribute.
include_metadata (bool, optional) – Indicates whether metadata should be included in the measurements. Defaults to True.
apply_post (bool, optional) – Whether to apply the configured PostMeasurement operations to the merged frame before returning. Defaults to True. Pass False to obtain the pre-post merged DataFrame — useful when the caller (e.g. the CLI) wants to persist a clean copy and apply post separately.

Returns:

A DataFrame containing the results of all performed measurements combined: on the same index.

Return type:

pd.DataFrame

Raises:

Exception – An exception is raised if a measurement operation fails while being applied to the image.

Extract measurements from a processed image using all registered measurement extractors. Results from all measurements are merged into a single DataFrame indexed by object labels.

Parameters:

image: Image instance to measure (must have detected objects)
include_metadata: If True, include image metadata in results (default: True)

Returns: pandas DataFrame with measurements for each detected object

Example:

# Measure after detection
detected = pipeline.apply(img)
measurements = pipeline.measure(detected, include_metadata=True)

# Inspect results
print(measurements.head())
print(measurements.columns)

# Filter by measurement criteria
large_colonies = measurements[measurements['area'] > 500]
bright_colonies = measurements[measurements['mean_intensity'] > 150]

ImagePipeline.apply_and_measure(image: Image, inplace: bool = False, reset: bool | None = None, include_metadata: bool = True, apply_post: bool = True) → pd.DataFrame#

Applies processing to the given image and measures the results.

This function first applies a processing method to the supplied image, adjusting it based on the given parameters. After processing, the resulting image is measured, and a DataFrame containing the measurement data is returned.

Parameters:

image (Image) – The image to process and measure.
inplace (bool) – Whether to modify the original image directly or work on a copy. Default is False.
reset (bool, optional) – Whether to reset any previous processing on the image before applying the current method. If None (default), uses the pipeline’s reset setting. If explicitly set, overrides the pipeline setting.
include_metadata (bool) – Whether to include metadata in the measurement results. Default is True.
apply_post (bool) – Forwarded to measure(). When False, the returned DataFrame skips PostMeasurement ops. Defaults to True.

Returns:

A DataFrame containing measurement data for the processed image.

Return type:

pd.DataFrame

Convenience method that applies all operations and then extracts all measurements in a single call. Equivalent to calling apply() followed by measure().

Parameters:

image: Image instance to process and measure
inplace: If True, modify image directly (default: False)
reset: If True, reset image state before processing (default: True)
include_metadata: If True, include metadata in measurements (default: True)

Returns: pandas DataFrame with measurements for each detected object

Example:

# Process and measure in one call
img = Image.imread('plate.jpg')
results = pipeline.apply_and_measure(img)

# Results ready for analysis
import seaborn as sns
sns.scatterplot(data=results, x='area', y='mean_intensity')

Performance Monitoring#

The ImagePipeline class provides comprehensive performance monitoring capabilities through benchmarking and verbose logging. These features help identify bottlenecks and optimize pipeline execution.

Benchmark Flag:

When benchmark=True, the pipeline tracks execution time for each operation and measurement. Times are stored internally and accessible via benchmark_results().

Verbose Flag:

When verbose=True and benchmark=True, the pipeline prints progress information during execution, including:

Current operation/measurement being executed
Execution time for each step
Progress bar (if tqdm is installed)
Total execution time

ImagePipeline.benchmark_results() → pandas.DataFrame#

Return execution times and memory usage for operations and measurements.

This method should be called after applying the pipeline on an image to get the execution times and memory consumption of the different processes.

When an operation is itself an ImagePipelineCore (nested pipeline), its sub-operations are expanded as indented sub-rows beneath the parent entry with names like "ParentOp > ChildOp".

Returns:

A DataFrame with columns Process Type,: Process Name, Execution Time (s), Memory Delta (MB), and RSS After (MB).

Return type:

pd.DataFrame

Return a DataFrame containing execution times for all operations and measurements from the most recent pipeline run. This method should be called after executing the pipeline with benchmark=True.

Returns: pandas DataFrame with columns:

Process Type: ‘Operation’ or ‘Measurement’
Process Name: Name of the operation/measurement
Execution Time (s): Time in seconds

Example:

# Create pipeline with benchmarking enabled
pipeline = ImagePipeline(
    ops=[GaussianBlur(), OtsuDetector(), RemoveSmallObjects()],
    meas=[MeasureShape(), MeasureIntensity()],
    benchmark=True,
    verbose=True
)

# Apply to image
img = Image.imread('plate.jpg')
results = pipeline.apply_and_measure(img)

# Get performance metrics
performance = pipeline.benchmark_results()
print(performance)
#   Process Type         Process Name  Execution Time (s)
# 0    Operation         GaussianBlur            0.0234
# 1    Operation        OtsuDetector            0.1523
# 2    Operation  RemoveSmallObjects            0.0456
# 3  Measurement         MeasureShape            0.0893
# 4  Measurement     MeasureIntensity            0.0234
# 5        Total        All Processes            0.3340

# Identify bottlenecks
slowest = performance.nlargest(3, 'Execution Time (s)')
print(f"Slowest steps:\n{slowest}")

Optimization Tips:

Based on benchmark results:

Detection is slow: Consider downsampling images or using faster detectors
Measurement is slow: Reduce number of measured features or measure fewer objects
Multiple operations slow: Consider parallelization with ImagePipelineBatch
File I/O slow: Use HDF5 format instead of pickle for large image sets

Progress Reporting:

When both benchmark and verbose are enabled, progress is displayed during execution:

Applying operations: 100%|████████████| 3/3 [00:00<00:00, 12.34it/s]
Operation: OtsuDetector  time=0.1523s
Measuring image properties...
Applying measurements: 100%|████████| 2/2 [00:00<00:00, 16.78it/s]

Serialization & State#

The ImagePipeline class supports serialization to JSON format, enabling pipeline configurations to be saved, shared, and reused across sessions. This ensures reproducibility and facilitates collaboration.

ImagePipeline.to_json(filepath: str | Path | None = None) → str | None#

Serialize the pipeline configuration to JSON format.

This method captures the pipeline’s operations and measurements. It excludes internal state (pydantic PrivateAttr fields) and pandas DataFrames to keep the serialization clean and focused on reproducible configuration.

Parameters:: filepath (str | Path | None) – Optional path to save the JSON. If None, returns JSON string. Can be a string or Path object.
Returns:: JSON string representation of the pipeline configuration.
Return type:: str

Example

Serialize a pipeline to JSON format:

>>> from phenotypic import ImagePipeline
>>> from phenotypic.detect import OtsuDetector
>>> from phenotypic.measure import MeasureShape
>>> pipe = ImagePipeline(pipe_cfgs=[OtsuDetector()], meas=[MeasureShape()])
>>> json_str = pipe.to_json()
>>> pipe.to_json('my_pipeline')  # Save to typed config file

Serialize the pipeline configuration to JSON format. Captures operations, measurements, and configuration flags (benchmark, verbose). Internal state and pandas DataFrames are excluded to keep serialization clean.

Parameters:

filepath: Optional path to save JSON (string or Path). If None, returns JSON string.

Returns: JSON string representation of pipeline configuration

Serialization includes:

Operation instances with their parameters
Measurement instances with their parameters
Benchmark and verbose flags
Operation/measurement order

Serialization excludes:

Internal state (attributes starting with ‘_’)
pandas DataFrames (measurement results)
Temporary computation results
Cached data

Example:

# Create and configure pipeline
pipeline = ImagePipeline(
    ops=[GaussianBlur(sigma=1.5), OtsuDetector()],
    meas=[MeasureShape(), MeasureIntensity()],
    benchmark=True,
    verbose=False
)

# Save to typed pipeline config file
pipeline.to_json('my_pipeline.json.pht-pipe')

# Get JSON string
json_str = pipeline.to_json()
print(json_str)

Example JSON Output:

{
  "ops": {
    "GaussianBlur": {
      "class": "GaussianBlur",
      "params": {"sigma": 1.5}
    },
    "OtsuDetector": {
      "class": "OtsuDetector",
      "params": {}
    }
  },
  "meas": {
    "MeasureShape": {
      "class": "MeasureShape",
      "params": {}
    },
    "MeasureIntensity": {
      "class": "MeasureIntensity",
      "params": {}
    }
  },
  "benchmark": true,
  "verbose": false
}

classmethod ImagePipeline.from_json(json_data: str | Path | dict, benchmark: bool = False, verbose: bool = False, *, skip_unknown_analyzers: bool = False, load_warnings: List[PipelineLoadWarning] | None = None) → ImagePipeline#

Deserialize a pipeline from JSON format.

This method reconstructs a pipeline from a JSON string or file, restoring all operations and measurements. Classes are imported from the phenotypic namespace and instantiated with their saved parameters via model_validate.

Parameters:

json_data (Union[str, Path, dict]) – A JSON string, path to a JSON file, or a pre-parsed dict.
benchmark (bool) – Whether to enable benchmarking for the pipeline. Defaults to False.
verbose (bool) – Whether to enable verbose output. Defaults to False.
skip_unknown_analyzers (bool) – When True, filter and model entries whose class cannot be resolved in the live phenotypic namespace are dropped silently instead of raising AttributeError. Useful for forward-compatible loaders (e.g. the analysis GUI) that prefer to render a partial pipeline plus a warning over a hard 500. Operations, measurements, and post entries are not covered — they still raise on unknown class. Defaults to False (raise on unknown analyzer class, preserving the historical contract).
load_warnings (Optional[List[PipelineLoadWarning]]) – Optional list to be populated with one PipelineLoadWarning per skipped analyzer entry. Only consulted when skip_unknown_analyzers is True. The on-disk JSON is never modified — the caller decides whether to re-save the pruned pipeline.

Returns:

A new pipeline instance with the loaded configuration.

Return type:

SerializablePipeline

Raises:

ValueError – If the JSON is invalid or cannot be parsed.
ImportError – If a required operation or measurement class cannot be imported.
AttributeError – If a class cannot be found in the phenotypic namespace.

Example

Deserialize a pipeline from JSON format:

>>> from phenotypic import ImagePipeline
>>> # Load from file
>>> pipe = ImagePipeline.from_json(saved_path)
>>> # Load from string
>>> json_str = '{"pipe_cfgs": {...}, "meas": {...}}'
>>> pipe = ImagePipeline.from_json(json_str)
>>> # Load with benchmarking enabled
>>> pipe = ImagePipeline.from_json(saved_path, benchmark=True)

Deserialize a pipeline from JSON format. Reconstructs operations, measurements, and configuration flags. Classes are imported from the phenotypic namespace and instantiated with their saved parameters.

Parameters:

json_data: Either a JSON string or a path to a JSON file

Returns: ImagePipeline instance with loaded configuration

Raises:

ValueError: If JSON is invalid or cannot be parsed
ImportError: If a required operation/measurement class cannot be imported
AttributeError: If a class cannot be found in phenotypic namespace

Example:

# Load from file
loaded_pipeline = ImagePipeline.from_json('my_pipeline.json.pht-pipe')

# Load from string
json_str = '{"ops": {...}, "meas": {...}}'
loaded_pipeline = ImagePipeline.from_json(json_str)

# Use loaded pipeline
img = Image.imread('plate.jpg')
results = loaded_pipeline.apply_and_measure(img)

Reproducibility Workflow:

# Researcher 1: Develop and save pipeline
pipeline = ImagePipeline(
    ops=[GaussianBlur(sigma=2.0), OtsuDetector()],
    meas=[MeasureShape()]
)
pipeline.to_json('colony_detection_v1.json.pht-pipe')

# Researcher 2: Load and apply to new data
pipeline = ImagePipeline.from_json('colony_detection_v1.json.pht-pipe')
results = []
for img_path in image_list:
    img = Image.imread(img_path)
    df = pipeline.apply_and_measure(img)
    results.append(df)

# Guaranteed identical processing!

Pipeline Versioning:

# Save pipeline with version tracking
import json
from datetime import datetime

config = json.loads(pipeline.to_json())
config['version'] = '1.0'
config['created'] = datetime.now().isoformat()
config['description'] = 'Colony detection for strain X'

with open('pipeline_v1.0.json', 'w') as f:
    json.dump(config, f, indent=2)

Configuration Persistence#

Pipeline configurations can be version controlled alongside analysis code, ensuring that:

Processing parameters are documented and reproducible
Changes to pipelines are tracked over time
Collaborators use identical processing workflows
Results can be regenerated with exact parameter settings

Best Practices:

Version pipeline files: Use descriptive names like detection_v1.0.json
Document parameters: Add comments explaining parameter choices
Test after loading: Verify loaded pipeline produces expected results
Commit to version control: Include pipeline JSON in git repositories
Archive with data: Store pipeline configuration alongside processed data

Batch Processing#

For processing multiple images efficiently, consider using ImagePipelineBatch which extends ImagePipeline with multiprocessing capabilities:

from phenotypic import ImagePipelineBatch, ImageSet

# Create batch pipeline with parallel processing
batch_pipeline = ImagePipelineBatch(
    ops=[GaussianBlur(), OtsuDetector()],
    meas=[MeasureShape(), MeasureIntensity()],
    n_jobs=4,  # Use 4 worker processes
    verbose=True
)

# Load image set
image_set = ImageSet.from_directory('plates/')

# Process in parallel
results = batch_pipeline.apply_and_measure(image_set)

See ImagePipelineBatch documentation for details on parallel processing, memory management, and progress tracking for large-scale analyses.

Comparison & Utility#

The ImagePipeline class provides standard comparison and utility methods:

String Representation:

pipeline = ImagePipeline(
    ops=[OtsuDetector()],
    meas=[MeasureShape()]
)

print(pipeline)
# Output: ImagePipeline(ops=1, meas=1, benchmark=False)

print(repr(pipeline))
# Output: <ImagePipeline: 1 operations, 1 measurements>

Accessing Operations and Measurements:

# Access operations dictionary
ops_dict = pipeline._ops
print(ops_dict.keys())  # ['OtsuDetector']

# Access measurements dictionary
meas_dict = pipeline._meas
print(meas_dict.keys())  # ['MeasureShape']

# Modify individual operation parameters
pipeline._ops['OtsuDetector'].threshold_multiplier = 1.2

Note

Direct modification of _ops and _meas dictionaries is possible but not recommended. Use set_ops() and set_meas() methods instead to ensure proper validation and naming.

ImagePipeline Class Methods#

Overview#

Initialization & Configuration#

Pipeline Execution#

Performance Monitoring#

Serialization & State#

Configuration Persistence#

Batch Processing#

Comparison & Utility#

This Page