Feature added: Get WSI at mpp

monai/data/wsi_reader.py

@@ -603,6 +603,25 @@ def get_mpp(self, wsi, level: int) -> tuple[float, float]:
         """
         return self.reader.get_mpp(wsi, level)
 
+    def get_wsi_at_mpp(self, wsi, mpp: tuple, atol: float = 0.00, rtol: float = 0.05) -> np.ndarray:
+        """
+        Returns the representation of the whole slide image at a given micro-per-pixel (mpp) resolution.
+        The optional tolerance parameters are considered at the level whose mpp value is closest to the one provided by the user.
+        If the user-provided mpp is larger than the mpp of the closest level,
+        the image is downscaled to a resolution that matches the user-provided mpp.
+        Otherwise, if the closest level's resolution is not sufficient to meet the user's requested resolution,
+        the next lower level (which has a higher resolution) is chosen.
+        The image from this level is then down-scaled to achieve a resolution at the user-provided mpp value.
+
+        Args:
+            wsi: whole slide image object from WSIReader
+            mpp: the resolution in micron per pixel at which the representation of the whole slide image should be extracted.
+            atol: the acceptable absolute tolerance for resolution in micro per pixel.
+            rtol: the acceptable relative tolerance for resolution in micro per pixel.
+
+        """
+        return self.reader.get_wsi_at_mpp(wsi, mpp, atol, rtol)
+
     def get_power(self, wsi, level: int) -> float:
         """
         Returns the micro-per-pixel resolution of the whole slide image at a given level.
@@ -744,6 +763,101 @@ def get_mpp(self, wsi, level: int) -> tuple[float, float]:
 
         raise ValueError("`mpp` cannot be obtained for this file. Please use `level` instead.")
 
