import collections
import logging
from typing import Optional, Union, Tuple, List, Any, Iterable, Dict
import numpy as np
logger = logging.getLogger("celldetective")
[docs]
def consume(iterator: Iterable[Any]) -> None:
"""
adapted from https://github.com/CSBDeep/CSBDeep/blob/main/csbdeep/utils/utils.py
Parameters
----------
iterator : iterable
The iterator to consume.
"""
collections.deque(iterator, maxlen=0)
[docs]
def axes_check_and_normalize(
axes: str,
length: Optional[int] = None,
disallowed: Optional[str] = None,
return_allowed: bool = False,
) -> Union[str, Tuple[str, str]]:
"""
adapted from https://github.com/CSBDeep/CSBDeep/blob/main/csbdeep/utils/utils.py
S(ample), T(ime), C(hannel), Z, Y, X
Parameters
----------
axes : str
The axes string to check and normalize.
length : int, optional
The expected length of the axes string. Default is None.
disallowed : str, optional
A string of disallowed axes. Default is None.
return_allowed : bool, optional
Whether to return the allowed axes string. Default is False.
"""
allowed = "STCZYX"
if axes is None:
raise ValueError("axis cannot be None.")
axes = str(axes).upper()
consume(a in allowed for a in axes)
disallowed is None or consume(a not in disallowed for a in axes)
consume(axes.count(a) == 1 for a in axes)
length is None or len(axes) == length
return (axes, allowed) if return_allowed else axes
[docs]
def axes_dict(axes: str) -> Dict[str, Optional[int]]:
"""
adapted from https://github.com/CSBDeep/CSBDeep/blob/main/csbdeep/utils/utils.py
from axes string to dict
Parameters
----------
axes : str
The axes string to convert to a dictionary.
"""
axes, allowed = axes_check_and_normalize(axes, return_allowed=True)
return {a: None if axes.find(a) == -1 else axes.find(a) for a in allowed}
# return collections.namedtuple('Axes',list(allowed))(*[None if axes.find(a) == -1 else axes.find(a) for a in allowed ])
[docs]
def move_image_axes(
x: np.ndarray, fr: str, to: str, adjust_singletons: bool = False
) -> np.ndarray:
"""
adapted from https://github.com/CSBDeep/CSBDeep/blob/main/csbdeep/utils/utils.py
x: ndarray
fr,to: axes string (see `axes_dict`)
Parameters
----------
x : ndarray
Input image.
fr : str
Source axes.
to : str
Target axes.
adjust_singletons : bool, optional
Whether to adjust singleton axes. Default is False.
"""
fr = axes_check_and_normalize(fr, length=x.ndim)
to = axes_check_and_normalize(to)
fr_initial = fr
x_shape_initial = x.shape
adjust_singletons = bool(adjust_singletons)
if adjust_singletons:
# remove axes not present in 'to'
slices = [slice(None) for _ in x.shape]
for i, a in enumerate(fr):
if (a not in to) and (x.shape[i] == 1):
# remove singleton axis
slices[i] = 0
fr = fr.replace(a, "")
x = x[tuple(slices)]
# add dummy axes present in 'to'
for i, a in enumerate(to):
if a not in fr:
# add singleton axis
x = np.expand_dims(x, -1)
fr += a
if set(fr) != set(to):
_adjusted = (
"(adjusted to %s and %s) " % (x.shape, fr) if adjust_singletons else ""
)
raise ValueError(
"image with shape %s and axes %s %snot compatible with target axes %s."
% (x_shape_initial, fr_initial, _adjusted, to)
)
ax_from, ax_to = axes_dict(fr), axes_dict(to)
if fr == to:
return x
return np.moveaxis(x, [ax_from[a] for a in fr], [ax_to[a] for a in fr])
[docs]
def estimate_unreliable_edge(
activation_protocol: Optional[List[List[Any]]] = [["gauss", 2], ["std", 4]]
) -> Optional[int]:
"""
Safely estimate the distance to the edge of an image in which the filtered image values can be artefactual.
Parameters
----------
activation_protocol : list of list, optional
A list of lists, where each sublist contains a string naming the filter function, followed by its arguments (usually a kernel size).
Default is [['gauss', 2], ['std', 4]].
Returns
-------
int or None
The sum of the kernel sizes in the activation protocol if the protocol
is not empty. Returns None if the activation protocol is empty.
Notes
-----
This function assumes that the second element of each sublist in the
activation protocol is a kernel size.
Examples
--------
>>> estimate_unreliable_edge([['gauss', 2], ['std', 4]])
6
>>> estimate_unreliable_edge([])
None
"""
if activation_protocol is None or activation_protocol == []:
return None
else:
edge = 0
for fct in activation_protocol:
if isinstance(fct[1], (int, np.int_)) and not fct[0] == "invert":
edge += fct[1]
return edge
[docs]
def unpad(img: np.ndarray, pad: int) -> np.ndarray:
"""
Remove padding from an image.
This function removes the specified amount of padding from the borders
of an image. The padding is assumed to be the same on all sides.
Parameters
----------
img : ndarray
The input image from which the padding will be removed.
pad : int
The amount of padding to remove from each side of the image. 0 or None leaves the
image unchanged.
Returns
-------
ndarray
The image with the padding removed.
Raises
------
ValueError
If `pad` is greater than or equal to half of the smallest dimension
of `img`.
See Also
--------
numpy.pad : Pads an array.
Notes
-----
This function assumes that the input image is a 2D array.
Examples
--------
>>> import numpy as np
>>> img = np.array([[0, 0, 0, 0, 0],
... [0, 1, 1, 1, 0],
... [0, 1, 1, 1, 0],
... [0, 1, 1, 1, 0],
... [0, 0, 0, 0, 0]])
>>> unpad(img, 1)
array([[1, 1, 1],
[1, 1, 1],
[1, 1, 1]])
"""
# img[0:-0] would be empty, and -None raises.
if not pad:
return img
return img[pad:-pad, pad:-pad]
[docs]
def mask_edges(binary_mask: np.ndarray, border_size: int) -> np.ndarray:
"""
Mask the edges of a binary mask.
This function sets the edges of a binary mask to False, effectively
masking out a border of the specified size.
Parameters
----------
binary_mask : ndarray
A 2D binary mask array where the edges will be masked.
border_size : int
The size of the border to mask (set to False) on all sides.
Returns
-------
ndarray
The binary mask with the edges masked out.
Raises
------
ValueError
If `border_size` is greater than or equal to half of the smallest
dimension of `binary_mask`.
Notes
-----
This function assumes that the input `binary_mask` is a 2D array. The
input mask is converted to a boolean array before masking the edges.
Examples
--------
>>> import numpy as np
>>> binary_mask = np.array([[1, 1, 1, 1, 1],
... [1, 1, 1, 1, 1],
... [1, 1, 1, 1, 1],
... [1, 1, 1, 1, 1],
... [1, 1, 1, 1, 1]])
>>> mask_edges(binary_mask, 1)
array([[False, False, False, False, False],
[False, True, True, True, False],
[False, True, True, True, False],
[False, True, True, True, False],
[False, False, False, False, False]])
"""
binary_mask = binary_mask.astype(bool)
binary_mask[:border_size, :] = False
binary_mask[(binary_mask.shape[0] - border_size) :, :] = False
binary_mask[:, :border_size] = False
binary_mask[:, (binary_mask.shape[1] - border_size) :] = False
return binary_mask
def _estimate_scale_factor(
spatial_calibration: Optional[float], required_spatial_calibration: Optional[float]
) -> Optional[float]:
"""
Estimates the scale factor needed to adjust spatial calibration to a required value.
This function calculates the scale factor by which spatial dimensions (e.g., in microscopy images)
should be adjusted to align with a specified calibration standard. This is particularly useful when
preparing data for analysis with models trained on data of a specific spatial calibration.
Parameters
----------
spatial_calibration : float or None
The current spatial calibration factor of the data, expressed as units per pixel (e.g., micrometers per pixel).
If None, indicates that the current spatial calibration is unknown or unspecified.
required_spatial_calibration : float or None
The spatial calibration factor required for compatibility with the model or analysis standard, expressed
in the same units as `spatial_calibration`. If None, indicates no adjustment is required.
Returns
-------
float or None
The scale factor by which the current data should be rescaled to match the required spatial calibration,
or None if no scaling is necessary or if insufficient information is provided.
Notes
-----
- A scale factor close to 1 (within a tolerance defined by `epsilon`) indicates that no significant rescaling
is needed, and the function returns None.
- The function issues a warning if a significant rescaling is necessary, indicating the scale factor to be applied.
Examples
--------
>>> scale_factor = _estimate_scale_factor(spatial_calibration=0.5, required_spatial_calibration=0.25)
# Each frame will be rescaled by a factor 2.0 to match with the model training data...
>>> scale_factor = _estimate_scale_factor(spatial_calibration=None, required_spatial_calibration=0.25)
# Returns None due to insufficient information about current spatial calibration.
"""
if (required_spatial_calibration is not None) * (spatial_calibration is not None):
scale = spatial_calibration / required_spatial_calibration
else:
scale = None
epsilon = 0.05
if scale is not None:
if not np.all([scale >= (1 - epsilon), scale <= (1 + epsilon)]):
logger.info(
f"Each frame will be rescaled by a factor {scale} to match with the model training data..."
)
else:
scale = None
return scale
#: Fraction by which a frame may differ from the model's sampling before it is
#: worth resampling at all. Shared by :func:`_estimate_scale_factor` and
#: :func:`_combined_scale_factor` so "close enough to 1" means one thing.
_SCALE_EPSILON = 0.05
def _combined_scale_factor(
spatial_calibration: Optional[float],
required_spatial_calibration: Optional[float],
cell_size: Optional[float] = None,
target_cell_size: Optional[float] = None,
) -> Optional[float]:
"""
The single factor a frame must be resized by before it reaches a network.
Two corrections decide it: the pixel sizes have to match (`spatial_calibration`
against `required_spatial_calibration`), and the objects have to arrive at the
size the network was trained on (`cell_size` against `target_cell_size`). They
multiply.
Applying :func:`_estimate_scale_factor` first and multiplying afterwards is
what this replaces, and it was wrong in one corner: that function collapses a
ratio within 5% of 1 to None, meaning "not worth resampling for". That
judgement is only true of the calibration correction *on its own*. Once a
cell-size correction is also being applied the frame is being resampled
regardless, and dropping the calibration factor silently biases the result by
up to 5% -- entirely depending on how close the model's own calibration
happens to sit to the images'. The two ratios are combined here first, and the
"close enough to 1" test is applied once, to the number actually used.
Parameters
----------
spatial_calibration : float or None
The images' pixel size. None when unknown, which leaves the
calibration correction out.
required_spatial_calibration : float or None
The pixel size the model was trained at.
cell_size : float or None, optional
The object size the model was trained on, in microns.
target_cell_size : float or None, optional
The object size in these images, in microns. Both sizes are needed
for the correction; either one missing leaves it out.
Returns
-------
float or None
The factor to resize by, or None when the frame is already close
enough to what the model expects to leave alone.
Examples
--------
>>> _combined_scale_factor(0.5, 0.25)
2.0
>>> _combined_scale_factor(0.2, 0.2075, 6.225, 20.0) is None
False
"""
scale = 1.0
corrected = False
if spatial_calibration is not None and required_spatial_calibration is not None:
scale *= spatial_calibration / required_spatial_calibration
corrected = True
if cell_size is not None and target_cell_size is not None:
scale *= cell_size / target_cell_size
corrected = True
if not corrected:
return None
if (1 - _SCALE_EPSILON) <= scale <= (1 + _SCALE_EPSILON):
return None
logger.info(
f"Each frame will be rescaled by a factor {scale} to match with the model training data..."
)
return scale
[docs]
def threshold_image(
img: np.ndarray,
min_threshold: float,
max_threshold: float,
foreground_value: float = 255.0,
fill_holes: bool = True,
edge_exclusion: Optional[int] = None,
) -> np.ndarray:
"""
Threshold the input image to create a binary mask.
Parameters
----------
img : ndarray
The input image to be thresholded.
min_threshold : float
The minimum threshold value.
max_threshold : float
The maximum threshold value.
foreground_value : float, optional
The value assigned to foreground pixels in the binary mask. Default is 255.
fill_holes : bool, optional
Whether to fill holes in the binary mask. If True, the binary mask will be processed to fill any holes.
If False, the binary mask will not be modified. Default is True.
edge_exclusion : int, optional
If provided, border pixels of this width will be masked (set to 0). Default is None.
Returns
-------
ndarray
The binary mask after thresholding.
Notes
-----
This function applies a threshold to the input image to create a binary mask. Pixels with values within the specified
threshold range are considered as foreground and assigned the `foreground_value`, while pixels outside the range are
considered as background and assigned 0. If `fill_holes` is True, the binary mask will be processed to fill any holes
using morphological operations.
Examples
--------
>>> image = np.random.rand(256, 256)
>>> binary_mask = threshold_image(image, 0.2, 0.8, foreground_value=1., fill_holes=True)
"""
from scipy import ndimage as ndi
binary = np.zeros_like(img).astype(bool)
binary[img == img] = (
(img[img == img] >= min_threshold)
* (img[img == img] <= max_threshold)
* foreground_value
)
if isinstance(edge_exclusion, (int, np.int_)):
binary = mask_edges(binary, edge_exclusion)
if fill_holes:
binary = ndi.binary_fill_holes(binary.astype(int))
return binary
[docs]
def pad_to_patch_size(
x: np.ndarray, y: Optional[np.ndarray] = None, patch_h: int = 256, patch_w: int = 256
) -> Tuple[np.ndarray, Optional[np.ndarray], bool, Tuple[int, int]]:
"""
Pad image x and optionally label y to match patch_h and patch_w if they are smaller.
Parameters
----------
x : ndarray
Input image.
y : ndarray, optional
Input label mask.
patch_h : int
Target patch height.
patch_w : int
Target patch width.
Returns
-------
x_padded : ndarray
Padded image.
y_padded : ndarray or None
Padded label mask (if y was provided).
padded : bool
True if padding was applied, False otherwise.
offsets : tuple of int
``(pad_h_top, pad_w_left)``, the top/left offset of the original image
within the padded array. ``(0, 0)`` when no padding was applied. Use
these to crop a result back to the original size instead of recomputing
the centering math at the call site.
"""
h, w = x.shape[:2]
if h < patch_h or w < patch_w:
pad_h = max(0, patch_h - h)
pad_w = max(0, patch_w - w)
pad_h_top = pad_h // 2
pad_h_bottom = pad_h - pad_h_top
pad_w_left = pad_w // 2
pad_w_right = pad_w - pad_w_left
pad_width_x = [(pad_h_top, pad_h_bottom), (pad_w_left, pad_w_right)]
if x.ndim > 2:
pad_width_x += [(0, 0)] * (x.ndim - 2)
x = np.pad(
x,
pad_width_x,
mode="constant",
constant_values=0.0,
)
if y is not None:
y = np.pad(
y,
((pad_h_top, pad_h_bottom), (pad_w_left, pad_w_right)),
mode="constant",
constant_values=0,
)
return x, y, True, (pad_h_top, pad_w_left)
return x, y, False, (0, 0)
[docs]
def pad_dataset_to_patch_size(
X: List[np.ndarray], Y: List[np.ndarray], patch_h: int, patch_w: int
) -> Tuple[List[np.ndarray], List[np.ndarray], int]:
"""
Pad a dataset of images X and label masks Y to match patch_h and patch_w if they are smaller.
Parameters
----------
X : List[ndarray]
List of input images.
Y : List[ndarray]
List of label masks.
patch_h : int
Target patch height.
patch_w : int
Target patch width.
Returns
-------
X_padded : List[ndarray]
Padded images.
Y_padded : List[ndarray]
Padded label masks.
padded_count : int
Number of images/labels padded.
"""
X_padded, Y_padded = [], []
padded_count = 0
for x, y in zip(X, Y):
xp, yp, padded, _ = pad_to_patch_size(x, y, patch_h, patch_w)
X_padded.append(xp)
Y_padded.append(yp)
if padded:
padded_count += 1
return X_padded, Y_padded, padded_count