"""Resolve, filter and cut the images used to train a classifier.
The crop source has two main values, named as in the other spaCR panels:
LOAD IMAGES and STREAM IMAGES. Training previously used its own terms
(``pre_generated`` / ``on_demand``) for the same choice, so the annotation
and training panels presented one setting under two sets of names.
``png`` — load pre-generated images (default)
Read crops previously written by the measurement workflow.
``path_string`` filters paths by substring and ``file_type`` filters by
image extension. This source performs no cropping.
``merged`` — stream images
Extract crops during training from ``merged/*.npy`` arrays.
``extract_channels`` selects the intensity planes and ``object_array``
selects the labelled mask plane that defines each object's extent. When
``coordinate_columns`` are configured, database coordinates instead
define fixed bounding-box crops.
``generate`` — generate images, then load them
Materialize a complete crop set on disk before training, then read it
through the same file-backed path as ``png``. This is a preprocessing
action rather than a streaming source.
**The stored values did not change.** ``png`` and ``merged`` are what
``spacr.crops.resolve_crop_source`` has always read, so a settings file
written under either vocabulary means what it always meant:
``pre_generated``, ``load_images`` and ``auto`` all arrive as LOAD IMAGES,
``on_demand`` and ``stream_images`` as STREAM IMAGES. :data:`CROP_SOURCE_ALIASES`
is that migration, in one place, so a value written by a renamed panel is
still accepted.
**Why streaming exists.** Pre-cutting every crop writes a copy of the dataset
to disk before a single epoch runs, and every change of crop size or channel
selection writes another. Cutting as training runs costs a slice per object
and no disk at all.
**Bounding box versus object.** A bounding box is the smallest rectangle
containing the object; an object crop masks everything outside it away. Both
are useful -- the background around a cell is sometimes signal and sometimes
contamination -- so it is a setting. Database-sourced objects can only ever be
bounding boxes: a coordinate has no outline to mask against.
"""
from __future__ import annotations
import logging
import os
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
import numpy as np
LOG = logging.getLogger("spacr.crop_source")
#: What a crop source can be, in the order the settings panel offers them.
#: LOAD IMAGES first, because it is the default.
CROP_SOURCES: Tuple[str, ...] = ("png", "merged", "generate")
#: Every spelling a settings file has ever carried, and the source it names.
#:
#: ACCEPTED, NOT REFUSED. Three panels and two renames have written this one
#: setting, and every value any of them wrote is still on somebody's disk:
#: ``pre_generated``/``on_demand`` from the training panel, ``load_images``/
#: ``stream_images`` from its rename, ``png``/``merged`` from the viewers, and
#: ``auto`` from before the question was asked out loud. They resolve here, so
#: this module accepts what the panels produce instead of raising on it --
#: which is what it did when the training panel was renamed and left this
#: reader behind: every computer-vision run refused at the door with
#: "crop_source='load_images' is not one of ...".
#:
#: ``auto`` means LOAD IMAGES rather than "whichever folder exists". It stays
#: readable because settings files hold it; it is not an answer a user is
#: offered, because "what is available here" is not an answer to which mode
#: they want.
CROP_SOURCE_ALIASES: Dict[str, str] = {
"auto": "png",
"png": "png",
"load_images": "png",
"pre_generated": "png",
"merged": "merged",
"stream": "merged",
"stream_images": "merged",
"on_demand": "merged",
"generate": "generate",
}
#: The choice as a panel shows it: the value stored, and the words shown.
#:
#: The same words the annotation and montage panels use, because it is the
#: same question. A panel that renders these renders LOAD IMAGES first.
CROP_SOURCE_OPTIONS: Tuple[Tuple[str, str], ...] = (
("png", "load images — crops already in data/"),
("merged", "stream images — cut from merged/"),
("generate", "generate crops — write a crop set, then load it"),
)
#: Image extensions a pre-generated crop may have. The setting is a FILTER on
#: the extension, which is what it always should have been -- it and
#: ``png_type`` were two names for a path filter, and one of them pretended to
#: name a file type.
IMAGE_FILE_TYPES: Tuple[str, ...] = (
"png", "tif", "tiff", "jpg", "jpeg", "bmp", "npy",
)
#: What a crop is cut as.
CROP_SHAPES: Tuple[str, ...] = ("bounding_box", "object")
#: Settings each source reads. Drives the greying -- a control the user can
#: edit that changes nothing is worse than one that is not there.
SOURCE_SETTINGS: Dict[str, Tuple[str, ...]] = {
"png": ("path_string", "file_type", "file_metadata", "tar_path"),
"merged": ("extract_channels", "object_array", "coordinate_columns",
"crop_shape", "image_size"),
"generate": ("extract_channels", "object_array", "crop_shape",
"image_size", "path_string", "file_type"),
}
[docs]
class CropSourceError(ValueError):
"""A crop source that cannot produce images, and why."""
[docs]
def resolve_source(settings: Mapping[str, Any]) -> str:
"""Return the canonical crop source (``png``, ``merged`` or ``generate``).
Every spelling in :data:`CROP_SOURCE_ALIASES` resolves, so a settings file
written by any released spaCR panel is accepted. An unset value means LOAD
IMAGES, the default in every panel.
:param settings: the settings mapping; its ``crop_source`` value is looked
up in :data:`CROP_SOURCE_ALIASES`, and an empty value means ``png``.
:raises CropSourceError: an unrecognised source. The source is not guessed,
because a wrong guess would train on different images without error.
"""
declared = str(settings.get("crop_source") or "").strip().lower()
if not declared:
return "png"
resolved = CROP_SOURCE_ALIASES.get(declared)
if resolved is None:
raise CropSourceError(
f"crop_source={settings.get('crop_source')!r} is not one of "
f"{list(CROP_SOURCES)} (load images, stream images, or generate "
f"a crop set); accepted spellings are "
f"{sorted(name for name in CROP_SOURCE_ALIASES if name)}")
return resolved
[docs]
def inapplicable_settings(source: str) -> Tuple[str, ...]:
"""Return the settings used only by the other crop sources, for greying out.
Greyed, never removed (INVARIANTS 6): a key absent from the dict makes the
pipeline fall back to its own default, which can differ from the value the
module needs, without a warning.
Any spelling :data:`CROP_SOURCE_ALIASES` knows is accepted, because what a
panel has in hand is the value stored in the settings file, not the name
this module resolved it to.
:param source: the crop source whose settings stay active; any spelling in
:data:`CROP_SOURCE_ALIASES`, matched case-insensitively. An unknown
source raises :class:`CropSourceError`.
"""
key = CROP_SOURCE_ALIASES.get(str(source).strip().lower(), "")
if key not in SOURCE_SETTINGS:
raise CropSourceError(
f"{source!r} is not one of {list(CROP_SOURCES)}")
mine = set(SOURCE_SETTINGS[key])
return tuple(dict.fromkeys(
s for name, keys in SOURCE_SETTINGS.items() if name != key
for s in keys if s not in mine))
[docs]
def normalise_extension(file_type: Any) -> str:
"""The extension ``file_type`` names, without its dot and lower-cased.
The setting previously held path substrings such as ``'cell_png'``,
duplicating ``png_type``. It now names only an extension: ``'.TIF'`` and
``'tif'`` both normalise to ``'tif'``, and ``'tiff'`` to ``'tiff'``.
:param file_type: an extension such as ``'.TIF'``, ``'tif'`` or
``'cell_png'`` (only the part after the last underscore is kept); empty
or None gives ``''``.
:raises CropSourceError: an extension spaCR cannot read, named alongside
the ones it can.
"""
text = str(file_type or "").strip().lower().lstrip(".")
if not text:
return ""
if "_" in text:
text = text.rsplit("_", 1)[-1]
if text not in IMAGE_FILE_TYPES:
raise CropSourceError(
f"{file_type!r} is not an image type spaCR reads; choose from "
f"{', '.join(IMAGE_FILE_TYPES)}")
return text
[docs]
def matches_path(path: str, *, path_string: str = "",
file_type: Any = "") -> bool:
"""Whether one crop belongs in the dataset.
Two independent tests are applied: the object type, as a substring of the
path (e.g. ``cell_png``), and the file format, as the extension. Keeping
them separate allows selections such as "every nucleus crop, any format"
or "every TIFF, any object".
:param path: the crop's file path, tested as a string for the
``path_string`` substring and for its extension.
"""
text = str(path)
if path_string and str(path_string) not in text:
return False
extension = normalise_extension(file_type)
if extension:
actual = os.path.splitext(text)[1].lower().lstrip(".")
if actual != extension and not (extension == "tif" and actual == "tiff"):
return False
return True
[docs]
def select_crops(paths: Iterable[str], settings: Mapping[str, Any]
) -> List[str]:
"""The crops a settings dict selects, in the order given.
:param paths: candidate crop file paths.
:param settings: the settings mapping; ``path_string`` (or, failing that,
``png_type``) and ``file_type`` are passed to :func:`matches_path`.
"""
path_string = settings.get("path_string") or settings.get("png_type") or ""
return [p for p in paths
if matches_path(p, path_string=path_string,
file_type=settings.get("file_type"))]
def _as_indices(value, what: str) -> List[int]:
"""Normalize one plane selection to integer indices.
:param value: One integer index or an iterable of integer-like indices.
:param what: Setting name used to identify an invalid selection.
:returns: Selected plane indices as ordinary integers.
:raises CropSourceError: The selection is unset or cannot be converted.
"""
if value is None:
raise CropSourceError(f"{what} is not set")
if isinstance(value, (int, np.integer)):
return [int(value)]
try:
return [int(v) for v in value]
except (TypeError, ValueError) as exc:
raise CropSourceError(f"{what}={value!r} is not a list of planes") from exc
def _positive_size(value: Any) -> int:
"""Return an integer crop side, refusing an empty or inverted image."""
try:
side = int(value)
except (TypeError, ValueError, OverflowError) as exc:
raise CropSourceError(
f"size={value!r} is not a positive integer") from exc
if side <= 0:
raise CropSourceError(f"size must be positive, got {value!r}")
return side
[docs]
def object_bounds(mask: np.ndarray, label: int) -> Optional[Tuple[int, int, int, int]]:
"""``(row0, row1, col0, col1)`` of one labelled object, or None if absent.
Half-open on the far edge, like every other slice in Python, so the caller
can index with it directly rather than remembering to add one.
:param mask: 2-D label image holding the object.
:param label: the label value of the object to bound.
"""
rows, cols = np.nonzero(mask == label)
if rows.size == 0:
return None
return int(rows.min()), int(rows.max()) + 1, int(cols.min()), int(cols.max()) + 1
[docs]
def crop_object(array: np.ndarray, mask: np.ndarray, label: int, *,
channels: Sequence[int], shape: str = "bounding_box",
size: Optional[int] = None,
padding: int = 0) -> Optional[np.ndarray]:
"""Cut one object out of a merged array.
:param array: the merged stack, ``(H, W, planes)``.
:param mask: the plane holding this object's labels.
:param label: which object.
:param channels: which planes become image channels, in order.
:param shape: ``bounding_box`` keeps the rectangle's contents;
``object`` zeroes everything outside the object itself. The background
around a cell is sometimes signal and sometimes contamination, which
is why this is a choice rather than a default.
:param size: resize the result to ``size × size`` when given. It must
be positive; ``None`` preserves the object's natural bounding box.
:returns: ``(h, w, len(channels))``, or None if the object is not there.
:raises CropSourceError: a plane the array does not have, or an explicit
``size`` that is not positive.
"""
if shape not in CROP_SHAPES:
raise CropSourceError(
f"crop_shape={shape!r} is not one of {list(CROP_SHAPES)}")
planes = _as_indices(channels, "extract_channels")
if array.ndim != 3:
raise CropSourceError(
f"a merged array is (height, width, planes); this one is "
f"{array.shape}")
if max(planes) >= array.shape[2]:
raise CropSourceError(
f"extract_channels asks for plane {max(planes)} but the merged "
f"array has {array.shape[2]}")
target_size = _positive_size(size) if size is not None else None
bounds = object_bounds(mask, label)
if bounds is None:
return None
row0, row1, col0, col1 = bounds
if padding:
row0 = max(0, row0 - padding)
col0 = max(0, col0 - padding)
row1 = min(mask.shape[0], row1 + padding)
col1 = min(mask.shape[1], col1 + padding)
cut = array[row0:row1, col0:col1, :][:, :, planes].astype(np.float32)
if shape == "object":
inside = (mask[row0:row1, col0:col1] == label)
cut = cut * inside[:, :, None]
if target_size is not None:
cut = _resize(cut, target_size)
return cut
[docs]
def crop_at(array: np.ndarray, row: float, column: float, *,
channels: Sequence[int], size: int) -> Optional[np.ndarray]:
"""Cut a fixed box centred on a coordinate.
Used for objects taken from a database. Only a bounding box is possible,
because a coordinate has no outline to mask against.
:param array: the merged stack, ``(H, W, planes)``. Its rank is not
checked as :func:`crop_object` checks it, so a 2-D image raises
``IndexError`` from the slice rather than a ``CropSourceError``.
:param row: centre on the FIRST axis -- an image row, not ``y``. Floats
are rounded half-to-even, Python's rule, so a centroid of ``2.5``
lands on row 2 while ``3.5`` lands on row 4.
:param column: centre on the second axis; rounded the same way.
:param channels: which planes become image channels, in the order given,
so ``[2, 0]`` returns them swapped. A bare int counts as a one-plane
list. Not bounds-checked here as it is in :func:`crop_object`: a plane
the array does not have raises ``IndexError``, and a negative one
silently counts back from the last plane. An empty list yields a
zero-channel crop rather than None.
:param size: the positive side length returned. Odd sizes keep the rounded
coordinate at index ``size // 2``. A box crossing an array edge is
zero-padded rather than clipped or rescaled, preserving its centre and
pixel scale.
:returns: ``(size, size, len(channels))`` as float32, or None when the box
falls entirely off the array.
:raises CropSourceError: ``channels`` is None or is not planes, or ``size``
is not positive.
"""
planes = _as_indices(channels, "extract_channels")
side = _positive_size(size)
r, c = int(round(float(row))), int(round(float(column)))
row0, row1 = r - side // 2, r - side // 2 + side
col0, col1 = c - side // 2, c - side // 2 + side
source_row0, source_row1 = max(0, row0), min(array.shape[0], row1)
source_col0, source_col1 = max(0, col0), min(array.shape[1], col1)
if source_row1 <= source_row0 or source_col1 <= source_col0:
return None
cut = array[source_row0:source_row1, source_col0:source_col1, :][
:, :, planes].astype(np.float32)
return np.pad(
cut,
((source_row0 - row0, row1 - source_row1),
(source_col0 - col0, col1 - source_col1),
(0, 0)),
mode="constant")
def _resize(image: np.ndarray, size: int) -> np.ndarray:
"""Nearest-neighbour resize, so a crop needs no image library.
Nearest rather than interpolated: a crop is usually being made SMALLER,
where interpolation blurs the boundary the classifier is being asked to
look at, and this module has to stay importable on a cluster.
"""
height, width = image.shape[:2]
if height == size and width == size:
return image
rows = np.clip((np.arange(size) * height) // max(1, size), 0, height - 1)
cols = np.clip((np.arange(size) * width) // max(1, size), 0, width - 1)
return image[rows][:, cols]
[docs]
def mask_plane_for(object_array: str, settings: Mapping[str, Any]) -> int:
"""Which plane of the merged array holds ``object_array``'s masks.
Read from the ``*_mask_dim`` settings the mask step already writes, so the
two cannot disagree about which plane is which.
:param object_array: the object name (e.g. ``cell``, ``nucleus``);
lower-cased and used to build the ``<name>_mask_dim`` setting key.
:param settings: the settings mapping the ``<name>_mask_dim`` plane index
is read from.
:raises CropSourceError: the object has no mask plane, naming the setting
that would give it one.
"""
name = str(object_array or "").strip().lower()
key = f"{name}_mask_dim"
plane = settings.get(key)
if plane is None:
raise CropSourceError(
f"object_array={object_array!r} has no mask plane: {key} is not "
f"set, so nothing says which plane of merged holds its objects")
try:
return int(plane)
except (TypeError, ValueError) as exc:
raise CropSourceError(f"{key}={plane!r} is not a plane number") from exc
[docs]
def crops_from_merged(array: np.ndarray, settings: Mapping[str, Any], *,
labels: Optional[Sequence[int]] = None
) -> List[Tuple[int, np.ndarray]]:
"""Every object's crop from one merged array.
:param array: merged ``(height, width, planes)`` image whose configured
object-mask plane supplies the labels and whose other planes supply
the crop pixels.
:param settings: crop settings used to choose the object mask, extracted
channels, crop shape and optional fixed image size.
:param labels: only these objects; by default every label in the mask.
:returns: ``(label, image)`` pairs, skipping objects that are not present.
:raises CropSourceError: a setting that makes cutting impossible.
"""
plane = mask_plane_for(settings.get("object_array", "cell"), settings)
if plane >= array.shape[2]:
raise CropSourceError(
f"the mask plane is {plane} but the merged array has "
f"{array.shape[2]} plane(s)")
mask = array[:, :, plane]
if labels is None:
found = np.unique(mask)
labels = [int(v) for v in found if v != 0]
channels = settings.get("extract_channels")
size = settings.get("image_size")
shape = str(settings.get("crop_shape") or "bounding_box")
out: List[Tuple[int, np.ndarray]] = []
for label in labels:
cut = crop_object(array, mask, int(label), channels=channels,
shape=shape, size=size)
if cut is not None:
out.append((int(label), cut))
return out
[docs]
def stream_planes(settings: Mapping[str, Any]) -> List[int]:
"""Which planes of a merged array become image channels, by either name.
``channel_arrays`` is the current spelling and ``extract_channels`` the
older one; both are a list of plane indices and both are still written to
settings files, so both are read here. The current spelling wins when a
file carries both, because that is the one the panel is editing.
:param settings: the settings mapping; ``channel_arrays`` is read first,
then ``extract_channels``.
:raises CropSourceError: neither is set, naming the one to set.
"""
for name in ("channel_arrays", "extract_channels"):
if settings.get(name) is not None:
return _as_indices(settings.get(name), name)
raise CropSourceError(
"channel_arrays is not set, so nothing says which planes of the "
"merged array become the image's channels")
[docs]
def validate(settings: Mapping[str, Any]) -> str:
"""Check that a settings dict can produce crops and return its source.
Run before training, so a missing plane setting is reported before
dataset building starts; the error message names the setting to fix.
:param settings: the training settings mapping to check: the crop source,
plus ``file_type`` for ``png`` or the plane, crop-shape, coordinate and
mask settings for a streamed source.
:raises CropSourceError: with what to change.
"""
source = resolve_source(settings)
if source == "png":
if settings.get("file_type"):
normalise_extension(settings.get("file_type"))
return source
stream_planes(settings)
shape = str(settings.get("crop_shape") or "bounding_box")
if shape not in CROP_SHAPES:
raise CropSourceError(
f"crop_shape={shape!r} is not one of {list(CROP_SHAPES)}")
coordinates = settings.get("coordinate_columns")
if coordinates:
if shape != "bounding_box":
raise CropSourceError(
"objects taken from a database can only be cut as bounding "
"boxes: a coordinate has no outline to mask against")
if not settings.get("image_size"):
raise CropSourceError(
"image_size is what decides how big a coordinate-centred crop "
"is; without it there is no box to cut")
return source
mask_plane_for(settings.get("object_array", "cell"), settings)
return source