from __future__ import annotations
from typing import TYPE_CHECKING, Any, List, Literal
import numpy as np
from pydantic import Field, field_validator
if TYPE_CHECKING:
from phenotypic._core._image import Image
from ..abc_ import ImageEnhancer
from ..sdk_.mixin import NormalizedOutputMixin
from ..sdk_.typing_ import NormOut, OperationField
from ._blur_gauss import BlurGauss
from ._median_filter import MedianFilter
CombineMode = Literal["max", "mean", "min", "median"]
[docs]
class CompositeEnhance(NormalizedOutputMixin, ImageEnhancer):
"""Enhance ``detect_mat`` by combining several enhancers' response maps pixel-wise.
Apply one or more enhancers (or preprocessing pipelines ending in an
enhancer) to the same plate image and reduce their ``detect_mat`` outputs
into a single response map. The immutable grayscale input can optionally
participate as another response source. This ensembles complementary
preprocessing strategies: where one enhancer responds strongly to faint
colonies and another to sharp edges, ``'max'`` keeps the strongest signal
from either. It is the enhancer-side analogue of
:class:`CompositeDetector`, which combines binary masks rather than
continuous response maps.
Each enhancer branch reads the *same* input ``detect_mat`` independently
(branches do not chain into one another), so the order of ``ops`` does not
affect the result for the commutative reductions used here.
Best For:
- Fusing complementary focus maps (e.g. an edge response and a blob
response) so colonies that only one branch emphasises survive.
- Averaging several denoisers (``'mean'`` / ``'median'``) to suppress
branch-specific artefacts before thresholding.
- Consensus preprocessing (``'min'``), where a pixel stays bright only
if every branch agrees it is signal.
Consider Also:
- A single :class:`ImageEnhancer` when one preprocessing step already
isolates colonies reliably.
- :class:`CompositeDetector` when you want to combine *detections*
(binary masks) rather than continuous enhancement maps.
- An :class:`ImagePipeline` of enhancers when the steps should be
applied *in sequence* rather than combined in parallel.
Args:
ops: List of :class:`~phenotypic.abc_.ImageEnhancer` or
:class:`~phenotypic.ImagePipeline` instances to combine. Pipelines
allow per-branch preprocessing before the response is collected.
Defaults to ``[BlurGauss(), MedianFilter()]`` when not
specified. ``None`` entries mark unfilled GUI-builder slots and
are skipped.
mode: Pixel-wise reduction across branch response maps. ``'max'``
keeps the brightest response (union of complementary signals,
the default). ``'min'`` keeps the darkest (consensus suppression).
``'mean'`` averages all branches. ``'median'`` takes the per-pixel
median (robust to a single outlier branch; most useful with three
or more enhancers). Default: ``'max'``.
include_gray: Include the immutable input ``image.gray`` layer as one
additional response map in the selected reduction. This allows a
branch response such as phase congruency to be fused with the
original grayscale signal before a downstream detector. Defaults
to ``False``.
norm: Output range policy applied after reduction. ``None`` (default)
passes values through untouched, which is what an arbitrarily-scaled
(non unit-normalised) response map needs. ``"clip"`` saturates
values outside [0, 1]; ``"rescale"`` remaps the full observed range
onto [0, 1].
Returns:
Image: Input image with ``detect_mat`` set to the combined response
map. ``rgb`` and ``gray`` are unchanged.
Raises:
ValueError: If ``ops`` is empty or contains only ``None`` slots while
``include_gray`` is ``False``.
Examples:
Combine two enhancers, keeping the strongest response per pixel:
>>> from phenotypic.data import load_synth_yeast_plate
>>> from phenotypic.enhance import (
... CompositeEnhance,
... BlurGauss,
... SubtractRollingBall,
... )
>>> image = load_synth_yeast_plate()
>>> combiner = CompositeEnhance(
... ops=[BlurGauss(sigma=1.5), SubtractRollingBall()],
... mode="max",
... )
>>> enhanced = combiner.apply(image)
>>> bool((image.gray[:] == enhanced.gray[:]).all()) # rgb/gray untouched
True
Average three branches and clamp the result to ``[0, 1]``:
>>> from phenotypic.enhance import MedianFilter, EnhanceLocalContrast
>>> combiner = CompositeEnhance(
... ops=[BlurGauss(), MedianFilter(), EnhanceLocalContrast()],
... mode="mean",
... norm="clip",
... )
>>> enhanced = combiner.apply(load_synth_yeast_plate())
>>> float(enhanced.detect_mat[:].max()) <= 1.0
True
"""
# Each branch may be an ``ImageEnhancer`` or a nested ``ImagePipeline``.
# ``OperationField`` keeps the concrete class of each entry across a JSON
# round-trip (a bare ``model_dump`` of the union would lose the subclass).
# ``| None`` permits an empty list slot that the GUI builder uses to mark
# an unfilled enhancer slot in an in-progress pipeline.
ops: List[OperationField | None] = Field(
default_factory=lambda: [BlurGauss(), MedianFilter()]
)
mode: CombineMode = "max"
include_gray: bool = False
norm: NormOut = None
@field_validator("ops", mode="before")
@classmethod
def _default_ops(cls, value: Any) -> Any:
"""Map an explicit ``None`` onto the default enhancer pair.
Mirrors :class:`CompositeDetector`: the ``default_factory`` covers the
omitted-argument case, and this validator preserves an explicit
``ops=None`` call by substituting the default pair.
"""
if value is None:
return [BlurGauss(), MedianFilter()]
return value
def _operate(self, image: Image) -> Image:
"""Reduce enhancer responses and optional grayscale pixel-wise."""
# Import here to avoid a circular dependency at module load.
from phenotypic import ImagePipeline
response_maps: list[np.ndarray] = []
if self.include_gray:
response_maps.append(np.asarray(image.gray[:], dtype=float))
for enhancer in self.ops:
if enhancer is None:
# An unfilled GUI-builder slot; nothing to apply -- skip it.
continue
if isinstance(enhancer, ImagePipeline):
enhanced = enhancer.apply(image, inplace=False, reset=False)
else:
enhanced = enhancer.apply(image, inplace=False)
response_maps.append(np.asarray(enhanced.detect_mat[:], dtype=float))
if not response_maps:
raise ValueError(
"At least one enhancer or include_gray=True must be provided "
"to CompositeEnhance"
)
combined = self._apply_norm(self._combine(response_maps))
original = image.detect_mat[:]
image.detect_mat[:] = combined.astype(original.dtype, copy=False)
return image
def _combine(self, response_maps: list[np.ndarray]) -> np.ndarray:
"""Reduce stacked branch response maps according to ``self.mode``."""
if self.mode == "max":
return np.maximum.reduce(response_maps)
if self.mode == "min":
return np.minimum.reduce(response_maps)
stack = np.stack(response_maps, axis=0)
if self.mode == "mean":
return np.mean(stack, axis=0)
# "median"
return np.median(stack, axis=0)
