image collage to functionals; change n_* -> num_*

histoprep/functional/__init__.py

@@ -2,6 +2,7 @@
 
 from ._dearray import *
 from ._draw import *
+from ._images import *
 from ._level import *
 from ._mean_std import *
 from ._metrics import *

histoprep/functional/_images.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+__all__ = [
+    "get_random_image_collage",
+    "_read_images_from_paths",
+    "_create_image_collage",
+]
+
+import random
+from collections.abc import Iterable
+from pathlib import Path
+
+import cv2
+import numpy as np
+from mpire import WorkerPool
+from PIL import Image
+
+
+def get_random_image_collage(
+    paths: Iterable[str | Path],
+    num_rows: int = 4,
+    num_cols: int = 16,
+    shape: tuple[int, int] = (64, 64),
+    num_workers: int = 1,
+) -> Image.Image:
+    """Create image collage from randomly sampled images from paths.
+
+    Args:
+        paths: Image paths.
+        num_rows: Number of rows in the collage. Ignored if there isn't enough images.
+            Defaults to 4.
+        num_cols: Number of columns in the collage. Defaults to 16.
+        shape: Shape of each image in the collage. Defaults to (64, 64).
+        num_workers: Number of image loading workers. Defaults to 1.
+
+    Returns:
+        Image collage.
+    """
+    if len(paths) > num_cols * num_rows:
+        paths = random.choices(paths, k=num_cols * num_rows)  # noqa
+    images = _read_images_from_paths(paths=paths, num_workers=num_workers)
+    return _create_image_collage(images=images, num_cols=num_cols, shape=shape)
+
+
+def _read_images_from_paths(
+    paths: Iterable[str | Path | None], num_workers: int
+) -> list[np.ndarray]:
+    """Read images from paths.
+
+    Args:
+        paths: Image paths.
+        num_workers: Number of image loading workers.
+
+    Returns:
+        List of numpy array images.
+    """
+    if num_workers <= 1:
+        return [_read_image(x) for x in paths]
+    with WorkerPool(n_jobs=num_workers) as pool:
+        output = list(pool.imap(_read_image, paths))
+    return output  # noqa
+
+
+def _create_image_collage(
+    images: list[np.ndarray], num_cols: int, shape: tuple[int, int]
+) -> Image.Image:
+    """Collect images into a collage.
+
+    Args:
+        images: List of array images.
+        num_cols: Number of columns. Number of rows is determined by
+            `ceil(images/num_cols)`.
+        shape: Shape for each image.
+
+    Returns:
+        Image collage.
+    """
+    if len(images) == 0:
+        return None
+    output, row = [], []
+    for img in images:
+        resized = cv2.resize(img, dsize=shape[::-1])
+        row.append(resized)
+        if len(row) == num_cols:
+            output.append(np.hstack(row))
+            row = []
+    if len(row) > 0:
+        row.extend([np.zeros_like(resized)] * (num_cols - len(row)))
+        output.append(np.hstack(row))
+    return Image.fromarray(np.vstack(output))
+
+
+def _read_image(path: str | None) -> np.ndarray:
+    """Parallisable."""
+    if path is None:
+        return None
+    return np.array(Image.open(path))

histoprep/utils/_histogram.py

@@ -3,7 +3,7 @@
 import matplotlib.pyplot as plt
 import numpy as np
 
