phenotypic.abc_.ImageOperation

class phenotypic.abc_.ImageOperation

Bases: BaseOperation, LazyWidgetMixin, ABC

Core abstract base class for all single-image transformation operations in PhenoTypic.

ImageOperation is the foundation of PhenoTypic’s algorithm system. It defines the interface for algorithms that transform an Image object by modifying specific components. Unlike GridOperation (which handles grid-aligned operations on plate images), ImageOperation acts on a single image independently.

What is ImageOperation?

ImageOperation manages the distinction between:

• apply() method: The user-facing interface that handles memory management (copy vs. in-place) and integrity validation

• _operate() method: The abstract algorithm-specific method that subclasses implement with the actual processing logic

This separation ensures consistent behavior, automatic memory tracking, and validation across all image operations.

The Operation Hierarchy

ImageOperation has four main subclass categories, each modifying different image components with different integrity guarantees:

ImageOperation (this class)
├── ImageEnhancer
│   └── Modifies ONLY image.enh_gray
│       ├── GaussianBlur, CLAHE, RankMedianEnhancer, ...
│       └── Use for: noise reduction, contrast, edge sharpening
│
├── ObjectDetector
│   └── Modifies ONLY image.objmask and image.objmap
│       ├── OtsuDetector, CannyDetector, RoundPeaksDetector, ...
│       └── Use for: discovering and labeling colonies/particles
│
├── ObjectRefiner
│   └── Modifies ONLY image.objmask and image.objmap
│       ├── Size filtering, merging, removing objects
│       └── Use for: cleaning up detection results
│
└── ImageCorrector
    └── Modifies ALL image components
        ├── GridAligner, rotation, color correction
        └── Use for: general-purpose transformations

When to inherit from each subclass:

• ImageEnhancer: You only modify image.enh_gray (enhanced grayscale). Original image.rgb and image.gray are protected by integrity checks. Typical use: preprocessing before detection.

• ObjectDetector: You analyze image data and produce only image.objmask (binary mask) and image.objmap (labeled object map). Input image data is protected. Typical use: colony detection, particle finding.

• ObjectRefiner: You edit the mask and map (filtering, merging, removing). Input image data is protected. Typical use: post-detection cleanup.

• ImageCorrector: You transform the entire Image (every component may change). No integrity checks are performed. Typical use: rotation, alignment, global color correction.

Never inherit directly from ImageOperation. Always choose one of the four subclasses above, as each provides appropriate integrity validation and shared utilities (e.g., _make_footprint() for morphology operations).

How apply() and _operate() work together

The user-facing method apply(image, inplace=False) is the entry point:

1. Calls ``_apply_to_single_image()`` with the operation logic

2. Handles copy/inplace semantics: - If inplace=False (default): Image is copied before modification,

   original unchanged

   • If inplace=True: Image is modified in-place for memory efficiency

3. Extracts matched parameters via _get_matched_operation_args() - Matches operation instance attributes to _operate() method parameters - Enables parallel execution in pipelines

4. Calls your _operate() static method with the image and matched parameters

5. Validates integrity (subclass-specific via @validate_operation_integrity) - Detects unexpected modifications to protected image components - Only enabled if VALIDATE_OPS=True in environment

Your subclass only needs to implement _operate(image, **kwargs) -> Image.

The _operate() method contract

_operate() is a static method (required for parallel execution):

• Signature: @staticmethod def _operate(image: Image, param1, param2=default) -> Image

• Parameters: All parameters except image are automatically matched to instance attributes via the _get_matched_operation_args() system

• Behavior: Modify only the allowed image components (determined by subclass)

• Returns: The modified Image object

• Must be static: This enables serialization and parallel execution

Example parameter matching:

class MyEnhancer(ImageEnhancer):
    def __init__(self, sigma: float):
        super().__init__()
        self.sigma = sigma  # Instance attribute

    @staticmethod
    def _operate(image: Image, sigma: float = 1.0) -> Image:
        # When apply() is called, 'sigma' is automatically passed from self.sigma
        image.enh_gray[:] = gaussian_filter(image.enh_gray[:], sigma=sigma)
        return image

The _apply_to_single_image() static method retrieves sigma from the instance (via _get_matched_operation_args()) and passes it to _operate().

Data access through accessors

Within _operate(), always access image data through accessors (never direct attribute modification). This ensures lazy evaluation, caching, and consistency:

Reading data:

• image.enh_gray[:] - Enhanced grayscale (for enhancers)

• image.rgb[:] - Original RGB data

• image.gray[:] - Luminance grayscale

• image.objmask[:] - Binary object mask

• image.objmap[:] - Labeled object map

• image.color.Lab[:], image.color.HSV[:] - Color spaces

Modifying data:

• image.enh_gray[:] = new_array - Set enhanced grayscale

• image.objmask[:] = binary_array - Set object mask

• image.objmap[:] = labeled_array - Set object map

Never do this:

