OutlierDetector

Class for exploring tile image metrics and detecting outliers.

Source code in histoslice/utils/_process.py

class OutlierDetector:
    """Class for exploring tile image metrics and detecting outliers."""

    def __init__(self, dataframe: pl.DataFrame) -> None:
        """Initialize `TileMetadata` class instance.

        Args:
            dataframe: Polars dataframe with image metrics.

        Raises:
            ValueError: Dataframe does not contain any metric columns.
        """
        self.__dataframe = dataframe
        self.__outliers = np.repeat([False], repeats=len(dataframe))
        self.__outlier_selections = []
        self.__metric_columns = [
            x for x in dataframe.columns if "_q" in x or (x.endswith(("_mean", "_std")))
        ]
        if len(self.__metric_columns) == 0:
            raise ValueError(ERROR_NO_METRICS)
        if "path" not in dataframe.columns:
            raise ValueError(ERROR_NO_PATHS)

    @classmethod
    def from_parquet(cls, *args, **kwargs) -> "OutlierDetector":
        """Wrapper around `polars.read_parquet` function."""
        return cls(pl.read_parquet(*args, **kwargs))

    @property
    def dataframe(self) -> pl.DataFrame:
        """Polars dataframe with metadata."""
        return self.__dataframe

    @property
    def dataframe_without_metrics(self) -> pl.DataFrame:
        """Polars dataframe without metadata."""
        forbidden = ["background", "black_pixels", "white_pixels", *self.metric_columns]
        return self.dataframe[[x for x in self.dataframe.columns if x not in forbidden]]

    @property
    def coordinates(self) -> np.ndarray:
        """Array of tile coordinates."""
        return self.dataframe[XYWH_COLUMNS].to_numpy()

    @property
    def paths(self) -> np.ndarray:
        """Array of tile paths."""
        return self["path"]

    @property
    def outliers(self) -> np.ndarray:
        """Array of outlier indices."""
        return self.__outliers

    @property
    def outlier_selections(self) -> list[dict[str, Union[np.ndarray, str]]]:
        """List of dicts with outlier selections and descriptions."""
        return self.__outlier_selections

    @property
    def metric_columns(self) -> np.ndarray:
        """Image metric columns."""
        return self.__metric_columns

    @property
    def metrics(self) -> np.ndarray:
        """Array of normalized image metrics (divided by 255)."""
        return self.dataframe[self.metric_columns].to_numpy() / 255

    @property
    def mean_and_std(
        self,
    ) -> tuple[tuple[float, float, float], tuple[float, float, float]]:
        """Means and standard deviations for RGB channels."""
        mean = tuple(
            float(x)
            for x in (self.dataframe[RGB_MEAN_COLUMNS].mean() / 255).to_numpy()[0]
        )
        std = tuple(
            float(x)
            for x in (self.dataframe[RGB_STD_COLUMNS].mean() / 255).to_numpy()[0]
        )
        return mean, std

    def add_outliers(self, selection: np.ndarray, *, desc: str) -> None:
        """Set outliers to `True` with selection, and append the selection to
        `outlier_selections` property.

        Args:
            selection: Selection for indexing `outliers` property.
            desc: Description for selection.
        """
        self.__outliers[selection] = True
        self.__outlier_selections.append({"selection": selection, "desc": desc})

    def random_image_collage(
        self,
        selection: np.ndarray,
        *,
        num_rows: int = 4,
        num_cols: int = 16,
        shape: tuple[int, int] = (64, 64),
        num_workers: int = 1,
    ) -> Image.Image:
        """Generate a random collage from `paths[selection]`.

        Args:
            selection: Selection for paths.
            num_rows: Number of rows in the collage image. Defaults to 4.
            num_cols: Number of columns in the collage image. Defaults to 16.
            shape: Size of each image in the collage. Defaults to (64, 64).
            num_workers: Number of image loading workers. Defaults to 1.

        Returns:
            Collage image of randomly samples images based on selection.
        """
        if selection.sum() == 0:
            raise ValueError("Empty selection.")
        rng = np.random.default_rng()
        sampled_paths = rng.choice(
            self["path"][selection],
            size=min(selection.sum(), num_cols * num_rows),
            replace=False,
        )
        return create_image_collage(
            images=read_images_from_paths(sampled_paths, num_workers),
            num_cols=num_cols,
            shape=shape,
        )

    def cluster_kmeans(self, num_clusters: int, **kwargs) -> np.ndarray:
        """Perform kmeans clustering on the metrics and order the clusters based on the
        distance from the mean cluster center.

        Args:
            num_clusters: Number of clusters.
            **kwargs: Passed on to `sklearn.cluster.MiniBatchKMeans`.

        Returns:
            Cluster assignments.
        """
        if "n_init" not in kwargs:
            kwargs["n_init"] = "auto"
        clust = MiniBatchKMeans(n_clusters=num_clusters, **kwargs)
        clusters = clust.fit_predict(self.metrics / 255)
        # Reorder based on distances from the mean cluster center.
        mean_center = clust.cluster_centers_.mean(0)
        distances = np.array(
            [np.linalg.norm(x - mean_center) for x in clust.cluster_centers_]
        )
        ordered_clusters = np.zeros_like(clusters)
        for new_idx, old_idx in enumerate(distances.argsort()[::-1]):
            ordered_clusters[clusters == old_idx] = new_idx
        return ordered_clusters

    def plot_histogram(
        self,
        column: str,
        min_value: Optional[float] = None,
        max_value: Optional[float] = None,
        *,
        num_bins: int = 20,
        num_images: int = 12,
        num_workers: int = 1,
        ax: Optional[plt.Axes] = None,
        **kwargs,
    ) -> plt.Axes:
        """Plot column values in a histogram with example images.

        Args:
            column: Column name.
            min_value: Minimum value for filtering. Defaults to None.
            max_value: Maximum value for filtering. Defaults to None.
            num_bins: Number of bins. Defaults to 20.
            num_images: Number of images per bin. Defaults to 12.
            num_workers: Number of image loading workers. Defaults to 1.
            ax: Axis for histogram. Cannot be passed when `num_images>0`. Defaults to
                None.
            **kwargs: Passed to `plt.hist`.

        Raises:
            ValueError: No difference between min and max values.
            ValueError: Passing an axis when `num_images>0`.

        Returns:
            Matplotlib axis or axes when `num_images>0`.
        """
        if ax is not None and num_images > 0:
            raise ValueError(ERROR_AX_WITH_IMAGES)
        values = self[column]
        if min_value is None:
            min_value = values.min()
        if max_value is None:
            max_value = values.max()
        if min_value == max_value:
            raise ValueError(ERROR_NO_DIFFERENCE.format(min_value, max_value, column))
        selection = (values >= min_value) & (values <= max_value)
        if "ec" not in kwargs:
            kwargs["ec"] = "black"
        if num_images == 0:
            # Plot only histogram.
            return _plot_histogram(values[selection], num_bins, ax=ax, **kwargs)
        plt.gca().remove()  # Auto removal depracated since 3.6
        # Initialize figure.
        ax_hist = plt.subplot2grid((4, num_bins), (0, 0), colspan=num_bins)
        ax_images = []
        for i in range(num_bins):
            ax_images.append(
                plt.subplot2grid((4, num_bins), (1, i), colspan=1, rowspan=3)
            )
        # Plot histogram.
        _plot_histogram(values[selection], num_bins, ax=ax_hist, **kwargs)
        # Plot images.
        bin_images = _get_bin_collages(
            paths=self["path"][selection],
            values=values[selection],
            num_bins=num_bins,
            num_images=num_images,
            rng=np.random.default_rng(),
            num_workers=num_workers,
        )
        for idx, bin_image in enumerate(bin_images):
            ax_images[idx].imshow(bin_image)
            ax_images[idx].axis("off")
        return np.array([ax_hist, *ax_images])

    def __len__(self) -> int:
        return len(self.dataframe)

    def __getitem__(self, key: str) -> np.ndarray:
        return self.dataframe.get_column(key).to_numpy()

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}(num_images={len(self.dataframe)}, "
            f"num_outliers={self.outliers.sum()})"
        )