-from ._images import _read_images
+from histoprep import functional as F
 
 
 def _plot_histogram(
@@ -68,7 +68,7 @@ def _read_bin_paths(
     """Read paths for each bin and generate vertically stacked array."""
     all_images = []
     resize_shape = None
-    for image in _read_images(bin_paths, num_workers=num_workers):
+    for image in F._read_images_from_paths(bin_paths, num_workers=num_workers):
         if resize_shape is None and image is not None:
             resize_shape = image.shape
         all_images.append(image)

histoprep/utils/_metadata.py

@@ -10,8 +10,9 @@
 from sklearn.cluster import MiniBatchKMeans
 from sklearn.decomposition import PCA
 
+from histoprep import functional as F
+
 from ._histogram import _get_bin_collages, _plot_histogram
-from ._images import _create_collage, _read_images
 
 ERROR_NO_METRICS = (
     "Metadata does not contain any metrics, make sure tiles are saved "
@@ -99,17 +100,17 @@ def random_collage(
         self,
         selection: np.ndarray,
         *,
-        n_rows: int = 4,
-        n_cols: int = 16,
+        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: `TileMetadata["path"][selection]`.
-            n_rows: Number of rows in the collage image. Defaults to 4.
-            n_cols: Number of columns in the collage image. Defaults to 16.
+            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.
 
@@ -121,27 +122,29 @@ def random_collage(
         rng = np.random.default_rng()
         sampled_paths = rng.choice(
             self["path"][selection],
-            size=min(selection.sum(), n_cols * n_rows),
+            size=min(selection.sum(), num_cols * num_rows),
             replace=False,
         )
-        return _create_collage(
-            _read_images(sampled_paths, num_workers), n_cols=n_cols, shape=shape
+        return F._create_image_collage(
+            images=F._read_images_from_paths(sampled_paths, num_workers),
+            num_cols=num_cols,
+            shape=shape,
         )
 
-    def cluster_kmeans(self, n_clusters: int, **kwargs) -> np.ndarray:
+    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:
-            n_clusters: Number of clusters.
+            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=n_clusters, **kwargs)
+        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)
@@ -197,6 +200,7 @@ def plot_histogram(
             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.

tests/image_test.py

@@ -0,0 +1,25 @@
+from PIL import Image
+
+import histoprep.functional as F
+from histoprep import SlideReader
+
+from ._utils import SLIDE_PATH_JPEG, TMP_DIRECTORY, clean_temporary_directory
+
+
+def test_image_collage() -> None:
+    clean_temporary_directory()
+    reader = SlideReader(SLIDE_PATH_JPEG)
+    tile_coords = reader.get_tile_coordinates(None, width=128)
+    metadata = reader.save_regions(TMP_DIRECTORY, tile_coords)
+    collage = F.get_random_image_collage(
+        metadata["path"], num_rows=4, num_cols=8, shape=(32, 32)
+    )
+    assert isinstance(collage, Image.Image)
+    assert collage.size == (8 * 32, 4 * 32)
+    # Not enough images for all rows.
+    collage = F.get_random_image_collage(
+        metadata["path"][:6], num_rows=4, num_cols=8, shape=(32, 32)
+    )
+    assert isinstance(collage, Image.Image)
+    assert collage.size == (8 * 32, 1 * 32)
+    clean_temporary_directory()

tests/metadata_test.py

@@ -77,8 +77,8 @@ def test_metadata_plot_pca() -> None:
 
 def test_metadata_plot_collage() -> None:
     metadata = TileMetadata(generate_metadata(clean_tmp=False))
-    assert metadata.random_collage(~metadata.outliers, n_rows=4).size == (1024, 256)
-    assert metadata.random_collage(~metadata.outliers, n_rows=2).size == (1024, 128)
+    assert metadata.random_collage(~metadata.outliers, num_rows=4).size == (1024, 256)
+    assert metadata.random_collage(~metadata.outliers, num_rows=2).size == (1024, 128)
     with pytest.raises(ValueError, match="Empty selection"):
         metadata.random_collage(metadata.outliers)
     clean_temporary_directory()
@@ -97,41 +97,3 @@ def test_metadata_cluster() -> None:
     metadata = TileMetadata(generate_metadata())
     clusters = metadata.cluster_kmeans(10)
     assert len(clusters) == len(metadata)
-
-
-# def test_tile_metadata() -> None:
-#     # Save tiles.
-#     clean_temporary_directory()
-#     reader = SlideReader(SLIDE_PATH_JPEG)
-#     threshold, tissue_mask = reader.get_tissue_mask()
-#     tiles = reader.get_tile_coordinates(tissue_mask, 64, max_background=0.75)
-#     dataframe = reader.save_regions(
-#         TMP_DIRECTORY, tiles, overwrite=True, save_metrics=True, threshold=threshold
-#     )
-#     # Initialize TileMetadata.
-#     metadata = TileMetadata(dataframe)
-#     # Test properites.
-#     assert np.equal(
-#         np.array(metadata.mean_and_std).round(3),
-#         np.array([[0.807, 0.63, 0.729], [0.135, 0.156, 0.122]]),
-#     ).all()
-#     assert len(metadata.metric_columns) == 64
-#     assert metadata.metrics.shape == (434, 64)
-#     assert len(metadata.outliers) == 434
-#     assert metadata.outliers.sum() == 0
-#     # test plotting and methods.
-#     metadata.plot_histogram("background")
-#     clusters = metadata.cluster_kmeans(10)
-#     metadata.plot_pca(clusters)
-#     with pytest.raises(ValueError):  # noqa
-#         metadata.plot_histogram("black_pixels")
-#     with pytest.raises(ValueError):  # noqa
-#         metadata.plot_histogram("black_pixels", ax="fake_axis", n_images=100)
-#     assert metadata.random_collage(
-#         selection=metadata["hue_std"] > 50, shape=(64, 64)
-#     ).size == (16 * 64, 4 * 64)
-#     # Add outliers.
-#     metadata.add_outliers(metadata["background"] > 0.7, desc="high background")
-#     assert metadata.outliers.sum() == 11
-#     assert metadata.__repr__() == "TileMetadata(n_images=434, n_outliers=11)"
-#     clean_temporary_directory()