# ✗ WRONG - direct attribute modification
image.rgb = new_data
image._enh_gray = new_data
image.objects_handler.enh_gray = new_data

Do this instead:

# ✓ CORRECT - use accessors
image.enh_gray[:] = new_data
image.objmask[:] = new_mask

Integrity validation with @validate_operation_integrity

Intermediate subclasses use the @validate_operation_integrity decorator to enforce that modifications are limited to specific components. For example:

class ImageEnhancer(ImageOperation, ABC):
    @validate_operation_integrity('image.rgb', 'image.gray')
    def apply(self, image: Image, inplace=False) -> Image:
        return super().apply(image=image, inplace=inplace)

This decorator:

1. Calculates MurmurHash3 signatures of protected arrays before apply()

2. Calls the parent apply() method

3. Recalculates signatures after operation completes

4. Raises OperationIntegrityError if any protected component changed

Only enabled if VALIDATE_OPS=True in environment (for performance).

Operation chaining and pipelines

Operations are designed for method chaining:

result = (GaussianBlur(sigma=2).apply(image)
         .apply_operation(OtsuDetector()))

Or use ImagePipeline for multi-step workflows with automatic benchmarking:

pipeline = ImagePipeline()
pipeline.add(GaussianBlur(sigma=2))
pipeline.add(OtsuDetector())
pipeline.add(GridFinder())

results = pipeline.operate([image1, image2, image3])

Parallel execution support

ImageOperation’s static method design enables parallel execution. When ImagePipeline runs with multiple images, it:

1. Extracts operation parameters via _get_matched_operation_args()

2. Serializes the operation instance (attributes only)

3. Sends to worker processes

4. Workers call _apply_to_single_image() in parallel

This is why _operate() must be static and all parameters must be instance attributes matching the method signature.

None

Type:

all operation state is stored in subclass instances as attributes

apply(image, inplace=False)

User-facing method that applies the operation. Handles copy/inplace logic and parameter matching.

Parameters:

image (Image)

Return type:

Image

_operate(image, **kwargs)

Abstract static method implemented by subclasses with algorithm logic. Parameters are automatically extracted from instance attributes via _get_matched_operation_args().

_apply_to_single_image(cls_name, image, operation, inplace, matched_args)

Static method that performs the actual apply operation. Handles copy/inplace logic and error handling. Called internally by apply(). Also called directly by ImagePipeline for parallel execution.

Notes

• No direct Image attribute modification: Never write to image.rgb, image.gray, or other attributes directly. Use the accessor pattern (image.component[:] = new_data).

• Immutability by default: Operations return modified copies by default. Original image is unchanged unless inplace=True is explicitly passed.

• Static _operate() is required: The method must be static (decorated with @staticmethod) to support parallel execution in pipelines. This enables ImagePipeline to serialize operations and execute them in worker processes.

• Parameter matching for parallelization: All _operate() parameters (except image) must exist as instance attributes with the same name. When apply() is called, _get_matched_operation_args() extracts these values and passes them to _operate(). This is why subclasses store operation parameters as self.param_name in __init__.

• Automatic memory/performance tracking: BaseOperation (parent class) automatically tracks memory usage and execution time when the logger is configured for INFO level or higher. Disable by setting logger to WARNING.

• Cross-platform compatibility: Some dependencies (rawpy, pympler) are platform-specific. Code must gracefully handle missing optional dependencies.

• Integrity validation is optional: The @validate_operation_integrity decorator only runs if VALIDATE_OPS=True in environment. This provides development-time safety without production overhead.

Examples

Implementing a custom ImageEnhancer with parameter matching

from phenotypic.abc_ import ImageEnhancer
from phenotypic import Image
from scipy.ndimage import gaussian_filter

class GaussianEnhancer(ImageEnhancer):
    '''Custom enhancer applying Gaussian blur to enh_gray.'''

    def __init__(self, sigma: float = 1.0):
        super().__init__()
        self.sigma = sigma  # Instance attribute matched to _operate()

    @staticmethod
    def _operate(image: Image, sigma: float = 1.0) -> Image:
        '''Apply Gaussian blur to enh_gray.'''
        # Read enhanced grayscale
        enh = image.enh_gray[:]

        # Apply Gaussian filter
        blurred = gaussian_filter(enh.astype(float), sigma=sigma)

        # Modify enh_gray through accessor
        image.enh_gray[:] = blurred.astype(enh.dtype)

        return image

# Usage
enhancer = GaussianEnhancer(sigma=2.5)
enhanced = enhancer.apply(image)  # Original unchanged
enhanced_inplace = enhancer.apply(image, inplace=True)  # Original modified

Implementing a custom ObjectDetector

from phenotypic.abc_ import ObjectDetector
from phenotypic import Image
from skimage.feature import peak_local_max
from skimage.measure import label as measure_label
import numpy as np

