Spaces:
No application file
No application file
File size: 9,672 Bytes
907462b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | # -*- coding: utf-8 -*-
# Process a batch of patches
#
# @ Fabian Hörst, fabian.hoerst@uk-essen.de
# Institute for Artifical Intelligence in Medicine,
# University Medicine Essen
import multiprocessing
from pathlib import Path
from typing import Dict, List, Tuple, Union
import numpy as np
from openslide import OpenSlide
from PIL import Image
from shapely.geometry import Polygon
from preprocessing.patch_extraction import logger
from preprocessing.patch_extraction.src.utils.patch_util import (
DeepZoomGeneratorOS,
calculate_background_ratio,
get_intersected_labels,
macenko_normalization,
pad_tile,
patch_to_tile_size,
standardize_brightness,
)
from utils.tools import module_exists
def process_batch(
batch: List[Tuple[int, int, float]],
*,
wsi_file: Union[Path, str],
wsi_metadata: dict,
patch_size: int,
patch_overlap: int,
level: int,
polygons: List[Polygon],
region_labels: List[str],
label_map: Dict[str, int],
min_intersection_ratio: float = 0.0,
save_only_annotated_patches: bool = False,
adjust_brightness: bool = False,
normalize_stains: bool = False,
normalization_vector_path: Union[str, Path] = None,
store_masks: bool = False,
overlapping_labels: bool = False,
context_scales: List[int] = None,
) -> Tuple[List[np.ndarray], List[dict], List[np.ndarray], dict[int, List[np.ndarray]]]:
"""Calcultes batch results for a list of coordinates
Patches are extracted according to their coordinate with given patch-settings (size, overlap).
Patch annotation masks can be stored, as well as context patches with the same shape retrieved.
Optionally, stains can be nornalized according to macenko normalization.
Args:
batch (List[Tuple[int, int, float]]): A batch of patch coordinates (row, col, backgropund ratio)
wsi_file (Union[Path, str]): Path to the WSI file from which the patches should be extracted from
wsi_metadata (dict): Dictionary with important WSI metadata
patch_size (int): The size of the patches in pixel that will be retrieved from the WSI, e.g. 256 for 256px
patch_overlap (int): The amount pixels that should overlap between two different patches
level (int): The tile level for sampling.
polygons (List[Polygon]): Annotations of this WSI as a list of polygons (referenced to highest level of WSI).
If no annotations, pass an empty list [].
region_labels (List[str]): List of labels for the annotations provided as polygons parameter.
If no annotations, pass an empty list [].
label_map (Dict[str, int]): Dictionary mapping the label names to an integer. Please ensure that background label has integer 0!
min_intersection_ratio(float, optional): Minimum ratio of intersection between annotation class and patch to be considered as class instance. Defaults to 0.0.
save_only_annotated_patches (bool, optional): If true only patches containing annotations will be stored. Defaults to False.
adjust_brightness (bool, optional): Normalize brightness in a batch by clipping to 90%. Defaults to False.
normalize_stains (bool, optional): Uses Macenko normalization on patches. Defaults to False.
normalization_vector_path (Union[str, Path], optional): The path to a JSON file where the normalization vectors are stored. Defaults to None.
store_masks (bool, optional): Set to store masks per patch. Defaults to False.
overlapping_labels (bool, optional): Per default, labels (annotations) are mutually exclusive.
If labels overlap, they are overwritten according to the label_map.json ordering (highest number = highest priority).
True means that the mask array is 3D with shape (patch_size, patch_size, len(label_map)), otherwise just (patch_size, patch_size).
Defaults to False.
context_scales (List[int], optional): Define context scales for context patches. Context patches are centered around a central patch.
The context-patch size is equal to the patch-size, but downsampling is different. Defaults to None.
Returns:
Tuple[List[np.ndarray], List[dict], List[np.ndarray], dict[int, List[np.ndarray]]]:
- List[np.ndarray]: List with patches as numpy arrays with shape (patch_size, patch_size, 3)
- List[dict]: List with metadata dictionary for each patch
- List[np.ndarray]: List with patch masks if store_masks is True. Shape is (256, 256) for non-overlapping labels
and (256, 256, num_classes) for overlapping labels. If masks should not be stored, returns an empty list
- dict[int, List[np.ndarray]]: Each key is a downsampling value for the context patch
and the entries are numpy array for context patches with shape [patch_size, patch_size, 3].
If no context patches, returns an empty dict
"""
logger.debug(f"Started process {multiprocessing.current_process().name}")
# Where the results of this batch will be stored
patches, metadata, patch_masks = [], [], []
# context patch
context_tiles = {}
if context_scales is not None:
context_patches = {scale: [] for scale in context_scales}
else:
context_patches = {}
# reopen slide
if module_exists("cucim", error="ignore"):
from cucim import CuImage
from preprocessing.patch_extraction.src.cucim_deepzoom import (
DeepZoomGeneratorCucim,
)
generator_module = DeepZoomGeneratorCucim
image_loader = CuImage
else:
generator_module = DeepZoomGeneratorOS
image_loader = OpenSlide
slide = OpenSlide(str(wsi_file))
slide_cu = image_loader(str(wsi_file))
tile_size = patch_to_tile_size(patch_size, patch_overlap)
tiles = generator_module(
osr=slide,
cucim_slide=slide_cu,
tile_size=tile_size,
overlap=patch_overlap,
limit_bounds=True,
)
if context_scales is not None:
for scale in context_scales:
overlap_context = int((scale - 1) * patch_size / 2) + patch_overlap
context_tiles[scale] = generator_module(
osr=slide,
cucim_slide=slide_cu,
tile_size=tile_size, # tile_size,
overlap=overlap_context, # (1-scale) * tile_size / 2,
limit_bounds=True,
)
# patches_count = 0
for row, col, _ in batch:
# OpenSlide: Address of the tile within the level as a (column, row) tuple
new_tile = np.array(tiles.get_tile(level, (col, row)), dtype=np.uint8)
# calculate background ratio for every patch
background_ratio = calculate_background_ratio(new_tile, patch_size)
# patch_label
if background_ratio > 1 - min_intersection_ratio:
intersected_labels = [] # Zero means background
ratio = []
patch_mask = np.zeros((tile_size, tile_size), dtype=np.uint8)
else:
intersected_labels, ratio, patch_mask = get_intersected_labels(
tile_size=tile_size,
patch_overlap=patch_overlap,
col=col,
row=row,
polygons=polygons,
label_map=label_map,
min_intersection_ratio=min_intersection_ratio,
region_labels=region_labels,
overlapping_labels=overlapping_labels,
store_masks=store_masks,
)
if len(intersected_labels) == 0 and save_only_annotated_patches:
continue
if store_masks:
patch_masks.append(patch_mask)
patches.append(pad_tile(new_tile, patch_size, col, row))
patch_metadata = {
"row": row,
"col": col,
"background_ratio": float(background_ratio),
"intersected_labels": intersected_labels,
"label_ratio": ratio,
"wsi_metadata": wsi_metadata,
}
if context_scales is not None:
patch_metadata["context_scales"] = []
for scale in context_scales:
context_patch = np.array(
context_tiles[scale].get_tile(level, (col, row)),
dtype=np.uint8, # TODO change back to level
)
context_patch = pad_tile(context_patch, patch_size * scale, col, row)
context_patch = np.array(
Image.fromarray(context_patch).resize((patch_size, patch_size)),
dtype=np.uint8,
)
context_patches[scale].append(context_patch)
patch_metadata["context_scales"].append(scale)
metadata.append(patch_metadata)
if len(patches) > 0:
if adjust_brightness:
patches = standardize_brightness(patches)
for scale, scale_patch in context_patches.items():
context_patches[scale] = standardize_brightness(scale_patch)
if normalize_stains:
patches, _, _ = macenko_normalization(
patches, normalization_vector_path=normalization_vector_path
)
for scale, scale_patch in context_patches.items():
context_patches[scale], _, _ = macenko_normalization(
scale_patch, normalization_vector_path=normalization_vector_path
)
logger.debug(
f"Process {multiprocessing.current_process().name} finished, found "
f"{len(patches)} patches."
)
return patches, metadata, patch_masks, context_patches
|