"""Illumination / flat-field correction for the measurement path.
The problem
-----------
No microscope lights a field of view evenly. A lamp profile, vignetting in
the objective, a tilted condenser and dirt on the optics together make the
same cell measure brighter at the centre of the field than at its edge --
routinely 10-40 % between the middle and a corner on a widefield screen.
Every intensity feature spaCR writes (``*_mean_intensity``,
``*_percentile_*``, the radial distribution, the texture channels that are
computed on intensity) therefore carries a **position-dependent bias**, and
because objects are not distributed identically over every well, that bias
does not average out: it survives into the per-well aggregate, and from there
into classification and regression as an effect that looks entirely real and
is entirely an artefact of the optics.
It is also the root cause of the plate-scale edge effects
:mod:`spacr.plate_qc` already detects and reports. Detecting them is useful;
removing them is what changes the answer.
The model
---------
Per channel, per plate::
observed(y, x) = dark + flat(y, x) * true(y, x)
``flat`` is the multiplicative illumination field, normalised so its mean is
1 -- the plate's overall intensity level is preserved, so a corrected number
stays on the same scale as an uncorrected one and only its *position
dependence* is removed. ``dark`` is the additive camera offset. The
correction the hook applies is::
corrected = (observed - dark) / flat
What is estimated, and why
--------------------------
**Retrospective, from the data itself**: a per-pixel median across many
fields of the plate, followed by a fit of a smooth low-order surface.
*Why the median across fields.* A pixel is covered by a cell in only a
minority of the fields on a plate, so the across-field median at that pixel
sees background almost every time and the objects drop out. Each field is
first divided by its own median, so a densely-seeded field does not pull the
estimate up because it has more cells in it -- what is being averaged
is the *relative* profile, not the brightness.
*Why the surface fit on top.* Illumination is a physically smooth,
low-frequency function of position: a lamp profile plus a vignette. Fitting a
low-order 2-D polynomial (default degree 4, 15 terms) to the per-pixel median
imposes exactly that prior, so residual object structure and photon noise
cannot leak into the gain map and be baked into every measurement on the
plate. The fit is trimmed twice against a MAD threshold, so a persistent
bright artefact -- a fluorescent speck in the same place on every field --
is rejected rather than smeared into the surface. For illumination that is
genuinely not polynomial (a dust shadow, a sharply structured lamp) pass
``estimator='smooth'``: the same per-pixel median, Gaussian-smoothed at
1/16 of the short side and interpolated back to full resolution.
*Why not BaSiC.* BaSiC's low-rank + sparse decomposition is the better
estimator when you have hundreds of fields and a genuine dark-field to
recover, but it is an iterative optimisation with its own convergence
failure modes, and its dark-field term is only identifiable because of those
extra assumptions. From a single acquisition, ``dark`` and ``flat`` are not
separately identifiable at all: multiplying ``flat`` by a constant and
absorbing it into the per-field brightness leaves every observation
unchanged. Estimating a dark-field anyway -- for instance from the per-pixel
minimum across fields, which is the usual shortcut -- returns *dark plus the
dimmest background*, and subtracting it removes real signal. So spaCR does
not guess: ``dark`` is 0 unless you supply the camera offset you measured
from a dark frame, via ``illumination_dark``.
Reaching the worker processes
-----------------------------
:func:`spacr.measure.measure_crop` measures fields in a
:class:`multiprocessing.Pool`. Under ``spawn`` / ``forkserver`` each worker is
a fresh interpreter with an empty hook registry, so a correction registered
only in the parent applies to **nothing** while the run looks perfectly
normal -- the single worst outcome for this feature, because the user then
believes their numbers are corrected. :func:`enable_illumination_correction`
therefore does not merely register the hook: it writes the model path to
:data:`MODEL_ENV_VAR` and appends ``spacr.illumination:install`` to
``SPACR_MEASURE_HOOKS``, which every start method inherits, so each worker
installs the correction for itself. :func:`worker_delivery_status` reports
whether that is actually in place, and :func:`enable_illumination_correction`
prints it.
Using it
--------
Off by default. From a settings dict (the keys are registered through the
:func:`spacr.settings.register_defaults` seam, see
:func:`illumination_settings`)::
settings = get_measure_crop_settings(settings={...})
settings['illumination_correction'] = True
prepare_illumination_correction(settings) # estimate, save, enable, QC
measure_crop(settings)
or explicitly::
model = estimate_illumination(src, channels=[0, 1, 2])
model.save('/data/plate1/illumination/illumination_model.npz')
illumination_qc(model, src, save_dir='/data/plate1/illumination')
enable_illumination_correction('/data/plate1/illumination/illumination_model.npz')
Nothing in this module runs unless one of those calls is made, and
:func:`disable_illumination_correction` returns the process to a state where
``measure_crop`` measures exactly what it measured before.
In the GUI the same two routes exist and end here: the "Illumination
Correction" category on the Measure panel, which is the switch thrown on the
run whose numbers it changes, and the Illumination button on that screen's
masthead, which opens this module's own settings form and Run button so the
field can be estimated and QC'd without measuring the plate. Neither is a
tile: the module folded into Measure and left the app registry with it.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import tempfile
import time
from dataclasses import dataclass
from dataclasses import field as _dataclass_field
from typing import Any, Dict, Iterable, Mapping, Optional, Sequence, Tuple
import numpy as np
from .errors import ConfigurationError
from .measure_hooks import (
HOOKS_ENV_VAR,
preprocessing_hooks,
register_preprocessing_hook,
unregister_preprocessing_hook,
)
__all__ = [
'APP_KEY',
'HOOK_NAME',
'HOOK_PRIORITY',
'INSTALLER_ENTRY',
'MODEL_ENV_VAR',
'ON_MISSING_ENV_VAR',
'IlluminationError',
'IlluminationField',
'IlluminationModel',
'IlluminationCorrector',
'PreparedIllumination',
'SegmentationIlluminationSession',
'estimate_illumination',
'load_illumination_model',
'plate_of_field',
'position_intensity_slope',
'illumination_qc',
'enable_illumination_correction',
'disable_illumination_correction',
'worker_delivery_status',
'install',
'prepare_illumination_model',
'prepare_illumination_correction',
'prepare_segmentation_illumination',
'load_segmentation_illumination_resume',
'validate_segmentation_illumination_resume',
'illumination_settings',
'register_illumination_settings',
]
#: Settings key namespace and :func:`spacr.settings.register_defaults` key.
APP_KEY = 'illumination'
#: Registry key the correction is registered under. Fixed, and always passed
#: explicitly: the parent process and a spawned worker both register the same
#: name, and :mod:`spacr.measure_hooks` replaces rather than appends, so a
#: field cannot be corrected twice however many routes installed the hook.
HOOK_NAME = 'spacr.illumination.correct'
#: Illumination correction runs before any other preprocessing hook. It is a
#: correction of the *sensor*, so anything else a user chains on top -- a
#: background subtraction, a ratio -- should see corrected pixels.
HOOK_PRIORITY = -100
#: What goes in ``SPACR_MEASURE_HOOKS`` so worker processes install it too.
INSTALLER_ENTRY = 'spacr.illumination:install'
#: Path to the saved model. Read by :func:`install` in each worker.
MODEL_ENV_VAR = 'SPACR_ILLUMINATION_MODEL'
#: ``'error'`` (default) or ``'skip'`` -- what a worker does with a field
#: whose plate has no estimated illumination field.
ON_MISSING_ENV_VAR = 'SPACR_ILLUMINATION_ON_MISSING'
#: Key used when the model is estimated across all plates at once.
ALL_PLATES = '*'
#: Below this many fields the across-field median is not a reliable object
#: rejector, so the estimate is still produced but loudly qualified.
MIN_FIELDS_FOR_A_TRUSTWORTHY_ESTIMATE = 10
#: A fitted surface is floored at this fraction of its own median before it is
#: inverted. Dividing by a gain that a polynomial fit dragged to ~0 in a corner
#: would turn a handful of pixels into astronomic intensities.
FLAT_FLOOR_FRACTION = 0.05
#: Clipping warnings printed per process before they are summarised instead.
_MAX_CLIP_WARNINGS = 5
[docs]
class IlluminationError(ConfigurationError):
"""Illumination correction was asked for and could not be delivered.
A :class:`spacr.errors.ConfigurationError`, not a per-field data error:
every failure this class reports (no fields to estimate from, a model that
does not cover the plate being measured, a channel the model was never
estimated for) is wrong for the whole run, and the alternative -- measuring
on quietly uncorrected pixels -- is the outcome this module exists to
prevent.
"""
@dataclass(frozen=True)
[docs]
class IlluminationField:
"""The illumination estimate for one plate, for one or more channels.
:param plate: the plate key, or :data:`ALL_PLATES` when the model was
estimated across every plate at once.
:param channels: source channel indices, in the order they index
:attr:`flatfield`'s first axis. These are indices into the *merged
stack*, i.e. exactly the values in ``settings['channels']``.
:param flatfield: ``(C, Y, X)`` float32 multiplicative field, normalised so
each channel's mean is 1.0. ``corrected = (observed - dark) /
flatfield``.
:param dark: per-channel additive offset subtracted before dividing. Zero
unless the user supplied a measured camera offset -- see the module
docstring for why it is not estimated.
:param n_fields: how many fields the estimate was made from.
:param estimator: ``'polynomial'`` or ``'smooth'``.
:param degree: polynomial degree, or 0 for the smooth estimator.
:param bin_size: the binning factor the per-pixel median was computed at.
Illumination is low-frequency, so binning costs nothing and buys both
the memory to hold many fields at once and a quieter statistic.
:param floored: pixels the fitted surface had to be floored at (see
:data:`FLAT_FLOOR_FRACTION`). Non-zero means the fit went negative
somewhere and the estimate should be looked at before it is trusted.
:param darkfield: optional ``(C, Y, X)`` spatial background subtracted on
top of ``dark``. Only a vendor profile that ships its own background
surface carries one; an estimated field never does.
"""
plate: str
channels: Tuple[int, ...]
flatfield: np.ndarray
dark: np.ndarray
n_fields: int
estimator: str
degree: int
bin_size: int
floored: int = 0
darkfield: Optional[np.ndarray] = None
@property
[docs]
def shape(self) -> Tuple[int, int]:
"""``(Y, X)`` pixel shape of the estimated field."""
return (int(self.flatfield.shape[1]), int(self.flatfield.shape[2]))
[docs]
def index_of(self, channel: int) -> int:
"""Position of source ``channel`` along :attr:`flatfield`'s first axis.
:param channel: a merged-stack channel index.
:raises IlluminationError: if the model was never estimated for it.
Correcting the channels that happen to be present and leaving the
rest alone would put corrected and uncorrected numbers in the same
table.
"""
try:
return self.channels.index(int(channel))
except ValueError:
raise IlluminationError(
f"the illumination model for plate {self.plate!r} covers "
f"channels {list(self.channels)}, but the run measures "
f"channel {channel}. Re-estimate with "
f"channels={sorted(set(self.channels) | {int(channel)})}."
) from None
[docs]
def gain_stack(self, channels: Sequence[int]) -> np.ndarray:
"""``(Y, X, C)`` multiplicative gains ``1 / flatfield`` for ``channels``.
Shaped for the array a preprocessing hook is handed, so applying the
correction is one broadcast multiply.
:param channels: source channel indices, in the order they appear
along the last axis of the array being corrected.
"""
gains = [1.0 / self.flatfield[self.index_of(c)] for c in channels]
return np.stack(gains, axis=-1).astype(np.float32, copy=False)
[docs]
def dark_stack(self, channels: Sequence[int]) -> np.ndarray:
"""``(C,)`` additive offsets for ``channels``, ready to broadcast.
:param channels: source channel indices in the order required by the
array being corrected.
"""
return np.asarray([self.dark[self.index_of(c)] for c in channels],
dtype=np.float32)
def _offset_stack(self, channels: Sequence[int]) -> np.ndarray:
"""The offsets to subtract: ``(C,)`` scalars, or ``(Y, X, C)`` planes.
A field with a spatial :attr:`darkfield` returns the scalar dark plus
that surface per channel; any other returns :meth:`dark_stack`.
:param channels: source channel indices in the order required by the
array being corrected.
"""
scalars = self.dark_stack(channels)
if self.darkfield is None:
return scalars
planes = [self.darkfield[self.index_of(c)] for c in channels]
return (np.stack(planes, axis=-1).astype(np.float32, copy=False)
+ scalars)
[docs]
def describe(self) -> str:
"""One line per channel: range, non-uniformity and how it was made."""
lines = []
nonuniform = self.nonuniformity()
for position, channel in enumerate(self.channels):
plane = self.flatfield[position]
lines.append(
f"plate {self.plate}, channel {channel}: gain field "
f"{plane.min():.3f}-{plane.max():.3f} (mean 1.000), "
f"non-uniformity {100 * nonuniform[int(channel)]:.1f}%, "
f"{self.estimator}"
f"{f' degree {self.degree}' if self.degree else ''} over "
f"{self.n_fields} field(s), dark={self.dark[position]:g}")
return '\n'.join(lines)
@dataclass
[docs]
class IlluminationModel:
"""Estimated illumination fields for every plate in a source folder.
:param fields: plate key -> :class:`IlluminationField`. A model estimated
with ``per_plate=False`` holds the single key :data:`ALL_PLATES`,
which matches every plate.
:param meta: provenance -- source folders, channels, when it was estimated,
the settings it was estimated with. Written into the ``.npz`` and read
back, so a model on disk can always say what produced it.
"""
fields: Dict[str, IlluminationField]
meta: Dict[str, Any] = _dataclass_field(default_factory=dict)
@property
[docs]
def per_plate(self) -> bool:
"""Whether the model holds one field per plate rather than one field."""
return ALL_PLATES not in self.fields
[docs]
def field_for(self, plate: str) -> IlluminationField:
"""The :class:`IlluminationField` that applies to ``plate``.
:param plate: plate key whose estimated illumination field is needed.
:raises IlluminationError: when nothing in the model covers it. This
is deliberately not a fall back to "some other plate's field":
illumination differs between acquisition sessions, which is the
whole reason the default is one field per plate.
"""
if ALL_PLATES in self.fields:
return self.fields[ALL_PLATES]
try:
return self.fields[plate]
except KeyError:
raise IlluminationError(
f"no illumination field was estimated for plate {plate!r}; "
f"the model covers {sorted(self.fields)}. Re-estimate over a "
f"source folder that contains this plate, or estimate one "
f"field for everything with illumination_per_plate=False."
) from None
[docs]
def describe(self) -> str:
"""Every field's :meth:`IlluminationField.describe`, one per line."""
return '\n'.join(self.fields[key].describe()
for key in sorted(self.fields))
[docs]
def save(self, path: str) -> str:
"""Write the model to ``path`` as a compressed ``.npz``.
The path is what :func:`enable_illumination_correction` puts in the
environment, and what each worker process loads: the model has to be
on disk for a ``spawn`` worker to be able to see it at all.
:param path: destination file. Parent folders are created.
:returns: the absolute path written.
"""
path = os.path.abspath(path)
parent = os.path.dirname(path)
os.makedirs(parent, exist_ok=True)
arrays = {}
index = {}
for key, item in self.fields.items():
slot = f'field{len(index)}'
index[slot] = {
'plate': item.plate,
'channels': [int(c) for c in item.channels],
'dark': [float(d) for d in np.asarray(item.dark).ravel()],
'n_fields': int(item.n_fields),
'estimator': item.estimator,
'degree': int(item.degree),
'bin_size': int(item.bin_size),
'floored': int(item.floored),
'key': key,
}
arrays[slot] = np.asarray(item.flatfield, dtype=np.float32)
if item.darkfield is not None:
arrays[f'{slot}_dark'] = np.asarray(item.darkfield,
dtype=np.float32)
payload = {'index': index, 'meta': self.meta, 'format': 1}
np.savez_compressed(path, manifest=np.asarray(json.dumps(payload)),
**arrays)
return path
@classmethod
[docs]
def load(cls, path: str) -> 'IlluminationModel':
"""Read a model written by :meth:`save`.
:param path: the ``.npz`` file.
:raises IlluminationError: when the file is missing or is not a
spaCR illumination model. A worker that cannot load the model must
say so rather than measure uncorrected pixels.
"""
if not os.path.isfile(path):
raise IlluminationError(
f"illumination model {path!r} does not exist. "
f"{MODEL_ENV_VAR} must point at a file "
f"IlluminationModel.save() wrote.")
try:
with np.load(path, allow_pickle=False) as handle:
payload = json.loads(str(handle['manifest']))
fields = {}
for slot, entry in payload['index'].items():
flat = np.asarray(handle[slot], dtype=np.float32)
darkfield = (np.asarray(handle[f'{slot}_dark'],
dtype=np.float32)
if f'{slot}_dark' in handle.files else None)
fields[entry['key']] = IlluminationField(
plate=entry['plate'],
channels=tuple(int(c) for c in entry['channels']),
flatfield=flat,
dark=np.asarray(entry['dark'], dtype=np.float32),
n_fields=int(entry['n_fields']),
estimator=str(entry['estimator']),
degree=int(entry['degree']),
bin_size=int(entry['bin_size']),
floored=int(entry.get('floored', 0)),
darkfield=darkfield)
except IlluminationError:
raise
except Exception as exc:
raise IlluminationError(
f"illumination model {path!r} could not be read: "
f"{type(exc).__name__}: {exc}") from exc
return cls(fields=fields, meta=dict(payload.get('meta', {})))
[docs]
def load_illumination_model(path: str) -> IlluminationModel:
"""Read a saved model. Thin alias for :meth:`IlluminationModel.load`.
:param path: the ``.npz`` written by :meth:`IlluminationModel.save`.
"""
return IlluminationModel.load(path)
[docs]
def plate_of_field(file_name: str) -> str:
"""The plate a merged field belongs to, from its file name.
spaCR names merged fields ``<plateID>_<wellID>_<fieldID>.npy`` (see
:mod:`spacr.io`), so the plate is the first underscore-separated token.
A name with no underscore is its own plate, which keeps a hand-assembled
folder working instead of silently pooling it.
:param file_name: file name or stem, with or without directories.
"""
stem = os.path.splitext(os.path.basename(str(file_name)))[0]
head, sep, _rest = stem.partition('_')
return head if sep else stem
def _source_folders(src) -> Tuple[str, ...]:
"""Normalise ``settings['src']`` -- a folder or a list of them -- to a tuple."""
if isinstance(src, (str, os.PathLike)):
return (os.path.abspath(str(src)),)
return tuple(os.path.abspath(str(item)) for item in src)
def _merged_files(src) -> Dict[str, list]:
"""Map plate key -> sorted list of merged ``.npy`` paths under ``src``.
Names starting with a dot are left out: a macOS ``._<field>.npy``
AppleDouble sidecar keeps the ``.npy`` ending and is not an array.
"""
grouped: Dict[str, list] = {}
for folder in _source_folders(src):
if not os.path.isdir(folder):
raise IlluminationError(
f"illumination correction was asked to estimate from "
f"{folder!r}, which is not a folder. Point it at the merged "
f"field folder measure_crop reads (settings['src']).")
for name in sorted(os.listdir(folder)):
if name.endswith('.npy') and not name.startswith('.'):
grouped.setdefault(plate_of_field(name), []).append(
os.path.join(folder, name))
return grouped
def _sample(paths: Sequence[str], limit: int) -> list:
"""Take at most ``limit`` paths, evenly spaced across the sorted list.
Evenly spaced rather than random: the files are sorted by well and field,
so this walks the whole plate instead of over-weighting whichever corner a
random draw happened to hit, and it is deterministic -- two runs of the
estimator on the same folder produce the same field, which matters when
the output is a correction applied to published numbers.
"""
paths = list(paths)
if limit <= 0 or len(paths) <= limit:
return paths
step = len(paths) / float(limit)
return [paths[min(len(paths) - 1, int(i * step))] for i in range(limit)]
def _bin_factor(shape: Tuple[int, int], grid: int) -> int:
"""Binning factor that puts the long side of ``shape`` at or below ``grid``."""
height, width = shape
factor = int(math.ceil(max(height, width) / float(max(1, grid))))
return max(1, min(factor, height, width))
def _bin_image(plane: np.ndarray, factor: int) -> np.ndarray:
"""Block-average ``plane`` by ``factor``, dropping any partial edge blocks."""
if factor == 1:
return np.asarray(plane, dtype=np.float32)
height = (plane.shape[0] // factor) * factor
width = (plane.shape[1] // factor) * factor
block = np.asarray(plane[:height, :width], dtype=np.float64)
block = block.reshape(height // factor, factor, width // factor, factor)
return block.mean(axis=(1, 3)).astype(np.float32)
def _bin_centres(length: int, factor: int) -> np.ndarray:
"""Full-resolution pixel coordinates of each block centre."""
count = length // factor
return np.arange(count, dtype=np.float64) * factor + (factor - 1) / 2.0
def _normalised(coords: np.ndarray, length: int) -> np.ndarray:
"""Map pixel coordinates onto ``[-1, 1]`` across an axis of ``length``."""
half = max((length - 1) / 2.0, 1e-9)
return (np.asarray(coords, dtype=np.float64) - half) / half
def _polynomial_terms(degree: int) -> Tuple[Tuple[int, int], ...]:
"""Exponent pairs ``(row_power, col_power)`` of a 2-D polynomial."""
return tuple((i, j) for i in range(degree + 1)
for j in range(degree + 1 - i))
def _fit_polynomial_surface(stat: np.ndarray, shape: Tuple[int, int],
factor: int, degree: int) -> np.ndarray:
"""Fit a trimmed 2-D polynomial to ``stat`` and evaluate it at ``shape``.
``stat`` is the binned per-pixel median; ``shape`` is the full-resolution
field. Two rounds of MAD trimming reject bins the surface cannot explain
-- a fluorescent speck that sits in the same place on every field, an
always-confluent corner -- rather than bending the surface towards them.
"""
height, width = shape
rows = _normalised(_bin_centres(height, factor), height)
cols = _normalised(_bin_centres(width, factor), width)
grid_rows, grid_cols = np.meshgrid(rows, cols, indexing='ij')
terms = _polynomial_terms(degree)
design = np.stack([(grid_rows ** i) * (grid_cols ** j)
for i, j in terms], axis=-1)
design = design.reshape(-1, len(terms))
target = np.asarray(stat, dtype=np.float64).reshape(-1)
keep = np.isfinite(target) & (target > 0)
if keep.sum() < 4 * len(terms):
keep = np.isfinite(target)
coefficients = None
for _round in range(3):
if keep.sum() < len(terms) + 1:
break
coefficients, *_ = np.linalg.lstsq(design[keep], target[keep],
rcond=None)
residual = target - design @ coefficients
scale = 1.4826 * np.median(np.abs(residual[keep] -
np.median(residual[keep])))
if not np.isfinite(scale) or scale <= 0:
break
tighter = keep & (np.abs(residual) <= 3.0 * scale)
if tighter.sum() < max(4 * len(terms), len(terms) + 1):
break
if tighter.sum() == keep.sum():
break
keep = tighter
if coefficients is None:
raise IlluminationError(
'the illumination surface could not be fitted: every binned '
'pixel was rejected as non-finite or non-positive. Check that '
'the channel being estimated actually holds image data.')
full_rows = _normalised(np.arange(height), height)
full_cols = _normalised(np.arange(width), width)
surface = np.zeros((height, width), dtype=np.float64)
for coefficient, (i, j) in zip(coefficients, terms):
surface += coefficient * np.outer(full_rows ** i, full_cols ** j)
return surface
def _smooth_surface(stat: np.ndarray, shape: Tuple[int, int],
factor: int) -> np.ndarray:
"""Gaussian-smooth ``stat`` and interpolate it up to ``shape``.
The alternative estimator, for illumination a polynomial cannot express.
Sigma is 1/16 of the binned short side: large enough that cell-scale
structure (a few binned pixels at most) cannot survive it, small enough to
keep a real lamp profile or a dust shadow.
"""
from scipy.ndimage import gaussian_filter, map_coordinates
sigma = max(1.0, min(stat.shape) / 16.0)
smoothed = gaussian_filter(np.asarray(stat, dtype=np.float64), sigma,
mode='nearest')
height, width = shape
rows = (np.arange(height, dtype=np.float64) - (factor - 1) / 2.0) / factor
cols = (np.arange(width, dtype=np.float64) - (factor - 1) / 2.0) / factor
grid_rows, grid_cols = np.meshgrid(rows, cols, indexing='ij')
return map_coordinates(smoothed, [grid_rows, grid_cols], order=1,
mode='nearest')
def _read_binned_field(path: str, channels: Sequence[int],
factor: Optional[int], grid: int
) -> Tuple[np.ndarray, int, Tuple[int, int]]:
"""Load one merged field and return its binned intensity channels.
:returns: ``(binned (C, y, x), factor, full (Y, X) shape)``.
"""
data = np.load(path, mmap_mode='r')
if data.ndim not in (3, 4):
raise IlluminationError(
f"{path!r} is a {data.ndim}-D array; a merged field is (Y, X, C) "
f"or (Z, Y, X, C).")
full_shape = (int(data.shape[-3]), int(data.shape[-2])) if data.ndim == 4 \
else (int(data.shape[0]), int(data.shape[1]))
if factor is None:
factor = _bin_factor(full_shape, grid)
planes = []
for channel in channels:
if channel >= data.shape[-1]:
raise IlluminationError(
f"{path!r} has {data.shape[-1]} channels, so channel "
f"{channel} does not exist. settings['channels'] names "
f"indices into the merged stack.")
plane = np.asarray(data[..., int(channel)])
if plane.ndim == 3:
plane = np.median(plane, axis=0)
planes.append(_bin_image(plane, factor))
return np.stack(planes, axis=0), factor, full_shape
def _relative_profile(stack: np.ndarray) -> np.ndarray:
"""Per-pixel median of fields, each normalised by its own median.
``stack`` is ``(K, y, x)``. Dividing each field by its own median before
the pixel-wise median is what makes this an estimate of the *profile*: a
field packed with bright cells then contributes its shape, not its
brightness.
"""
levels = np.median(stack.reshape(stack.shape[0], -1), axis=1)
usable = np.isfinite(levels) & (levels > 0)
if not usable.any():
raise IlluminationError(
'every field used for the illumination estimate had a median of '
'zero or worse; there is no signal to estimate a profile from.')
scaled = stack[usable] / levels[usable][:, None, None]
return np.median(scaled, axis=0)
[docs]
def estimate_illumination(src, channels: Sequence[int], *,
per_plate: bool = True,
estimator: str = 'polynomial',
degree: int = 4,
max_fields: int = 50,
grid: int = 256,
dark: float = 0.0,
verbose: bool = True) -> IlluminationModel:
"""Estimate the illumination field from the merged fields in ``src``.
Retrospective: the data corrects itself. See the module docstring for what
is estimated and why.
:param src: the merged field folder ``measure_crop`` reads, or a list of
them (``settings['src']`` accepts both).
:param channels: merged-stack channel indices to estimate, i.e.
``settings['channels']``. One field is estimated per channel: two
fluorophores go through different filters and vignette differently.
:param per_plate: one field per plate (default) or one for everything.
Per plate is the default because illumination differs between
acquisition sessions -- lamp age, a re-seated filter cube, a different
objective -- and pooling two sessions estimates neither.
:param estimator: ``'polynomial'`` (default) or ``'smooth'``.
:param degree: polynomial degree. 4 gives 15 terms: enough for a lamp
profile plus a vignette and a tilt, far too few to fit a cell.
:param max_fields: fields per plate to estimate from, sampled evenly
across the sorted file list. 50 is well past the point where the
across-field median stops moving, and bounds the memory.
:param grid: the per-pixel median is computed on a binned grid whose long
side is at most this. Illumination is low-frequency, so binning loses
nothing, quiets the photon noise and is what makes 50 fields fit in
memory. The fitted surface is returned at full resolution.
:param dark: additive camera offset subtracted before dividing, in raw
counts. Not estimated -- see the module docstring.
:param verbose: print one line per estimated field.
:returns: an :class:`IlluminationModel`.
:raises IlluminationError: when ``src`` holds no merged fields, or a field
is unreadable in a way that would make the estimate meaningless.
"""
channels = [int(c) for c in channels]
if not channels:
raise IlluminationError(
'illumination correction needs at least one channel to estimate; '
"settings['channels'] was empty.")
if estimator not in ('polynomial', 'smooth'):
raise IlluminationError(
f"unknown illumination estimator {estimator!r}; use 'polynomial' "
f"(a trimmed low-order surface) or 'smooth' (a Gaussian-smoothed "
f"median).")
grouped = _merged_files(src)
if not grouped:
raise IlluminationError(
f"no .npy fields found under {list(_source_folders(src))}; there "
f"is nothing to estimate an illumination field from.")
if not per_plate:
pooled = [path for paths in grouped.values() for path in paths]
grouped = {ALL_PLATES: sorted(pooled)}
fields = {}
for plate in sorted(grouped):
paths = _sample(grouped[plate], max_fields)
stack = []
factor = None
full_shape = None
skipped = 0
for path in paths:
binned, factor, shape = _read_binned_field(path, channels, factor,
grid)
if full_shape is None:
full_shape = shape
elif shape != full_shape:
skipped += 1
continue
stack.append(binned)
if not stack:
raise IlluminationError(
f"plate {plate!r} contributed no usable field to the "
f"illumination estimate ({skipped} had a different shape to "
f"the first).")
stack = np.stack(stack, axis=0)
planes = []
floored_total = 0
for position in range(len(channels)):
profile = _relative_profile(stack[:, position])
if estimator == 'polynomial':
surface = _fit_polynomial_surface(profile, full_shape, factor,
degree)
else:
surface = _smooth_surface(profile, full_shape, factor)
floor = FLAT_FLOOR_FRACTION * float(np.median(surface))
floored = int(np.count_nonzero(surface < floor))
floored_total += floored
if floored:
surface = np.maximum(surface, floor)
mean = float(surface.mean())
if not np.isfinite(mean) or mean <= 0:
raise IlluminationError(
f"the illumination surface for plate {plate!r} channel "
f"{channels[position]} has mean {mean!r}; it cannot be "
f"normalised or inverted.")
planes.append((surface / mean).astype(np.float32))
item = IlluminationField(
plate=str(plate),
channels=tuple(channels),
flatfield=np.stack(planes, axis=0),
dark=np.full(len(channels), float(dark), dtype=np.float32),
n_fields=int(stack.shape[0]),
estimator=estimator,
degree=int(degree) if estimator == 'polynomial' else 0,
bin_size=int(factor),
floored=floored_total)
fields[str(plate)] = item
if verbose:
print(item.describe())
if item.n_fields < MIN_FIELDS_FOR_A_TRUSTWORTHY_ESTIMATE:
print(f"WARNING: plate {plate!r} was estimated from only "
f"{item.n_fields} field(s). The across-field median "
f"rejects objects because a pixel is covered by a cell "
f"in a minority of fields; with this few, cells can "
f"survive into the gain map. Check the QC image.")
if item.floored:
print(f"WARNING: {item.floored} pixel(s) of the fitted "
f"surface for plate {plate!r} fell below "
f"{FLAT_FLOOR_FRACTION:g} of its median and were "
f"floored. Look at the QC image before trusting this "
f"field, or use estimator='smooth'.")
if skipped:
print(f"NOTE: {skipped} field(s) of plate {plate!r} were a "
f"different pixel shape and sat the estimate out.")
meta = {
'src': list(_source_folders(src)),
'channels': channels,
'per_plate': bool(per_plate),
'estimator': estimator,
'degree': int(degree),
'max_fields': int(max_fields),
'grid': int(grid),
'dark': float(dark),
'created': time.strftime('%Y-%m-%d %H:%M:%S'),
'application_contract_version': 1,
'channel_index_space': 'persisted-intensity-axis',
'estimated_from_intensity_state': 'raw',
}
return IlluminationModel(fields=fields, meta=meta)
_VENDOR_IMAGE_SUFFIXES = ('.tif', '.tiff', '.npy', '.czi')
def _lenient_profile_json(text: str) -> Dict[str, Any]:
"""Parse one Harmony ``FlatfieldProfile`` blob.
Harmony writes the profile as JSON in some versions and as a JSON-like
map with bare keys and bare words in others. Strict JSON is tried first;
otherwise bare keys and bare-word values are quoted and it is read again.
:param text: the element text.
:returns: the parsed map.
:raises IlluminationError: when neither form parses.
"""
import re
text = str(text or '').strip()
try:
return json.loads(text)
except ValueError:
pass
quoted = text.replace("'", '"')
quoted = re.sub(r'([{,]\s*)([A-Za-z_][\w ]*?)\s*:', r'\1"\2":', quoted)
def _value(match):
"""Quote a bare profile value while preserving JSON booleans and null."""
word = match.group(2).strip()
if word in ('true', 'false', 'null'):
return match.group(0)
return f'{match.group(1)}"{word}"'
quoted = re.sub(r'(:\s*)([A-Za-z_][^,}\]"]*?)\s*(?=[,}])', _value, quoted)
quoted = re.sub(r':\s*(?=[,}])', ': ""', quoted)
try:
return json.loads(quoted)
except ValueError as exc:
raise IlluminationError(
f"a Harmony FlatfieldProfile could not be parsed: {exc}. The "
f"profile starts {text[:80]!r}.") from exc
def _harmony_surface(profile: Any) -> Optional[np.ndarray]:
"""Evaluate one Harmony polynomial profile on its pixel grid.
Harmony stores each surface as ``Coefficients`` grouped by total degree,
``Dims`` (width, height), ``Origin`` (x, y) and ``Scale`` (x, y). With
``x = (column - Origin[0]) * Scale[0]`` and
``y = (row - Origin[1]) * Scale[1]`` the surface is
``sum_i sum_j Coefficients[i][j] * x**(i - j) * y**j``.
:param profile: the ``Foreground`` or ``Background`` entry.
:returns: ``(height, width)`` float64 surface, or None when the entry
carries no polynomial (Harmony's "no correction" state).
:raises IlluminationError: for a profile type other than polynomial.
"""
inner = profile.get('Profile') if isinstance(profile, Mapping) else None
if not isinstance(inner, Mapping) or not inner.get('Coefficients'):
return None
kind = str(inner.get('Type', 'Polynomial'))
if kind.lower() != 'polynomial':
raise IlluminationError(
f"Harmony flat-field profile type {kind!r} is not supported; "
f"only polynomial profiles can be evaluated.")
width, height = (int(v) for v in inner['Dims'])
origin = [float(v) for v in inner.get('Origin', (0.0, 0.0))]
scale = [float(v) for v in inner.get('Scale', (1.0, 1.0))]
x = (np.arange(width, dtype=np.float64) - origin[0]) * scale[0]
y = (np.arange(height, dtype=np.float64) - origin[1]) * scale[1]
xx, yy = np.meshgrid(x, y)
surface = np.zeros((height, width), dtype=np.float64)
for degree, row in enumerate(inner['Coefficients']):
for power, coefficient in enumerate(row):
surface += float(coefficient) * xx ** (degree - power) * yy ** power
return surface
def _harmony_background_mean(profile: Any) -> Optional[float]:
"""The ``Mean`` of a Harmony background profile, or None when absent.
A profile with a background polynomial but no foreground polynomial is
Harmony's additive correction: the background surface ``B`` is
normalised to about 1 and Harmony's corrected image is
``raw - Mean * (B - 1)``, which keeps the field's mean intensity.
:param profile: the ``Background`` entry.
:returns: the mean in raw counts, or None when it is missing or not a
number.
"""
if not isinstance(profile, Mapping):
return None
try:
return float(profile['Mean'])
except (KeyError, TypeError, ValueError):
return None
def _harmony_profiles(path: str) -> Dict[int, Dict[str, Any]]:
"""Every channel's flat-field profile in a Harmony XML file.
Reads any element whose tag ends in ``FlatfieldProfile`` -- the entries
of an Operetta or Opera Phenix ``Index.idx.xml`` / ``Index.xml`` export,
or of a saved FFC profile file.
:param path: the XML file.
:returns: Harmony channel number -> ``{'foreground', 'background',
'degree', 'name'}``, the surfaces evaluated. A background-only
profile with a ``Mean`` becomes a unit foreground and the additive
offset ``Mean * (B - 1)``.
:raises IlluminationError: when the file holds no profile.
"""
import xml.etree.ElementTree as ElementTree
try:
root = ElementTree.parse(path).getroot()
except (OSError, ElementTree.ParseError) as exc:
raise IlluminationError(
f"vendor flat-field profile {path!r} could not be read as XML: "
f"{exc}") from exc
parents = {child: parent for parent in root.iter() for child in parent}
profiles: Dict[int, Dict[str, Any]] = {}
for element in root.iter():
if not str(element.tag).endswith('FlatfieldProfile'):
continue
blob = _lenient_profile_json(element.text)
owner = parents.get(element)
channel = blob.get('Channel')
if channel is None and owner is not None:
channel = owner.get('ChannelID')
if channel is None:
channel = len(profiles) + 1
foreground = _harmony_surface(blob.get('Foreground'))
background = _harmony_surface(blob.get('Background'))
if foreground is not None:
coefficients = blob['Foreground']['Profile']['Coefficients']
else:
mean = _harmony_background_mean(blob.get('Background'))
if background is None or mean is None:
continue
coefficients = blob['Background']['Profile']['Coefficients']
with np.errstate(over='ignore', invalid='ignore'):
background = mean * (background - 1.0)
foreground = np.ones_like(background)
candidate = {
'foreground': foreground,
'background': background,
'degree': max(len(coefficients) - 1, 0),
'name': str(blob.get('ChannelName', '') or ''),
}
channel = int(channel)
if channel in profiles:
previous = profiles[channel]
same_foreground = np.array_equal(previous['foreground'], foreground)
old_dark, new_dark = previous['background'], candidate['background']
same_background = ((old_dark is None and new_dark is None)
or (old_dark is not None and new_dark is not None
and np.array_equal(old_dark, new_dark)))
if not same_foreground or not same_background:
raise IlluminationError(
f"{path!r} contains conflicting Harmony flat-field "
f"profiles for channel {channel}; select an unambiguous export.")
continue
profiles[channel] = candidate
if not profiles:
raise IlluminationError(
f"{path!r} contains no Harmony FlatfieldProfile with a "
f"foreground polynomial or a background polynomial and mean. Point illumination_vendor_profile at the "
f"export's Index.idx.xml (or Index.xml), or at a saved FFC "
f"profile file.")
return profiles
def _vendor_reference_image(path: str) -> np.ndarray:
"""Read a vendor shading reference image as ``(C, Y, X)`` float64.
A ZEN shading reference, or a Nikon or Olympus flat-field image, exported
as ``.tif``/``.tiff``/``.czi``, or saved as ``.npy``. A 2-D image is one
plane.
:param path: the image.
:raises IlluminationError: when it cannot be read or is not 2-D or 3-D
once singleton axes are dropped.
"""
suffix = os.path.splitext(path)[1].lower()
try:
if suffix == '.npy':
image = np.load(path, allow_pickle=False)
elif suffix == '.czi':
import czifile
image = czifile.imread(path)
else:
import tifffile
image = tifffile.imread(path)
except ImportError as exc:
raise IlluminationError(
f"reading {path!r} needs {exc.name}; install it with "
f"'pip install {exc.name}'.") from exc
except Exception as exc:
raise IlluminationError(
f"vendor shading reference {path!r} could not be read: "
f"{type(exc).__name__}: {exc}") from exc
image = np.squeeze(np.asarray(image, dtype=np.float64))
if image.ndim == 2:
image = image[np.newaxis]
if image.ndim != 3:
raise IlluminationError(
f"vendor shading reference {path!r} has shape {image.shape}; "
f"expected one plane (Y, X) or one plane per channel (C, Y, X).")
return image
def _parse_vendor_channel_map(value: str, channels: Sequence[int]) -> Dict[int, int]:
"""Parse an explicit intensity-axis to vendor-profile assignment.
:param value: comma-separated zero-based:one-based channel pairs, or blank.
:param channels: persisted intensity-axis positions being corrected.
:returns: integer mapping, or an empty dict for the legacy default order.
:raises IlluminationError: for malformed, duplicated or incomplete pairs.
"""
text = str(value or '').strip()
if not text:
return {}
mapping = {}
for pair in text.split(','):
match = re.fullmatch(r'\s*(\d+)\s*:\s*(\d+)\s*', pair)
if match is None:
raise IlluminationError(
"illumination_vendor_channel_map must use comma-separated "
"zero-based intensity:one-based vendor pairs, for example 0:2,1:1.")
channel, vendor = (int(value) for value in match.groups())
if vendor < 1 or channel in mapping:
raise IlluminationError(
"illumination_vendor_channel_map needs one assignment per intensity "
"channel and vendor channel/plane numbers starting at 1.")
mapping[channel] = vendor
missing = sorted(set(int(channel) for channel in channels) - set(mapping))
if missing:
raise IlluminationError(
"illumination_vendor_channel_map does not cover corrected intensity "
f"channel(s) {missing}; add an explicit assignment for each.")
return mapping
def _vendor_illumination(path: str, channels: Sequence[int], *,
dark: float = 0.0,
channel_map: str = '',
verbose: bool = True) -> IlluminationModel:
"""Build an :class:`IlluminationModel` from a vendor flat-field file.
Two kinds of file are read:
* **Harmony XML** (PerkinElmer/Revvity Operetta and Opera Phenix). Merged
channel ``c`` takes Harmony channel ``c + 1``, the order Harmony's
``-ch1``, ``-ch2`` file names merge in. The foreground polynomial is the
gain and the background polynomial the spatial offset, both kept at
Harmony's own scale, so ``(observed - background) / foreground`` is
exactly what Harmony applies. A profile with only a background
polynomial and its ``Mean`` is applied additively as
``observed - Mean * (background - 1)``, as Harmony does. ``dark`` is
added to the background.
* **A shading reference image** (ZEN shading reference, Nikon or Olympus
flat-field image). Plane ``c`` is merged channel ``c``; a single plane
serves every channel. Each plane has ``dark`` subtracted and is
normalised to mean 1, like an estimated field.
One field covers every plate, because a vendor profile describes the
instrument rather than any one plate.
:param path: the vendor file.
:param channels: merged-stack channel indices to correct.
:param dark: camera offset in raw counts.
:param channel_map: optional comma-separated intensity:vendor pairs, e.g.
``0:2,1:1``. Intensity positions are zero-based, Harmony IDs and image
planes one-based. Every corrected channel must be covered; entries
for other intensity channels may be retained when selecting a subset.
Blank keeps the existing channel order and single-plane broadcasting.
:param verbose: print the resulting field's description.
:returns: model with resolved channel mapping in its saved metadata.
:raises IlluminationError: when the file does not cover a channel, or a
plane cannot be inverted.
"""
path = os.path.abspath(str(path))
if not os.path.isfile(path):
raise IlluminationError(
f"vendor flat-field profile {path!r} does not exist.")
channels = [int(c) for c in channels]
if not channels:
raise IlluminationError(
'a vendor flat-field profile needs at least one channel to apply '
'to; settings["channels"] is empty.')
try:
with np.errstate(over='ignore', invalid='ignore'):
scalar_dark = np.float32(float(dark))
except (TypeError, ValueError, OverflowError) as exc:
raise IlluminationError('vendor camera offset must be a finite number') from exc
if not np.isfinite(scalar_dark):
raise IlluminationError('vendor camera offset must be finite in float32')
explicit_mapping = _parse_vendor_channel_map(channel_map, channels)
resolved_mapping = {c: explicit_mapping.get(c, c + 1) for c in channels}
suffix = os.path.splitext(path)[1].lower()
darkfield = None
if suffix == '.xml':
profiles = _harmony_profiles(path)
missing = [c for c in channels if resolved_mapping[c] not in profiles]
if missing:
raise IlluminationError(
f"{path!r} has Harmony profiles for channels "
f"{sorted(profiles)} (1-based), but merged channel(s) "
f"{missing} need Harmony channel(s) "
f"{[resolved_mapping[c] for c in missing]}.")
planes = [profiles[resolved_mapping[c]]['foreground'] for c in channels]
backgrounds = [profiles[resolved_mapping[c]]['background'] for c in channels]
estimator = 'harmony'
degree = max(profiles[resolved_mapping[c]]['degree'] for c in channels)
elif suffix in _VENDOR_IMAGE_SUFFIXES:
image = _vendor_reference_image(path)
if image.shape[0] == 1 and not explicit_mapping:
resolved_mapping = {c: 1 for c in channels}
planes = [image[0] - float(dark) for _ in channels]
else:
missing = [c for c in channels
if not 1 <= resolved_mapping[c] <= image.shape[0]]
if missing:
raise IlluminationError(
f"{path!r} has {image.shape[0]} plane(s), but merged "
f"channel(s) {missing} request vendor plane(s) "
f"{[resolved_mapping[c] for c in missing]} (1-based).")
planes = [image[resolved_mapping[c] - 1] - float(dark) for c in channels]
planes = [plane / max(float(plane.mean()), 1e-12) for plane in planes]
estimator = 'vendor image'
degree = 0
else:
raise IlluminationError(
f"{path!r} is not a vendor flat-field file spaCR reads: use a "
f"Harmony .xml, or a shading reference image "
f"({', '.join(_VENDOR_IMAGE_SUFFIXES)}).")
shapes = {plane.shape for plane in planes}
if len(shapes) != 1:
raise IlluminationError(
f"the profiles in {path!r} differ in size between channels "
f"({sorted(shapes)}); one correction cannot cover them all.")
if suffix == '.xml' and any(b is not None for b in backgrounds):
for channel, plane, background in zip(channels, planes, backgrounds):
if background is not None and background.shape != plane.shape:
raise IlluminationError(
f"the Harmony background for channel {resolved_mapping[channel]} "
f"has shape {background.shape}, but its foreground has "
f"shape {plane.shape}; calibration grids must match exactly.")
with np.errstate(over='ignore', invalid='ignore'):
darkfield = np.stack(
[np.zeros_like(planes[i]) if b is None else b
for i, b in enumerate(backgrounds)]).astype(np.float32)
if not np.isfinite(darkfield).all():
raise IlluminationError(
f"the vendor spatial background in {path!r} must be finite in float32.")
with np.errstate(over='ignore', invalid='ignore'):
flatfield = np.stack(planes).astype(np.float32)
if not np.isfinite(flatfield).all():
raise IlluminationError(
f"the vendor flat field in {path!r} must be finite in float32.")
low = float(flatfield.min())
if low <= 0:
raise IlluminationError(
f"the vendor flat field in {path!r} reaches {low!r}; a gain map "
f"that is not strictly positive cannot be inverted.")
item = IlluminationField(
plate=ALL_PLATES,
channels=tuple(channels),
flatfield=flatfield,
dark=np.full(len(channels), float(dark), dtype=np.float32),
n_fields=0,
estimator=estimator,
degree=int(degree),
bin_size=1,
darkfield=darkfield)
if verbose:
print(f"illumination field read from vendor profile {path}")
print(item.describe())
meta = {
'vendor_profile': path,
'vendor_channel_map': {str(c): resolved_mapping[c] for c in channels},
'vendor_channel_map_explicit': bool(explicit_mapping),
'channels': channels,
'per_plate': False,
'estimator': estimator,
'degree': int(degree),
'dark': float(dark),
'created': time.strftime('%Y-%m-%d %H:%M:%S'),
'application_contract_version': 1,
'channel_index_space': 'persisted-intensity-axis',
'estimated_from_intensity_state': 'raw',
}
return IlluminationModel(fields={ALL_PLATES: item}, meta=meta)
[docs]
class IlluminationCorrector:
"""The preprocessing hook that applies an :class:`IlluminationModel`.
Registered through
:func:`spacr.measure_hooks.register_preprocessing_hook`, so it is handed
exactly the array the intensity measurements see -- the channels named by
``settings['channels']``, selected out of the merged stack, before a
single feature is computed.
**The dtype round trip is this class's decision, and it is made here
rather than in the hook machinery on purpose** (see
:func:`spacr.measure_hooks.apply_preprocessing_hooks`). Integer input is
corrected in float32 and returned by *rounding to nearest* and then
clipping to the dtype's range:
* **round, not truncate.** Truncation would shave a mean of 0.5 counts off
every corrected pixel. Averaged over a 500-pixel object that does not
wash out -- it is a systematic, one-directional shift of exactly the
kind this feature exists to remove. Rounding is unbiased.
* **clip, and count.** A gain above 1 at the edge of the field can push a
near-full-scale pixel past the top of a uint16. Clipping is the only
option that keeps the dtype the hook contract requires, but silently
clipping real signal is a lie about the data, so every pixel that was
*below* full scale before the correction and lands *at* full scale after
it is counted and reported. Pixels that were already saturated are not
counted: they were destroyed by the microscope, not by this class.
Float input is returned in its own float dtype with no rounding and no
clipping at all -- there is nothing to round to and no range to leave.
:param model: the estimated :class:`IlluminationModel`.
:param on_missing: ``'error'`` (default) or ``'skip'`` for a field whose
plate the model does not cover. The default is to fail the field:
a half-corrected table is worse than a failed one.
:param verbose: print the clipping reports.
"""
def __init__(self, model: IlluminationModel, *, on_missing: str = 'error',
verbose: bool = True) -> None:
"""Arm a corrector over a fitted illumination model.
:param model: the fitted model to divide fields by.
:param on_missing: what to do with a field the model has no profile for
-- ``'error'`` fails the field, ``'skip'`` measures it uncorrected
and counts it.
:param verbose: report the first few clipping events.
:raises IlluminationError: if ``on_missing`` is neither of the two.
"""
if on_missing not in ('error', 'skip'):
raise IlluminationError(
f"on_missing={on_missing!r}; use 'error' (fail the field) or "
f"'skip' (measure it uncorrected).")
self.model = model
self.on_missing = on_missing
self.verbose = bool(verbose)
#: fields corrected, fields skipped, pixels clipped, fields that clipped.
self.stats = {'corrected': 0, 'skipped': 0,
'clipped_pixels': 0, 'clipped_fields': 0}
self._warned = 0
self._cache: Dict[Tuple[str, Tuple[int, ...]], Tuple] = {}
[docs]
def __call__(self, channel_arrays: np.ndarray, context) -> np.ndarray:
"""Return ``channel_arrays`` corrected, in the same shape and dtype.
:param channel_arrays: ``(Y, X, C)`` or ``(Z, Y, X, C)`` intensities.
:param context: the :class:`spacr.measure_hooks.PreprocessingContext`.
"""
array = np.asarray(channel_arrays)
plate = plate_of_field(context.file_name)
try:
item = self.model.field_for(plate)
except IlluminationError:
if self.on_missing == 'error':
raise
self.stats['skipped'] += 1
if self.verbose and self.stats['skipped'] <= _MAX_CLIP_WARNINGS:
print(f"WARNING: no illumination field for plate {plate!r}; "
f"{context.file_name} is measured UNCORRECTED "
f"(illumination_on_missing='skip').")
return channel_arrays
gains, darks = self._gains_for(item, context.channels)
spatial = tuple(array.shape[-3:-1])
if spatial != item.shape:
raise IlluminationError(
f"the illumination field for plate {plate!r} is "
f"{item.shape[0]}x{item.shape[1]} but {context.file_name} is "
f"{spatial[0]}x{spatial[1]}. A gain map cannot be stretched "
f"onto a different sensor geometry; re-estimate over these "
f"fields.")
corrected = (array.astype(np.float32) - darks) * gains
return self._to_input_dtype(corrected, array, context)
def _gains_for(self, item: IlluminationField, channels: Sequence[int]):
"""Cache the ``(Y, X, C)`` gain stack per (plate, channel order)."""
key = (item.plate, tuple(int(c) for c in channels))
cached = self._cache.get(key)
if cached is None:
cached = (item.gain_stack(channels), item._offset_stack(channels))
self._cache[key] = cached
return cached
def _to_input_dtype(self, corrected: np.ndarray, original: np.ndarray,
context) -> np.ndarray:
"""Cast back to ``original.dtype``; see the class docstring."""
dtype = original.dtype
if not np.issubdtype(dtype, np.integer):
self.stats['corrected'] += 1
return corrected.astype(dtype, copy=False)
info = np.iinfo(dtype)
rounded = np.rint(corrected)
result = np.clip(rounded, info.min, info.max).astype(dtype)
lost = int(np.count_nonzero((rounded > info.max) &
(original < info.max)))
lost += int(np.count_nonzero((rounded < info.min) &
(original > info.min)))
self.stats['corrected'] += 1
if lost:
self.stats['clipped_pixels'] += lost
self.stats['clipped_fields'] += 1
self._report_clipping(lost, corrected.size, context)
return result
def _report_clipping(self, lost: int, total: int, context) -> None:
"""Say that real signal was clipped, without printing it 384 times."""
if not self.verbose:
return
self._warned += 1
if self._warned > _MAX_CLIP_WARNINGS:
return
tail = ('' if self._warned < _MAX_CLIP_WARNINGS else
' Further clipping warnings are suppressed; the running total '
'is on the corrector\'s .stats.')
print(f"WARNING: illumination correction clipped {lost} pixel(s) "
f"({100.0 * lost / max(total, 1):.4f}% of "
f"{context.file_name}) at the top of its dtype. Those pixels "
f"were not saturated before the correction, so real signal was "
f"lost. Re-acquire with more headroom, or measure on a wider "
f"dtype.{tail}")
[docs]
def report(self) -> str:
"""One line summarising what this corrector has done so far."""
return (f"illumination correction: {self.stats['corrected']} field(s) "
f"corrected, {self.stats['skipped']} skipped, "
f"{self.stats['clipped_pixels']} pixel(s) clipped across "
f"{self.stats['clipped_fields']} field(s)")
@dataclass(frozen=True)
[docs]
class PreparedIllumination:
"""One fitted/loaded model and the stage-neutral objects derived from it.
The saved model is deliberately not tagged ``measurement`` or
``segmentation``: both stages may reuse the same optical estimate. The
stage that *applies* it owns that provenance separately.
:param model: loaded or newly estimated illumination model.
:param corrector: corrector configured with the requested missing-plate
policy, but not registered as a Measure preprocessing hook.
:param model_path: absolute path of the saved model.
:param model_sha256: digest of the exact saved bytes at ``model_path``.
:param qc_artifacts: QC figure paths written while preparing the model.
"""
model: IlluminationModel
corrector: IlluminationCorrector
model_path: str
model_sha256: str
qc_artifacts: Tuple[str, ...] = ()
def _file_sha256(path: str) -> str:
"""Return the SHA-256 digest of ``path`` without loading it into memory."""
digest = hashlib.sha256()
with open(path, 'rb') as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b''):
digest.update(chunk)
return digest.hexdigest()
_SEGMENTATION_MODEL_META = {
'application_contract_version': 1,
'channel_index_space': 'persisted-intensity-axis',
'estimated_from_intensity_state': 'raw',
}
_SEGMENTATION_IMMUTABLE_RECORD_KEYS = (
'schema_version', 'model_path', 'model_sha256', 'pipeline_style',
'source_intensity_state', 'target_scope', 'correction_depth',
'raw_persisted_intensities_modified',
)
def _segmentation_pipeline_style(pipeline_style: str) -> str:
"""Validate and normalise the segmentation pipeline style.
:param pipeline_style: the style.
:returns: it, lowercased.
:raises IlluminationError: if it is neither ``'v1'`` nor ``'v2'`` -- the
two write different provenance, so a third value would produce a
record nothing can read back.
"""
style = str(pipeline_style).strip().lower()
if style not in {'v1', 'v2'}:
raise IlluminationError(
"segmentation illumination pipeline_style must be 'v1' or "
f"'v2', not {style!r}.")
return style
def _validate_segmentation_model(prepared: PreparedIllumination) -> None:
"""Check a prepared illumination model may be applied to segmentation inputs.
:param prepared: the prepared model and its QC artefacts.
:raises IlluminationError: if it was not prepared for this use. The
correction is applied to segmentation INPUTS only and never to the
persisted intensities, and a model prepared under other terms would
silently break that guarantee.
"""
try:
saved_digest = _file_sha256(prepared.model_path)
except OSError as exc:
raise IlluminationError(
'segmentation illumination cannot verify its saved model: '
f'{prepared.model_path}: {exc}') from exc
if saved_digest != prepared.model_sha256:
raise IlluminationError(
'segmentation illumination model bytes changed after preparation; '
'the saved SHA-256 no longer matches. Re-load the recorded model '
'or start a clean mask run.')
incompatible = [
key for key, value in _SEGMENTATION_MODEL_META.items()
if prepared.model.meta.get(key) != value
]
if incompatible:
raise IlluminationError(
'segmentation illumination cannot use this legacy or '
'incompatible model because its saved provenance does not prove '
'raw persisted-intensity-axis inputs '
f'({", ".join(incompatible)} missing or changed). Re-estimate '
'the illumination model from raw merged fields and start a clean '
'mask run.')
def _segmentation_application_record(
prepared: PreparedIllumination, provenance_path: str,
pipeline_style: str, completed_fields: Iterable[str],
application_state: str,
) -> Dict[str, Any]:
"""Build the provenance record for a segmentation-illumination session.
:param prepared: the prepared model.
:param provenance_path: where the record lives.
:param pipeline_style: which segmentation pipeline this corrects for.
:param completed_fields: the fields corrected so far.
:param application_state: where the session has got to.
:returns: the record.
"""
model_path = os.path.relpath(
prepared.model_path, os.path.dirname(provenance_path))
return {
'schema_version': 1,
'model_path': model_path,
'model_sha256': prepared.model_sha256,
'pipeline_style': pipeline_style,
'source_intensity_state': 'raw',
'target_scope': 'segmentation-input-only',
'correction_depth': 1,
'raw_persisted_intensities_modified': False,
'application_state': application_state,
'completed_fields': sorted({str(item) for item in completed_fields}),
'qc_artifacts': list(prepared.qc_artifacts),
}
def _read_segmentation_application(
prepared: PreparedIllumination, provenance_path: str,
pipeline_style: str) -> Tuple[Dict[str, Any], set]:
"""Read a previous session's record and check it describes this one.
The immutable keys -- the model hash, the pipeline style, the scope --
are compared rather than trusted: resuming against a record written for
a DIFFERENT model would report fields as corrected that were corrected
by something else.
:param prepared: the prepared model.
:param provenance_path: where the record lives.
:param pipeline_style: this session's pipeline style.
:returns: the record and the set of fields it says are complete.
:raises IlluminationError: if the record describes a different run.
"""
existing = _load_segmentation_application(provenance_path)
wanted = _segmentation_application_record(
prepared, provenance_path, pipeline_style, (), 'prepared')
mismatched = [
key for key in _SEGMENTATION_IMMUTABLE_RECORD_KEYS
if existing.get(key) != wanted[key]
]
if mismatched:
raise IlluminationError(
'cannot resume segmentation illumination with different '
f'provenance ({", ".join(mismatched)} changed); re-run '
'preprocessing as a clean mask run.')
completed = existing.get('completed_fields', [])
if not isinstance(completed, list):
raise IlluminationError(
'segmentation illumination provenance completed_fields must be '
'a list.')
application_state = existing.get('application_state')
if application_state not in {'prepared', 'running', 'complete'}:
raise IlluminationError(
'segmentation illumination provenance application_state must be '
"'prepared', 'running', or 'complete'.")
return existing, {str(field_id) for field_id in completed}
def _load_segmentation_application(
provenance_path: str) -> Dict[str, Any]:
"""Load a provenance record from disk.
:param provenance_path: the record file.
:returns: the parsed record.
:raises IlluminationError: if it is missing or unreadable -- a resume
with no record to resume from is a mistake worth stopping for, not a
fresh start.
"""
try:
with open(provenance_path, encoding='utf-8') as handle:
existing = json.load(handle)
except FileNotFoundError as exc:
raise IlluminationError(
'preprocess=False with segmentation illumination requires an '
'existing compatible segmentation_application.json; no record '
f'exists at {provenance_path}. Re-run preprocessing cleanly.') \
from exc
except (OSError, json.JSONDecodeError) as exc:
raise IlluminationError(
f"segmentation illumination provenance is unreadable: "
f"{provenance_path}: {exc}") from exc
if not isinstance(existing, dict):
raise IlluminationError(
'segmentation illumination provenance must be a JSON object.')
return existing
[docs]
def validate_segmentation_illumination_resume(
prepared: PreparedIllumination, *, provenance_path: str,
pipeline_style: str, expected_fields: Iterable[str]
) -> Dict[str, Any]:
"""Validate a ``preprocess=False`` mask resume without writing anything.
Normalised mask NPZ files cannot prove which intensity state Cellpose saw.
A bypassed preprocessing stage therefore proceeds only when a prior
application record names the same model bytes and pipeline style and
covers exactly the fields already on disk. This function never fits a
model, creates a record, corrects pixels, or updates the run journal.
:param prepared: the loaded model and corrector whose model path and digest
the record must name.
:param provenance_path: path of the prior segmentation application record
(JSON) to read.
:param pipeline_style: segmentation pipeline style the record must match,
``'v1'`` or ``'v2'`` (case-insensitive).
:param expected_fields: identifiers of the mask fields already on disk; the
record's completed-field set, compared as strings, must equal them
exactly.
:returns: the validated existing application record.
:raises IlluminationError: for an absent or incompatible record, model,
pipeline style, or completed-field set.
"""
_validate_segmentation_model(prepared)
style = _segmentation_pipeline_style(pipeline_style)
path = os.path.abspath(str(provenance_path))
existing, completed = _read_segmentation_application(
prepared, path, style)
if existing['application_state'] != 'complete':
raise IlluminationError(
'preprocess=False cannot trust an illumination application that '
f"is only {existing['application_state']!r}; finish the mask "
'preprocessing run or start it cleanly.')
expected = {str(field_id) for field_id in expected_fields}
if completed != expected:
missing = sorted(expected - completed)
extra = sorted(completed - expected)
raise IlluminationError(
'preprocess=False illumination provenance does not cover exactly '
f'the existing mask fields: missing={missing}, unexpected={extra}. '
'Re-run preprocessing as a clean mask run.')
return existing
[docs]
def load_segmentation_illumination_resume(
settings: Mapping[str, Any], *, provenance_path: str,
pipeline_style: str, expected_fields: Iterable[str],
verbose: Optional[bool] = None) -> PreparedIllumination:
"""Read and validate prior segmentation illumination without side effects.
This is the ``preprocess=False`` entry point. The application record is
authoritative: its exact model path is loaded and its digest, metadata,
pipeline style, and completed-field set are checked before the caller may
trust existing normalised mask NPZ files. No model is fitted, no QC or
application record is written, no pixels are corrected, and no Measure
hook is installed.
:param settings: run settings; ``illumination_correction`` must be true,
and ``illumination_model`` (when set), ``illumination_on_missing`` and
``verbose`` are read.
:param provenance_path: path of the prior segmentation application record
(JSON); its ``model_path`` is resolved relative to this file's folder
when not absolute.
:param pipeline_style: segmentation pipeline style, ``'v1'`` or ``'v2'``
(case-insensitive); any other value raises :class:`IlluminationError`.
:param expected_fields: identifiers of the mask fields already on disk; the
record's completed-field set must equal them exactly.
"""
if not settings.get('illumination_correction', False):
raise IlluminationError(
'load_segmentation_illumination_resume requires '
'illumination_correction=True.')
style = _segmentation_pipeline_style(pipeline_style)
path = os.path.abspath(str(provenance_path))
existing = _load_segmentation_application(path)
recorded_model = existing.get('model_path')
if not isinstance(recorded_model, str) or not recorded_model.strip():
raise IlluminationError(
'segmentation illumination provenance has no usable model_path; '
're-run preprocessing as a clean mask run.')
model_path = recorded_model
if not os.path.isabs(model_path):
model_path = os.path.join(os.path.dirname(path), model_path)
model_path = os.path.abspath(model_path)
requested_model = str(settings.get('illumination_model', '') or '').strip()
if (requested_model and
os.path.abspath(requested_model) != model_path):
raise IlluminationError(
"settings['illumination_model'] does not name the model recorded "
'for these masks; use the recorded model or re-run preprocessing '
'as a clean mask run.')
try:
digest = _file_sha256(model_path)
except OSError as exc:
raise IlluminationError(
f'the recorded segmentation illumination model cannot be read: '
f'{model_path}: {exc}') from exc
if digest != existing.get('model_sha256'):
raise IlluminationError(
'the recorded segmentation illumination model hash does not '
'match the model bytes on disk; re-run preprocessing as a clean '
'mask run.')
qc_artifacts = existing.get('qc_artifacts', [])
if not isinstance(qc_artifacts, list):
raise IlluminationError(
'segmentation illumination provenance qc_artifacts must be a '
'list.')
talk = settings.get('verbose', True) if verbose is None else verbose
model = IlluminationModel.load(model_path)
prepared = PreparedIllumination(
model=model,
corrector=IlluminationCorrector(
model,
on_missing=str(settings.get('illumination_on_missing', 'error')),
verbose=talk,
),
model_path=model_path,
model_sha256=digest,
qc_artifacts=tuple(str(item) for item in qc_artifacts),
)
validate_segmentation_illumination_resume(
prepared,
provenance_path=path,
pipeline_style=style,
expected_fields=expected_fields,
)
return prepared
[docs]
class SegmentationIlluminationSession:
"""Apply one illumination model exactly once per segmentation field.
Correction always receives a private copy. :meth:`correct` records that
an in-memory input was corrected; :meth:`mark_completed` is deliberately
separate and is the only operation that persists a field id. A pipeline
therefore marks a field only *after* its durable NPZ/mask output exists.
:param prepared: model/corrector returned by
:func:`prepare_illumination_model`.
:param provenance_path: destination ``segmentation_application.json``.
:param pipeline_style: ``'v1'`` or ``'v2'`` for the audit record.
:param resume: restore explicitly completed fields from an existing
compatible record without rewriting it; absence is an error. False
starts a fresh regenerated-output session and atomically replaces any
old completion claim with an explicit ``prepared`` record.
"""
STAGE_ID = 'illumination.segmentation_input'
STAGE_LABEL = 'Illumination correction — segmentation input'
def __init__(self, prepared: PreparedIllumination, *,
provenance_path: str, pipeline_style: str,
resume: bool = False) -> None:
"""Open a segmentation-input correction session and stamp its provenance.
:param prepared: the validated illumination model and its QC artefacts.
:param provenance_path: where the application record is written.
:param pipeline_style: which segmentation pipeline this corrects for.
:param resume: continue a previous session, reading back which fields it
already completed. Without it a fresh record is written stating that
no field has yet been corrected -- a new preprocessing run
invalidates any earlier completion claim, and saying so explicitly
is what stops a half-finished run being read as a finished one.
"""
_validate_segmentation_model(prepared)
pipeline_style = _segmentation_pipeline_style(pipeline_style)
self.prepared = prepared
self.provenance_path = os.path.abspath(str(provenance_path))
self.pipeline_style = pipeline_style
self._applied_fields = set()
self._completed_fields = set()
self._duplicate_attempts = 0
self._application_state = 'prepared'
if resume:
self._load_completed_fields()
else:
self._write_provenance(self._completed_fields, 'prepared')
self._record_stage('running')
@property
[docs]
def completed_fields(self) -> Tuple[str, ...]:
"""Durably completed field ids in stable order."""
return tuple(sorted(self._completed_fields))
@property
[docs]
def applied_fields(self) -> Tuple[str, ...]:
"""Field ids corrected during this process, in stable order."""
return tuple(sorted(self._applied_fields))
[docs]
def correct(self, field_id: str, channel_arrays: np.ndarray,
context) -> np.ndarray:
"""Correct a private copy of one raw field, refusing a second pass.
:param field_id: identifier of the raw field, compared as a string; a
field already corrected or marked complete raises
:class:`IlluminationError`.
:param channel_arrays: raw field intensities, ``(Y, X, C)`` or
``(Z, Y, X, C)``; a copy is corrected and the input is left
unchanged.
:param context: the :class:`spacr.measure_hooks.PreprocessingContext`
passed to the prepared corrector; its file name selects the plate's
model.
"""
field_id = str(field_id)
if (field_id in self._applied_fields or
field_id in self._completed_fields):
self._duplicate_attempts += 1
self._record_stage('failed')
raise IlluminationError(
f"illumination correction was requested twice for segmentation "
f"field {field_id!r}; correction_depth must remain 1.")
private = np.array(channel_arrays, copy=True)
skipped_before = int(self.prepared.corrector.stats['skipped'])
corrected = self.prepared.corrector(private, context)
if int(self.prepared.corrector.stats['skipped']) > skipped_before:
self._record_stage('failed')
raise IlluminationError(
f"segmentation field {field_id!r} has no illumination model; "
"an uncorrected field cannot be recorded as correction_depth=1. "
"Use illumination_on_missing='error' or estimate a model that "
'covers every segmentation plate.')
self._applied_fields.add(field_id)
self._application_state = 'running'
self._record_stage('running')
return corrected
[docs]
def mark_completed(self, field_id: str) -> bool:
"""Persist ``field_id`` after its corrected pipeline output is durable.
:param field_id: identifier of a field already passed to
:meth:`correct`, compared as a string; an uncorrected field raises
:class:`IlluminationError`.
:returns: ``True`` when the record changed, ``False`` when the same
completed field was marked again.
"""
field_id = str(field_id)
if field_id in self._completed_fields:
return False
if field_id not in self._applied_fields:
raise IlluminationError(
f"cannot mark segmentation field {field_id!r} complete before "
f"its illumination correction was applied.")
completed = set(self._completed_fields)
completed.add(field_id)
self._write_provenance(completed, 'running')
self._completed_fields = completed
self._application_state = 'running'
self._record_stage('running')
return True
[docs]
def finish(self, expected_fields: Iterable[str]) -> None:
"""Mark the journal stage done after every expected field is durable.
:param expected_fields: identifiers of every field the run should have
completed; the set, compared as strings, must equal the completed
set exactly or :class:`IlluminationError` is raised.
"""
expected = {str(field_id) for field_id in expected_fields}
if expected != self._completed_fields:
missing = sorted(expected - self._completed_fields)
extra = sorted(self._completed_fields - expected)
raise IlluminationError(
'cannot finish segmentation illumination provenance: '
f'missing={missing}, unexpected={extra}.')
self._write_provenance(self._completed_fields, 'complete')
self._application_state = 'complete'
self._record_stage('done')
def _record(self, completed_fields: Iterable[str],
application_state: str) -> Dict[str, Any]:
"""Build the provenance record for a set of completed fields.
:param completed_fields: the fields corrected so far.
:param application_state: where the session has got to.
:returns: the record, ready to serialise.
"""
return _segmentation_application_record(
self.prepared,
self.provenance_path,
self.pipeline_style,
completed_fields,
application_state,
)
def _load_completed_fields(self) -> None:
"""Read back which fields a previous session already corrected.
The record is validated against this session's model and pipeline style,
so a resume cannot silently adopt another run's completions.
"""
existing, completed = _read_segmentation_application(
self.prepared, self.provenance_path, self.pipeline_style)
self._completed_fields = completed
self._application_state = existing['application_state']
def _write_provenance(self, completed_fields: Iterable[str],
application_state: str) -> None:
"""Write the provenance record atomically.
Written to a temporary file in the same directory, flushed and fsynced,
then renamed over the target -- a crash mid-write must not leave a
truncated record that reads as a valid claim about which fields were
corrected. The temporary file is removed on any failure.
:param completed_fields: the fields corrected so far.
:param application_state: where the session has got to.
"""
parent = os.path.dirname(self.provenance_path)
os.makedirs(parent, exist_ok=True)
descriptor, temporary = tempfile.mkstemp(
prefix='.segmentation_application_', suffix='.json', dir=parent)
try:
with os.fdopen(descriptor, 'w', encoding='utf-8') as handle:
json.dump(self._record(completed_fields, application_state), handle,
indent=2, sort_keys=True)
handle.write('\n')
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, self.provenance_path)
except BaseException:
try:
os.remove(temporary)
except OSError:
pass
raise
def _record_stage(self, state: str) -> None:
"""Report this stage to the run journal, if a run is open.
Every failure is swallowed: provenance must not replace a scientific
result or the error that would otherwise have been raised.
:param state: the stage state to record.
"""
try:
from .run_journal import current_run
run = current_run()
if run is None:
return
run._record_stage(
self.STAGE_ID,
label=self.STAGE_LABEL,
state=state,
metrics={
'applied': bool(
self._applied_fields or self._completed_fields),
'application_state': self._application_state,
'model_sha256': self.prepared.model_sha256,
'pipeline_style': self.pipeline_style,
'source_intensity_state': 'raw',
'target_scope': 'segmentation-input-only',
'correction_depth': 1,
'raw_persisted_intensities_modified': False,
'fields_corrected_once': len(self._completed_fields),
'duplicate_attempts': self._duplicate_attempts,
'qc_artifacts': list(self.prepared.qc_artifacts),
},
)
except Exception:
return
def _env_entries(value: str) -> list:
"""Split a ``SPACR_MEASURE_HOOKS`` value into its non-empty entries."""
return [item.strip() for item in str(value or '').split(',') if item.strip()]
[docs]
def install() -> str:
"""Install the correction in **this** process from the environment.
This is the zero-argument installer ``SPACR_MEASURE_HOOKS`` names, and the
only route that survives a ``spawn`` / ``forkserver`` worker: the worker is
a fresh interpreter, so it imports this module and calls this function for
itself, reading the model from :data:`MODEL_ENV_VAR`.
:returns: the registry key the hook was registered under.
:raises IlluminationError: if the environment does not name a readable
model. Refusing loudly is the point -- a worker that cannot load the
model must not go on to measure uncorrected pixels.
"""
path = os.environ.get(MODEL_ENV_VAR, '').strip()
if not path:
raise IlluminationError(
f"{INSTALLER_ENTRY} is in {HOOKS_ENV_VAR} but {MODEL_ENV_VAR} is "
f"not set, so there is no illumination model to install. Call "
f"spacr.illumination.enable_illumination_correction(path), which "
f"sets both.")
model = IlluminationModel.load(path)
on_missing = os.environ.get(ON_MISSING_ENV_VAR, 'error').strip() or 'error'
corrector = IlluminationCorrector(model, on_missing=on_missing)
return register_preprocessing_hook(corrector, name=HOOK_NAME,
priority=HOOK_PRIORITY)
[docs]
def enable_illumination_correction(model, *, path: Optional[str] = None,
on_missing: str = 'error',
verbose: bool = True) -> str:
"""Turn the correction on, here and in every worker process.
Three things happen, and the third is the one that matters:
1. the model is saved to disk if it is not already there -- a ``spawn``
worker can only reach it through the file system;
2. :data:`MODEL_ENV_VAR` and :data:`ON_MISSING_ENV_VAR` are set;
3. ``spacr.illumination:install`` is appended to ``SPACR_MEASURE_HOOKS``
(appended, not assigned -- another extension may already be in there),
and the registry is then consulted so that this process installs the
hook through that same environment route.
:param model: an :class:`IlluminationModel`, or the path to a saved one.
:param path: where to save ``model`` when it is not already a path.
Defaults to ``<first src>/../illumination/illumination_model.npz``.
:param on_missing: ``'error'`` or ``'skip'``; see
:class:`IlluminationCorrector`.
:param verbose: print what was enabled and whether workers will see it.
:returns: the registry key the hook is registered under.
:raises IlluminationError: when the model cannot be saved or loaded.
"""
if isinstance(model, (str, os.PathLike)):
model_path = os.path.abspath(str(model))
IlluminationModel.load(model_path)
else:
if path is None:
sources = model.meta.get('src') or []
base = (os.path.join(os.path.dirname(str(sources[0])),
'illumination')
if sources else os.path.join(os.getcwd(), 'illumination'))
path = os.path.join(base, 'illumination_model.npz')
model_path = model.save(path)
os.environ[MODEL_ENV_VAR] = model_path
os.environ[ON_MISSING_ENV_VAR] = on_missing
entries = _env_entries(os.environ.get(HOOKS_ENV_VAR, ''))
if INSTALLER_ENTRY not in entries:
entries.append(INSTALLER_ENTRY)
os.environ[HOOKS_ENV_VAR] = ','.join(entries)
registered = [entry.name for entry in preprocessing_hooks()]
if HOOK_NAME not in registered:
install()
if verbose:
ok, message = worker_delivery_status()
print(f"illumination correction ENABLED from {model_path}")
print((' ' if ok else ' WARNING: ') + message)
return HOOK_NAME
[docs]
def disable_illumination_correction() -> bool:
"""Turn the correction off, here and for any worker started afterwards.
Unregisters the hook and removes this module's entry from
``SPACR_MEASURE_HOOKS`` -- leaving any other extension's entries alone --
plus the two model variables.
:returns: True if a correction was registered and has been removed.
"""
removed = unregister_preprocessing_hook(HOOK_NAME)
entries = [item for item in _env_entries(os.environ.get(HOOKS_ENV_VAR, ''))
if item != INSTALLER_ENTRY]
if entries:
os.environ[HOOKS_ENV_VAR] = ','.join(entries)
else:
os.environ.pop(HOOKS_ENV_VAR, None)
os.environ.pop(MODEL_ENV_VAR, None)
os.environ.pop(ON_MISSING_ENV_VAR, None)
return removed
[docs]
def worker_delivery_status(start_method: Optional[str] = None
) -> Tuple[bool, str]:
"""Whether the correction will actually reach ``measure_crop``'s workers.
The failure this answers is silent by construction: a hook registered only
in the parent process is a no-op in every ``spawn`` worker, the run
completes, and the numbers are uncorrected while the user believes they
are not.
:param start_method: the pool start method to judge against. Defaults to
whatever ``SPACR_START_METHOD`` selects, falling back to the platform
default -- i.e. what :func:`spacr.measure.measure_crop` will use.
:returns: ``(ok, message)``. ``ok`` is False whenever a field could be
measured uncorrected without anything saying so.
"""
if start_method is None:
import multiprocessing as mp
start_method = (os.environ.get('SPACR_START_METHOD', '').strip().lower()
or mp.get_start_method())
registered = {entry.name: entry for entry in preprocessing_hooks()}
entry = registered.get(HOOK_NAME)
if entry is None:
return False, ('illumination correction is not registered in this '
'process; measure_crop will measure uncorrected '
'pixels.')
in_env = INSTALLER_ENTRY in _env_entries(os.environ.get(HOOKS_ENV_VAR, ''))
model = os.environ.get(MODEL_ENV_VAR, '').strip()
if in_env and model and os.path.isfile(model):
return True, (f"workers install it themselves from {HOOKS_ENV_VAR}="
f"'{INSTALLER_ENTRY}' and {MODEL_ENV_VAR}='{model}', so "
f"a '{start_method}' pool is covered.")
if start_method == 'fork':
return True, (f"a 'fork' pool inherits this process's registry, so the "
f"correction reaches the workers -- but {HOOKS_ENV_VAR} "
f"does not name {INSTALLER_ENTRY}, so it would NOT "
f"survive SPACR_START_METHOD=spawn.")
return False, (f"the correction is registered in this process only and the "
f"pool starts workers with '{start_method}', which does not "
f"inherit it: every field would be measured uncorrected. "
f"Call enable_illumination_correction(), which sets "
f"{HOOKS_ENV_VAR} and {MODEL_ENV_VAR}.")
[docs]
def position_intensity_slope(intensities: Sequence[float],
coordinates: np.ndarray,
shape: Tuple[int, int]) -> float:
"""Least-squares slope of intensity against distance from the field centre.
This is the number illumination correction exists to drive to zero, and
the same statistic is used on pixels (the QC images) and on objects (the
scientific claim: the same cell must not measure brighter in the middle of
the field).
Intensities are divided by their own mean and the radius is normalised so
that 0 is the centre of the field and 1 is a corner, so the slope reads as
*the fraction of the mean intensity gained or lost between the centre of
the field and its corner*. -0.30 means a corner object measures 30 % of
the mean below a central one.
:param intensities: one value per position.
:param coordinates: ``(N, 2)`` array of ``(row, col)`` positions in pixels.
:param shape: ``(Y, X)`` shape of the field the positions live in.
:returns: the slope, or 0.0 when there is nothing to fit.
"""
values = np.asarray(intensities, dtype=np.float64).ravel()
coordinates = np.asarray(coordinates, dtype=np.float64).reshape(-1, 2)
if values.size < 2 or values.size != len(coordinates):
return 0.0
mean = values.mean()
if not np.isfinite(mean) or mean == 0:
return 0.0
height, width = shape
rows = (coordinates[:, 0] - (height - 1) / 2.0) / max((height - 1) / 2.0, 1e-9)
cols = (coordinates[:, 1] - (width - 1) / 2.0) / max((width - 1) / 2.0, 1e-9)
radius = np.sqrt(rows ** 2 + cols ** 2) / math.sqrt(2.0)
design = np.stack([np.ones_like(radius), radius], axis=1)
solution, *_ = np.linalg.lstsq(design, values / mean, rcond=None)
return float(solution[1])
def _image_slope(image: np.ndarray, factor: int,
shape: Tuple[int, int]) -> float:
""":func:`position_intensity_slope` over every binned pixel of ``image``."""
rows = _bin_centres(shape[0], factor)[:image.shape[0]]
cols = _bin_centres(shape[1], factor)[:image.shape[1]]
grid_rows, grid_cols = np.meshgrid(rows, cols, indexing='ij')
coordinates = np.stack([grid_rows.ravel(), grid_cols.ravel()], axis=1)
return position_intensity_slope(image.ravel(), coordinates, shape)
[docs]
def illumination_qc(model: IlluminationModel, src, *,
channels: Optional[Sequence[int]] = None,
save_dir: Optional[str] = None,
max_fields: int = 25,
verbose: bool = True,
stage: Optional[str] = None) -> Dict[str, Any]:
"""Show that the correction worked, and say by how much.
Three things, per plate and per channel:
* **the estimated field as an image**, so a lamp profile that is really a
dirty objective is visible rather than inferred;
* **the position-versus-intensity trend before and after**, measured on
the same fields the estimate came from -- the curve that should be flat
after correction and is not before it;
* **a number**: the residual slope of intensity against distance from the
centre of the field, before and after, and the percentage of that bias
the correction removed.
:param model: the estimated model.
:param src: the merged field folder(s) to measure the trend on.
:param channels: channels to report; defaults to the model's own.
:param save_dir: where the figure goes. Defaults to
``<first src>/../illumination``. Pass ``''`` to skip the figure and
compute only the numbers.
:param max_fields: fields per plate to measure the trend over.
:param verbose: print the per-channel summary.
:param stage: optional consumer label such as ``'segmentation_input'``.
When supplied it appears in the figure title and filename, preventing
segmentation and measurement QC artifacts from being confused.
:returns: ``{plate: {channel: {...metrics...}}}`` with, per channel,
``slope_before``, ``slope_after``, ``bias_removed_pct``,
``nonuniformity_pct``, ``gain_min``, ``gain_max`` and ``n_fields``;
plus, when a figure was written, one extra key ``'_figures'`` mapping
each plate to the path of its PNG.
"""
grouped = _merged_files(src)
if model.per_plate:
plates = {plate: paths for plate, paths in grouped.items()
if plate in model.fields}
else:
plates = {ALL_PLATES: [path for paths in grouped.values()
for path in paths]}
if save_dir is None:
sources = _source_folders(src)
save_dir = os.path.join(os.path.dirname(sources[0]), 'illumination')
report: Dict[str, Any] = {}
stage = str(stage or '').strip() or None
if stage is not None:
report['_stage'] = stage
for plate in sorted(plates):
item = model.field_for(plate)
wanted = [int(c) for c in (channels if channels is not None
else item.channels)]
paths = _sample(sorted(plates[plate]), max_fields)
stack = []
factor = item.bin_size
for path in paths:
binned, _factor, shape = _read_binned_field(path, wanted, factor,
256)
if shape == item.shape:
stack.append(binned)
if not stack:
continue
stack = np.stack(stack, axis=0)
metrics = {}
panels = {}
for position, channel in enumerate(wanted):
observed = _relative_profile(stack[:, position])
gain_full = 1.0 / item.flatfield[item.index_of(channel)]
gain_binned = _bin_image(gain_full, factor)
corrected = observed * gain_binned[:observed.shape[0],
:observed.shape[1]]
before = _image_slope(observed, factor, item.shape)
after = _image_slope(corrected, factor, item.shape)
removed = (100.0 * (1.0 - abs(after) / abs(before))
if abs(before) > 1e-12 else 0.0)
plane = item.flatfield[item.index_of(channel)]
low, high = np.percentile(plane, [2, 98])
metrics[int(channel)] = {
'slope_before': before,
'slope_after': after,
'bias_removed_pct': removed,
'nonuniformity_pct': float(100 * (high - low) /
max(plane.mean(), 1e-12)),
'gain_min': float(plane.min()),
'gain_max': float(plane.max()),
'n_fields': int(stack.shape[0]),
}
panels[int(channel)] = (plane, observed, corrected)
if verbose:
print(f"plate {plate}, channel {channel}: "
f"position-intensity slope {before:+.4f} -> "
f"{after:+.4f} per corner-radius "
f"({removed:.1f}% of the bias removed), field "
f"non-uniformity "
f"{metrics[int(channel)]['nonuniformity_pct']:.1f}%")
report[plate] = metrics
if save_dir:
report.setdefault('_figures', {})[plate] = _write_qc_figure(
plate, item, wanted, panels, metrics, save_dir, factor,
stage=stage)
return report
def _write_qc_figure(plate, item, channels, panels, metrics, save_dir,
factor, *, stage: Optional[str] = None) -> str:
"""Render one figure per plate: field, trend before/after, residual.
Uses the object-oriented matplotlib API rather than pyplot: this can be
called from a worker or a headless run, and pyplot's global figure
registry is a leak in both.
"""
from matplotlib.figure import Figure
os.makedirs(save_dir, exist_ok=True)
rows = max(len(channels), 1)
figure = Figure(figsize=(13, 3.4 * rows), dpi=120)
from .figures.bundle import _register_figure_data
_register_figure_data(figure, lambda: [np.asarray(panels[int(c)][0]) for c in channels], kind="image", title=str(stage or "Illumination correction"))
if stage:
figure.suptitle(f'Illumination correction — {stage.replace("_", " ")}')
for index, channel in enumerate(channels):
plane, observed, corrected = panels[int(channel)]
stats = metrics[int(channel)]
axis = figure.add_subplot(rows, 3, index * 3 + 1)
image = axis.imshow(plane, cmap='viridis')
axis.set_title(f'plate {plate} ch{channel}: estimated field\n'
f'gain {stats["gain_min"]:.2f}-{stats["gain_max"]:.2f}, '
f'non-uniformity {stats["nonuniformity_pct"]:.1f}%',
fontsize=9)
axis.set_xticks([])
axis.set_yticks([])
figure.colorbar(image, ax=axis, fraction=0.046)
axis = figure.add_subplot(rows, 3, index * 3 + 2)
radius, before = _radial_profile(observed, factor, item.shape)
_, after = _radial_profile(corrected, factor, item.shape)
axis.plot(radius, before, 'o-', ms=3, label='before')
axis.plot(radius, after, 's-', ms=3, label='after')
axis.axhline(1.0, color='0.6', lw=0.8, ls='--')
axis.set_xlabel('distance from field centre (0 = centre, 1 = corner)',
fontsize=8)
axis.set_ylabel('mean intensity / field mean', fontsize=8)
axis.set_title(f'position-intensity trend\nslope '
f'{stats["slope_before"]:+.3f} -> '
f'{stats["slope_after"]:+.3f} '
f'({stats["bias_removed_pct"]:.1f}% removed)',
fontsize=9)
axis.legend(fontsize=8)
axis = figure.add_subplot(rows, 3, index * 3 + 3)
span = float(np.max(np.abs(observed - 1.0))) or 0.1
image = axis.imshow(corrected, cmap='coolwarm', vmin=1 - span,
vmax=1 + span)
axis.set_title('after correction (same scale as the\nobserved '
'deviation before it)', fontsize=9)
axis.set_xticks([])
axis.set_yticks([])
figure.colorbar(image, ax=axis, fraction=0.046)
figure.tight_layout(rect=(0, 0, 1, 0.97) if stage else None)
safe_stage = (''.join(character if character.isalnum() else '_'
for character in stage).strip('_')
if stage else '')
infix = f'{safe_stage}_' if safe_stage else ''
path = os.path.join(save_dir, f'illumination_qc_{infix}{plate}.png')
from .plot import save_figure
return save_figure(figure, path, fmt="png")
def _radial_profile(image: np.ndarray, factor: int,
shape: Tuple[int, int], bins: int = 20):
"""Mean of ``image`` in ``bins`` rings of normalised corner-radius."""
rows = _bin_centres(shape[0], factor)[:image.shape[0]]
cols = _bin_centres(shape[1], factor)[:image.shape[1]]
grid_rows, grid_cols = np.meshgrid(rows, cols, indexing='ij')
normalised_rows = ((grid_rows - (shape[0] - 1) / 2.0) /
max((shape[0] - 1) / 2.0, 1e-9))
normalised_cols = ((grid_cols - (shape[1] - 1) / 2.0) /
max((shape[1] - 1) / 2.0, 1e-9))
radius = np.sqrt(normalised_rows ** 2 + normalised_cols ** 2) / math.sqrt(2)
edges = np.linspace(0, radius.max() + 1e-9, bins + 1)
which = np.clip(np.digitize(radius.ravel(), edges) - 1, 0, bins - 1)
values = image.ravel() / max(float(image.mean()), 1e-12)
centres, means = [], []
for index in range(bins):
selected = values[which == index]
if selected.size:
centres.append(0.5 * (edges[index] + edges[index + 1]))
means.append(float(selected.mean()))
return np.asarray(centres), np.asarray(means)
[docs]
def prepare_illumination_model(
settings: Mapping[str, Any], *, src=None,
channels: Optional[Sequence[int]] = None,
qc_stage: Optional[str] = None,
verbose: Optional[bool] = None) -> Optional[PreparedIllumination]:
"""Prepare one reusable optical model without installing a Measure hook.
This is the direct consumer of the ``illumination_*`` settings. It
estimates, loads or reads a vendor flat-field profile into the model once, ensures a fitted model is saved, hashes
the exact saved bytes, optionally writes stage-labelled QC, and builds a
corrector. Applying that corrector belongs to the caller's stage.
:param settings: settings carrying the illumination controls. A
non-empty ``illumination_vendor_profile`` replaces the estimate with
the vendor's own flat field; ``illumination_vendor_channel_map``
optionally assigns persisted channels to vendor channels/planes.
``illumination_model`` still wins over both.
:param src: optional raw field folder override. Defaults to ``settings['src']``.
:param channels: optional persisted intensity-axis positions. Defaults to
``settings['channels']``.
:param qc_stage: optional stage label included in QC filenames/titles.
:param verbose: override ``settings['verbose']``.
:returns: a prepared model/corrector, or ``None`` when correction is off.
"""
talk = settings.get('verbose', True) if verbose is None else verbose
if not settings.get('illumination_correction', False):
if talk:
print("illumination correction is OFF (illumination_correction "
"is False), so no field was estimated and every intensity "
"feature keeps its position-dependent bias.")
return None
source = src if src is not None else settings.get('src')
if not source:
raise IlluminationError(
"illumination_correction is on but settings['src'] is empty; "
"there is nothing to estimate the illumination field from.")
wanted = (list(channels) if channels is not None
else list(settings.get('channels') or []))
folder = os.path.join(
os.path.dirname(_source_folders(source)[0]), 'illumination')
existing = str(settings.get('illumination_model', '') or '').strip()
vendor = str(settings.get('illumination_vendor_profile', '') or '').strip()
if existing:
model = IlluminationModel.load(existing)
model_path = os.path.abspath(existing)
elif vendor:
model = _vendor_illumination(
vendor, wanted,
dark=float(settings.get('illumination_dark', 0.0)),
channel_map=settings.get('illumination_vendor_channel_map', ''),
verbose=talk)
model_path = os.path.join(folder, 'illumination_model.npz')
else:
model = estimate_illumination(
source,
channels=wanted,
per_plate=bool(settings.get('illumination_per_plate', True)),
estimator=str(settings.get('illumination_estimator', 'polynomial')),
degree=int(settings.get('illumination_degree', 4)),
max_fields=int(settings.get('illumination_max_fields', 50)),
dark=float(settings.get('illumination_dark', 0.0)),
verbose=talk)
model_path = os.path.join(folder, 'illumination_model.npz')
qc_artifacts = ()
if settings.get('illumination_qc', True):
report = illumination_qc(
model, source, channels=wanted, save_dir=folder, verbose=talk,
stage=qc_stage)
qc_artifacts = tuple(sorted(
str(path) for path in report.get('_figures', {}).values()))
if not existing:
model_path = model.save(model_path)
on_missing = str(settings.get('illumination_on_missing', 'error'))
corrector = IlluminationCorrector(
model, on_missing=on_missing, verbose=talk)
return PreparedIllumination(
model=model,
corrector=corrector,
model_path=model_path,
model_sha256=_file_sha256(model_path),
qc_artifacts=qc_artifacts,
)
[docs]
def prepare_illumination_correction(settings: Mapping[str, Any], *,
verbose: Optional[bool] = None):
"""Estimate, save, enable and QC the Measure correction.
The one call a pipeline makes before ``measure_crop``, and the one the
Illumination button on the Measure masthead runs on its own -- the model
and its QC figures in minutes, before committing hours to the measure run
that will reuse them through ``illumination_model``.
It does nothing at all -- and returns None -- unless
``settings['illumination_correction']`` is True, which is NOT the shipped
default: the correction is opt-in, so a run that never mentions it is
measured uncorrected. Being asked to run with the switch off is a
no-op worth hearing about rather than a silent one, because from a
settings form it looks exactly like a Run button that does nothing, so a
verbose call says which switch was not thrown.
:param settings: a ``measure_crop`` settings dict. Reads
``illumination_correction``, ``illumination_model``,
``illumination_estimator``, ``illumination_degree``,
``illumination_per_plate``, ``illumination_max_fields``,
``illumination_dark``, ``illumination_on_missing``,
``illumination_qc``, ``illumination_vendor_profile``,
``illumination_vendor_channel_map``, plus ``src`` and ``channels``.
:param verbose: overrides ``settings['verbose']``.
:returns: the :class:`IlluminationModel` that was enabled, or None.
"""
talk = settings.get('verbose', True) if verbose is None else verbose
prepared = prepare_illumination_model(settings, verbose=verbose)
if prepared is None:
return None
enable_illumination_correction(
prepared.model_path,
on_missing=str(settings.get('illumination_on_missing', 'error')),
verbose=talk)
return prepared.model
[docs]
def prepare_segmentation_illumination(
settings: Mapping[str, Any], *, src=None,
channels: Optional[Sequence[int]] = None,
pipeline_style: str,
verbose: Optional[bool] = None
) -> Optional[SegmentationIlluminationSession]:
"""Prepare correction for segmentation inputs without changing raw data.
The returned session is not a Measure hook. V1/V2 adapters hand it raw
field copies, then explicitly mark fields complete after their durable
segmentation output exists. The application record is stored beside the
current run's source folder, even when the optical model is shared from an
external path, so two runs cannot overwrite one another's completion set.
:param settings: run settings carrying the illumination controls read by
:func:`prepare_illumination_model`, and ``src`` when ``src`` is not
given.
:param pipeline_style: segmentation pipeline style, ``'v1'`` or ``'v2'``
(case-insensitive); any other value raises :class:`IlluminationError`.
"""
prepared = prepare_illumination_model(
settings, src=src, channels=channels,
qc_stage='segmentation_input', verbose=verbose)
if prepared is None:
return None
source = src if src is not None else settings.get('src')
run_root = os.path.dirname(_source_folders(source)[0])
provenance_path = os.path.join(
run_root, 'illumination',
'segmentation_application.json')
return SegmentationIlluminationSession(
prepared,
provenance_path=provenance_path,
pipeline_style=pipeline_style,
)
[docs]
def illumination_settings(settings=None):
"""Defaults for illumination correction. Registered through the seam.
:param settings: values to seed, exactly like every ``set_default_*`` in
:mod:`spacr.settings`.
:returns: the settings dict with the illumination defaults applied.
"""
settings = dict(settings or {})
settings.setdefault('src', '')
settings.setdefault('channels', [0, 1, 2])
settings.setdefault('illumination_correction', False)
settings.setdefault('illumination_model', '')
settings.setdefault('illumination_estimator', 'polynomial')
settings.setdefault('illumination_degree', 4)
settings.setdefault('illumination_per_plate', True)
settings.setdefault('illumination_max_fields', 50)
settings.setdefault('illumination_dark', 0.0)
settings.setdefault('illumination_on_missing', 'error')
settings.setdefault('illumination_qc', True)
settings.setdefault('illumination_vendor_profile', '')
settings.setdefault('illumination_vendor_channel_map', '')
return settings
_TOOLTIPS = {
'illumination_correction': (
'(bool) - Estimate the uneven illumination of the microscope from the '
'fields themselves and divide it out of the pixels at every '
'enabled stage. Off by default. On, the same cell looks and measures '
'the same wherever it sits in the field of view, which is what '
'removes the position-dependent bias behind plate edge effects. '
'Default False.'),
'illumination_model': (
'(str) - Path to an illumination model saved earlier. Empty means '
'estimate a fresh one from the fields in src, which is what you want '
'unless you are re-measuring a plate and must reproduce the exact '
'correction an earlier run applied. Default empty.'),
'illumination_estimator': (
"(str) - How the smooth field is fitted to the across-field median: "
"'polynomial' fits a low-order surface, which cannot bend around a "
"cell and is the right choice for a lamp profile plus a vignette; "
"'smooth' Gaussian-blurs the median instead and can follow a dust "
"shadow the polynomial would miss. Default polynomial."),
'illumination_degree': (
'(int) - Order of the fitted illumination surface. 4 gives fifteen '
'terms, enough for a lamp profile, a vignette and a tilt. Raising it '
'lets the surface follow finer structure and, past about 6, start '
'absorbing the cells you are trying to measure. Default 4.'),
'illumination_per_plate': (
'(bool) - Estimate one illumination field per plate rather than one '
'for every plate together. Lamp age, a re-seated filter cube or a '
'different objective change the field between acquisition sessions, '
'so pooling two sessions estimates neither of them well. Default True.'),
'illumination_max_fields': (
'(int) - How many fields per plate the estimate reads, sampled evenly '
'across the plate. More fields make the across-field median a better '
'object rejector and cost linear time; below about ten, cells start '
'surviving into the gain map. Default 50.'),
'illumination_dark': (
'(float) - Camera dark offset in raw counts, subtracted before the '
'gain is applied. Leave at zero unless you measured it from a dark '
'frame: it is not identifiable from the images themselves, and '
'an estimated value can subtract genuine background signal. Default 0.0.'),
'illumination_on_missing': (
"(str) - What to do with a field whose plate the model does not "
"cover: 'error' fails that field and stamps the run incomplete, "
"'skip' measures it uncorrected. Default error, because corrected and "
"uncorrected rows sharing one table is worse than a failed field."),
'illumination_qc': (
'(bool) - Write the QC figure beside the model: the estimated field '
'as an image, the intensity-versus-position trend before and after, '
'and the percentage of the position bias the correction removed. '
'The figure has low computational cost and provides direct '
'verification of the correction. Default True.'),
'illumination_vendor_channel_map': (
'(str) - Assign saved intensity channels to a vendor flat-field calibration. '
'Example 0:2,1:1 assigns intensity channels 0 and 1 to calibration channels 2 and 1. '
'Intensity channel numbers start at 0. Calibration channel or image-layer '
'numbers start at 1. Include every corrected intensity channel. '
'An empty value keeps the original channel order. With an empty value, '
'a single calibration image layer applies to all channels. Select a vendor calibration '
'file first. A saved illumination model takes priority. '
'Default empty.'),
'illumination_vendor_profile': (
'(str) - Path to the flat-field correction the microscope software '
'saved, used instead of estimating one from the fields: a Harmony '
'Index.idx.xml or FFC profile (Operetta, Opera Phenix), or a shading '
'reference image (.tif, .czi, .npy) from ZEN, Nikon or Olympus. '
'Harmony profiles are applied at Harmony\'s own scale, so the '
'corrected pixels match Harmony\'s corrected export. Empty means '
'estimate. Default empty.'),
}
_TYPES = {
'illumination_correction': bool,
'illumination_model': str,
'illumination_estimator': str,
'illumination_degree': int,
'illumination_per_plate': bool,
'illumination_max_fields': int,
'illumination_dark': float,
'illumination_on_missing': str,
'illumination_qc': bool,
'illumination_vendor_profile': str,
'illumination_vendor_channel_map': str,
}
_DESCRIPTION = (
'Illumination / flat-field correction. Estimates the microscope\'s '
'uneven illumination once from the fields of a plate and divides it '
'out of the pixels at every enabled stage, so the same cell looks '
'and measures the same wherever it sat in the field of view.'
)
[docs]
def register_illumination_settings(replace: bool = False) -> bool:
"""Register the illumination settings through the defaults seam.
Called once at import. Uses
:func:`spacr.settings.register_defaults` rather than appending to
``spacr/settings.py``, so this module owns its own knobs.
Types, tooltips and the module description are contributed; **categories
deliberately are not**. ``spacr.settings.categories`` is one shared,
ordered map that every settings panel walks, and its growth is guarded by
an exact-equality test against a hand-kept list. A key contributed at
*import* time is in that map only in a session that imported this module,
so contributing categories would make that test's result depend on which
files pytest was pointed at.
The illumination keys ARE filed under a heading -- "Illumination Correction" in
``spacr.settings.categories`` -- and Measure's panel offers every one of
them, because ``measure_crop`` calls
:func:`prepare_illumination_correction` itself and these are the keys
that call reads. The heading is written in that map by hand for the
reason above: it has to exist for every process that groups the Measure
settings, not only for one that happened to import this module.
:param replace: re-register over an existing registration.
:returns: True if it registered, False if it was already registered.
"""
from .settings import has_registered_defaults, register_defaults
if has_registered_defaults(APP_KEY) and not replace:
return False
register_defaults(
APP_KEY, illumination_settings, replace=replace,
expected_types=_TYPES, tooltips=_TOOLTIPS,
description=_DESCRIPTION)
return True
register_illumination_settings()