Add scan-editing CLI and alignment tests; extend Manual Alignment correlation

Continues the Manual Alignment work: refines the FFT cross-correlation and
mask handling, adds sras_edit_scans.py (drop/renumber bad angle scans),
tools/test_alignment.py (registration ground-truth suite), and a rotating
test fixture in make_test_sras.py.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
2026-08-06 09:35:30 -05:00
parent bbb075ff34
commit d5028db445
8 changed files with 1619 additions and 597 deletions
+31 -30
View File
@@ -282,10 +282,10 @@ class BatchCacheWorker(QObject):
class AngleAlignmentWorker(QObject):
"""Computes rotation+translation alignment for every angle in *sras*,
referenced to *ref_angle_idx*, from each angle's binarized CH4 mask.
Rotation is analytic (from sras.angles_deg); only translation is found by
phase correlation.
"""Computes the rigid (rotation + translation, never scale) alignment for
every angle in *sras* against *ref_angle_idx*, by cross-correlating each
angle's CH4 image against the reference's. Both the rotation and the
translation are found from image content — see compute_angle_alignment.
"""
progress = pyqtSignal(int) # 0–100
finished = pyqtSignal(object, str) # AlignmentResult|None, error ("" = success)
@@ -349,52 +349,53 @@ class Ch4MaskWorker(QObject):
class CrossCorrelateWorker(QObject):
"""FFT phase-correlation translation for each of *angle_indices* against
*ref_angle_idx*, for ManualAlignmentDialog's Auto Cross-Correlate button.
"""Rigid registration (rotation + translation, never scale) of each of
*angle_indices* against *ref_angle_idx*, for ManualAlignmentDialog's Auto
Cross-Correlate button.
Runs on a background thread — a real many-angle, high-resolution scan's
correlation (even at its downsampled working resolution) can take long
enough that doing all of them on the GUI thread would visibly freeze the
dialog. Rotation is set to the same analytic scan-angle delta Auto
De-rotate uses alongside the correlated shift, since a translation
search is only meaningful once both angles' content is already oriented
the same way. dc4_mv/pivot_mm are the dialog's own already-in-memory
per-angle images/pivots — this worker does no fetching of its own.
Runs on a background thread — registering a real many-angle,
high-resolution scan takes long enough that doing it on the GUI thread
would visibly freeze the dialog. Rotation is *searched*, not taken from the
stage's reported angle: see compute.register_angle_to_reference, which
seeds from that angle but scores both of its signs and refines from there.
dc4_mv is the dialog's own already-in-memory per-angle CH4 image — this
worker does no fetching of its own.
"""
angle_done = pyqtSignal(int, float, float, float) # angle_idx, rotation_deg, shift_x_mm, shift_y_mm
# angle_idx, rotation_deg, shift_x_mm, shift_y_mm, score, source
angle_done = pyqtSignal(int, float, float, float, float, str)
finished = pyqtSignal()
error = pyqtSignal(str)
def __init__(self, sras: SrasFile, ref_angle_idx: int, angle_indices: list[int],
dc4_mv: dict[int, np.ndarray], pivot_mm: dict[int, tuple[float, float]],
*, use_mask: bool, dc_threshold_mv: float, margin_frac: float):
dc4_mv: dict[int, np.ndarray], *,
sources: tuple[str, ...], dc_threshold_mv: float,
search_deg: float):
super().__init__()
self._sras = sras
self._ref = ref_angle_idx
self._angles = angle_indices
self._dc4_mv = dc4_mv
self._pivot_mm = pivot_mm
self._use_mask = use_mask
self._sources = sources
self._threshold = dc_threshold_mv
self._margin = margin_frac
self._search_deg = search_deg
def _one(self, a: int) -> tuple[int, float, float, float]:
theta = compute._theta_deg(self._sras, a, self._ref)
dx, dy = compute.correlate_translation_mm(
self._sras, a, self._ref, self._dc4_mv, self._pivot_mm,
use_mask=self._use_mask, dc_threshold_mv=self._threshold,
margin_frac=self._margin)
return a, theta, dx, dy
def _one(self, a: int) -> tuple[int, compute.RigidFit]:
return a, compute.register_angle_to_reference(
self._sras, a, self._ref, self._dc4_mv,
dc_threshold_mv=self._threshold, sources=self._sources,
search_deg=self._search_deg)
def run(self):
try:
n_workers, _budget = compute.plan_angle_level(self._sras)
n_workers = compute._registration_workers(
self._sras, compute._DEFAULT_FINE_DIM)
pool = ThreadPoolExecutor(max_workers=max(1, n_workers))
try:
futures = [pool.submit(self._one, a) for a in self._angles]
for fut in as_completed(futures):
a, theta, dx, dy = fut.result()
self.angle_done.emit(a, theta, dx, dy)
a, fit = fut.result()
self.angle_done.emit(a, fit.rotation_deg, fit.shift_mm[0],
fit.shift_mm[1], fit.score, fit.source)
finally:
pool.shutdown(wait=True)
self.finished.emit()