coordinates property

Array of tile coordinates.

dataframe property

Polars dataframe with metadata.

dataframe_without_metrics property

Polars dataframe without metadata.

mean_and_std property

Means and standard deviations for RGB channels.

metric_columns property

Image metric columns.

metrics property

Array of normalized image metrics (divided by 255).

outlier_selections property

List of dicts with outlier selections and descriptions.

outliers property

Array of outlier indices.

paths property

Array of tile paths.

__init__(dataframe)

Initialize TileMetadata class instance.

Parameters:

Name	Type	Description	Default
dataframe	DataFrame	Polars dataframe with image metrics.	required

Raises:

Type	Description
ValueError	Dataframe does not contain any metric columns.

Source code in histoslice/utils/_process.py

def __init__(self, dataframe: pl.DataFrame) -> None:
    """Initialize `TileMetadata` class instance.

    Args:
        dataframe: Polars dataframe with image metrics.

    Raises:
        ValueError: Dataframe does not contain any metric columns.
    """
    self.__dataframe = dataframe
    self.__outliers = np.repeat([False], repeats=len(dataframe))
    self.__outlier_selections = []
    self.__metric_columns = [
        x for x in dataframe.columns if "_q" in x or (x.endswith(("_mean", "_std")))
    ]
    if len(self.__metric_columns) == 0:
        raise ValueError(ERROR_NO_METRICS)
    if "path" not in dataframe.columns:
        raise ValueError(ERROR_NO_PATHS)

