import dataclasses
import multiprocessing
import shutil
import tempfile
import uuid
import warnings
from abc import ABC, abstractmethod
from functools import partial
from math import floor
from numbers import Number
from pathlib import Path
from typing import (
Any,
Callable,
Dict,
Generator,
Iterable,
Iterator,
List,
Optional,
Tuple,
TypeVar,
Union,
)
import numpy as np
import zarr
from wsic import __version__ as wsic_version
from wsic.codecs import register_codecs
from wsic.enums import Codec, ColorSpace
from wsic.metadata import ngff
from wsic.readers import DICOMWSIReader, Reader, TIFFReader
from wsic.tile_iterators import DaskTileIterator
from wsic.tile_iterators import (
PersistentMultiProcessTileIterator as MultiProcessTileIterator,
)
from wsic.typedefs import PathLike
from wsic.utils import (
downsample_shape,
mean_pool,
mosaic_shape,
mpp2ppu,
scale_to_fit,
tile_slices,
warn_unused,
)
[docs]class Writer(ABC):
"""Base class for image writers.
Args:
path (PathLike):
Path to the output file.
shape (Tuple[int, int]):
Shape of the output image.
tile_size (Tuple[int, int], optional):
A (width, height) tuple of output tile size in pixels.
Defaults to (256, 256).
dtype (np.dtype, optional):
Data type of the output image. Defaults to np.uint8.
color_space (ColorSpace, optional):
Color space the output image.
Defaults to "rgb".
compression (str, optional):
Compression codec to use. Defaults to None. Not all
writers support compression.
compression_level (int, optional):
Compression level to use. Defaults to 0 (lossless /
maximum).
microns_per_pixel (Tuple[float, float], optional):
A (width, height) tuple of microns per pixel. Defaults to
None.
pyramid_downsamples (List[int], optional):
A list of downsamples to use in the pyramid. Defaults to
None. Not all writers support pyramids.
overwrite (bool, optional):
Overwrite output file if it exists.
Defaults to False.
verbose (bool, optional):
Print more output. Defaults to False.
"""
T = TypeVar("T")
def __init__(
self,
path: Optional[PathLike],
shape: Tuple[int, int],
tile_size: Tuple[int, int] = (256, 256),
dtype: np.dtype = np.uint8,
color_space: Optional[ColorSpace] = ColorSpace.RGB,
codec: Optional[Codec] = None,
compression_level: int = 0,
microns_per_pixel: Tuple[float, float] = None,
pyramid_downsamples: Optional[List[int]] = None,
overwrite: bool = False,
verbose: bool = False,
):
self.path = Path(path) if path is not None else path
self.shape = shape
self.tile_size = tile_size
self.dtype = dtype
self.color_space = color_space or ColorSpace.RGB
self.codec = Codec.from_string(codec) if codec else None
self.compression_level = compression_level or 0
self.microns_per_pixel = microns_per_pixel
self.pyramid_downsamples = (
list(pyramid_downsamples) if pyramid_downsamples else []
)
self.overwrite = overwrite
self.verbose = verbose
if path is not None and self.path.exists() and not self.overwrite:
raise FileExistsError(f"{self.path} already exists")
[docs] def reader_tile_iterator(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Tuple[int, int] = None,
yield_tile_size: Tuple[int, int] = None,
intermediate: zarr.Group = None,
timeout: float = 10.0,
) -> Iterator[np.ndarray]:
"""Returns an iterator which returns tiles generated by reader.
Args:
reader (Reader):
Reader to read tiles from.
num_workers (int, optional):
Number of workers to use. Defaults to 2.
read_tile_size (Tuple[int, int], optional):
A (width, height) tuple of read tile size in pixels.
Defaults to self.tile_size.
intermediate (np.ndarray, optional):
An intermediate image to write tiles to.
Returns:
Iterator: Iterator which returns tiles generated by reader.
"""
if read_tile_size is None:
read_tile_size = self.tile_size
if hasattr(reader, "_tzyxc_dataset"):
return DaskTileIterator(
reader=reader,
read_tile_size=read_tile_size,
yield_tile_size=yield_tile_size or self.tile_size,
intermediate=intermediate,
num_workers=num_workers,
verbose=self.verbose,
timeout=timeout,
match_tile_sizes=not isinstance(self, ZarrWriter),
)
return MultiProcessTileIterator(
reader=reader,
read_tile_size=read_tile_size,
yield_tile_size=yield_tile_size or self.tile_size,
intermediate=intermediate,
num_workers=num_workers,
verbose=self.verbose,
timeout=timeout,
match_tile_sizes=not isinstance(self, ZarrWriter),
)
def __setitem__(
self, index: Tuple[Union[int, slice], ...], value: np.ndarray
) -> None:
"""Return pixel data at index."""
raise NotImplementedError() # pragma: no cover
[docs] @abstractmethod
def copy_from_reader(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Tuple[int, int] = None,
timeout: float = 10.0,
downsample_method: Optional[str] = None,
) -> None:
"""Write pixel data to by copying from a Reader.
Args:
reader (Reader):
Reader object.
num_workers (int, optional):
Number of workers to use. Defaults to 2.
read_tile_size (Tuple[int, int], optional):
Tile size to read. Defaults to None.
This will use the tile size of the writer if None.
timeout (float, optional):
Timeout for workers. Defaults to 10s.
downsample_method (str, optional):
Downsample method to use when building pyramid levels.
Defaults to None. Valid downsample methods are: "cv2",
"scipy", "np", None.
"""
if self.path.exists() and not self.overwrite:
raise FileExistsError(f"{self.path} exists and overwrite is False.")
[docs] def transcode_from_reader(
self,
reader: Union[TIFFReader, DICOMWSIReader],
downsample_method: Optional[str] = None,
) -> None:
raise NotImplementedError()
[docs] @staticmethod
def level_progress(iterable: Iterable[T], **kwargs) -> Iterator[T]:
"""Wrapper for a tile yeilding iterable when writing a level.
Used to display progress when copying from a reader.
Some of the tqdm defaults are overridden but can be changed by
passing values as kwargs. Parameters which differ to the tqdm
defaults here are:
- `smoothing = 0.1`
- `colour = "magenta"`
Args:
iterable (Iterable):
The iterable to wrap.
**kwargs (dict):
Extra kwargs for tqdm. Overrides defaults.
"""
tqdm_kwargs = {
"colour": "magenta",
"smoothing": 0.01,
"desc": "Writing",
}
tqdm_kwargs.update(kwargs)
try:
from tqdm.auto import tqdm
return tqdm(iterable, **tqdm_kwargs)
except ImportError:
return iterable
[docs] @staticmethod
def pyramid_progress(iterable: Iterable, **kwargs) -> Iterator:
"""Wrap an iterable in a progress bar.
Used to display progress when copying from a reader.
Some of the tqdm defaults are overridden but can be changed by
passing values as kwargs. Parameters which differ to the tqdm
defaults here are:
- `smoothing = 0`
- `colour = "magenta"`
Args:
iterable (Iterable):
The iterable to wrap.
**kwargs (dict):
Extra kwargs for tqdm. Overrides defaults.
"""
tqdm_kwargs = {
"colour": "blue",
# Bar format with no ETA
"bar_format": "{l_bar}{bar}| {n_fmt}/{total_fmt}",
"desc": "Building Pyramid",
}
tqdm_kwargs.update(kwargs)
try:
from tqdm.auto import tqdm
return tqdm(iterable, **tqdm_kwargs)
except ImportError:
return iterable
[docs] @staticmethod
def transcode_progress(iterable: Iterable, **kwargs) -> Iterable:
"""Progress bar for transcoding.
Args:
iterable (Iterable):
Iterable to wrap.
**kwargs:
Keyword arguments to pass to the progress bar.
Returns:
Iterable:
"""
try:
from tqdm.auto import tqdm
return tqdm(
iterable,
desc="Transcoding",
colour="green",
**kwargs,
)
except ImportError:
return iterable
[docs]class JP2Writer(Writer):
"""Tile-wise JP2 writer using glymur.
Note that when writing tiled JP2 files, the tiles must all be the
same size and must be written in the order left-to-right, then
top-to-bottom (row-by-row). Tiles cannot be skipped.
Args:
path (PathLike):
Path to output file.
shape (Tuple[int, int]):
A (width, height) tuple of image size in pixels.
tile_size (Tuple[int, int], optional):
A (width, height) tuple of tile size in pixels.
Defaults to (256, 256).
dtype (np.dtype, optional):
Data type of output image. Defaults to np.uint8.
color_space (ColorSpace, optional):
Color space the output image.
Defaults to "rgb".
compression (str, optional):
Compression type. Currently only JPEG 2000 compression is
supported. Defaults to None.
compression_level (int, optional):
Compression level. Currently unused. Defaults to None.
microns_per_pixel (Tuple[float, float], optional):
A (width, height) tuple of microns per pixel.
Defaults to None.
pyramid_downsamples (List[int], optional):
A list of downsamples to create. Unused but included
for API consistency. Defaults to None.
overwrite (bool, optional):
Overwrite existing file. Defaults to False.
verbose (bool, optional):
Print more output. Defaults to False.
"""
def __init__(
self,
path: PathLike,
shape: Tuple[int, int],
tile_size: Tuple[int, int] = (256, 256),
dtype: np.dtype = np.uint8,
color_space: Optional[ColorSpace] = ColorSpace.RGB,
codec: Optional[Codec] = "jpeg2000", # Currently unused
compression_level: int = 0,
microns_per_pixel: Optional[Tuple[float, float]] = None, # Currently unused
pyramid_downsamples: Optional[List[int]] = None,
overwrite: bool = False,
verbose: bool = False,
**kwargs,
) -> None:
if codec != "jpeg2000":
warn_unused(codec)
pyramid_downsamples = pyramid_downsamples or []
if not np.array_equal(
pyramid_downsamples,
[2 ** (x + 1) for x in range(len(pyramid_downsamples))],
):
raise ValueError(
"Pyramid downsamples must be consecutive powers of 2 for JP2."
)
super().__init__(
path=path,
shape=shape,
tile_size=tile_size,
dtype=dtype,
color_space=color_space,
codec=codec,
compression_level=compression_level,
microns_per_pixel=microns_per_pixel,
pyramid_downsamples=pyramid_downsamples,
overwrite=overwrite,
verbose=verbose,
)
def __setitem__(self, index: Tuple[int, ...], value: np.ndarray) -> None:
"""Write pixel data at index. Not supported for JP2Writer."""
raise NotImplementedError("JP2 files do not support random access writes.")
[docs] def copy_from_reader(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Optional[Tuple[int, int]] = None,
timeout: float = 10.0,
downsample_method: Optional[str] = None,
) -> None:
"""Write pixel data to by copying from a Reader.
Args:
reader (Reader):
Reader object.
num_workers (int, optional):
Number of workers to use. Defaults to 2.
read_tile_size (Tuple[int, int], optional):
Tile size to read. Defaults to None.
This will use the tile size of the writer if None.
timeout (float, optional):
Timeout for workers. Defaults to 10s.
downsample_method (str, optional):
Downsample method to use. Defaults to None. Not used for
JP2Writer, but included for API consistency.
"""
warn_unused(downsample_method, ignore_falsey=True)
super().copy_from_reader(
reader=reader,
num_workers=num_workers,
read_tile_size=read_tile_size,
timeout=timeout,
downsample_method=downsample_method,
)
import glymur
numres = len(self.pyramid_downsamples) + 1 if self.pyramid_downsamples else None
psnr = (
[self.compression_level]
if isinstance(self.compression_level, Number)
else self.compression_level
)
mpp = self.microns_per_pixel or reader.microns_per_pixel
capture_resolution = tuple(mpp2ppu(x, "m") for x in mpp) if mpp else None
jp2 = glymur.Jp2k(
self.path,
shape=reader.shape,
tilesize=self.tile_size,
verbose=self.verbose,
numres=numres,
psnr=psnr,
colorspace=self.color_space,
capture_resolution=capture_resolution,
)
with ZarrIntermediate(
None,
reader.shape,
zero_after_read=True,
) as intermediate:
reader_tile_iterator = self.reader_tile_iterator(
reader=reader,
num_workers=num_workers,
intermediate=intermediate,
read_tile_size=read_tile_size or self.tile_size,
timeout=timeout,
)
reader_tile_iterator = iter(self.level_progress(reader_tile_iterator))
for tile_writer in jp2.get_tilewriters():
try:
tile_writer[:] = next(reader_tile_iterator)
except StopIteration as error:
raise StopIteration(
"Reader tile iterator stopped early. "
"Glymur is expecting more tiles to be written."
) from error
[docs]class TIFFWriter(Writer):
"""Tile-wise TIFF writer using tifffile.
Note that when writing tiled TIFF files, the tiles must all be the
same size and must be written in the order left-to-right, then
top-to-bottom (row-by-row). Tiles cannot be skipped.
Notes:
The following notes are from the TIFF 6.0 Specification.
- TileWidth and TileLength (height) must each be a multiple of 16.
- "Offsets [bytes from the start of file to each tile blob and
therefore the tile ordering when writing] are ordered
left-to-right and top-to-bottom."
- "For PlanarConfiguration = 2, the offsets for the first
component plane are stored first, followed by all the offsets
for the second component plane, and so on."
The full specification is available at:
https://web.archive.org/web/20210108174645/https://www.adobe.io/content/dam/udp/en/open/standards/tiff/TIFF6.pdf
Args:
path (PathLike):
Path to output file.
shape (Tuple[int, int]):
A (width, height) tuple of image size in pixels.
tile_size (Tuple[int, int], optional):
A (width, height) tuple of tile size in pixels.
Defaults to (256, 256).
dtype (np.dtype, optional):
Data type of output image. Defaults to np.uint8.
color_space (ColorSpace, optional):
color_space. Defaults to "rgb".
compression (str, optional):
Compression type.
Defaults to "jpeg".
compression_level (int, optional):
Compression level. Defaults to -1 (highest / lossless).
microns_per_pixel (Tuple[float, float], optional):
A (width, height) tuple of microns per pixel.
Defaults to None.
pyramid_downsamples (List[int], optional):
A list of downsamples to create. Should be strictly
inceasing for maximum compatibility.
Defaults to None.
overwrite (bool, optional):
Overwrite existing file. Defaults to False.
verbose (bool, optional):
Print more output. Defaults to False.
ome (bool):
Write OME-TIFF metadata. Defaults to False.
"""
def __init__(
self,
path: Path,
shape: Tuple[int, int],
tile_size: Tuple[int, int] = (256, 256),
dtype: np.dtype = np.uint8, # Currently unused
color_space: Optional[ColorSpace] = "rgb",
codec: Optional[Codec] = "jpeg",
compression_level: int = -1, # Currently unused
microns_per_pixel: Tuple[float, float] = None,
pyramid_downsamples: Optional[List[int]] = None,
overwrite: bool = False,
verbose: bool = False,
*,
ome: bool = False,
) -> None:
if dtype != np.uint8:
warn_unused(dtype)
super().__init__(
path=path,
shape=shape,
tile_size=tile_size,
dtype=dtype,
color_space=color_space,
codec=codec,
compression_level=compression_level,
microns_per_pixel=microns_per_pixel,
pyramid_downsamples=pyramid_downsamples,
overwrite=overwrite,
verbose=verbose,
)
self.ome = ome
def __setitem__(self, index: Tuple[int, ...], value: np.ndarray) -> None:
"""Write pixel data at index. Not supported for TIFFWriter.
In theory this is possible but it can be complex. If the new tile
is larger in bytes, the tile will have to be added to the end of the
file. The old tile will remain in the file and waste space unless it
is later overwritten by another of length smaller or equal to the
original tile.
"""
raise NotImplementedError(
"Compressed tiled TIFF files do not support random access writes."
)
[docs] def copy_from_reader(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Optional[Tuple[int, int]] = None,
timeout: float = 10.0,
downsample_method: Optional[str] = None,
) -> None:
"""Write pixel data to by copying from a Reader.
Args:
reader (Reader):
Reader object.
num_workers (int, optional):
Number of workers to use. Defaults to 2.
read_tile_size (Tuple[int, int], optional):
Tile size to read. Defaults to None.
This will use the tile size of the writer if None.
timeout (float, optional):
Timeout for workers. Defaults to 10s.
downsample_method (str, optional):
Downsample method to use when building pyramid levels.
Defaults to None. Valid downsample methods are: "cv2",
"scipy", "np", None.
"""
super().copy_from_reader(
reader=reader,
num_workers=num_workers,
read_tile_size=read_tile_size,
timeout=timeout,
downsample_method=downsample_method,
)
import tifffile
microns_per_pixel = self.microns_per_pixel or reader.microns_per_pixel
resolution = (
(
round(mpp2ppu(microns_per_pixel[0], "cm")),
round(mpp2ppu(microns_per_pixel[1], "cm")),
)
if microns_per_pixel
else None
)
tile_size = self.tile_size
with ZarrIntermediate(
None, reader.shape, zero_after_read=False
) as intermediate:
reader_tile_iterator = self.reader_tile_iterator(
reader=reader,
num_workers=num_workers,
intermediate=intermediate,
read_tile_size=read_tile_size or self.tile_size,
timeout=timeout,
)
reader_tile_iterator = self.level_progress(reader_tile_iterator)
# Write baseline (level 0)
with tifffile.TiffWriter(
self.path,
bigtiff=True,
ome=self.ome,
) as tif:
metadata = {}
if self.ome and self.microns_per_pixel:
metadata["PhysicalSizeXUnit"] = "µm"
metadata["PhysicalSizeYUnit"] = "µm"
metadata["PhysicalSizeX"] = self.microns_per_pixel[0]
metadata["PhysicalSizeY"] = self.microns_per_pixel[1]
resolution = resolution or (1.0, 1.0)
self.validate_write_args(
tile_size=tile_size,
resolution=resolution,
)
tif.write(
data=iter(reader_tile_iterator),
tile=self.tile_size,
shape=reader.shape,
dtype=reader.dtype,
photometric=self.color_space,
compression=(self.codec.condensed(), self.compression_level),
# tifffile uses 1.0 by default but we also set it explicitly
resolution=resolution,
# Default unit is inches if resolution is not None
resolutionunit="centimeter" if resolution else None,
subifds=len(self.pyramid_downsamples),
metadata=metadata,
)
# Write pyramid resolutions
with multiprocessing.Pool(num_workers) as pool:
for level, downsample in self.pyramid_progress(
enumerate(self.pyramid_downsamples),
total=len(self.pyramid_downsamples),
):
level_shape = tuple(
floor(s / downsample) for s in reader.shape[:2]
) + (reader.shape[-1],)
level_tiles_shape = mosaic_shape(
level_shape,
self.tile_size,
)
func = partial(
get_level_tile,
tile_size=self.tile_size,
downsample=downsample,
read_intermediate_path=intermediate.path,
downsample_method=downsample_method,
)
tile_generator = pool.imap(
func=func,
iterable=np.ndindex(level_tiles_shape),
)
tile_generator = self.level_progress(
tile_generator,
total=int(np.product(level_tiles_shape)),
desc=f"Level {level + 1}",
leave=False,
)
tif.write(
data=iter(tile_generator),
tile=tile_size,
shape=level_shape,
dtype=reader.dtype,
photometric=self.color_space,
compression=(
self.codec.condensed(),
self.compression_level,
),
subfiletype=1, # Subfile type: reduced resolution
)
[docs] def transcode_from_reader(
self,
reader: Union[TIFFReader, DICOMWSIReader],
downsample_method: Optional[str] = None,
) -> None:
# Validate input
if not reader.tile_shape:
raise ValueError(
"Reader must have a known tile shape/size and implement get_tile."
)
import tifffile
microns_per_pixel = self.microns_per_pixel or reader.microns_per_pixel
resolution = (
(
round(mpp2ppu(microns_per_pixel[0], "cm")),
round(mpp2ppu(microns_per_pixel[1], "cm")),
)
if microns_per_pixel
else None
)
tile_size = reader.tile_shape[:2][::-1]
reader_mosaic_shape = mosaic_shape(reader.original_shape[:2], tile_size)
tile_generator = (
reader.get_tile(tile_index, decode=False)
for tile_index in np.ndindex(reader_mosaic_shape)
)
tile_iterator = self.level_progress(
tile_generator, total=np.product(reader_mosaic_shape)
)
metadata = {}
if self.ome and self.microns_per_pixel:
metadata["PhysicalSizeXUnit"] = "µm"
metadata["PhysicalSizeYUnit"] = "µm"
metadata["PhysicalSizeX"] = self.microns_per_pixel[0]
metadata["PhysicalSizeY"] = self.microns_per_pixel[1]
# Fall back to (1.0, 1.0) as a resolution is required.
# tifffile uses 1.0 by default but we also set it explicitly
resolution = resolution or (1.0, 1.0)
self.validate_write_args(
tile_size=tile_size,
resolution=resolution,
)
with tifffile.TiffWriter(
self.path,
bigtiff=True,
ome=self.ome,
) as tiff:
tiff.write(
data=iter(tile_iterator),
tile=tile_size,
shape=reader.original_shape,
dtype=reader.dtype,
photometric=reader.color_space.to_tiff(),
jpegtables=reader.jpeg_tables,
compression=reader.codec.condensed(),
metadata=metadata,
resolution=resolution,
# Default unit is inches if resolution is not None
resolutionunit="centimeter" if resolution else None,
)
[docs] @staticmethod
def validate_write_args(
tile_size: Tuple[int, int],
resolution: Optional[Tuple[float, float]] = None,
) -> None:
"""Validate write arguments."""
from wsic.utils import ppu2mpp
from wsic.validation import check_mpp
# Check that tile size is a multiple of 16
if any(s % 16 for s in tile_size):
raise ValueError(
"Tile size must be a multiple of 16 pixels for a TIFF file."
)
# Check if resolution is present
if not resolution:
warnings.warn(
"No resolution data. Output file will not have any "
"resolution metadata.",
stacklevel=2,
)
return
# Check that resolution is a sensible value
for r in resolution:
check_mpp(mpp=ppu2mpp(r, "cm"))
[docs]class SVSWriter(Writer):
"""Aperio SVS writer using tifffile.
Notes:
- When writing tiled TIFF files, the tiles must all be the same
size and must be written in the order left-to-right, then
top-to-bottom (row-by-row). Tiles cannot be skipped.
- Microns per pixel (MPP) is taken from microns per pixel of the
reader if not provided. If microns per pixel is set to None,
microns per pixel will not be written to the file.
- The SVS MPP metadata is the mean of `microns_per_pixel` from
init or the reader.
- Apparent magnification (AppMag) can be specified as a float to
the optional "app_mag" kwarg.
- If only an MPP resolution is given, the AppMag will be
approximated from the MPP (by AppMag = 10 / MPP) and rounded
to the nearest common AppMag (10, 20,40, 50, 60, 80, 100, 125,
150, 200, 250, 312.5, 375, 500, 600, 750, 1000, 1250). If
neither MPP or AppMag are given, no resolution will be written
to the file.
Args:
path (PathLike):
Path to output file.
shape (Tuple[int, int]):
A (width, height) tuple of image size in pixels.
tile_size (Tuple[int, int], optional):
A (width, height) tuple of tile size in pixels.
Defaults to (256, 256).
dtype (np.dtype, optional):
Data type of output image. Defaults to np.uint8.
color_space (ColorSpace, optional):
color_space. Defaults to "rgb".
compression (str, optional):
Compression type.
Defaults to "jpeg".
compression_level (int, optional):
Compression level. Defaults to 95. Currently unused.
microns_per_pixel (Tuple[float, float], optional):
A (width, height) tuple of microns per pixel.
Defaults to None.
pyramid_downsamples (List[int], optional):
A list of downsamples to create. Should be strictly
inceasing for maximum compatibility.
Defaults to None.
overwrite (bool, optional):
Overwrite existing file. Defaults to False.
verbose (bool, optional):
Print more output. Defaults to False.
ome (bool):
Write OME-TIFF metadata. Defaults to False.
app_mag (float):
Apparent magnification. Defaults to None.
"""
def __init__(
self,
path: Path,
shape: Tuple[int, int],
tile_size: Tuple[int, int] = (256, 256),
dtype: np.dtype = np.uint8, # Currently unused
color_space: Optional[ColorSpace] = ColorSpace.RGB,
codec: Optional[Codec] = Codec.JPEG,
compression_level: int = 0, # Currently unused
microns_per_pixel: Tuple[float, float] = None,
pyramid_downsamples: Optional[List[int]] = None,
overwrite: bool = False,
verbose: bool = False,
**kwargs,
) -> None:
# Validate inputs
if dtype is not np.uint8:
raise ValueError(f"SVSWriter only supports uint8, not {dtype}")
if color_space not in (ColorSpace.RGB, ColorSpace.YCBCR):
raise ValueError(
f"SVSWriter only supports RGB and YCbCr, not {color_space}"
)
codec = Codec.from_string(codec)
if codec not in (Codec.JPEG,): # aperio_jp2000_ycbc not working
raise ValueError(
"SVSWriter currently only supports JPEG compession," f" not {codec}"
)
# Super
super().__init__(
path=path,
shape=shape,
tile_size=tile_size,
dtype=dtype,
color_space=color_space,
codec=codec,
compression_level=compression_level,
microns_per_pixel=microns_per_pixel,
pyramid_downsamples=pyramid_downsamples,
overwrite=overwrite,
verbose=verbose,
)
# Apparent magnification
self.app_mag = kwargs.get("app_mag")
def __setitem__(self, index: Tuple[int, ...], value: np.ndarray) -> None:
"""Write pixel data at index. Not supported for SVSWriter.
In theory this is possible but it can be complex. If the new tile
is larger in bytes, the tile will have to be added to the end of the
file. The old tile will remain in the file and waste space unless it
is later overwritten by another of length smaller or equal to the
original tile.
"""
raise NotImplementedError(
"Compressed tiled TIFF (SVS) files do not support random access writes."
)
[docs] def copy_from_reader(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Optional[Tuple[int, int]] = None,
timeout: float = 10.0,
downsample_method: Optional[str] = None,
) -> None:
"""Write pixel data to by copying from a Reader.
Args:
reader (Reader):
Reader object.
num_workers (int, optional):
Number of workers to use. Defaults to 2.
read_tile_size (Tuple[int, int], optional):
Tile size to read. Defaults to None.
This will use the tile size of the writer if None.
timeout (float, optional):
Timeout for workers. Defaults to 10s.
downsample_method (str, optional):
Downsample method to use when building pyramid levels.
Defaults to None. Valid downsample methods are: "cv2",
"scipy", "np", None.
"""
super().copy_from_reader(
reader=reader,
num_workers=num_workers,
read_tile_size=read_tile_size,
timeout=timeout,
downsample_method=downsample_method,
)
import tifffile
microns_per_pixel = self.microns_per_pixel or reader.microns_per_pixel
resolution = (
(
round(mpp2ppu(microns_per_pixel[0], "cm")),
round(mpp2ppu(microns_per_pixel[1], "cm")),
"CENTIMETER",
)
if microns_per_pixel
else None
)
with ZarrIntermediate(
None, reader.shape, zero_after_read=False
) as intermediate:
reader_tile_iterator = self.reader_tile_iterator(
reader=reader,
num_workers=num_workers,
intermediate=intermediate,
read_tile_size=read_tile_size or self.tile_size,
timeout=timeout,
)
reader_tile_iterator = self.level_progress(reader_tile_iterator)
# Write baseline (level 0)
with tifffile.TiffWriter(
self.path,
bigtiff=True,
shaped=False,
) as tif:
# Construct the pipe separated Aperio description
# Example description:
# skipcq: PYL-W0105
"""
Aperio Image Library v11.2.1\n
46000x32914 [42673,5576 2220x2967] (240x240) JPEG/RGB Q=30;
Aperio Image Library v10.0.51\n
46920x33014 [0,100 46000x32914] (256x256) JPEG/RGB Q=30|
AppMag = 20|
StripeWidth = 2040|
ScanScope ID = CPAPERIOCS|
Filename = CMU-1|
Date = 12/29/09|Time = 09:59:15|
User = b414003d-95c6-48b0-9369-8010ed517ba7|
Parmset = USM Filter|
MPP = 0.4990|
Left = 25.691574|Top = 23.449873|
LineCameraSkew = -0.000424|
LineAreaXOffset = 0.019265|LineAreaYOffset = -0.000313|
Focus Offset = 0.000000|
ImageID = 1004486|
OriginalWidth = 46920|Originalheight = 33014|
Filtered = 5|
OriginalWidth = 46000|
OriginalHeight = 32914
"""
aperio_desc_compression = {
Codec.JPEG: f"JPEG/{self.color_space.upper()}",
Codec.JPEG2000: "J2K/YUV16",
"APERIO_JP2000_YCBC": "J2K/YUV16",
}
software = f"Aperio wsic Library v{wsic_version}"
# Using reader shape for now, could be user specified in future
original_height = reader.shape[0]
original_width = reader.shape[1]
# Not sure what these are copying original for now
mystery_height = original_height
mystery_width = original_width
# Get compression in Aperio header format
aperio_header_compression = aperio_desc_compression[self.codec.upper()]
headers = [
(
f"{software} \n"
f"{original_width}x{original_height} "
f"[0,100 {mystery_width}x{mystery_height}] "
f"({self.tile_size[0]}x{self.tile_size[1]})"
f" {aperio_header_compression} "
f"Q={self.compression_level}"
)
]
common_mags = np.array(
list(range(1, 10))
+ [
10,
20,
40,
50,
60,
80,
100,
125,
150,
200,
250,
312.5,
375,
500,
600,
750,
1000,
1250,
],
dtype=int,
)
def mpp2appmag(mpp: Optional[float]) -> Optional[int]:
"""Convert microns per pixel to app mag.
This is a rough conversion, but it is the best we can do
without knowing more (e.g. the scan scope ID).
The formula is: app_map = 10 / mpp.
"""
if mpp is None:
return None
return common_mags[np.argmin(np.abs(common_mags - 10 / mpp))]
mpp = np.mean(microns_per_pixel) if microns_per_pixel else None
app_mag = self.app_mag or mpp2appmag(mpp) or None
key_values = {
"AppMag": app_mag, # e.g. 20
"StripeWidth": None,
"ScanScopeID": None, # e.g. CPAPERIOCS
"Filename": None,
"Date": None, # e.g. 01/01/22
"Time": None, # e.g. 09:00:00
"User": None, # e.g. UUID4
"Parmset": None, # e.g. USM Filter
"MPP": mpp, # e.g. 0.5
"Left": None,
"Top": None,
"LineCameraSkew": None, # e.g. -0.003035
"LineAreaXOffset": None, # e.g. 0.000000
"LineAreaYOffset": None, # e.g. 0.000000
"Focus Offset": None, # e.g. -0.001000
"DSR ID": None, # e.g. homer
"ImageID": None, # e.g. 1234
"OriginalWidth": original_width,
"OriginalHeight": original_height,
"Filtered": None, # e.g. 5
}
description = (
"\n".join(headers)
+ ("|" if any(key_values.values()) else "")
+ "|".join(
f"{key} = {value}"
for key, value in key_values.items()
if value is not None
)
)
compression = self.codec.condensed()
# Write baseline (level 0, 1st IFD)
tif.write(
data=iter(reader_tile_iterator),
tile=self.tile_size,
shape=reader.shape,
dtype=reader.dtype,
photometric=self.color_space,
compressionargs={"outcolorspace": self.color_space},
compression=(compression, self.compression_level),
resolution=resolution,
description=description,
subfiletype=0,
)
reader_tile_iterator.close()
# Write the thumbnail (2nd IFD)
# NOTE: Assuming YXC order
print("Writing thumbnail")
thumb_scale = scale_to_fit(
reader.shape[:2],
np.maximum(
np.floor_divide(reader.shape[:2], self.tile_size[::-1]),
(1024, 1024),
),
)
thumb_shape = tuple(floor(s * thumb_scale) for s in reader.shape[:2])
# Ignore warnings about `approx_ok` being unused
warnings.filterwarnings("ignore", message=".*approx_ok.*")
thumbnail = reader.thumbnail(thumb_shape, approx_ok=True)
warnings.resetwarnings()
tif.write(
data=thumbnail,
rowsperstrip=16,
dtype=np.uint8,
photometric=ColorSpace.RGB,
compressionargs={"outcolorspace": self.color_space},
compression=Codec.JPEG,
description=software,
subfiletype=0,
)
# Write pyramid resolutions
with multiprocessing.Pool(num_workers) as pool:
for level, downsample in self.pyramid_progress(
enumerate(self.pyramid_downsamples),
total=len(self.pyramid_downsamples),
):
level_shape = tuple(
floor(s / downsample) for s in reader.shape[:2]
) + (reader.shape[-1],)
level_tiles_shape = mosaic_shape(
level_shape,
self.tile_size,
)
func = partial(
get_level_tile,
tile_size=self.tile_size,
downsample=downsample,
read_intermediate_path=intermediate.path,
downsample_method=downsample_method,
)
tile_generator = pool.imap(
func=func,
iterable=np.ndindex(level_tiles_shape),
)
tile_generator = self.level_progress(
tile_generator,
total=int(np.product(level_tiles_shape)),
desc=f"Level {level + 1}",
leave=False,
)
tif.write(
data=iter(tile_generator),
tile=self.tile_size,
shape=level_shape,
dtype=reader.dtype,
photometric=self.color_space,
compressionargs={"outcolorspace": self.color_space},
compression=(
compression,
self.compression_level,
),
description=software, # Optional for OpenSlide
subfiletype=0, # Subfile type: 1 = reduced resolution
)
[docs]class ZarrWriter(Writer, Reader):
"""Zarr reader and writer.
Args:
path (PathLike, optional):
Path to the output zarr. May be None if `store` is
provided.
shape (Tuple[int, int]):
Shape of the output zarr.
tile_size (Tuple[int, int], optional):
A (width, height) tuple of zarr chunks in pixels.
Defaults to (256, 256).
dtype (np.dtype, optional):
Data type of the output zarr. Defaults to np.uint8.
codec (str, optional):
Compression codec to use. Defaults to None. Not all
writers support compression.
color_space (ColorSpace, optional):
Color space. Defaults to RGB.
compression_level (int, optional):
Compression level to use. Defaults to 0 (lossless /
maximum).
microns_per_pixel (Tuple[float, float], optional):
A (width, height) tuple of microns per pixel. Defaults to
None.
pyramid_downsamples (List[int], optional):
A list of downsamples to use in the pyramid. Defaults to
None.
overwrite (bool, optional):
Overwrite output file if it exists.
Defaults to False.
verbose (bool, optional):
Print more output. Defaults to False.
ome (bool):
Write OME-NGFF metadata. Defaults to False.
Currently not implemented.
store (zarr.StoreLike, optional):
Zarr storage backend to use. Defaults to None, which passes
the `path` argument `zarr.open`. May be a string or a
zarr.StoreLike instance (e.g. MutableMapping). If None, the
path is passed to the `zarr.open` convenince function. See
https://zarr.readthedocs.io/en/stable/api/storage.html and
https://zarr.readthedocs.io/en/stable/api/convenience.html
for more information.
"""
def __init__(
self,
path: Optional[Path] = None,
shape: Tuple[int, int] = None,
tile_size: Tuple[int, int] = (256, 256),
dtype: np.dtype = np.uint8,
color_space: Optional[ColorSpace] = ColorSpace.RGB, # Currently unused
codec: Union[str, Codec] = Codec.BLOSC,
compression_level: int = 9,
microns_per_pixel: Tuple[float, float] = None, # Currently unused
pyramid_downsamples: Optional[List[int]] = None, # Currently unused
overwrite: bool = False,
verbose: bool = False,
*,
ome: bool = False,
store: Optional[zarr.storage.StoreLike] = None,
) -> None:
warn_unused(microns_per_pixel)
super().__init__(
path=path,
shape=shape,
tile_size=tile_size,
dtype=dtype,
color_space=color_space,
codec=codec,
compression_level=compression_level,
microns_per_pixel=microns_per_pixel,
pyramid_downsamples=pyramid_downsamples,
overwrite=True,
verbose=verbose,
)
self.ome = ome
self.overwrite = overwrite
register_codecs()
self.compressor = self.get_codec(codec, compression_level)
# If only path is given, pass this to zarr.open
if store is None:
store = path
# Else check that if path is not None, it matches store.path (if
# store has a path attr).
elif path is not None and (
not hasattr(store, "path") or Path(path) != Path(store.path)
):
raise ValueError(
f"ZarrWriter path {path} not None and does not match " f"store {store}"
)
if store is None:
raise ValueError("ZarrWriter requires either path or store")
self.zarr = zarr.open(store, mode="a")
self.tile_shape = tile_size[::-1]
@property
def mosaic_shape(self) -> Optional[Tuple[int, int]]:
return mosaic_shape(self.shape, self.tile_shape)
[docs] @staticmethod
def get_codec(
codec: Union[str, Codec],
level: int,
**kwargs: Dict[str, Any],
) -> Callable[[bytes], bytes]:
"""Get a codec for the given compression method and compression level."""
import numcodecs
config = Codec.from_string(codec).to_numcodecs_config(level=level)
config.update(kwargs)
return numcodecs.get_codec(config)
def __setitem__(self, index: Tuple[int, ...], value: np.ndarray) -> None:
"""Write pixel data at index."""
self.zarr[0][index] = value
def __getitem__(self, index: Tuple[int, ...]) -> np.ndarray:
"""Read pixel data at index."""
return self.zarr[0][index]
[docs] def copy_from_reader(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Optional[Tuple[int, int]] = None,
timeout: float = 10.0,
downsample_method: Optional[str] = None,
) -> None:
"""Write pixel data to by copying from a Reader.
Args:
reader (Reader):
Reader object.
num_workers (int, optional):
Number of workers to use. Defaults to 2.
read_tile_size (Tuple[int, int], optional):
Tile size to read. Defaults to None.
This will use the tile size of the writer if None.
timeout (float, optional):
Timeout for workers. Defaults to 10s.
downsample_method (str, optional):
Downsample method to use when building pyramid levels.
Defaults to None. Valid downsample methods are: "cv2",
"scipy", "np", None.
"""
# Ensure there is a zarr to write to
self.zarr.create_dataset(
name="0",
shape=self.shape,
dtype=self.dtype,
chunks=(*self.tile_shape, reader.shape[-1]),
compressor=self.compressor,
)
# Validate and normalise inputs
read_tile_size = read_tile_size or self.tile_shape[:2][::-1]
yield_tile_size = self.tile_shape[:2][::-1]
self._validate_pre_write(read_tile_size)
with ZarrIntermediate(
None, shape=reader.shape, zero_after_read=False
) as intermediate:
tile_sizes_match = read_tile_size == yield_tile_size
# Create a reader tile iterator
reader_tile_iterator = self.reader_tile_iterator(
reader,
read_tile_size=read_tile_size,
yield_tile_size=yield_tile_size,
num_workers=num_workers,
timeout=timeout,
intermediate=None if tile_sizes_match else intermediate,
)
reader_tile_iterator = self.level_progress(reader_tile_iterator)
# Write the reader tile iterator to the writer
tiles_shape = mosaic_shape(
reader.shape,
yield_tile_size[::-1],
)
tiles_index = np.ndindex(tiles_shape)
for ji, tile in zip(tiles_index, reader_tile_iterator):
level_0 = self.zarr[0]
level_0[tile_slices(ji, yield_tile_size)] = tile
self._build_pyramid(downsample_method)
self._write_ome_metadata(reader.microns_per_pixel or self.microns_per_pixel)
def _validate_pre_write(self, read_tile_size: Tuple[int, ...]) -> None:
"""Validate parameters before writing to disk.
Args:
read_tile_size (Tuple[int, ...]):
Tile size to read.
Raises:
ValueError:
If compression is lossy and the write tile writesize is
not a multiple of the read tile size
"""
lossy_codecs = ["jpeg"]
optionally_lossy_codecs = ["jpeg2000", "webp", "jpegls", "jpegxl", "jpegxr"]
lossy = self.codec.condensed().lower() in lossy_codecs or (
self.codec in optionally_lossy_codecs and self.compression_level > 0
)
write_multiple_of_read = all(np.mod(read_tile_size, self.tile_size) == 0)
if lossy and not write_multiple_of_read:
raise ValueError(
"Lossy compression requires that the tile write size is a "
"multiple of the read tile size."
)
def _write_ome_metadata(self, mpp: Tuple[float, ...]) -> None:
"""Write OME-NGFF metadata to the .zattrs file in the root.
This is based on version 0.4: https://ngff.openmicroscopy.org/0.4/.
"""
if self.ome:
multiscales = [
ngff.Multiscale(
datasets=[
ngff.Dataset(
path=str(level),
coordinateTransformations=[
ngff.CoordinateTransformation(
"scale",
[
mpp[0] * downsample,
mpp[1] * downsample,
1,
],
)
]
if mpp is not None
else [ngff.CoordinateTransformation("identity")],
)
for level, downsample in enumerate(
[1] + self.pyramid_downsamples
)
],
axes=[
ngff.Axis("y", "space", "micrometer"),
ngff.Axis("x", "space", "micrometer"),
ngff.Axis("c", "channel", None),
],
)
]
# Convert dataclasses
meta_dict = dataclasses.asdict(
ngff.Zattrs(multiscales=multiscales),
# Exclude None values
dict_factory=lambda x: {k: v for (k, v) in x if v is not None},
)
# Set the attrs
for key, value in meta_dict.items():
self.zarr.attrs[key] = value
def _build_pyramid(self, downsample_method: Optional[str] = None):
"""Build the pyramid.
Constructs additional levels of the pyramid from the first level.
Args:
downsample_method (str, optional):
Downsample method to use. Defaults to None.
"""
previous_level = self.zarr[0]
previous_downsample = 1
for level, downsample in self.pyramid_progress(
enumerate(self.pyramid_downsamples, start=1),
):
inter_level_downsample = downsample // previous_downsample
# NOTE: Assuming length three shape with channels last
level_shape = downsample_shape(self.shape, (downsample, downsample, 1))
level_tiles_shape = mosaic_shape(
level_shape,
self.tile_size,
)
level_array = self.zarr.zeros(
name=level,
shape=level_shape,
chunks=(*self.tile_shape, self.shape[-1]),
dtype=self.dtype,
compressor=self.compressor,
)
level_tiles_index = np.ndindex(level_tiles_shape)
level_read_tile_size = np.multiply(self.tile_size, inter_level_downsample)
# Write tiles to the level by copying from the previous level
for ji in self.level_progress(level_tiles_index):
read_slices = tile_slices(ji, level_read_tile_size)
tile = previous_level[read_slices]
down_tile = downsample_tile(
tile, inter_level_downsample, method=downsample_method
)
write_slices = tile_slices(ji, self.tile_size)
level_array[write_slices] = down_tile
previous_level = level_array
previous_downsample = downsample
[docs] def transcode_from_reader(
self,
reader: Union[TIFFReader, DICOMWSIReader],
downsample_method: Optional[str] = None,
) -> None:
"""Losslessly transform into a new format from a supported Reader.
Repackages tiles from the Reader to a zarr. Currently only
supports transcoding from:
- JPEG compressed SVS (:class:`wsic.readers.TIFFReader`)
- J2K compressed SVS (:class:`wsic.readers.TIFFReader`)
- JPEG compressed OME-TIFF (:class:`wsic.readers.TIFFReader`)
- JPEG compressed DICOM WSI (:class:`wsic.readers.DICOMWSIReader`)
- JPEG compressed NDPI (Hamamatsu)
Currently only outputs a single resolution level (level 0).
It may also be possible to transcode the tiles themselves (e.g.
JPEG to JPEG XL) or perform simple geometric transforms (flip,
rotate, etc). However, this is not yet implemented. Currently,
they are simply copied into a new structure.
Args:
reader (Reader):
Reader object.
downsample_method (str, optional):
Downsample method to use for new reduced resolutions.
Defaults to None.
Valid downsample methods are: "cv2", "scipy", "np", None.
"""
transcode_supported = self._can_transcode_from_reader(reader)
if not transcode_supported:
raise ValueError(
"Currently only SVS, NDPI, OME-TIFF, and WSI DICOM "
"(with JPEG, JPEG2000, or WebP compression) "
"are supported for transcoding."
)
register_codecs()
codec = self.get_transcode_codec(reader)
self.shape = reader.shape
self.dtype = reader.dtype
self.zarr = zarr.open_group(zarr.NestedDirectoryStore(self.path))
self.zarr.create_dataset(
name="0",
shape=self.shape,
dtype=self.dtype,
chunks=(*self.tile_shape, reader.shape[-1]),
compressor=codec,
)
# Copy tiles
for index in self.transcode_progress(
np.ndindex(reader.mosaic_shape),
total=np.prod(reader.mosaic_shape),
):
tile_path = self.path / "0" / str(index[0]) / str(index[1]) / "0"
tile_path.parent.mkdir(parents=True, exist_ok=True)
tile_bytes = reader.get_tile(index, decode=False)
with open(tile_path, "wb") as file_handle:
file_handle.write(tile_bytes)
self._build_pyramid(downsample_method)
self._write_ome_metadata(reader.microns_per_pixel or self.microns_per_pixel)
def _can_transcode_from_reader(self, reader: Reader) -> bool:
"""Determine if a reader supports from to the current writer.
Args:
reader (Reader):
Reader object.
Returns:
bool:
Whether the reader supports being transcoded from.
"""
# 1. A valid get_tile(decode=False)
try:
reader.get_tile((0, 0), decode=False)
except (NotImplementedError, AttributeError) as error:
raise ValueError(
"Reader must have a get_tile method which can return encoded tiles"
" (decoded=False)."
) from error
# 2. Compatible tile sizes
if self.tile_size != reader.tile_shape[:2][::-1]:
raise ValueError(
"Tile size must match the reader tile size for transcoding."
)
# 3. Matching data types
if self.dtype != reader.dtype:
raise ValueError("Dtype must match the reader dtype for transcoding.")
# 4. Compatible compression
supported_compression = (
Codec.from_string(reader.codec) in (Codec.JPEG, Codec.JPEG2000, Codec.WEBP),
)
# 5. Supported Reader (WSIDICOM or a TIFF with supported format)
return all(
[
isinstance(reader, (TIFFReader, DICOMWSIReader)),
supported_compression,
]
)
[docs] @staticmethod
def get_transcode_codec(reader: TIFFReader) -> Any:
"""Get the codec to use for transcoding.
Args:
reader (TiffReader):
Reader object.
Returns:
numcodecs.Codec:
Codec to use for transcoding.
"""
from imagecodecs.numcodecs import Jpeg, Jpeg2k, Webp
# Try to get the compression level from the reader if known
try:
level = reader.compression_level
except AttributeError:
level = None
# Create the codec object
if reader.codec == Codec.JPEG:
return Jpeg(
tables=reader.jpeg_tables,
colorspace_jpeg=reader.color_space,
colorspace_data=ColorSpace.RGB,
level=reader.compression_level,
)
if reader.codec == Codec.JPEG2000:
return Jpeg2k(codecformat="JP2", colorspace=reader.color_space, level=level)
if reader.codec == Codec.WEBP:
return Webp(level=level)
# Out of options
raise ValueError(
f"Codec {reader.codec} is not supported for transcoding. "
"Currently only JPEG, J2K (JPEG-2000), and WebP compression "
" are supported for transcoding."
)
[docs] def close(self) -> None:
"""Close the writer."""
if hasattr(self.zarr.store, "close"):
self.zarr.store.close()
[docs]class DICOMWSIWriter(Writer):
"""Writer for DICOM WSI images using wsidicom.
Notes:
- Supports JPEG and JPEG2000 compression.
- DICOM Whole Slide Imaging: https://dicom.nema.org/Dicom/DICOMWSI/
"""
def __init__(
self,
path: PathLike,
shape: Tuple[int, int],
tile_size: Tuple[int, int] = (256, 256),
dtype: np.dtype = np.uint8,
color_space: Optional[ColorSpace] = None,
codec: Optional[Codec] = None,
compression_level: int = 0,
microns_per_pixel: Tuple[float, float] = None,
pyramid_downsamples: Optional[List[int]] = None,
overwrite: bool = False,
verbose: bool = False,
**kwargs,
) -> None:
if color_space != "rgb":
warn_unused(color_space)
warn_unused(codec)
warn_unused(compression_level, ignore_falsey=True)
warn_unused(microns_per_pixel)
warn_unused(pyramid_downsamples, ignore_falsey=True)
for key, value in kwargs.items():
warn_unused(key, value)
super().__init__(
path=path,
shape=shape,
tile_size=tile_size,
dtype=dtype,
color_space=color_space,
codec=codec,
compression_level=compression_level,
microns_per_pixel=microns_per_pixel,
pyramid_downsamples=pyramid_downsamples,
overwrite=overwrite,
verbose=verbose,
)
[docs] def copy_from_reader(
self,
reader: Reader,
num_workers: int = 2,
read_tile_size: Tuple[int, int] = None,
timeout: float = 10,
downsample_method: Optional[str] = None,
) -> None:
from pydicom import FileDataset, dcmwrite
from wsic.dicom import append_frames, create_vl_wsi_dataset
warn_unused(downsample_method, ignore_falsey=True)
width = self.shape[1]
height = self.shape[0]
photometric_interpretation = (
ColorSpace.YCBCR.to_dicom_photometric_interpretation((4, 2, 2))
)
mpp: Tuple[float, float] = (
self.microns_per_pixel or reader.microns_per_pixel or (1.0, 1.0)
)
self.validate_write_args(microns_per_pixel=mpp)
meta, dataset = create_vl_wsi_dataset(
size=(width, height),
tile_size=self.tile_size,
microns_per_pixel=mpp,
photometric_interpretation=photometric_interpretation,
)
file_dataset = FileDataset(
str(self.path),
dataset=dataset,
preamble=b"\0" * 128,
file_meta=meta,
is_implicit_VR=False,
is_little_endian=True,
)
dcmwrite(
dataset=file_dataset,
filename=file_dataset.filename,
write_like_original=False,
)
with ZarrIntermediate(
None, reader.shape, zero_after_read=False
) as intermediate:
reader_tile_iterator = self.reader_tile_iterator(
reader=reader,
num_workers=num_workers,
intermediate=intermediate,
read_tile_size=read_tile_size or self.tile_size,
timeout=timeout,
)
def jpeg_generator(tile_iterator) -> Generator[bytes, None, None]:
"""Encodes arrays to JPEG bytes."""
import imagecodecs
for tile in tile_iterator:
yield imagecodecs.jpeg_encode(
tile,
level=self.compression_level,
colorspace=self.color_space,
outcolorspace=ColorSpace.YCBCR,
)
tile_iterator = iter(
self.level_progress(
jpeg_generator(reader_tile_iterator),
total=len(reader_tile_iterator),
)
)
append_frames(self.path, tile_iterator, len(reader_tile_iterator))
[docs] def transcode_from_reader(
self,
reader: Union[TIFFReader, DICOMWSIReader],
downsample_method: Optional[str] = None,
) -> None:
from pydicom import FileDataset, dcmwrite
from wsic.dicom import append_frames, create_vl_wsi_dataset
warn_unused(downsample_method, ignore_falsey=True)
width = self.shape[1]
height = self.shape[0]
photometric_interpretation = (
reader.color_space.to_dicom_photometric_interpretation(
(4, 2, 2) if reader.color_space == ColorSpace.YCBCR else None
)
)
mpp: Tuple[float, float] = (
self.microns_per_pixel or reader.microns_per_pixel or (1.0, 1.0)
)
self.validate_write_args(microns_per_pixel=mpp)
if reader.codec != Codec.JPEG:
raise ValueError(
f"Currenly only JPEG compression is supported. " f"Got {reader.codec}."
)
meta, dataset = create_vl_wsi_dataset(
size=(width, height),
tile_size=self.tile_size,
microns_per_pixel=mpp,
photometric_interpretation=photometric_interpretation,
)
file_dataset = FileDataset(
str(self.path),
dataset=dataset,
preamble=b"\0" * 128,
file_meta=meta,
is_implicit_VR=False,
is_little_endian=True,
)
dcmwrite(
dataset=file_dataset,
filename=file_dataset.filename,
write_like_original=False,
)
tile_count = np.prod(reader.mosaic_shape)
def tile_generator() -> Generator[bytes, None, None]:
"""Yields tiles as bytes from the reader."""
for xy in self.transcode_progress(
np.ndindex(reader.mosaic_shape), total=tile_count
):
yield reader.get_tile(xy, decode=False)
append_frames(self.path, tile_generator(), tile_count)
[docs] @staticmethod
def validate_write_args(
microns_per_pixel: Optional[Tuple[float, float]],
):
from wsic.validation import check_mpp
if microns_per_pixel is None:
warnings.warn(
"Resolution (PixelSpacing) is None but it is a required "
"(Type 1) attribute for DICOM VL Whole Slide Image files. "
"A default of (1.0, 1.0) microns-per-mm or "
"(1000, 1000) microns-per-pixel will be used.",
stacklevel=3,
)
else:
for r in microns_per_pixel:
check_mpp(r)
def _cv2_downsample(image: np.ndarray, factor: int) -> np.ndarray:
"""Resample an image using OpenCV.
Args:
image (np.ndarray):
The image to resample.
factor (int):
The downsampling factor.
Returns:
np.ndarray:
The resampled image.
"""
import cv2
return cv2.resize(
image,
(image.shape[1] // factor, image.shape[0] // factor),
interpolation=cv2.INTER_AREA,
)
def _pil_downsample(image: np.ndarray, factor: int) -> np.ndarray:
"""Resample an image using PIL.
Args:
image (np.ndarray):
The image to resample.
factor (int):
The downsampling factor.
Returns:
np.ndarray:
The resampled image.
"""
import PIL.Image
return PIL.Image.fromarray(image).resize(
(image.shape[1] // factor, image.shape[0] // factor),
resample=PIL.Image.Resampling.Box,
)
def _scipy_downsample(image: np.ndarray, factor: int) -> np.ndarray:
"""Resample an image using SciPy.
Args:
image (np.ndarray):
The image to resample.
factor (int):
The downsampling factor.
Returns:
np.ndarray:
The resampled image.
"""
from scipy import ndimage
return ndimage.zoom(image, (1 / factor, 1 / factor, 1), order=1)
def _np_downsample(image: np.ndarray, factor: int) -> np.ndarray:
"""Resample an image using NumPy.
Args:
image (np.ndarray):
The image to resample.
factor (int):
The downsampling factor.
Returns:
np.ndarray:
The resampled image.
"""
return mean_pool(image.astype(float), factor).clip(0, 255).astype(np.uint8)
[docs]def downsample_tile(
image: np.ndarray, factor: int, method: Optional[str] = None
) -> np.array:
"""Downsample an image by a factor.
Args:
image (np.ndarray):
The image to downsample.
factor (int):
The downsampling factor.
method (str):
The downsampling method (library) to use.
Defaults to None, which tries cv2, then scipy, and falls
back to numpy.
Valid options are: "cv2", "pillow", "scipy", "np", None.
"""
methods = {
"cv2": _cv2_downsample,
"pillow": _pil_downsample,
"scipy": _scipy_downsample,
"np": _np_downsample,
}
if method is not None and method not in methods:
raise ValueError(f"Invalid method: {method}")
if method in methods:
return methods[method](image, factor)
for method_name, func in methods.items():
try:
return func(image, factor)
except ImportError:
warnings.warn(
f"Failed to import library for {method_name} for downsampling. "
"It may not be installed.",
stacklevel=2,
)
allowed_methods = {method} if method else set(methods)
raise ImportError(
f"Failed to use any allowed downsampling method: {allowed_methods}."
)
[docs]def get_level_tile(
yx: Tuple[int, int],
tile_size: Tuple[int, int],
downsample: int,
read_intermediate_path: PathLike,
downsample_method: Optional[str] = None,
) -> np.ndarray:
"""Generate tiles for a downsampled level.
Args:
yx (Tuple[int, int]):
The tile coordinates.
tile_size (Tuple[int, int]):
The tile size.
downsample (int):
The downsampling factor.
read_intermediate_path (PathLike):
The path to the intermediate file (zarr).
downsample_method (str):
The downsampling method (library) to use.
"""
y, x = yx
w, h = tile_size
tile_index = (
slice(y * h * downsample, (y + 1) * h * downsample),
slice(x * w * downsample, (x + 1) * w * downsample),
)
reader = zarr.open(read_intermediate_path, mode="r")
tile = reader[tile_index]
return downsample_tile(tile, downsample, method=downsample_method)