"""One well, one cycle, from a folder of tiles to a table of positions.
:mod:`spacr.ops_layout` says which tiles touch, :mod:`spacr.ops_register` says
how far apart a touching pair is, :mod:`spacr.ops_solve` says where each tile
ends up, and this runs the three of them over a well and reports what happened.
The output is coordinates, not pixels. A well of this acquisition is 26,855 x
26,865 px, enough to exhaust a machine's memory; the transform table is a few
kilobytes and every later phase reads through it. A canvas, if anybody wants
one, is rendered from this at whatever downsample suits the screen.
It reports three numbers, not two. An early end-to-end run of this pipeline
reported
333 of 333 placed residual median 0.2 px canvas 35,374 x 35,367
and was wrong: the driver negated the correlation's shift, the unwrap then
chose the representative one period out, identically in every edge of a
direction, and the well came out one pitch per column too large in both axes.
Every edge still agreed with every other edge, so the residual was perfect and
the count was perfect. A uniform error is invisible to a residual. Only the
canvas said otherwise, and only because 21 columns of a known pitch has an
arithmetic answer -- 20 x 1267 + 1480 = 26,820 -- to check it against.
So :meth:`StitchedWell.summary` gives the count, the residual and the canvas in
one line, and :attr:`StitchedWell.expected_canvas` carries what the layout says
the canvas should be, because a number with nothing to compare it to is not a
check.
Typical use::
from spacr.ops_stitch import stitch_well
well = stitch_well(read_tile, overlap=213, tolerance=4)
print(well.summary())
positions = well.placements # site -> (y, x), in well pixels
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Callable, Dict, Optional, Sequence, Tuple
import numpy as np
from .ops_layout import WellLayout, round_well_layout
LOG = logging.getLogger(__name__)
__all__ = ["StitchedWell", "stitch_well"]
@dataclass
[docs]
class StitchedWell:
"""Result of one well's stitch: positions, registrations and geometry.
:ivar placements: ``site -> (y, x)`` in well pixels, from the solve.
:ivar edges: ``(a, b) -> Registration`` for every pair the layout proposed,
accepted or not. A refused pair is a fact about the acquisition and
belongs in `ops_geometry`, not in a debug log.
:ivar tile_shape: ``(height, width)`` of one tile.
:ivar layout: the well model the pairs came from.
:ivar overlap: the overlap in pixels the run was told to expect, or
None when it was not told.
"""
placements: Dict[int, Tuple[float, float]]
edges: Dict[Tuple[int, int], object]
tile_shape: Tuple[int, int]
layout: WellLayout
overlap: Optional[int] = None
_residuals: Optional[np.ndarray] = field(default=None, repr=False)
@property
[docs]
def placed(self) -> int:
"""How many tiles got a position."""
return len(self.placements)
@property
[docs]
def accepted(self) -> int:
"""How many proposed pairs registered."""
return sum(1 for one in self.edges.values()
if getattr(one, "accepted", True))
@property
[docs]
def proposed(self) -> int:
"""How many pairs the layout offered."""
return len(self.edges)
@property
[docs]
def residuals(self) -> np.ndarray:
"""Per-edge disagreement between the solve and the measurement.
The distance, in pixels, between where an accepted edge said its two
tiles sit relative to each other and where the solve put them.
Sub-pixel across a well is what says the geometry closed.
"""
if self._residuals is None:
self._residuals = self._measure_residuals()
return self._residuals
def _measure_residuals(self) -> np.ndarray:
"""How far each accepted edge's measurement is from the solve.
Only an edge that registered and whose two tiles both got a
position can contribute, so this is shorter than the edge list
whenever a pair was rejected or a tile went unplaced, and empty
when none of them survived -- which is not the same answer as a
disagreement of zero, and is left as an empty array so it cannot
be read as one.
:returns: one distance in pixels per contributing edge, between
the shift that edge measured and the shift the placements
imply.
"""
found = []
for (a, b), one in self.edges.items():
if not getattr(one, "accepted", True):
continue
if a not in self.placements or b not in self.placements:
continue
dy, dx = getattr(one, "shift", one)
ay, ax = self.placements[a]
by, bx = self.placements[b]
found.append(float(np.hypot((by - ay) - dy, (bx - ax) - dx)))
return np.asarray(found, dtype=float)
@property
[docs]
def canvas(self) -> Tuple[int, int]:
"""``(height, width)`` the placed tiles span, in pixels."""
if not self.placements:
return (0, 0)
ys = [y for y, _x in self.placements.values()]
xs = [x for _y, x in self.placements.values()]
height, width = self.tile_shape
return (int(round(max(ys) - min(ys) + height)),
int(round(max(xs) - min(xs) + width)))
@property
[docs]
def expected_canvas(self) -> Tuple[int, int]:
"""What the layout says the canvas should be.
This check catches a uniform placement error that residuals miss.
Columns minus one pitches plus one tile, and the same down the tallest
column. A measured canvas with nothing to compare it to is a number,
not a check.
"""
height, width = self.tile_shape
pitch_y = height - (self.overlap or 0)
pitch_x = width - (self.overlap or 0)
columns = sum(1 for one in self.layout.heights if one)
rows = max(self.layout.heights) if self.layout.heights else 0
return (int(max(0, rows - 1) * pitch_y + height),
int(max(0, columns - 1) * pitch_x + width))
[docs]
def summary(self) -> str:
"""Return the placed count, residual and canvas on one line."""
residuals = self.residuals
if residuals.size:
middle = float(np.median(residuals))
worst = float(residuals.max())
residual = f"residual median {middle:.2f} px max {worst:.2f} px"
else:
residual = "residual n/a (no accepted edge placed both tiles)"
canvas = self.canvas
expected = self.expected_canvas
return (f"{self.placed} of {self.layout.site_count} placed "
f"{self.accepted}/{self.proposed} edges {residual} "
f"canvas {canvas[0]:,} x {canvas[1]:,} "
f"(layout says {expected[0]:,} x {expected[1]:,})")
[docs]
def canvas_agrees(self, tolerance: float = 0.01) -> bool:
"""Whether the measured canvas matches the layout's arithmetic.
:param tolerance: allowed fractional difference on either axis.
One per cent, because the layout's number is exact and the
stitch's is not: it carries the real stage's skew.
:returns: True when both axes agree within ``tolerance``.
"""
measured, expected = self.canvas, self.expected_canvas
for got, want in zip(measured, expected):
if want <= 0:
return False
if abs(got - want) / want > tolerance:
return False
return True
[docs]
def stitch_well(tiles, layout: Optional[WellLayout] = None, *,
overlap: Optional[int] = None,
tolerance: Optional[int] = None,
skew: Optional[int] = None,
gpu: bool = True,
sites: Optional[Sequence[int]] = None,
**kwargs) -> StitchedWell:
"""Register a well's tiles against each other and solve their positions.
One well at a time, and one tile pair at a time within it. ``tiles`` is
normally a callable, because 333 tiles of 1480 px is 2.9 GB and nothing
here needs two of them resident: the caller decides what to keep and what
to re-read.
:param tiles: ``site -> 2-D array``, or a callable taking a site.
:param layout: the well model. None fits the circle to ``sites`` or,
failing that, to the measured 333-field well.
:param overlap: the raster's overlap in pixels. Given, it sets the
registration's expectation and the layout's canvas arithmetic;
omitted, both fall back to weaker answers, so pass it.
:param tolerance: how far along the raster an edge may land from the
layout's prediction and still be accepted. This is the acceptance the
real well used -- 624 of 624 -- and without it the peak ratio decides,
which on a real plate it cannot.
:param skew: how far across it may. The stage's skew is real and constant
-- 9 px on the measured plate -- so it is a separate number from the
tolerance. None takes `spacr.ops_register.SKEW_PX`.
:param gpu: passed through to the registration.
:param sites: which sites to place. None uses every site the layout
holds.
:param kwargs: passed to :func:`spacr.ops_register.register_edge`.
:returns: the stitch, its edges and its residuals.
:raises ValueError: when the well holds no tiles at all, which is a
caller error rather than an empty result.
"""
from .ops_register import Registration, register_edge
from .ops_solve import solve_placements
if layout is None:
layout = (round_well_layout(len(sites)) if sites is not None
else round_well_layout())
every_site = list(sites) if sites is not None else list(
range(layout.site_count))
if not every_site:
raise ValueError("a well with no sites cannot be stitched")
read: Callable = tiles if callable(tiles) else tiles.__getitem__
first = np.asarray(read(every_site[0]))
if first.ndim != 2:
raise ValueError(
f"a tile is a 2-D field; site {every_site[0]} is {first.shape}. "
"Pick the DAPI plane before stitching -- the geometry is solved "
"on one channel and PART 10 says which.")
tile_shape = (int(first.shape[0]), int(first.shape[1]))
known = set(every_site)
edges: Dict[Tuple[int, int], object] = {}
for a, b, axis in layout.pairs():
if a not in known or b not in known:
continue
try:
edges[(a, b)] = register_edge(
read(a), read(b), axis, expected_overlap=overlap,
tolerance=tolerance, gpu=gpu,
**({} if skew is None else {"skew": skew}), **kwargs)
except Exception: # noqa: BLE001
LOG.debug("could not register the pair %s-%s", a, b,
exc_info=True)
edges[(a, b)] = Registration(dy=0, dx=0, peak_ratio=0.0,
accepted=False, backend="none")
accepted = {pair: one.shift for pair, one in edges.items()
if getattr(one, "accepted", True)}
placements = solve_placements(accepted, every_site)
return StitchedWell(placements=placements, edges=edges,
tile_shape=tile_shape, layout=layout,
overlap=overlap)