add_outliers(selection, *, desc)

Set outliers to True with selection, and append the selection to outlier_selections property.

Parameters:

Name	Type	Description	Default
selection	ndarray	Selection for indexing outliers property.	required
desc	str	Description for selection.	required

Source code in histoslice/utils/_process.py

def add_outliers(self, selection: np.ndarray, *, desc: str) -> None:
    """Set outliers to `True` with selection, and append the selection to
    `outlier_selections` property.

    Args:
        selection: Selection for indexing `outliers` property.
        desc: Description for selection.
    """
    self.__outliers[selection] = True
    self.__outlier_selections.append({"selection": selection, "desc": desc})

cluster_kmeans(num_clusters, **kwargs)

Perform kmeans clustering on the metrics and order the clusters based on the distance from the mean cluster center.

Parameters:

Name	Type	Description	Default
num_clusters	int	Number of clusters.	required
**kwargs		Passed on to sklearn.cluster.MiniBatchKMeans.	{}

Returns:

Type	Description
ndarray	Cluster assignments.

Source code in histoslice/utils/_process.py

def cluster_kmeans(self, num_clusters: int, **kwargs) -> np.ndarray:
    """Perform kmeans clustering on the metrics and order the clusters based on the
    distance from the mean cluster center.

    Args:
        num_clusters: Number of clusters.
        **kwargs: Passed on to `sklearn.cluster.MiniBatchKMeans`.

    Returns:
        Cluster assignments.
    """
    if "n_init" not in kwargs:
        kwargs["n_init"] = "auto"
    clust = MiniBatchKMeans(n_clusters=num_clusters, **kwargs)
    clusters = clust.fit_predict(self.metrics / 255)
    # Reorder based on distances from the mean cluster center.
    mean_center = clust.cluster_centers_.mean(0)
    distances = np.array(
        [np.linalg.norm(x - mean_center) for x in clust.cluster_centers_]
    )
    ordered_clusters = np.zeros_like(clusters)
    for new_idx, old_idx in enumerate(distances.argsort()[::-1]):
        ordered_clusters[clusters == old_idx] = new_idx
    return ordered_clusters

from_parquet(*args, **kwargs) classmethod

Wrapper around polars.read_parquet function.

Source code in histoslice/utils/_process.py

@classmethod
def from_parquet(cls, *args, **kwargs) -> "OutlierDetector":
    """Wrapper around `polars.read_parquet` function."""
    return cls(pl.read_parquet(*args, **kwargs))

plot_histogram(column, min_value=None, max_value=None, *, num_bins=20, num_images=12, num_workers=1, ax=None, **kwargs)

Plot column values in a histogram with example images.

Parameters:

Name	Type	Description	Default
column	str	Column name.	required
min_value	Optional[float]	Minimum value for filtering. Defaults to None.	None
max_value	Optional[float]	Maximum value for filtering. Defaults to None.	None
num_bins	int	Number of bins. Defaults to 20.	20
num_images	int	Number of images per bin. Defaults to 12.	12
num_workers	int	Number of image loading workers. Defaults to 1.	1
ax	Optional[Axes]	Axis for histogram. Cannot be passed when num_images>0. Defaults to None.	None
**kwargs		Passed to plt.hist.	{}

Raises:

Type	Description
ValueError	No difference between min and max values.
ValueError	Passing an axis when num_images>0.

Returns:

