Add FFT cross-correlation to Manual Alignment, with tunable options

Manual mode previously offered only Auto De-rotate (rotation from known
scan angles) plus keyboard nudging, so every angle's translation had to
be found entirely by eye — fine for a small correction, unworkable for
the tens-of-mm scatter a real many-angle scan can have between angles
before any translation search runs at all.

Add an "Auto Cross-Correlate (vs Reference)" action that, for every
non-reference angle, sets rotation to the known analytic angle (same as
Auto De-rotate) and translation to the FFT-phase-correlation best fit
against angle #0 (always the reference/ground truth). This is meant to
land every angle's outline roughly stacked on the others so keyboard
nudging only has to make small corrections afterward, per the intended
workflow: cross-correlate first, then nudge.

Exposes two tunable options, since real data may correlate better one
way than the other: "Correlate on" (raw CH4 signal, minus its own
minimum -- the default, robust to a shared threshold not suiting every
angle's real signal level -- or the thresholded binary mask), and
"Search margin" (how far a shift the phase correlation searches before
wraparound bias becomes a risk).

Runs on a background thread (new CrossCorrelateWorker in
sras_workers.py) since correlating many high-resolution angles can take
long enough to visibly freeze the dialog otherwise.

The underlying per-angle correlation math is factored out of
compute_angle_alignment's Step 2 into a new, reusable
correlate_translation_mm, so the existing automatic Fusion -> Angle
Alignment action and the new manual button now share one implementation
instead of two copies of the same phase-correlation logic.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
2026-07-31 11:31:14 -05:00
parent 52e91d46fd
commit bbb075ff34
4 changed files with 258 additions and 54 deletions
+54
View File
@@ -346,3 +346,57 @@ class Ch4MaskWorker(QObject):
self.finished.emit()
except Exception as exc:
self.error.emit(str(exc))
class CrossCorrelateWorker(QObject):
"""FFT phase-correlation translation for 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.
"""
angle_done = pyqtSignal(int, float, float, float) # angle_idx, rotation_deg, shift_x_mm, shift_y_mm
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):
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._threshold = dc_threshold_mv
self._margin = margin_frac
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 run(self):
try:
n_workers, _budget = compute.plan_angle_level(self._sras)
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)
finally:
pool.shutdown(wait=True)
self.finished.emit()
except Exception as exc:
self.error.emit(str(exc))