<reponame>vladpopovici/CP_mri<filename>cp_mri/masks.py
#   -*- coding: utf-8 -*-
#
#  --------------------------------------------------------------------
#  Copyright (c) 2022 <NAME> <<EMAIL>>
#
#  Licensed under the MIT License. See LICENSE file in root folder.
#  --------------------------------------------------------------------

# MASKS are single channel images (arrays) storing information about a
# corresponding pixel in the image. A binary mask, for example, indicates
# that the pixels corresponding to "True" values in the mask have a common
# property (e.g. they belong to the same object).
# Here we use PyVIPS for creating and storing large masks (potentially
# larger than available RAM). Also, pyramids are generated from a given
# mask to match the levels of the corresponding MRI.
#
# Usage philosophy: a mask is created to match the image shape (width and
# height) of the image currently processed. This mask is stored as a temporary
# ZARR array, in <tmp_folder>. Upon completion of the processing, the user must
# request that the mask is written to its final destination: either a large
# TIFF file or a ZARR pyramid. The temporary file is automatically deleted
# when the mask is closed (mask object expires).

import numpy as np
from pathlib import Path
import shutil
import zarr
from hashlib import md5
from time import localtime
import tifffile as tif
from skimage.transform import resize


class BinaryMask(object):
    def __init__(self, shape: dict[str, int],
                 dtype=np.uint8,
                 tmp_folder: str='/tmp',
                 tmp_prefix: str='tmp_binarymask_'):
        """Initialize a binary mask.

        Args:
            shape: (dict) {'width', 'height'} of the mask
            dtype: (np.dtype) type of values stored
            tmp_folder: (str) path where the temp mask is stored
            tmp_prefix: (str) prefix for the temp mask
        """
        self._shape = {'width' : shape['width'], 'height' : shape['height']}
        self._dtype = dtype
        self._white = np.iinfo(self._dtype).max
        if not Path(tmp_folder).exists():
            tmp_folder = '/tmp'
        random_file_name = tmp_prefix + md5(str(localtime()).encode('utf-8')).hexdigest() + '.zarr'
        self._mask_storage_path = Path(tmp_folder) / Path(random_file_name)

        chunks = (min(4096, self._shape['height']), min(4096, self._shape['width']))
        self._mask_storage = zarr.open(str(self._mask_storage_path), mode='w',
                                       shape=(self._shape['height'], self._shape['width']),
                                       chunks=chunks,
                                       dtype=self._dtype)
        self._mask_storage[:] = 0

        return

    def __del__(self):
        self._mask_storage.store.close()
        if self._mask_storage_path.exists():
            shutil.rmtree(self._mask_storage_path)
        return

    def get_temp_path(self) -> Path:
        return self._mask_storage_path

    @property
    def mask(self) -> zarr.Array:
        return self._mask_storage

    def to_image(self, dst_path: Path):
        self._mask_storage.set_mask_selection(self._mask_storage[:] > 0,
                                              self._white) # 0 - black, everything else - white
        tif.imwrite(dst_path.with_suffix('.tiff'), self._mask_storage,
                    bigtiff=True, photometric='minisblack',
                    compression='zlib', metadata={'axes': 'CYX'})
        return

    def to_pyramid(self, dst_path: Path,
                   current_level: int,
                   max_level: int,
                   min_level: int=0
                   ):
        """Generate a multi-resolution pyramid from the current mask.

        The pyramid will have the current mask as its <current_level> image, all
        other levels being generated via down-/up-sampling from it. The number
        of levels to be generated is controlled via <min_level> (non-negative) and
        <max_level>.

        Args:
            dst_path: (Path) where to save the pyramid.zarr dataset
            current_level: (int) the level in the pyramid represented by the mask
            max_level: (int) maximum level (smallest image) to be generated
            min_level: (int) minimum level (largest image) to be generated
        """


        min_level = max(min_level, 0)

        # now, find a max_level that does not shrink the image below 128px in either
        # width or height
        max_level = max(current_level, max_level)
        d = min(self._mask_storage.shape)
        l = max_level - current_level
        while d // 2**l < 128 and l >= 0:
            l -= 1
        if l < 0:
            l = 0
        max_level = current_level + l

        with zarr.open_group(str(dst_path.with_suffix('.zarr')), mode='w') as zroot:
            pyramid_info = []
            # up-sampling:
            for level in range(min_level, max_level):
                factor = 2**(current_level - level)
                mask_shape = (int(factor * self._mask_storage.shape[0]),
                              int(factor * self._mask_storage.shape[1]))
                chunks = (min(4096, mask_shape[0]), min(4096, mask_shape[1]))
                new_mask = zroot.create_dataset(str(level),
                                                shape=mask_shape,
                                                chunks=chunks,
                                                dtype=self._mask_storage.dtype)
                if level == current_level:
                    # just copy:
                    new_mask[:] = self._mask_storage[:]
                else:
                    new_mask[:] = resize(self._mask_storage[:], mask_shape, order=0,
                                         mode='reflect', anti_aliasing=False)

                new_mask.set_mask_selection(new_mask[:] > 0, self._white)

                pyramid_info.append({
                    'level': level,
                    'width': mask_shape[1],
                    'height' : mask_shape[0],
                    'downsample_factor': 2**(level - min_level)
                })
            zroot.attrs['pyramid'] = pyramid_info
            zroot.attrs['pyramid_desc'] = 'generated for binary mask'

        return