+    def get_wsi_at_mpp(self, wsi, mpp: tuple, atol: float = 0.00, rtol: float = 0.05) -> Any:
+        """
+        Returns the representation of the whole slide image at a given micro-per-pixel (mpp) resolution.
+        The optional tolerance parameters are considered at the level whose mpp value is closest to the one provided by the user.
+        If the user-provided mpp is larger than the mpp of the closest level,
+        the image is downscaled to a resolution that matches the user-provided mpp.
+        Otherwise, if the closest level's resolution is not sufficient to meet the user's requested resolution,
+        the next lower level (which has a higher resolution) is chosen.
+        The image from this level is then down-scaled to achieve a resolution at the user-provided mpp value.
+
+        Args:
+            wsi: whole slide image object from WSIReader
+            mpp: the resolution in micron per pixel at which the representation of the whole slide image should be extracted.
+            atol: the acceptable absolute tolerance for resolution in micro per pixel.
+            rtol: the acceptable relative tolerance for resolution in micro per pixel.
+
+        """
+
+        cucim_resize, _ = optional_import("cucim.skimage.transform", name="resize")
+        cp, _ = optional_import("cupy")
+
+        user_mpp_x, user_mpp_y = mpp
+        mpp_list = [self.get_mpp(wsi, lvl) for lvl in range(wsi.resolutions["level_count"])]
+        closest_lvl = self._find_closest_level("mpp", mpp, mpp_list, 0, 5)
+        # -> Should not throw ValueError, instead just return the closest value; how to select tolerances?
+
+        mpp_closest_lvl = mpp_list[closest_lvl]
+        closest_lvl_dim = wsi.resolutions["level_dimensions"][closest_lvl]
+
+        print(f"Closest Level: {closest_lvl} with MPP: {mpp_closest_lvl}")
+        mpp_closest_lvl_x, mpp_closest_lvl_y = mpp_closest_lvl
+
+        # Define tolerance intervals for x and y of closest level
+        lower_bound_x = mpp_closest_lvl_x * (1 - rtol) - atol
+        upper_bound_x = mpp_closest_lvl_x * (1 + rtol) + atol
+        lower_bound_y = mpp_closest_lvl_y * (1 - rtol) - atol
+        upper_bound_y = mpp_closest_lvl_y * (1 + rtol) + atol
+
+        # Check if user-provided mpp_x and mpp_y fall within the tolerance intervals for closest level
+        within_tolerance_x = (user_mpp_x >= lower_bound_x) & (user_mpp_x <= upper_bound_x)
+        within_tolerance_y = (user_mpp_y >= lower_bound_y) & (user_mpp_y <= upper_bound_y)
+        within_tolerance = within_tolerance_x & within_tolerance_y
+
+        if within_tolerance:
+            # Take closest_level and continue with returning img at level
+            print(f"User-provided MPP lies within tolerance of level {closest_lvl}, returning wsi at this level.")
+            closest_lvl_wsi = wsi.read_region(
+                (0, 0), level=closest_lvl, size=closest_lvl_dim, num_workers=self.num_workers
+            )
+
+        else:
+            # If mpp_closest_level < mpp -> closest_level has higher res than img at mpp => downscale from closest_level to mpp
+            closest_level_is_bigger_x = mpp_closest_lvl_x < user_mpp_x
+            closest_level_is_bigger_y = mpp_closest_lvl_y < user_mpp_y
+            closest_level_is_bigger = closest_level_is_bigger_x & closest_level_is_bigger_y
+
+            if closest_level_is_bigger:
+                ds_factor_x = mpp_closest_lvl_x / user_mpp_x
+                ds_factor_y = mpp_closest_lvl_y / user_mpp_y
+
+                closest_lvl_wsi = wsi.read_region(
+                    (0, 0), level=closest_lvl, size=closest_lvl_dim, num_workers=self.num_workers
+                )
+                wsi_arr = cp.array(closest_lvl_wsi)
+
+                target_res_x = int(np.round(closest_lvl_dim[1] * ds_factor_x))
+                target_res_y = int(np.round(closest_lvl_dim[0] * ds_factor_y))
+
+                closest_lvl_wsi = cucim_resize(wsi_arr, (target_res_x, target_res_y), order=0)
+                print(f"Case 1: Downscaling using factor {(ds_factor_x, ds_factor_y)}")
+
+            else:
+                # Else: increase resolution (ie, decrement level) and then downsample
+                closest_lvl = closest_lvl - 1
+                mpp_closest_lvl = mpp_list[closest_lvl]
+                mpp_closest_lvl_x, mpp_closest_lvl_y = mpp_closest_lvl
+
+                ds_factor_x = mpp_closest_lvl_x / user_mpp_x
+                ds_factor_y = mpp_closest_lvl_y / user_mpp_y
+
+                closest_lvl_dim = wsi.resolutions["level_dimensions"][closest_lvl]
+                closest_lvl_wsi = wsi.read_region(
+                    (0, 0), level=closest_lvl, size=closest_lvl_dim, num_workers=self.num_workers
+                )
+                wsi_arr = cp.array(closest_lvl_wsi)
+
+                target_res_x = int(np.round(closest_lvl_dim[1] * ds_factor_x))
+                target_res_y = int(np.round(closest_lvl_dim[0] * ds_factor_y))
+
+                closest_lvl_wsi = cucim_resize(wsi_arr, (target_res_x, target_res_y), order=0)
+                print(f"Case 2: Downscaling using factor {(ds_factor_x, ds_factor_y)}, now from level {closest_lvl}")
+
+        wsi_arr = cp.asnumpy(closest_lvl_wsi)
+        return wsi_arr
+
     def get_power(self, wsi, level: int) -> float:
         """
         Returns the objective power of the whole slide image at a given level.
@@ -940,6 +1054,91 @@ def get_mpp(self, wsi, level: int) -> tuple[float, float]:
 
         raise ValueError("`mpp` cannot be obtained for this file. Please use `level` instead.")
 
+    def get_wsi_at_mpp(self, wsi, mpp: tuple, atol: float = 0.00, rtol: float = 0.05) -> np.ndarray:
+        """
+        Returns the representation of the whole slide image at a given micro-per-pixel (mpp) resolution.
+        The optional tolerance parameters are considered at the level whose mpp value is closest to the one provided by the user.
+        If the user-provided mpp is larger than the mpp of the closest level,
+        the image is downscaled to a resolution that matches the user-provided mpp.
+        Otherwise, if the closest level's resolution is not sufficient to meet the user's requested resolution,
+        the next lower level (which has a higher resolution) is chosen.
+        The image from this level is then down-scaled to achieve a resolution at the user-provided mpp value.
+
+        Args:
+            wsi: whole slide image object from WSIReader
+            mpp: the resolution in micron per pixel at which the representation of the whole slide image should be extracted.
+            atol: the acceptable absolute tolerance for resolution in micro per pixel.
+            rtol: the acceptable relative tolerance for resolution in micro per pixel.
+
+        """
+
+        pil_image, _ = optional_import("PIL", name="Image")
+        user_mpp_x, user_mpp_y = mpp
+        mpp_list = [self.get_mpp(wsi, lvl) for lvl in range(wsi.level_count)]
+        closest_lvl = self._find_closest_level("mpp", mpp, mpp_list, 0, 5)
+        # -> Should not throw ValueError, instead just return the closest value; how to select tolerances?
+
+        mpp_closest_lvl = mpp_list[closest_lvl]
+        closest_lvl_dim = wsi.level_dimensions[closest_lvl]
+
+        print(f"Closest Level: {closest_lvl} with MPP: {mpp_closest_lvl}")
+        mpp_closest_lvl_x, mpp_closest_lvl_y = mpp_closest_lvl
+
+        # Define tolerance intervals for x and y of closest level
+        lower_bound_x = mpp_closest_lvl_x * (1 - rtol) - atol
+        upper_bound_x = mpp_closest_lvl_x * (1 + rtol) + atol
+        lower_bound_y = mpp_closest_lvl_y * (1 - rtol) - atol
+        upper_bound_y = mpp_closest_lvl_y * (1 + rtol) + atol
+
+        # Check if user-provided mpp_x and mpp_y fall within the tolerance intervals for closest level
+        within_tolerance_x = (user_mpp_x >= lower_bound_x) & (user_mpp_x <= upper_bound_x)
+        within_tolerance_y = (user_mpp_y >= lower_bound_y) & (user_mpp_y <= upper_bound_y)
+        within_tolerance = within_tolerance_x & within_tolerance_y
+
+        if within_tolerance:
+            # Take closest_level and continue with returning img at level
+            print(f"User-provided MPP lies within tolerance of level {closest_lvl}, returning wsi at this level.")
+            closest_lvl_wsi = wsi.read_region((0, 0), level=closest_lvl, size=closest_lvl_dim)
+
+        else:
+            # If mpp_closest_level < mpp -> closest_level has higher res than img at mpp => downscale from closest_level to mpp
+            closest_level_is_bigger_x = mpp_closest_lvl_x < user_mpp_x
+            closest_level_is_bigger_y = mpp_closest_lvl_y < user_mpp_y
+            closest_level_is_bigger = closest_level_is_bigger_x & closest_level_is_bigger_y
+
+            if closest_level_is_bigger:
+                ds_factor_x = mpp_closest_lvl_x / user_mpp_x
+                ds_factor_y = mpp_closest_lvl_y / user_mpp_y
+
+                closest_lvl_wsi = wsi.read_region((0, 0), level=closest_lvl, size=closest_lvl_dim)
+
+                target_res_x = int(np.round(closest_lvl_dim[0] * ds_factor_x))
+                target_res_y = int(np.round(closest_lvl_dim[1] * ds_factor_y))
+
+                closest_lvl_wsi = closest_lvl_wsi.resize((target_res_x, target_res_y), pil_image.BILINEAR)
+                print(f"Case 1: Downscaling using factor {(ds_factor_x, ds_factor_y)}")
+
+            else:
+                # Else: increase resolution (ie, decrement level) and then downsample
+                closest_lvl = closest_lvl - 1
+                mpp_closest_lvl = mpp_list[closest_lvl]
+                mpp_closest_lvl_x, mpp_closest_lvl_y = mpp_closest_lvl
+
+                ds_factor_x = mpp_closest_lvl_x / user_mpp_x
+                ds_factor_y = mpp_closest_lvl_y / user_mpp_y
+
+                closest_lvl_dim = wsi.level_dimensions[closest_lvl]
+                closest_lvl_wsi = wsi.read_region((0, 0), level=closest_lvl, size=closest_lvl_dim)
+
+                target_res_x = int(np.round(closest_lvl_dim[0] * ds_factor_x))
+                target_res_y = int(np.round(closest_lvl_dim[1] * ds_factor_y))
+
+                closest_lvl_wsi = closest_lvl_wsi.resize((target_res_x, target_res_y), pil_image.BILINEAR)
+                print(f"Case 2: Downscaling using factor {(ds_factor_x, ds_factor_y)}, now from level {closest_lvl}")
+
+        wsi_arr = np.array(closest_lvl_wsi)
+        return wsi_arr
+
     def get_power(self, wsi, level: int) -> float:
         """
         Returns the objective power of the whole slide image at a given level.
