Source code for phenotypic.detect._filfinder_detector

"""Lazy FilFinder 1.8 raster-product adapter."""

from __future__ import annotations

from concurrent.futures import Future, ProcessPoolExecutor
import re
from typing import (
    TYPE_CHECKING,
    Annotated,
    Any,
    Callable,
    Generic,
    overload,
    cast,
    TypeVar,
)
import warnings

import numpy as np
from pydantic import Field
from scipy import ndimage
from typing_extensions import Self

from phenotypic.abc_ import ObjectDetector
from phenotypic.sdk_.typing_ import (
    FilFinderOutput,
    FilFinderPruneCriteria,
    TuneSpec,
)

if TYPE_CHECKING:
    from types import TracebackType

    from phenotypic._core._image import Image
    from phenotypic._core._grid_image import GridImage


EXPECTED_SUPPLIED_MASK_WARNING = (
    "Using inputted mask. Skipping creation of anew mask."
)
_TOPOLOGY_IMPORT_ERROR = (
    "FilFinderDetector requires FilFinder and Astropy. Install PhenoTypic "
    "with the `topology` extra before applying a nonempty detection."
)
_T = TypeVar("_T")
_WarningRecord = tuple[str, type[Warning], str, int]
_WorkerResult = tuple[int, str, _T, list[_WarningRecord]]


def _execute_with_warning_capture(
    task_index: int,
    function: Callable[..., _T],
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
) -> _WorkerResult[_T]:
    """Execute one process task and serialize its keyed warning records."""
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        result = function(*args, **kwargs)
    records: list[_WarningRecord] = [
        (str(item.message), item.category, item.filename, item.lineno)
        for item in caught
    ]
    return task_index, function.__qualname__, result, records


class _WarningForwardingFuture(Generic[_T]):
    """Expose a source-compatible future that retains and re-emits warnings."""

    def __init__(
        self,
        future: Future[_WorkerResult[_T]],
        task_index: int,
        warning_sink: dict[int, dict[str, object]],
    ) -> None:
        self._future = future
        self._task_index = task_index
        self._warning_sink = warning_sink

    def result(self, timeout: float | None = None) -> _T:
        """Return the child result after keyed parent warning forwarding."""
        if timeout is None:
            task_index, function_name, result, records = self._future.result()
        else:
            task_index, function_name, result, records = self._future.result(
                timeout=timeout
            )
        if task_index != self._task_index:
            raise RuntimeError("FilFinder worker task order changed")
        self._warning_sink[task_index] = {
            "task_index": task_index,
            "function": function_name,
            "warnings": records,
        }
        for message, category, filename, lineno in records:
            warnings.warn_explicit(message, category, filename, lineno)
        return result


class _WarningForwardingProcessPool:
    """Own one real process while preserving keyed worker-warning visibility."""

    def __init__(self, *, max_workers: int) -> None:
        self._executor = ProcessPoolExecutor(max_workers=max_workers)
        self._next_task_index = 0
        self.warning_records_by_task: dict[int, dict[str, object]] = {}

    def submit(
        self,
        function: Callable[..., _T],
        /,
        *args: Any,
        **kwargs: Any,
    ) -> _WarningForwardingFuture:
        """Submit one indexed task through the warning-capture trampoline."""
        task_index = self._next_task_index
        self._next_task_index += 1
        future = self._executor.submit(
            _execute_with_warning_capture,
            task_index,
            function,
            args,
            kwargs,
        )
        return _WarningForwardingFuture(
            future, task_index, self.warning_records_by_task
        )

    def shutdown(
        self,
        wait: bool = True,
        *,
        cancel_futures: bool = False,
    ) -> None:
        """Shut down the owned process executor."""
        self._executor.shutdown(wait=wait, cancel_futures=cancel_futures)

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        del exc_type, exc_value, traceback
        self.shutdown(wait=True)


def _create_warning_forwarding_pool() -> _WarningForwardingProcessPool:
    """Return the contract's fresh one-process warning-forwarding executor."""
    return _WarningForwardingProcessPool(max_workers=1)


def _load_filfinder_runtime() -> tuple[Any, Any]:
    """Import the optional runtime only for a nonempty application."""
    try:
        import astropy.units as units  # type: ignore[import-untyped, import-not-found]
        from fil_finder import FilFinder2D  # type: ignore[import-untyped, import-not-found]
    except ImportError as error:
        raise ImportError(_TOPOLOGY_IMPORT_ERROR) from error
    return FilFinder2D, units


def _copy_float32_source(detect_mat: np.ndarray) -> np.ndarray:
    """Apply the ImageData float32 seam before making the float64 source copy."""
    quantized = np.asarray(detect_mat, dtype=np.float32)
    return np.array(quantized, dtype=np.float64, copy=True)


def _create_mask_with_narrow_warning_policy(filfinder: Any) -> None:
    """Create the supplied mask while suppressing one exact source warning."""
    with warnings.catch_warnings():
        warnings.filterwarnings(
            "ignore",
            message=f"^{re.escape(EXPECTED_SUPPLIED_MASK_WARNING)}$",
            category=UserWarning,
        )
        filfinder.create_mask(use_existing_mask=True)