#   -*- coding: utf-8 -*-
#
#  --------------------------------------------------------------------
#  Copyright (c) 2022 <NAME> <<EMAIL>>
#
#  Licensed under the MIT License. See LICENSE file in root folder.
#  --------------------------------------------------------------------

# MASKS are single channel images (arrays) storing information about a
# corresponding pixel in the image. A binary mask, for example, indicates
# that the pixels corresponding to "True" values in the mask have a common
# property (e.g. they belong to the same object).
# Here we use PyVIPS for creating and storing large masks (potentially
# larger than available RAM). Also, pyramids are generated from a given
# mask to match the levels of the corresponding MRI.
#
# Usage philosophy: a mask is created to match the image shape (width and
# height) of the image currently processed. This mask is stored as a temporary
# ZARR array, in <tmp_folder>. Upon completion of the processing, the user must
# request that the mask is written to its final destination: either a large
# TIFF file or a ZARR pyramid. The temporary file is automatically deleted
# when the mask is closed (mask object expires).

import numpy as np
from pathlib import Path
import shutil
import zarr
from hashlib import md5
from time import localtime
import tifffile as tif
from skimage.transform import resize


class BinaryMask(object):
    def __init__(self, shape: dict[str, int],
                 dtype=np.uint8,
                 tmp_folder: str='/tmp',
                 tmp_prefix: str='tmp_binarymask_'):
        """Initialize a binary mask.

        Args:
            shape: (dict) {'width', 'height'} of the mask
            dtype: (np.dtype) type of values stored
            tmp_folder: (str) path where the temp mask is stored
            tmp_prefix: (str) prefix for the temp mask
        """
        self._shape = {'width' : shape['width'], 'height' : shape['height']}
        self._dtype = dtype
        self._white = np.iinfo(self._dtype).max
        if not Path(tmp_folder).exists():
            tmp_folder = '/tmp'
        random_file_name = tmp_prefix + md5(str(localtime()).encode('utf-8')).hexdigest() + '.zarr'
        self._mask_storage_path = Path(tmp_folder) / Path(random_file_name)

        chunks = (min(4096, self._shape['height']), min(4096, self._shape['width']))
        self._mask_storage = zarr.open(str(self._mask_storage_path), mode='w',
                                       shape=(self._shape['height'], self._shape['width']),
                                       chunks=chunks,
                                       dtype=self._dtype)
        self._mask_storage[:] = 0

        return

    def __del__(self):
        self._mask_storage.store.close()
        if self._mask_storage_path.exists():
            shutil.rmtree(self._mask_storage_path)
        return

    def get_temp_path(self) -> Path:
        return self._mask_storage_path

    @property
    def mask(self) -> zarr.Array:
        return self._mask_storage

    def to_image(self, dst_path: Path):
        self._mask_storage.set_mask_selection(self._mask_storage[:] > 0,
                                              self._white) # 0 - black, everything else - white
        tif.imwrite(dst_path.with_suffix('.tiff'), self._mask_storage,
                    bigtiff=True, photometric='minisblack',
                    compression='zlib', metadata={'axes': 'CYX'})
        return

    def to_pyramid(self, dst_path: Path,
                   current_level: int,
                   max_level: int,
                   min_level: int=0
                   ):
        """Generate a multi-resolution pyramid from the current mask.

        The pyramid will have the current mask as its <current_level> image, all
        other levels being generated via down-/up-sampling from it. The number
        of levels to be generated is controlled via <min_level> (non-negative) and
        <max_level>.

        Args:
            dst_path: (Path) where to save the pyramid.zarr dataset
            current_level: (int) the level in the pyramid represented by the mask
            max_level: (int) maximum level (smallest image) to be generated
            min_level: (int) minimum level (largest image) to be generated
        """


        min_level = max(min_level, 0)

        # now, find a max_level that does not shrink the image below 128px in either
        # width or height
        max_level = max(current_level, max_level)
        d = min(self._mask_storage.shape)
        l = max_level - current_level
        while d // 2**l < 128 and l >= 0:
            l -= 1
        if l < 0:
            l = 0
        max_level = current_level + l

        with zarr.open_group(str(dst_path.with_suffix('.zarr')), mode='w') as zroot:
            pyramid_info = []
            # up-sampling:
            for level in range(min_level, max_level):
                factor = 2**(current_level - level)
                mask_shape = (int(factor * self._mask_storage.shape[0]),
                              int(factor * self._mask_storage.shape[1]))
                chunks = (min(4096, mask_shape[0]), min(4096, mask_shape[1]))
                new_mask = zroot.create_dataset(str(level),
                                                shape=mask_shape,
                                                chunks=chunks,
                                                dtype=self._mask_storage.dtype)
                if level == current_level:
                    # just copy:
                    new_mask[:] = self._mask_storage[:]
                else:
                    new_mask[:] = resize(self._mask_storage[:], mask_shape, order=0,
                                         mode='reflect', anti_aliasing=False)

                new_mask.set_mask_selection(new_mask[:] > 0, self._white)

                pyramid_info.append({
                    'level': level,
                    'width': mask_shape[1],
                    'height' : mask_shape[0],
                    'downsample_factor': 2**(level - min_level)
                })
            zroot.attrs['pyramid'] = pyramid_info
            zroot.attrs['pyramid_desc'] = 'generated for binary mask'

        return