@@ -1096,8 +1295,10 @@ def get_mpp(self, wsi, level: int) -> tuple[float, float]:
             and wsi.pages[level].tags["YResolution"].value
         ):
             unit = wsi.pages[level].tags.get("ResolutionUnit")
-            if unit is not None:
+            if unit is not None:  # Needs to be improved
                 unit = str(unit.value)[8:]
+                # unit = str(unit.value.name).lower()  # TODO: Merge both methods
+
             else:
                 warnings.warn("The resolution unit is missing. `micrometer` will be used as default.")
                 unit = "micrometer"
@@ -1110,6 +1311,96 @@ def get_mpp(self, wsi, level: int) -> tuple[float, float]:
 
         raise ValueError("`mpp`  cannot be obtained for this file. Please use `level` instead.")
 
+    def get_wsi_at_mpp(self, wsi, mpp: tuple, atol: float = 0.00, rtol: float = 0.05) -> np.ndarray:
+        """
+        Returns the representation of the whole slide image at a given micro-per-pixel (mpp) resolution.
+        The optional tolerance parameters are considered at the level whose mpp value is closest to the one provided by the user.
+        If the user-provided mpp is larger than the mpp of the closest level,
+        the image is downscaled to a resolution that matches the user-provided mpp.
+        Otherwise, if the closest level's resolution is not sufficient to meet the user's requested resolution,
+        the next lower level (which has a higher resolution) is chosen.
+        The image from this level is then down-scaled to achieve a resolution at the user-provided mpp value.
+
+        Args:
+            wsi: whole slide image object from WSIReader
+            mpp: the resolution in micron per pixel at which the representation of the whole slide image should be extracted.
+            atol: the acceptable absolute tolerance for resolution in micro per pixel.
+            rtol: the acceptable relative tolerance for resolution in micro per pixel.
+
+        """
+
+        pil_image, _ = optional_import("PIL", name="Image")
+        user_mpp_x, user_mpp_y = mpp
+        mpp_list = [self.get_mpp(wsi, lvl) for lvl in range(len(wsi.pages))]
+        closest_lvl = self._find_closest_level("mpp", mpp, mpp_list, 0, 5)
+        # -> Should not throw ValueError, instead just return the closest value; how to select tolerances?
+
+        mpp_closest_lvl = mpp_list[closest_lvl]
+
+        lvl_dims = [self.get_size(wsi, lvl) for lvl in range(len(wsi.pages))]
+        closest_lvl_dim = lvl_dims[closest_lvl]
+        closest_lvl_dim = (closest_lvl_dim[1], closest_lvl_dim[0])
+
+        print(f"Closest Level: {closest_lvl} with MPP: {mpp_closest_lvl}")
+        mpp_closest_lvl_x, mpp_closest_lvl_y = mpp_closest_lvl
+
+        # Define tolerance intervals for x and y of closest level
+        lower_bound_x = mpp_closest_lvl_x * (1 - rtol) - atol
+        upper_bound_x = mpp_closest_lvl_x * (1 + rtol) + atol
+        lower_bound_y = mpp_closest_lvl_y * (1 - rtol) - atol
+        upper_bound_y = mpp_closest_lvl_y * (1 + rtol) + atol
+
+        # Check if user-provided mpp_x and mpp_y fall within the tolerance intervals for closest level
+        within_tolerance_x = (user_mpp_x >= lower_bound_x) & (user_mpp_x <= upper_bound_x)
+        within_tolerance_y = (user_mpp_y >= lower_bound_y) & (user_mpp_y <= upper_bound_y)
+        within_tolerance = within_tolerance_x & within_tolerance_y
+
+        if within_tolerance:
+            # Take closest_level and continue with returning img at level
+            print(f"User-provided MPP lies within tolerance of level {closest_lvl}, returning wsi at this level.")
+            closest_lvl_wsi = wsi.read_region((0, 0), level=closest_lvl, size=closest_lvl_dim)
+
+        else:
+            # If mpp_closest_level < mpp -> closest_level has higher res than img at mpp => downscale from closest_level to mpp
+            closest_level_is_bigger_x = mpp_closest_lvl_x < user_mpp_x
+            closest_level_is_bigger_y = mpp_closest_lvl_y < user_mpp_y
+            closest_level_is_bigger = closest_level_is_bigger_x & closest_level_is_bigger_y
+
+            if closest_level_is_bigger:
+                ds_factor_x = mpp_closest_lvl_x / user_mpp_x
+                ds_factor_y = mpp_closest_lvl_y / user_mpp_y
+
+                closest_lvl_wsi = pil_image.fromarray(wsi.pages[closest_lvl].asarray())  # Might be suboptimal
+
+                target_res_x = int(np.round(closest_lvl_dim[0] * ds_factor_x))
+                target_res_y = int(np.round(closest_lvl_dim[1] * ds_factor_y))
+
+                closest_lvl_wsi = closest_lvl_wsi.resize((target_res_x, target_res_y), pil_image.BILINEAR)
+                print(f"Case 1: Downscaling using factor {(ds_factor_x, ds_factor_y)}")
+
+            else:
+                # Else: increase resolution (ie, decrement level) and then downsample
+                closest_lvl = closest_lvl - 1
+                mpp_closest_lvl = mpp_list[closest_lvl]  # Update MPP
+                mpp_closest_lvl_x, mpp_closest_lvl_y = mpp_closest_lvl
+
+                ds_factor_x = mpp_closest_lvl_x / user_mpp_x
+                ds_factor_y = mpp_closest_lvl_y / user_mpp_y
+
+                closest_lvl_dim = lvl_dims[closest_lvl]
+                closest_lvl_dim = (closest_lvl_dim[1], closest_lvl_dim[0])
+
+                closest_lvl_wsi = pil_image.fromarray(wsi.pages[closest_lvl].asarray())  # Might be suboptimal
+
+                target_res_x = int(np.round(closest_lvl_dim[0] * ds_factor_x))
+                target_res_y = int(np.round(closest_lvl_dim[1] * ds_factor_y))
+
+                closest_lvl_wsi = closest_lvl_wsi.resize((target_res_x, target_res_y), pil_image.BILINEAR)
+                print(f"Case 2: Downscaling using factor {(ds_factor_x, ds_factor_y)}, now from level {closest_lvl}")
+
+        wsi_arr = np.array(closest_lvl_wsi)
+        return wsi_arr
+
     def get_power(self, wsi, level: int) -> float:
         """
         Returns the objective power of the whole slide image at a given level.

monai/transforms/regularization/array.py

@@ -22,7 +22,6 @@
 
 
 class Mixer(RandomizableTransform):
-
     def __init__(self, batch_size: int, alpha: float = 1.0) -> None:
         """
         Mixer is a base class providing the basic logic for the mixup-class of

tests/test_regularization.py

@@ -59,6 +59,7 @@ def test_mixupd(self):
 
 
 class TestCutMix(unittest.TestCase):
+
     def setUp(self) -> None:
         set_determinism(seed=0)