class PeakDetector(ObjectDetector):
    '''Detector using local peak finding to locate colonies.'''

    def __init__(self, min_distance: int = 10, threshold_abs: int = 100):
        super().__init__()
        self.min_distance = min_distance
        self.threshold_abs = threshold_abs

    @staticmethod
    def _operate(image: Image, min_distance: int = 10,
                 threshold_abs: int = 100) -> Image:
        '''Find peaks in enh_gray and create object mask/map.'''
        # Find local maxima (colony peaks)
        coords = peak_local_max(
            image.enh_gray[:],
            min_distance=min_distance,
            threshold_abs=threshold_abs
        )

        # Create binary mask from peaks
        mask = np.zeros(image.enh_gray.shape, dtype=bool)
        for y, x in coords:
            mask[y, x] = True

        # Create labeled map from mask
        labeled_map = measure_label(mask)

        # Set detection results
        image.objmask[:] = mask
        image.objmap[:] = labeled_map

        return image

# Usage - automatic integrity validation in ImageDetector
detector = PeakDetector(min_distance=15, threshold_abs=120)
detected = detector.apply(image)
colonies = detected.objects
print(f"Detected {len(colonies)} colonies")

Understanding inplace parameter and memory efficiency

from phenotypic.enhance import GaussianBlur
from phenotypic import Image

image = Image.from_image_path('colony_plate.jpg')
enhancer = GaussianBlur(sigma=2.0)

# Default: inplace=False (safe, creates copy)
enhanced = enhancer.apply(image)
print(f"Same object? {id(image) == id(enhanced)}")  # False

# For memory efficiency with large images
result = enhancer.apply(image, inplace=True)
print(f"Same object? {id(image) == id(result)}")  # True

# inplace=True is useful in pipelines with many large images
# to minimize memory overhead, but modifies the original

Using operations in a processing pipeline

from phenotypic import Image, ImagePipeline
from phenotypic.enhance import GaussianBlur
from phenotypic.detect import OtsuDetector
from phenotypic.grid import GridFinder

# Load image
image = Image.from_image_path('colony_plate.jpg')

# Sequential chaining
enhanced = GaussianBlur(sigma=2).apply(image)
detected = OtsuDetector().apply(enhanced)
grid = GridFinder().apply(detected)

# Or use ImagePipeline for batch processing
pipeline = ImagePipeline()
pipeline.add(GaussianBlur(sigma=2))
pipeline.add(OtsuDetector())
pipeline.add(GridFinder())

# Process multiple images with automatic parallelization
images = [Image.from_image_path(f) for f in image_files]
results = pipeline.operate(images)
# Results are fully processed images

How parameter matching enables parallel execution

from phenotypic.abc_ import ImageOperation
from phenotypic import Image

class CustomThreshold(ImageOperation):
    def __init__(self, threshold: int, min_size: int = 5):
        super().__init__()
        self.threshold = threshold  # Matched to _operate
        self.min_size = min_size    # Matched to _operate

    @staticmethod
    def _operate(image: Image, threshold: int,
                 min_size: int = 5) -> Image:
        # 'threshold' and 'min_size' automatically passed
        binary = image.enh_gray[:] > threshold
        image.objmask[:] = binary
        return image

# When apply() is called:
op = CustomThreshold(threshold=100, min_size=10)

# apply() internally:
# 1. Calls _get_matched_operation_args()
# 2. Extracts: {'threshold': 100, 'min_size': 10}
# 3. Calls _apply_to_single_image(..., matched_args=...)
# 4. _apply_to_single_image passes kwargs to _operate()

result = op.apply(image)

Methods

__init__
apply	Applies the operation to an image, either in-place or on a copy.
dispose_widgets	Drop references to the UI widgets.
sync_widgets_from_state	Push internal state into widgets.
widget	Return (and optionally display) the root widget.

apply(image: Image, inplace=False) → Image

Applies the operation to an image, either in-place or on a copy.

Parameters:

• image (Image) – The arr image to apply the operation on.

• inplace (bool) – If True, modifies the image in place; otherwise, operates on a copy of the image.

Returns:

The modified image after applying the operation.

Return type:

Image

__del__()

Automatically stop tracemalloc when the object is deleted.

__getstate__()

Prepare the object for pickling by disposing of any widgets.

This ensures that UI components (which may contain unpickleable objects like input functions or thread locks) are cleaned up before serialization.

Note

This method modifies the object state by calling dispose_widgets(). Any active widgets will be detached from the object.

dispose_widgets() → None

Drop references to the UI widgets.

Return type:

None

sync_widgets_from_state() → None

Push internal state into widgets.

Return type:

None

widget(image: Image | None = None, show: bool = False) → Widget

Return (and optionally display) the root widget.

Parameters:

• image (Image | None) – Optional image to visualize. If provided, visualization controls will be added to the widget.

• show (bool) – Whether to display the widget immediately. Defaults to False.

Returns:

The root widget.

Return type:

ipywidgets.Widget

Raises:

ImportError – If ipywidgets or IPython are not installed.