#   -*- coding: utf-8 -*- # #  -------------------------------------------------------------------- #  Copyright (c) 2022 <NAME> <<EMAIL>> # #  Licensed under the MIT License. See LICENSE file in root folder. #  -------------------------------------------------------------------- # MASKS are single channel images (arrays) storing information about a # corresponding pixel in the image. A binary mask, for example, indicates # that the pixels corresponding to "True" values in the mask have a common # property (e.g. they belong to the same object). # Here we use PyVIPS for creating and storing large masks (potentially # larger than available RAM). Also, pyramids are generated from a given # mask to match the levels of the corresponding MRI. # # Usage philosophy: a mask is created to match the image shape (width and # height) of the image currently processed. This mask is stored as a temporary # ZARR array, in <tmp_folder>. Upon completion of the processing, the user must # request that the mask is written to its final destination: either a large # TIFF file or a ZARR pyramid. The temporary file is automatically deleted # when the mask is closed (mask object expires). Initialize a binary mask.          Args:             shape: (dict) {'width', 'height'} of the mask             dtype: (np.dtype) type of values stored             tmp_folder: (str) path where the temp mask is stored             tmp_prefix: (str) prefix for the temp mask # 0 - black, everything else - white Generate a multi-resolution pyramid from the current mask.          The pyramid will have the current mask as its <current_level> image, all         other levels being generated via down-/up-sampling from it. The number         of levels to be generated is controlled via <min_level> (non-negative) and         <max_level>.          Args:             dst_path: (Path) where to save the pyramid.zarr dataset             current_level: (int) the level in the pyramid represented by the mask             max_level: (int) maximum level (smallest image) to be generated             min_level: (int) minimum level (largest image) to be generated # now, find a max_level that does not shrink the image below 128px in either # width or height # up-sampling: # just copy: