"""Import externally generated masks and finish a spaCR Measure project.
This is the entry point for segmentation performed outside spaCR. It accepts
one or more mixed folders/files, detects intensity images and label images,
lets callers override every proposed role/object type, builds the canonical
``stack/``, ``masks/`` and ``merged/`` folders, and then delegates all feature
extraction and crop generation to :func:`spacr.measure.measure_crop`.
The output is therefore the same contract Annotate expects from Measure::
destination/
data/.../<object>_png/*.png
images/*.tif
stack/*.npy
masks/<object>_mask_stack/*.npy
merged/*.npy
measurements/measurements.db
Only object types supplied by the user are measured. ``cytoplasm`` is
derived by Measure from a cell mask and any supplied interior masks; it is
not an input mask plane.
"""
from __future__ import annotations
from .schema import object_type_summary
import json
import os
import re
import sqlite3
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
import numpy as np
class _LazyConvert:
"""``spacr.convert``, imported on first attribute access.
THIS MODULE IS ON THE STARTUP PATH, three links down: the settings model
imports the external-mask widget, which imports this, and `convert`
imports pandas. The packaged smoke test asserts Home crosses no
operation-only import boundary and was failing on pandas for exactly that
chain.
A proxy rather than sixteen function-level imports, so every existing
``cv.something`` reads the same. Safe because this module declares
``from __future__ import annotations``: the annotations that name
``cv.ConversionPlan`` are strings and never evaluated at import.
"""
__slots__ = ()
def __getattr__(self, name: str):
"""Forward to :mod:`spacr.convert`, imported on first use."""
from . import convert
return getattr(convert, name)
cv = _LazyConvert()
from . import crops
from .errors import ConfigurationError
from .object_roles import ORGANELLE_ROLES
SUPPORTED_SUFFIXES = (
".tif", ".tiff", ".ome.tif", ".ome.tiff",
".png", ".jpg", ".jpeg", ".bmp",
)
OBJECT_TYPES = tuple(crops.MASK_PLANE_ORDER)
ROLES = ("image", "mask", "ignore")
#: What may sit either side of a name token. Path separators count, so a
#: folder named only ``organelle_2/`` or ``cell_masks/`` (with plain
#: ``fov001.tif`` files inside, or added as a subfolder of a project) names
#: its object type. Before 2026-09-30 (item 76) only ``_-. `` and a space
#: did, and such folders were proposed as nothing, or as the wrong slot.
_B = r"[_\-. /\\]"
_MASK_WORDS = re.compile(
rf"(?i)(?:^|{_B})(?:mask|masks|label|labels|labelled|labeled|"
rf"instance|instances|seg|segmentation|outline)(?:$|{_B})"
)
_ORGANELLE_PATTERN = re.compile(
rf"(?i)(?:^|{_B})"
r"(?:organelle(?:[_\-. ]?(?P<number>\d+)|(?P<letters>[a-z]+))?"
r"|organell|mitochondria|mitochondrion)"
rf"(?={_B}|$)"
)
#: Excitation laser lines. ``organelle_488_masks`` names the 488 nm channel,
#: not Organelle 488; such a token counts as a bare ``organelle`` (item 76,
#: 2026-09-30). Only these numbers are read as wavelengths, so every other
#: slot number from 1 to 702 still names its slot.
_LASER_LINES = frozenset({
355, 375, 405, 440, 445, 457, 458, 473, 488, 491, 505, 514, 515, 532,
543, 552, 555, 561, 568, 588, 594, 633, 635, 637, 638, 640, 642, 647,
660, 685, 730, 750, 785,
})
_ORGANELLE_TAIL = re.compile(
r"(?i)(?:^|[_\-. ])"
r"(?:organelle(?:[_\-. ]?\d+|[a-z]+)?|organell|mitochondria|mitochondrion)$"
)
_ORGANELLE_ROLE_SET = frozenset(ORGANELLE_ROLES)
_ORGANELLE_PLURAL = "organelles"
_OBJECT_PATTERNS = (
("nucleus", re.compile(rf"(?i)(?:^|{_B})(?:nucleus|nuclei|nuclear|nuc)(?:$|{_B})")),
("pathogen", re.compile(rf"(?i)(?:^|{_B})(?:pathogen|parasite|bacteria|bacterial)(?:$|{_B})")),
("organelle", _ORGANELLE_PATTERN),
("cell", re.compile(rf"(?i)(?:^|{_B})(?:cell|cells|wholecell|cytoplasm)(?:$|{_B})")),
)
@dataclass
@dataclass(frozen=True)
[docs]
class MaskMatch:
"""One label-mask file matched to an intensity-image field.
:param path: absolute path of the matched source label-mask file; the
importer reads this file and reports it in validation errors.
:param object_type: valid spaCR object role assigned to the mask group,
such as ``cell`` or ``nucleus``.
:param stem: canonical ``<plate>_<well>_<field>`` stem of the intensity
field to which the mask was paired.
:param match: pairing rule that succeeded, currently ``"exact"`` or
``"normalised"`` after stripping mask/object-role suffixes.
"""
path: str
object_type: str
stem: str
match: str
@dataclass
[docs]
class ExternalMaskPlan:
"""Read-only preview of an external-mask import.
:param groups: normalized image, mask, and ignored input groups reviewed for
this import.
:param images: proposed intensity-image conversion plan.
:param masks: spaCR object types mapped to canonical field stems and their
matched label-mask files.
:param destination: root of the spaCR project that would be written.
:param n_channels: number of intensity-image channels in each generated
merged field.
:param mask_dims: merged-array plane index assigned to each supplied mask
type.
:param errors: blocking problems that make the preview unrunnable.
:param warnings: non-blocking ambiguities shown before the import proceeds.
"""
groups: List[InputGroup]
images: cv.ConversionPlan
masks: Dict[str, Dict[str, MaskMatch]]
destination: str
n_channels: int
mask_dims: Dict[str, int]
errors: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
@property
[docs]
def stems(self) -> List[str]:
"""Return sorted field stems covered by every supplied mask type."""
return sorted(set.intersection(
*(set(per_stem) for per_stem in self.masks.values())
)) if self.masks else []
@property
[docs]
def object_types(self) -> List[str]:
"""Return supplied mask roles in canonical object-type order."""
return [name for name in OBJECT_TYPES if name in self.masks]
@property
[docs]
def ok(self) -> bool:
"""Return whether images, shared fields, and importer checks are valid."""
return bool(self.images.ok and self.stems and not self.errors)
[docs]
def summary(self) -> str:
"""Return a multiline read-only preview of mappings and problems."""
lines = [
"External masks → Measure project (preview; nothing written)",
f" intensity mappings: {len(self.images)}",
f" fields ready: {len(self.stems)}",
f" intensity channels: {self.n_channels}",
f" mask types: {object_type_summary(self.object_types) or 'none'}",
f" destination: {self.destination}",
]
for name in self.object_types:
lines.append(f" {name}: {len(self.masks[name])} paired mask(s), "
f"merged plane {self.mask_dims[name]}")
if self.warnings:
lines.append("Warnings:")
lines.extend(f" - {message}" for message in self.warnings)
if self.errors or self.images.errors:
lines.append("Blocking problems:")
lines.extend(f" - {message}"
for message in [*self.images.errors, *self.errors])
return "\n".join(lines)
@dataclass
[docs]
class ExternalMaskResult:
"""Files and database produced by :func:`prepare_external_masks`.
:param destination: root of the generated spaCR project.
:param merged: paths of generated merged image-and-mask arrays.
:param db_path: path of the generated measurements database.
:param tables: measurement-table names written to that database.
:param data_dir: generated annotation-crop directory.
:param plan: validated read-only import plan used to produce these outputs.
"""
destination: str
merged: List[str]
db_path: str
tables: List[str]
data_dir: str
plan: ExternalMaskPlan
[docs]
def summary(self) -> str:
"""Return one line naming materialized fields and output locations."""
return (
f"Prepared {len(self.merged)} field(s) in {self.destination}. "
f"measurements.db tables: {', '.join(self.tables) or 'none'}. "
f"Annotation crops: {self.data_dir}"
)
def _all_files(path: Path, recursive: bool) -> List[Path]:
"""Collect supported files from a file or directory input.
:param path: input file or directory; missing paths produce no files.
:param recursive: descend through directory children when true.
"""
if path.is_file():
return [path] if _supported(path) else []
if not path.is_dir():
return []
iterator = path.rglob("*") if recursive else path.glob("*")
return sorted(candidate for candidate in iterator
if candidate.is_file() and _supported(candidate))
def _supported(path: Path) -> bool:
"""Return whether a path has a supported suffix, case-insensitively."""
name = path.name.lower()
return any(name.endswith(suffix) for suffix in SUPPORTED_SUFFIXES)
def _suggest_object(name: str) -> Optional[str]:
"""Infer the first matching spaCR object role from path-name tokens.
:param name: filename or path whose stem and parent tokens are inspected.
:returns: canonical object type, or ``None`` when no pattern matches.
"""
stem = cv._split_ext(str(name))[0]
for object_type, pattern in _OBJECT_PATTERNS:
if pattern is _ORGANELLE_PATTERN:
slot = _suggest_organelle_slot(stem)
if slot is not None:
return slot
elif pattern.search(stem):
return object_type
return None
def _suggest_organelle_slot(stem: str) -> Optional[str]:
"""Name the organelle slot a filename token spells, for every slot.
A bare ``organelle`` (or ``organell``, ``mitochondria``,
``mitochondrion``) is slot 1. A number after it, with or without one
separator -- ``organelle_5``, ``organelle 12``, ``organelle5`` -- is the
slot as the user counts it, so ``organelle_5`` is ``organellee``. A
letter suffix is the role spelling spaCR itself writes, so a folder such
as ``organellee_mask_stack`` imports back into its own slot.
Every slot of :data:`spacr.object_roles.ORGANELLE_ROLES` is reachable,
rather than the first four a fixed table of patterns listed. A numbered
token that names no slot -- ``organelle_0``, ``organelle_2024`` -- and the
English plural ``organelles`` propose nothing, because neither says which
slot is meant; the object type stays editable in the import table. A
number that is a laser line (:data:`_LASER_LINES`, e.g. ``organelle_488``)
names a channel, so it counts as a bare token.
A token that names a slot outranks a bare one wherever each sits in the
path, so ``organelle_masks/fov001_organelle_2.tif`` and
``organelle_2/fov001_organelle_2_mask.tif`` are slot 2, not slot 1. A
bare token is slot 1 only when no token in the path names another.
:param stem: filename stem, optionally with parent-folder tokens.
:returns: an organelle role, or ``None`` when no token names a slot.
"""
bare = False
for match in _ORGANELLE_PATTERN.finditer(stem):
number = match.group("number")
letters = match.group("letters")
if number is not None:
index = int(number)
if index in _LASER_LINES:
bare = True
continue
if 1 <= index <= len(ORGANELLE_ROLES):
return ORGANELLE_ROLES[index - 1]
continue
if letters is not None:
role = f"organelle{letters.lower()}"
if role in _ORGANELLE_ROLE_SET and role != _ORGANELLE_PLURAL:
return role
continue
bare = True
return ORGANELLE_ROLES[0] if bare else None
def _label_likelihood(path: Path) -> Tuple[bool, float, str]:
"""Inspect one small sample and decide whether it resembles labels."""
try:
from .foreign import _read_mask
array = np.asarray(_read_mask(str(path)))
except Exception as exc:
return False, 0.0, f"could not sample pixels: {exc}"
if array.ndim != 2 or array.size == 0:
return False, 0.0, f"shape {array.shape} is not a 2-D label plane"
if not (np.issubdtype(array.dtype, np.integer)
or np.issubdtype(array.dtype, np.bool_)):
return False, 0.0, f"{array.dtype} is not an integer label dtype"
stride = max(int(np.sqrt(array.size / 250_000)), 1)
sampled = array[::stride, ::stride]
values = np.unique(sampled)
nonnegative = bool(values.size == 0 or values[0] >= 0)
compact = len(values) <= max(64, int(sampled.size * 0.002))
background = bool(np.any(values == 0))
likely = nonnegative and compact and background
confidence = 0.88 if likely else 0.85
return likely, confidence, (
f"{len(values)} unique integer values in {sampled.size} sampled "
f"pixels; {'zero background' if background else 'no zero background'}"
)
def _coerce_groups(value: Any, *, recursive: bool) -> List[InputGroup]:
"""Normalize path-like or serialized inputs into reviewed groups.
:param value: absent input, one path, path sequence, group instances, or
serialized group mappings.
:param recursive: directory traversal rule used when detecting paths.
"""
if value is None:
return []
if isinstance(value, (str, os.PathLike)):
return detect_inputs([value], recursive=recursive)
values = list(value)
if not values:
return []
if all(isinstance(item, (str, os.PathLike)) for item in values):
return detect_inputs(values, recursive=recursive)
return [InputGroup.from_value(item) for item in values]
def _scan_group(group: InputGroup, *, layout: str = "auto"
) -> List[cv.SourceImage]:
"""Scan a group's root once and keep only its selected paths."""
selected = {os.path.abspath(path) for path in group.paths}
sources = cv.scan(group.root, layout=layout)
return [source for source in sources
if os.path.abspath(source.path) in selected]
def _stem(mapping: cv.Mapping) -> str:
"""Return the canonical ``plate_well_integer-field`` destination stem."""
return f"{mapping.plate}_{mapping.well}_{int(mapping.field)}"
def _pair_masks(image_plan: cv.ConversionPlan,
groups: Sequence[InputGroup],
*,
layout: str = "auto",
) -> Tuple[Dict[str, Dict[str, MaskMatch]], List[str], List[str]]:
"""Pair reviewed mask sources to image fields by canonical identity.
Exact source identities are preferred before normalized loose-field
matching. The return tuple contains mappings by object type, blocking
errors, and non-blocking ambiguity warnings.
Normalizing drops a trailing mask word and the group's object-type name
from a mask field. For a group typed as any organelle slot it drops any
trailing organelle token -- ``organelle``, ``organelle_7``,
``organelleb``, ``mitochondria`` -- not only the assigned slot's own
spelling, so a group re-typed in the import table still pairs, and
``fov001_mitochondria_mask`` pairs with ``fov001``.
:param image_plan: reviewed intensity-image conversion plan.
:param groups: reviewed inputs, including zero or more mask groups.
:param layout: filename-layout rule forwarded while scanning each group.
"""
errors: List[str] = []
warnings: List[str] = []
source_to_stem: Dict[Tuple[str, str, str], str] = {}
loose_fields: Dict[str, set] = {}
all_stems = set()
for mapping in image_plan.mappings:
key = (mapping.source_plate, mapping.source_well,
mapping.source_field)
stem = _stem(mapping)
source_to_stem[key] = stem
loose_fields.setdefault(mapping.source_field, set()).add(stem)
all_stems.add(stem)
by_type: Dict[str, Dict[str, MaskMatch]] = {}
for group in groups:
if group.role != "mask":
continue
object_type = group.object_type
if object_type not in OBJECT_TYPES:
fixed = [role for role in OBJECT_TYPES
if role not in _ORGANELLE_ROLE_SET]
errors.append(
f"{group.key}: choose whether these masks are "
f"{', '.join(fixed)} or an organelle slot "
f"({ORGANELLE_ROLES[0]}, {ORGANELLE_ROLES[1]}, "
f"{ORGANELLE_ROLES[2]}, …).")
continue
matched = by_type.setdefault(object_type, {})
for source in _scan_group(group, layout=layout):
field = source.field
normalised = field
for token in (
"mask", "masks", "label", "labels", "labelled", "labeled",
"instance", "instances", "seg", "segmentation", object_type,
"nuclei" if object_type == "nucleus" else object_type,
"parasite" if object_type == "pathogen" else object_type,
"organell" if object_type == "organelle" else object_type,
):
normalised = re.sub(
rf"(?i)(?:^|[_\-. ]){re.escape(token)}$",
"", normalised).strip("_-. ")
if object_type in _ORGANELLE_ROLE_SET:
normalised = _ORGANELLE_TAIL.sub(
"", normalised).strip("_-. ")
candidates = [
((source.plate, source.well, field), "exact"),
((source.plate, source.well, normalised), "normalised"),
]
hit: Optional[Tuple[str, str]] = None
for key, how in candidates:
if key in source_to_stem:
hit = (source_to_stem[key], how)
break
loose = loose_fields.get(key[2], set())
if len(loose) == 1:
hit = (next(iter(loose)), how)
break
if hit is None:
warnings.append(
f"{source.path}: no intensity field has the same "
f"plate/well/field name; mask not imported.")
continue
stem, how = hit
if stem in matched:
errors.append(
f"{matched[stem].path} and {source.path} both map to "
f"{stem} as {object_type} masks.")
continue
matched[stem] = MaskMatch(
path=source.path, object_type=object_type, stem=stem,
match=how)
for object_type, matched in by_type.items():
missing = sorted(all_stems - set(matched))
if missing:
errors.append(
f"{object_type}: {len(missing)} intensity field(s) have no "
f"matching mask ({', '.join(missing[:5])}"
f"{'…' if len(missing) > 5 else ''}).")
return by_type, errors, warnings
[docs]
def default_settings(settings: Optional[Mapping[str, Any]] = None
) -> Dict[str, Any]:
"""Return importer settings plus the complete Measure setting contract."""
from .settings import get_measure_crop_settings
resolved = get_measure_crop_settings({})
for key in ("src", *(f"{role}_mask_dim" for role in OBJECT_TYPES)):
resolved.pop(key, None)
resolved.update({
"inputs": [],
"dst": None,
"recursive": True,
"layout": "auto",
"z_handling": cv.Z_MAX,
"plate_naming": "index",
"overwrite": False,
"preview_only": False,
"channels": [],
"png_dims": [],
"experiment": "external_masks",
"save_measurements": True,
"save_png": True,
"cytoplasm": True,
})
resolved.update(dict(settings or {}))
return resolved
[docs]
def plan_external_masks(settings: Optional[Mapping[str, Any]] = None
) -> ExternalMaskPlan:
"""Validate and preview an external image/mask import without writing.
:param settings: Partial settings mapping accepted by
:func:`default_settings`.
:returns: Pairing plan containing canonical image mappings, per-object
mask matches, warnings, and blocking errors.
"""
resolved = default_settings(settings)
groups = _coerce_groups(
resolved.get("inputs"), recursive=bool(resolved.get("recursive", True)))
destination = os.path.abspath(str(
resolved.get("dst") or
(f"{groups[0].root}_spacr" if groups else "external_masks_spacr")))
errors: List[str] = []
warnings: List[str] = []
image_groups = [group for group in groups if group.role == "image"]
if not image_groups:
errors.append("No intensity-image group is selected.")
mask_groups = [group for group in groups if group.role == "mask"]
if not mask_groups:
errors.append("No label-mask group is selected.")
invalid_roles = sorted({group.role for group in groups
if group.role not in ROLES})
if invalid_roles:
errors.append(f"Unknown input roles: {', '.join(invalid_roles)}.")
layout = str(resolved.get("layout") or "auto")
if layout not in cv.LAYOUTS:
errors.append(
f"Unknown input layout {layout!r}; choose one of "
f"{', '.join(cv.LAYOUTS)}.")
layout = "auto"
sources: List[cv.SourceImage] = []
seen = set()
for group in image_groups:
for source in _scan_group(group, layout=layout):
marker = (source.path, source.meta.get("series", 0))
if marker not in seen:
sources.append(source)
seen.add(marker)
image_plan = cv.plan(
sources,
z_handling=str(resolved.get("z_handling") or cv.Z_MAX),
plate_naming=str(resolved.get("plate_naming") or "index"),
)
n_channels = len({mapping.channel for mapping in image_plan.mappings})
if not n_channels:
errors.append("No readable intensity channels were detected.")
mapping_slots = [
(_stem(mapping), int(mapping.channel))
for mapping in image_plan.mappings
]
if len(mapping_slots) != len(set(mapping_slots)):
errors.append(
"Multiple time or Z planes map to the same field/channel. "
"Choose z_handling='max' or 'first', or export each 2-D plane "
"as a separately named field before importing.")
masks_by_type, pair_errors, pair_warnings = _pair_masks(
image_plan, mask_groups, layout=layout)
errors.extend(pair_errors)
warnings.extend(pair_warnings)
mask_dims = {
object_type: n_channels + index
for index, object_type in enumerate(
name for name in OBJECT_TYPES if name in masks_by_type)
}
if os.path.exists(os.path.join(destination, "measurements",
"measurements.db")):
errors.append(
f"{destination} already has measurements/measurements.db. "
"Choose a new destination; existing measurements are never "
"silently replaced.")
return ExternalMaskPlan(
groups=groups, images=image_plan, masks=masks_by_type,
destination=destination, n_channels=n_channels,
mask_dims=mask_dims, errors=errors, warnings=warnings)
def _save_npy(path: str, array: np.ndarray) -> str:
"""Atomically replace one NumPy destination through a hidden partial file.
An interrupted write cannot look like a finished field to array readers.
Ordinary save/replace failures remove the scratch file and preserve the
previous destination; a killed process can leave only a hidden partial.
:param path: destination ``.npy`` path.
:param array: array to serialize.
:returns: ``path`` after replacement succeeds.
"""
from .io import _save_array_atomic
return _save_array_atomic(path, array)
def _tables(path: str) -> List[str]:
"""Return sorted SQLite table names, or none for a missing database."""
if not os.path.isfile(path):
return []
with sqlite3.connect(path, timeout=30) as connection:
rows = connection.execute(
"SELECT name FROM sqlite_master WHERE type='table' "
"ORDER BY name").fetchall()
return [str(row[0]) for row in rows]
[docs]
def run_external_masks(plan: ExternalMaskPlan,
settings: Optional[Mapping[str, Any]] = None
) -> ExternalMaskResult:
"""Materialize ``plan`` and call the standard Measure pipeline.
:param plan: Validated plan from :func:`plan_external_masks`, whose
``ok`` property must be True.
:param settings: Partial settings mapping accepted by
:func:`default_settings`; it carries the full Measure contract and
supplies ``overwrite`` for the intensity conversion and
``channels``, ``png_dims``, ``crop_mode`` and ``cytoplasm`` for the
Measure call.
:returns: Result describing the written project, its merged arrays,
measurements database and tables, and the plan used.
:raises ConfigurationError: If the plan is not ``ok``, if a field's
channel count, shapes, dtypes or label IDs violate the Measure
uint16 contract, or if Measure finishes without a required table.
"""
if not plan.ok:
raise ConfigurationError(
"External-mask import refused; nothing was written:\n "
+ "\n ".join([*plan.images.errors, *plan.errors]))
resolved = default_settings(settings)
dst = plan.destination
os.makedirs(dst, exist_ok=True)
images_dir = os.path.join(dst, "images")
conversion = cv.convert(plan.images, images_dir,
overwrite=bool(resolved.get("overwrite", False)))
available = {mapping.target for mapping in conversion.written}
available.update(mapping.target for mapping in conversion.existing)
mappings_by_stem: Dict[str, Dict[int, cv.Mapping]] = {}
for mapping in plan.images.mappings:
if _stem(mapping) not in plan.stems:
continue
mappings_by_stem.setdefault(_stem(mapping), {})[
int(mapping.channel)] = mapping
merged_paths: List[str] = []
from .foreign import _read_mask
for stem in plan.stems:
channels = mappings_by_stem.get(stem, {})
if len(channels) != plan.n_channels:
raise ConfigurationError(
f"{stem}: expected {plan.n_channels} channels, found "
f"{len(channels)}.")
image_planes = []
for channel in sorted(channels):
mapping = channels[channel]
if mapping.target not in available:
raise ConfigurationError(
f"{stem}: converted intensity image is missing: "
f"{mapping.target}")
image_planes.append(np.asarray(
cv._read_source(cv.SourceImage(
path=os.path.join(images_dir, mapping.target),
plate="", well="", field="",
meta={"ext": cv._split_ext(mapping.target)[1]},
))
).squeeze())
shape = image_planes[0].shape
if any(plane.shape != shape for plane in image_planes):
raise ConfigurationError(
f"{stem}: intensity channels have inconsistent shapes.")
mask_planes = []
for object_type in plan.object_types:
match = plan.masks[object_type][stem]
array = np.asarray(_read_mask(match.path))
if array.shape != shape:
raise ConfigurationError(
f"{match.path}: {object_type} mask shape {array.shape} "
f"does not match intensity shape {shape} for {stem}.")
if np.any(array < 0):
raise ConfigurationError(
f"{match.path}: label masks cannot contain negative IDs.")
maximum = int(np.max(array, initial=0))
if maximum > np.iinfo(np.uint16).max:
raise ConfigurationError(
f"{match.path}: label ID {maximum} exceeds the maximum "
"65535 supported by the Measure array contract.")
integer = array.astype(np.uint16, copy=False)
mask_planes.append(integer)
_save_npy(os.path.join(
dst, "masks", f"{object_type}_mask_stack", f"{stem}.npy"),
integer)
for plane in image_planes:
if np.issubdtype(plane.dtype, np.floating):
if not np.all(np.isfinite(plane)):
raise ConfigurationError(
f"{stem}: intensity data contain NaN or infinity.")
if not np.all(plane == np.floor(plane)):
raise ConfigurationError(
f"{stem}: floating-point intensities would lose "
"precision in Measure's uint16 arrays. Rescale and "
"export them as 8- or 16-bit images first.")
if float(np.min(plane, initial=0)) < 0 or \
float(np.max(plane, initial=0)) > np.iinfo(np.uint16).max:
raise ConfigurationError(
f"{stem}: intensity values must fit the Measure uint16 "
"contract (0–65535). Rescale the source images first.")
stack = np.stack([plane.astype(np.uint16, copy=False)
for plane in image_planes], axis=-1)
_save_npy(os.path.join(dst, "stack", f"{stem}.npy"), stack)
merged = np.stack(
[plane.astype(np.uint16, copy=False)
for plane in [*image_planes, *mask_planes]], axis=-1)
merged_paths.append(_save_npy(
os.path.join(dst, "merged", f"{stem}.npy"), merged))
plane_layout = {
"version": 1,
"intensity_channels": list(range(plan.n_channels)),
"mask_plane_order": list(plan.object_types),
"mask_dims": dict(plan.mask_dims),
}
with open(os.path.join(dst, "merged", crops.MERGED_LAYOUT_SIDECAR),
"w", encoding="utf-8") as handle:
json.dump(plane_layout, handle, indent=2, sort_keys=True)
handle.write("\n")
manifest = {
"module": "external_masks",
"destination": dst,
"n_channels": plan.n_channels,
"mask_dims": plan.mask_dims,
"groups": [group.to_dict() for group in plan.groups],
"merged": [os.path.basename(path) for path in merged_paths],
}
with open(os.path.join(dst, "external_mask_import.json"), "w",
encoding="utf-8") as handle:
json.dump(manifest, handle, indent=2, sort_keys=True)
measure_settings = dict(resolved)
for key in (
"inputs", "dst", "recursive", "layout", "z_handling",
"plate_naming", "overwrite", "preview_only",
):
measure_settings.pop(key, None)
measure_settings["src"] = os.path.join(dst, "merged")
measure_settings["channels"] = (
list(range(plan.n_channels))
if not measure_settings.get("channels")
else list(measure_settings["channels"])
)
measure_settings["png_dims"] = (
list(range(min(plan.n_channels, 3)))
if not measure_settings.get("png_dims")
else list(measure_settings["png_dims"])
)
for object_type in OBJECT_TYPES:
measure_settings[f"{object_type}_mask_dim"] = (
plan.mask_dims.get(object_type))
available_crops = list(plan.object_types)
if "cell" in plan.object_types and measure_settings.get("cytoplasm"):
available_crops.append("cytoplasm")
requested = measure_settings.get("crop_mode") or []
if isinstance(requested, str):
requested = [requested]
requested = [name for name in requested if name in available_crops]
measure_settings["crop_mode"] = (
requested or available_crops[:1])
from .measure import measure_crop
measure_crop(measure_settings)
db_path = os.path.join(dst, "measurements", "measurements.db")
tables = _tables(db_path)
expected = set(plan.object_types)
if "cell" in expected and measure_settings.get("cytoplasm"):
expected.add("cytoplasm")
if measure_settings.get("save_png"):
expected.add("png_list")
missing = sorted(expected - set(tables))
if missing:
raise ConfigurationError(
"Measure finished without required output table(s): "
+ ", ".join(missing))
return ExternalMaskResult(
destination=dst, merged=merged_paths, db_path=db_path,
tables=tables, data_dir=os.path.join(dst, "data"), plan=plan)
[docs]
def prepare_external_masks(settings: Optional[Mapping[str, Any]] = None
) -> Any:
"""Plan an external-mask import, print the preview, and run it.
The plan summary is always printed to stdout. Unless ``preview_only``
is set, the project is written and Measure is run, and the result
summary is printed too.
:param settings: Partial settings mapping accepted by
:func:`default_settings`.
:returns: The :class:`ExternalMaskPlan` when ``preview_only`` is set,
otherwise the :class:`ExternalMaskResult` from
:func:`run_external_masks`.
"""
resolved = default_settings(settings)
plan = plan_external_masks(resolved)
print(plan.summary())
if resolved.get("preview_only"):
return plan
result = run_external_masks(plan, resolved)
print(result.summary())
return result
[docs]
def register_settings(replace: bool = False) -> bool:
"""Publish this module's settings help through the shared registry.
External Masks predates the module-registration seam, so its seven
importer-specific controls were the only displayed settings in the app
registry without authored help. Registering beside the defaults keeps
the inventory, validation types, and tooltip prose together; the Measure
settings returned by :func:`default_settings` retain their existing
shared declarations.
:param replace: replace this module's existing defaults registration.
:returns: whether a new registration was made.
"""
from .settings import has_registered_defaults, register_defaults
from .settings import tooltips as shared_tooltips
if has_registered_defaults("external_masks") and not replace:
return False
types = {
"inputs": list,
"recursive": bool,
"layout": str,
"z_handling": str,
"plate_naming": str,
"overwrite": bool,
"preview_only": bool,
}
tips = {
"inputs":
"(list) - Image and external label-mask files or folders to "
"import. The preview groups them by source and proposes whether "
"each group is an intensity image, an object mask, or ignored. "
"Default [].",
"recursive":
"(bool) - Search inside subfolders of every input folder. Turn "
"this off when only files directly inside each selected folder "
"belong to the import. Default True.",
"layout":
"(str) - Naming-layout hint used to read plate, well, field, "
"channel, Z, and time identifiers from source files. 'auto' "
"detects the supported layout from the filenames. Default "
"'auto'.",
"z_handling":
"(str) - How multiple Z planes become a 2-D Measure input. "
"'max' takes a maximum-intensity projection and 'first' keeps "
"only the first plane; inputs that still contain separate planes "
"are rejected. Default 'max'.",
"plate_naming":
"(str) - How imported plates are named when the source does not "
"provide one. 'index' assigns stable plate numbers in discovered "
"input order. Default 'index'.",
"overwrite":
"(bool) - Allow the importer to replace files in an existing "
"destination project. Leave this off to stop before previously "
"written images, masks, or measurements can be replaced. Default "
"False.",
"preview_only":
"(bool) - Build and print the complete input-to-mask assignment "
"plan without writing a project or running Measure. Use this to "
"review automatic role and object-type detection first. Default "
"False.",
}
tips = {key: value for key, value in tips.items()
if key not in shared_tooltips}
register_defaults(
"external_masks", default_settings, replace=replace,
expected_types=types, tooltips=tips,
description=(
"Import images and externally generated label masks as a "
"measured spaCR project ready for annotation."),
)
return True
register_settings()
__all__ = [
"InputGroup", "MaskMatch", "ExternalMaskPlan", "ExternalMaskResult",
"SUPPORTED_SUFFIXES", "OBJECT_TYPES", "ROLES",
"detect_inputs", "default_settings", "plan_external_masks",
"run_external_masks", "prepare_external_masks", "register_settings",
]