Type	Description
Axes	Matplotlib axis or axes when num_images>0.

Source code in histoslice/utils/_process.py

def plot_histogram(
    self,
    column: str,
    min_value: Optional[float] = None,
    max_value: Optional[float] = None,
    *,
    num_bins: int = 20,
    num_images: int = 12,
    num_workers: int = 1,
    ax: Optional[plt.Axes] = None,
    **kwargs,
) -> plt.Axes:
    """Plot column values in a histogram with example images.

    Args:
        column: Column name.
        min_value: Minimum value for filtering. Defaults to None.
        max_value: Maximum value for filtering. Defaults to None.
        num_bins: Number of bins. Defaults to 20.
        num_images: Number of images per bin. Defaults to 12.
        num_workers: Number of image loading workers. Defaults to 1.
        ax: Axis for histogram. Cannot be passed when `num_images>0`. Defaults to
            None.
        **kwargs: Passed to `plt.hist`.

    Raises:
        ValueError: No difference between min and max values.
        ValueError: Passing an axis when `num_images>0`.

    Returns:
        Matplotlib axis or axes when `num_images>0`.
    """
    if ax is not None and num_images > 0:
        raise ValueError(ERROR_AX_WITH_IMAGES)
    values = self[column]
    if min_value is None:
        min_value = values.min()
    if max_value is None:
        max_value = values.max()
    if min_value == max_value:
        raise ValueError(ERROR_NO_DIFFERENCE.format(min_value, max_value, column))
    selection = (values >= min_value) & (values <= max_value)
    if "ec" not in kwargs:
        kwargs["ec"] = "black"
    if num_images == 0:
        # Plot only histogram.
        return _plot_histogram(values[selection], num_bins, ax=ax, **kwargs)
    plt.gca().remove()  # Auto removal depracated since 3.6
    # Initialize figure.
    ax_hist = plt.subplot2grid((4, num_bins), (0, 0), colspan=num_bins)
    ax_images = []
    for i in range(num_bins):
        ax_images.append(
            plt.subplot2grid((4, num_bins), (1, i), colspan=1, rowspan=3)
        )
    # Plot histogram.
    _plot_histogram(values[selection], num_bins, ax=ax_hist, **kwargs)
    # Plot images.
    bin_images = _get_bin_collages(
        paths=self["path"][selection],
        values=values[selection],
        num_bins=num_bins,
        num_images=num_images,
        rng=np.random.default_rng(),
        num_workers=num_workers,
    )
    for idx, bin_image in enumerate(bin_images):
        ax_images[idx].imshow(bin_image)
        ax_images[idx].axis("off")
    return np.array([ax_hist, *ax_images])

random_image_collage(selection, *, num_rows=4, num_cols=16, shape=(64, 64), num_workers=1)

Generate a random collage from paths[selection].

Parameters:

Name	Type	Description	Default
selection	ndarray	Selection for paths.	required
num_rows	int	Number of rows in the collage image. Defaults to 4.	4
num_cols	int	Number of columns in the collage image. Defaults to 16.	16
shape	tuple[int, int]	Size of each image in the collage. Defaults to (64, 64).	(64, 64)
num_workers	int	Number of image loading workers. Defaults to 1.	1

Returns:

Type	Description
Image	Collage image of randomly samples images based on selection.

Source code in histoslice/utils/_process.py

def random_image_collage(
    self,
    selection: np.ndarray,
    *,
    num_rows: int = 4,
    num_cols: int = 16,
    shape: tuple[int, int] = (64, 64),
    num_workers: int = 1,
) -> Image.Image:
    """Generate a random collage from `paths[selection]`.

    Args:
        selection: Selection for paths.
        num_rows: Number of rows in the collage image. Defaults to 4.
        num_cols: Number of columns in the collage image. Defaults to 16.
        shape: Size of each image in the collage. Defaults to (64, 64).
        num_workers: Number of image loading workers. Defaults to 1.

    Returns:
        Collage image of randomly samples images based on selection.
    """
    if selection.sum() == 0:
        raise ValueError("Empty selection.")
    rng = np.random.default_rng()
    sampled_paths = rng.choice(
        self["path"][selection],
        size=min(selection.sum(), num_cols * num_rows),
        replace=False,
    )
    return create_image_collage(
        images=read_images_from_paths(sampled_paths, num_workers),
        num_cols=num_cols,
        shape=shape,
    )