[docs] class FilFinderDetector(ObjectDetector): """Detect filament rasters through the pinned FilFinder 1.8 workflow. Threshold ``detect_mat`` into a supplied FilFinder mask, then return the existing mask, pre-prune medial skeleton, or analyzed longest-path raster. The selected product becomes consecutive 8-connected object labels. Best For: - Connected linear growth for which a medial skeleton is informative. - Comparing an inclusive threshold mask with its skeleton. - Topology workflows that require FilFinder's pruned longest path. Consider Also: - :class:`OtsuDetector` for compact colonies separated by one threshold. - :class:`FilamentousFungiDetector` for the complete fungal workflow. Args: threshold: Inclusive ``detect_mat`` threshold in ``[0, 1]``. Equality is foreground. Default: 0.5. output: Raster product. Accepted values are ``"mask"``, ``"skeleton"``, and ``"longest_path"``. Default: ``"mask"``. beamwidth_px: Positive finite FilFinder beam width in pixels. Default: 1.0. prune_criteria: Branch-pruning criterion. Accepted values are ``"all"``, ``"intensity"``, and ``"length"``. Default: ``"all"``. relative_intensity_threshold: Relative branch-intensity cutoff in ``(0, 1]``. Default: 0.2. branch_threshold_px: Positive finite branch-length cutoff in pixels, or ``None`` for FilFinder's three-beam-width default. Default: None. max_prune_iterations: Positive pruning-iteration cap. Default: 10. rng_seed: Nonnegative medial-axis tie-breaking seed. Default: 0. Returns: Image: A copy by default with ``objmask`` equal to the selected raster and ``objmap`` equal to deterministic 8-connected labels. Raises: ImportError: A nonempty application cannot import the ``topology`` extra. Examples: Construct the operation without importing optional dependencies: >>> from phenotypic.detect._filfinder_detector import FilFinderDetector >>> detector = FilFinderDetector(output="skeleton", rng_seed=7) >>> (detector.output, detector.rng_seed) ('skeleton', 7) References: Koch, E. W., and Rosolowsky, E. W. (2015), "Filament identification through mathematical morphology," MNRAS, 452(4), 3435-3450. """ threshold: Annotated[ float, Field(ge=0.0, le=1.0, allow_inf_nan=False), ] = 0.5 output: FilFinderOutput = "mask" beamwidth_px: Annotated[ float, Field(gt=0.0, allow_inf_nan=False), ] = 1.0 prune_criteria: FilFinderPruneCriteria = "all" relative_intensity_threshold: Annotated[ float, Field(gt=0.0, le=1.0, allow_inf_nan=False), ] = 0.2 branch_threshold_px: Annotated[ float | None, Field(gt=0.0, allow_inf_nan=False), TuneSpec(tunable=False), ] = None max_prune_iterations: Annotated[int, Field(ge=1, strict=True)] = 10 rng_seed: Annotated[int, Field(ge=0, strict=True)] = 0 @overload def apply(self, image: GridImage, inplace: bool = False) -> GridImage: ... @overload def apply(self, image: Image, inplace: bool = False) -> Image: ...
[docs] def apply( self, image: Image | GridImage, inplace: bool = False ) -> Image | GridImage: """Apply detection while preserving the dependency error type.""" try: return super().apply(image=image, inplace=inplace) except RuntimeError as error: cause: BaseException | None = error while cause is not None: if ( isinstance(cause, ImportError) and str(cause) == _TOPOLOGY_IMPORT_ERROR ): raise ImportError(_TOPOLOGY_IMPORT_ERROR) from cause cause = cause.__cause__ raise
def _operate(self, image: Image) -> Image: """Run the frozen FilFinder stage graph and label its selected raster.""" source_image = _copy_float32_source(image.detect_mat[:]) threshold_mask = source_image >= self.threshold if not threshold_mask.any(): image.objmask[:] = np.zeros_like(threshold_mask, dtype=bool) image.objmap[:] = np.zeros_like(threshold_mask, dtype=np.int32) return image FilFinder2D, units = _load_filfinder_runtime() pool = _create_warning_forwarding_pool() try: filfinder = FilFinder2D( source_image.copy(), beamwidth=self.beamwidth_px * units.pix, mask=threshold_mask.copy(), pool=pool, ) _create_mask_with_narrow_warning_policy(filfinder) if self.output == "mask": selected = np.asarray(filfinder.mask, dtype=bool).copy() else: filfinder.medskel(rng=self.rng_seed) if self.output == "skeleton": selected = np.asarray( filfinder.skeleton, dtype=bool ).copy() else: branch_threshold = ( None if self.branch_threshold_px is None else self.branch_threshold_px * units.pix ) filfinder.analyze_skeletons( prune_criteria=self.prune_criteria, relintens_thresh=self.relative_intensity_threshold, skel_thresh=1.0 * units.pix, branch_thresh=branch_threshold, max_prune_iter=self.max_prune_iterations, ) selected = np.asarray( filfinder.skeleton_longpath, dtype=bool, ).copy() finally: pool.shutdown(wait=True) objmap, _ = cast( tuple[np.ndarray, int], ndimage.label( selected, structure=np.ones((3, 3), dtype=np.uint8), ), ) image.objmap[:] = objmap image.objmask[:] = objmap > 0 return image
FilFinderDetector.apply.__doc__ = FilFinderDetector.__doc__