Add Export Fused ROI CSV export

Lets a user export the current ROI as one CSV with a column per selected
angle's value (RF peak freq, Bias A, Bias B, or velocity), once angles share
a common (x, y) grid -- either via a live Fusion alignment result or because
the open file is itself a previous Alignment Wizard export whose angles
already share one grid on disk. Never triggers a background compute; angles
without cached/stored data for the chosen value type are simply unavailable
in the picker.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
2026-08-09 19:11:11 -05:00
parent d5914b5793
commit 191d1b8946
6 changed files with 659 additions and 30 deletions
+18
View File
@@ -770,6 +770,24 @@ class SrasFile:
def y_positions_mm(self, angle_idx: int) -> np.ndarray:
return self._y_pos_per_angle[angle_idx]
def angles_share_raw_grid(self) -> bool:
"""True iff every angle's raw (x, y) pixel grid is literally the same
array as angle 0's -- the case for a .sras file the viewer's own
Alignment Wizard exported (see scan_format.md, "Files written by the
viewer's Alignment Wizard"): the writer packs one Per-Angle Geometry
record and one Row Table span and repeats those same bytes for every
angle, so re-parsed arrays are bit-identical copies rather than
independently re-derived numbers -- a bare np.array_equal is the
correct test here, no tolerance needed.
"""
if self.n_angles <= 1:
return True
x0 = self.x_axis_mm(0)
y0 = self.y_positions_mm(0)
return all(np.array_equal(self.x_axis_mm(a), x0)
and np.array_equal(self.y_positions_mm(a), y0)
for a in range(1, self.n_angles))
def time_axis_ns(self) -> np.ndarray:
return np.arange(self.samples_per_frame) / self.sample_rate_hz * 1e9
+3 -1
View File
@@ -23,5 +23,7 @@ from .canvases import ( # noqa: E4
AlignOverlayCanvas, ImageCanvas, RoiQuad, WaveformCanvas,
)
from .common import CH_LABELS, CMAPS, VELOCITY_MODE_IDX # noqa: E402,F401
from .dialogs import FftOptionsDialog, RowAverageFftOptionsDialog # noqa: E402,F401
from .dialogs import ( # noqa: E402,F401
FftOptionsDialog, FusedRoiExportDialog, RowAverageFftOptionsDialog,
)
from .main_window import SrasViewerWindow, main # noqa: E402,F401
+181 -3
View File
@@ -5,14 +5,18 @@ alignment wizard's first page (see align_wizard.py), which needed the same
mask-overlay editor plus the crop and export steps.
"""
from pathlib import Path
from PyQt6.QtWidgets import (
QButtonGroup, QDialog, QDialogButtonBox, QGroupBox, QHBoxLayout, QLabel,
QRadioButton, QSpinBox, QVBoxLayout,
QButtonGroup, QCheckBox, QDialog, QDialogButtonBox, QFileDialog,
QGroupBox, QHBoxLayout, QLabel, QLineEdit, QPushButton, QRadioButton,
QScrollArea, QSpinBox, QVBoxLayout, QWidget,
)
from sras_compute import PYFFTW_AVAILABLE
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, CH_NAMES
from .common import _CSS_HINT, _make_dspin
from .common import CH_LABELS, VELOCITY_MODE_IDX, _CSS_HINT, _CSS_WARN, _group, _make_dspin, _wrap_label
# ---------------------------------------------------------------------------
@@ -235,3 +239,177 @@ class RowAverageFftOptionsDialog(QDialog):
def get_threshold_mv(self) -> float:
return self._spin_threshold.value()
# ---------------------------------------------------------------------------
# Export Fused ROI dialog
# ---------------------------------------------------------------------------
class FusedRoiExportDialog(QDialog):
"""Choose a value type and which angles to fuse for Export Fused ROI.
Every angle in the file is listed (a live AlignmentResult's per_angle
always covers every angle, and the raw-shared-grid path needs no per-
angle transform at all). Each checkbox is disabled — and auto-unchecked
— whenever *availability_fn(angle_idx, ch_idx)* is False for the
currently selected value type; switching the value-type radio
re-evaluates every checkbox live, since availability is per (angle,
value type) rather than just per angle — e.g. DC may be ready
everywhere while FFT is ready nowhere.
"""
_VALUE_MODES = (CH1_IDX, CH3_IDX, CH4_IDX, VELOCITY_MODE_IDX)
def __init__(self, parent=None, *,
angles: list[tuple[int, float]],
availability_fn,
default_ch_idx: int,
out_dir: str,
stem: str,
grid_note: str):
super().__init__(parent)
self.setWindowTitle("Export Fused ROI")
self.setModal(True)
self.setMinimumWidth(420)
self._availability_fn = availability_fn
self._out_dir = out_dir
self._stem = stem
self._path_user_chosen = False
layout = QVBoxLayout(self)
layout.addWidget(_wrap_label(grid_note, _CSS_HINT))
# ---- Value type --------------------------------------------------
grp_val, vl = _group("Value to Export")
self._val_group = QButtonGroup(self)
self._val_buttons: dict[int, QRadioButton] = {}
for ch_idx, label in zip(self._VALUE_MODES, CH_LABELS):
rb = QRadioButton(label)
self._val_group.addButton(rb, id=ch_idx)
self._val_buttons[ch_idx] = rb
vl.addWidget(rb)
self._val_buttons[default_ch_idx].setChecked(True)
self._val_group.idClicked.connect(self._on_value_type_changed)
layout.addWidget(grp_val)
# ---- Angles --------------------------------------------------
grp_ang, al = _group("Angles to Include")
sel_row = QHBoxLayout()
btn_all = QPushButton("Select All Available")
btn_none = QPushButton("Select None")
btn_all.clicked.connect(self._on_select_all_available)
btn_none.clicked.connect(self._on_select_none)
sel_row.addWidget(btn_all)
sel_row.addWidget(btn_none)
al.addLayout(sel_row)
scroll_inner = QWidget()
scroll_layout = QVBoxLayout(scroll_inner)
self._angle_checks: dict[int, QCheckBox] = {}
for angle_idx, angle_deg in angles:
cb = QCheckBox(f"{angle_deg:.1f}° (angle {angle_idx})")
self._angle_checks[angle_idx] = cb
cb.toggled.connect(self._update_accept_enabled)
scroll_layout.addWidget(cb)
scroll = QScrollArea()
scroll.setWidget(scroll_inner)
scroll.setWidgetResizable(True)
scroll.setMaximumHeight(220)
al.addWidget(scroll)
self._lbl_none_available = _wrap_label("", _CSS_WARN)
al.addWidget(self._lbl_none_available)
layout.addWidget(grp_ang)
# ---- Output path --------------------------------------------------
grp_out, ol = _group("Output File")
path_row = QHBoxLayout()
self._edit_path = QLineEdit()
self._edit_path.setReadOnly(True)
path_row.addWidget(self._edit_path, 1)
btn_browse = QPushButton("Browse…")
btn_browse.clicked.connect(self._on_browse)
path_row.addWidget(btn_browse)
ol.addLayout(path_row)
layout.addWidget(grp_out)
# ---- Buttons -----------------------------------------------------
buttons = QDialogButtonBox()
self._btn_export = buttons.addButton(
"Export", QDialogButtonBox.ButtonRole.AcceptRole)
self._btn_export.clicked.connect(self.accept)
buttons.addButton("Cancel", QDialogButtonBox.ButtonRole.RejectRole
).clicked.connect(self.reject)
layout.addWidget(buttons)
self._refresh_default_path()
self._apply_availability()
# ---- internals -----------------------------------------------------
def _current_ch_idx(self) -> int:
return self._val_group.checkedId()
def _apply_availability(self):
ch_idx = self._current_ch_idx()
n_ok = 0
for angle_idx, cb in self._angle_checks.items():
ok = self._availability_fn(angle_idx, ch_idx)
cb.setEnabled(ok)
if ok:
n_ok += 1
cb.setToolTip("")
else:
cb.setChecked(False)
cb.setToolTip(
f"No cached/stored {CH_LABELS[ch_idx]} data for this "
"angle yet — view it in the main window (or run "
"Batch Compute) first.")
self._lbl_none_available.setText(
"" if n_ok else "No angle has this value type ready yet.")
self._update_accept_enabled()
def _on_value_type_changed(self, _id: int):
self._apply_availability()
if not self._path_user_chosen:
self._refresh_default_path()
def _on_select_all_available(self):
for cb in self._angle_checks.values():
if cb.isEnabled():
cb.setChecked(True)
def _on_select_none(self):
for cb in self._angle_checks.values():
cb.setChecked(False)
def _refresh_default_path(self):
ch_idx = self._current_ch_idx()
name = f"{self._stem}_fused_roi_{CH_NAMES[ch_idx]}.csv"
self._edit_path.setText(str(Path(self._out_dir) / name))
self._update_accept_enabled()
def _on_browse(self):
path, _ = QFileDialog.getSaveFileName(
self, "Export Fused ROI as CSV", self._edit_path.text(),
"CSV files (*.csv);;All files (*)")
if path:
self._edit_path.setText(path)
self._path_user_chosen = True
self._update_accept_enabled()
def _update_accept_enabled(self):
any_checked = any(cb.isChecked() for cb in self._angle_checks.values())
self._btn_export.setEnabled(any_checked and bool(self._edit_path.text()))
# ---- getters ---------------------------------------------------------
def get_ch_idx(self) -> int:
return self._current_ch_idx()
def get_selected_angles(self) -> list[int]:
return sorted(a for a, cb in self._angle_checks.items() if cb.isChecked())
def get_output_path(self) -> str:
return self._edit_path.text()
+194 -22
View File
@@ -34,7 +34,9 @@ from .common import (
_wrap_label,
)
from .align_wizard import AlignmentWizard
from .dialogs import FftOptionsDialog, RowAverageFftOptionsDialog
from .dialogs import (
FftOptionsDialog, FusedRoiExportDialog, RowAverageFftOptionsDialog,
)
# ---------------------------------------------------------------------------
# Main window
@@ -335,6 +337,11 @@ class SrasViewerWindow(QMainWindow):
self.btn_export_roi.clicked.connect(self._on_export_roi_csv)
rl.addWidget(self.btn_export_roi)
self.btn_export_fused_roi = QPushButton("Export Fused ROI…")
self.btn_export_fused_roi.setEnabled(False)
self.btn_export_fused_roi.clicked.connect(self._on_export_fused_roi_csv)
rl.addWidget(self.btn_export_fused_roi)
self.lbl_roi_center = _wrap_label("centroid: —", _CSS_HINT)
self.lbl_roi_size = _wrap_label("bbox: —", _CSS_HINT)
self.lbl_roi_npix = _wrap_label("pixels inside: —", _CSS_HINT)
@@ -826,6 +833,91 @@ class SrasViewerWindow(QMainWindow):
self.statusBar().showMessage(
f"Exported ROI ({n_pix} pixels) to {Path(path).name}")
def _on_export_fused_roi_csv(self):
s = self._sras
if s is None:
return
roi = self.image_canvas.get_roi()
if roi is None or not self._fused_grid_ready():
return # button is disabled in these states; defensive no-op
angles = [(a, float(s.angles_deg[a])) for a in range(s.n_angles)]
grid_note = (
"Using the live Fusion alignment canvas."
if self._alignment_result is not None else
"Using the file's shared raw grid (no live alignment result — "
"this file's angles already share one grid).")
dlg = FusedRoiExportDialog(
self, angles=angles,
availability_fn=lambda a, c: self._cached_value_image(a, c) is not None,
default_ch_idx=self.combo_channel.currentIndex(),
out_dir=str(s.path.parent), stem=s.path.stem, grid_note=grid_note)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
self._write_fused_roi_csv(
roi, dlg.get_ch_idx(), dlg.get_selected_angles(), dlg.get_output_path())
def _write_fused_roi_csv(self, roi, ch_idx: int, angle_idxs: list[int],
out_path: str):
"""Mask once on the fused grid, then one value column per selected
angle. No row/frame index columns: once angles are fused onto one
shared canvas there is no single meaningful raw (row, frame) per
output pixel, unlike the single-angle _on_export_roi_csv above."""
s = self._sras
if s is None or not angle_idxs or not out_path:
return
x_axis, y_axis = self._fused_export_axes()
mask = roi.mask_for_grid(x_axis, y_axis)
if not mask.any():
self.statusBar().showMessage(
"ROI does not overlap any pixel on the fused grid")
return
columns: list[np.ndarray] = []
used: list[int] = []
skipped: list[int] = []
for a in angle_idxs:
img = self._fused_value_image(a, ch_idx)
if img is None or img.shape != mask.shape:
# Defensive only: nothing else can mutate the caches between
# dialog-accept and this synchronous call in a single-
# threaded GUI callback, so this should never trigger — skip
# the angle rather than abort the whole export.
skipped.append(a)
continue
columns.append(img[mask].astype(np.float64))
used.append(a)
if not columns:
self.statusBar().showMessage(
"Nothing to export — no selected angle had data")
return
X, Y = np.meshgrid(np.asarray(x_axis, dtype=np.float64),
np.asarray(y_axis, dtype=np.float64))
data = np.column_stack([X[mask], Y[mask], *columns])
ch_name = CH_NAMES[ch_idx]
corners_str = " ".join(f"({p[0]:.6g},{p[1]:.6g})" for p in roi.corners())
angle_list_str = ", ".join(f"{s.angles_deg[a]:.4g}°" for a in used)
header_cols = ["x_mm", "y_mm"] + [f"v_{s.angles_deg[a]:.4g}deg" for a in used]
header = (
f"# ROI quad corners (BL BR TR TL) mm: {corners_str}\n"
f"# source: {s.path.name}, value={ch_name}, "
f"grid={'aligned canvas' if self._alignment_result is not None else 'shared raw grid'}\n"
f"# angles (deg): {angle_list_str}\n"
f"# n_pixels={int(mask.sum())}\n"
+ ",".join(header_cols)
)
np.savetxt(out_path, data, delimiter=",",
fmt=["%.6g"] * data.shape[1], header=header, comments="")
note = (f" ({len(skipped)} angle(s) skipped — no longer available)"
if skipped else "")
self.statusBar().showMessage(
f"Exported fused ROI ({int(mask.sum())} pixels, {len(used)} "
f"angle(s)) to {Path(out_path).name}{note}")
# ------------------------------------------------------------------
# ROI
# ------------------------------------------------------------------
@@ -855,6 +947,7 @@ class SrasViewerWindow(QMainWindow):
self.lbl_roi_npix.setText("pixels inside: —")
self.btn_clear_roi.setEnabled(False)
self.btn_export_roi.setEnabled(False)
self._update_fused_export_enabled()
return
cen = roi.centroid()
@@ -878,6 +971,26 @@ class SrasViewerWindow(QMainWindow):
self.lbl_roi_npix.setText(f"pixels inside: {npix}")
self.btn_clear_roi.setEnabled(True)
self.btn_export_roi.setEnabled(self._current_image is not None and npix > 0)
self._update_fused_export_enabled()
def _update_fused_export_enabled(self):
s = self._sras
roi = self.image_canvas.get_roi() if s is not None else None
if s is None:
ready, reason = False, "Open a file first."
elif roi is None:
ready, reason = False, "Draw an ROI first."
elif not self._fused_grid_ready():
ready, reason = False, (
"Angles need a common grid to fuse: run Fusion → Alignment "
"Wizard, or open a file the wizard already exported.")
else:
ready, reason = True, (
"Export the ROI as one CSV with a column per selected "
"angle's value (choose which value and which angles in "
"the dialog).")
self.btn_export_fused_roi.setEnabled(ready)
self.btn_export_fused_roi.setToolTip(reason)
# ------------------------------------------------------------------
# Display
@@ -934,6 +1047,46 @@ class SrasViewerWindow(QMainWindow):
self._aligned_cache[key] = cached
return cached
# ------------------------------------------------------------------
# Fused ROI export (Export Fused ROI…)
# ------------------------------------------------------------------
def _fused_grid_ready(self) -> bool:
"""Is there a common (x, y) grid to fuse angles onto right now?
Either a live alignment result exists (angles are reconciled onto its
canvas via apply_alignment), or — with no alignment run this session
— the open file's raw per-angle grids already coincide, which is true
for a .sras file the Alignment Wizard itself previously exported (see
scan_format.md, "Files written by the viewer's Alignment Wizard")."""
return (self._sras is not None
and (self._alignment_result is not None
or self._sras.angles_share_raw_grid()))
def _fused_export_axes(self) -> tuple[np.ndarray, np.ndarray]:
"""(x_axis, y_axis) a fused ROI export masks and labels against: the
alignment canvas when a live result exists — the more current,
deliberate source of truth even if the raw grids happen to already
match too — else the grid every angle already shares."""
if self._alignment_result is not None:
return self._aligned_canvas_axes()
s = self._sras
return s.x_axis_mm(0), s.y_positions_mm(0)
def _fused_value_image(self, angle_idx: int, ch_idx: int) -> np.ndarray | None:
"""One angle's image on the fused grid for (angle, channel), scaled
exactly like the on-screen display — so a Velocity export and the
on-screen Velocity image share the same scaling code. None if
nothing is cached/stored for this angle/channel without a compute."""
raw = self._cached_value_image(angle_idx, ch_idx)
if raw is None:
return None
img = (self._scale_for_display(raw, ch_idx)
if ch_idx in CH1_DERIVED_MODES else raw)
if self._alignment_result is not None:
return self._get_aligned_display_image(img, angle_idx, ch_idx)
return img # angles already share the raw grid — no resample needed
def _stored_fft_image(self, angle_idx: int) -> np.ndarray | None:
"""The open file's own stored peak-frequency image for this angle,
masked and ready to display, or None if the file has nothing stored
@@ -967,6 +1120,42 @@ class SrasViewerWindow(QMainWindow):
dc4_mv=self._dc_cache.get((angle_idx, CH4_IDX)),
allow_dc_recompute=False)
def _cached_value_image(self, angle_idx: int, ch_idx: int) -> np.ndarray | None:
"""The already-available (no compute) image for (angle, channel):
this window's own in-session dicts first, then the file's stored
v5/v7 cache blocks — the same two tiers, in the same cost order, that
_refresh_display and Export Fused ROI both need, so both share this
one lookup rather than drifting apart. None means genuinely nothing
is cached/stored, i.e. only a real compute could produce it — and
neither caller here is allowed to trigger one: _refresh_display falls
back to _start_compute() itself, and a fused export simply treats the
angle as unavailable.
For CH1/Velocity this is the unscaled, already-masked peak-frequency
(MHz) image; callers displaying/exporting Velocity must still run it
through _scale_for_display.
"""
s = self._sras
if s is None:
return None
if ch_idx in CH1_DERIVED_MODES:
key = self._fft_cache_key(angle_idx)
raw = self._fft_cache.get(key)
if raw is None:
raw = self._stored_fft_image(angle_idx)
if raw is not None:
# Masking the stored image is cheap but not free; keep the
# result so revisiting this angle costs nothing at all.
self._fft_cache[key] = raw
return raw
# A stored DC image needs no post-processing, so the file's own
# parsed array is served directly — as the compute path already
# does, _current_image is treated as read-only by every consumer.
cached = self._dc_cache.get((angle_idx, ch_idx))
if cached is None:
cached = s.cached_dc_mv(angle_idx, ch_idx)
return cached
def _refresh_display(self):
"""Show the image for the current angle/channel/threshold, using
cached data whenever possible and only falling back to a background
@@ -983,28 +1172,11 @@ class SrasViewerWindow(QMainWindow):
angle_idx = self.spin_angle.value()
ch_idx = self.combo_channel.currentIndex()
if ch_idx in CH1_DERIVED_MODES:
key = self._fft_cache_key(angle_idx)
raw = self._fft_cache.get(key)
if raw is None:
raw = self._stored_fft_image(angle_idx)
raw = self._cached_value_image(angle_idx, ch_idx)
if raw is not None:
# Masking the stored image is cheap but not free; keep the
# result so revisiting this angle costs nothing at all.
self._fft_cache[key] = raw
if raw is not None:
self._show_image_now(self._scale_for_display(raw, ch_idx),
angle_idx, ch_idx)
return
else:
# A stored DC image needs no post-processing, so the file's own
# parsed array is served directly — as the compute path already
# does, _current_image is treated as read-only by every consumer.
cached = self._dc_cache.get((angle_idx, ch_idx))
if cached is None:
cached = self._sras.cached_dc_mv(angle_idx, ch_idx)
if cached is not None:
self._show_image_now(cached, angle_idx, ch_idx)
img = (self._scale_for_display(raw, ch_idx)
if ch_idx in CH1_DERIVED_MODES else raw)
self._show_image_now(img, angle_idx, ch_idx)
return
# Nothing cached for these settings — need a real compute. Changing
+75
View File
@@ -0,0 +1,75 @@
"""SrasFile.angles_share_raw_grid(): the no-alignment-needed gating path for
Export Fused ROI.
A plain multi-angle scan gives each angle its own bounding box and stage
x_start (scan_format.md's whole reason v6 geometry is per-angle), so it must
read as "not shareable" without a live alignment. A file the viewer's own
Alignment Wizard exported repeats one Per-Angle Geometry record and one Row
Table span for every angle (scan_format.md, "Files written by the viewer's
Alignment Wizard"), so it must read as "shareable" with no alignment needed
at all.
No Qt: this exercises sras_format/sras_compute/sras_align_export directly,
mirroring tests/test_align_export.py.
"""
import sras_align_export as export
import sras_compute as compute
from sras_format import SrasFile
import tools.make_test_sras as gen
_THRESHOLD_MV = 80.0
def test_single_angle_file_always_shares_its_grid(tmp_path):
path = tmp_path / "one_angle.sras"
gen.write(path, n_angles=1)
sras = SrasFile(str(path))
assert sras.angles_share_raw_grid()
def test_plain_multi_angle_file_does_not_share_its_grid(tmp_path):
"""tools.make_test_sras.write gives every angle its own geometry and
stage x_start (build()'s `x_start = -0.5 + 0.1 * a`), matching how real
v6 scans vary per angle — so this must read as "not shareable"."""
path = tmp_path / "plain.sras"
gen.write(path, n_angles=3)
sras = SrasFile(str(path))
assert not sras.angles_share_raw_grid()
def test_wizard_exported_file_shares_its_grid(tmp_path):
src_path = tmp_path / "rotating.sras"
meta = gen.write_rotating(src_path, n_angles=4)
sras = SrasFile(str(src_path))
params = {a: compute.ManualAngleParams(rot, shift)
for a, (rot, shift) in meta["truth"].items()}
result = compute.build_manual_alignment(sras, 0, _THRESHOLD_MV, params)
out_path = tmp_path / "rotating_aligned.sras"
export.write_aligned_sras(sras, result, out_path)
out = SrasFile(str(out_path))
assert not sras.angles_share_raw_grid(), (
"sanity check: the *source* rotating scan must NOT already share a "
"grid, or this test would not actually exercise the wizard export")
assert out.angles_share_raw_grid()
def test_mutated_angle_breaks_the_shared_grid(tmp_path):
"""A loaded SrasFile's x_start_mm is a public per-angle array (as
tests/test_gui.py::test_alignment_geometry_is_stage_independent also
relies on) -- mutating one angle's start must be visible here too."""
src_path = tmp_path / "rotating.sras"
meta = gen.write_rotating(src_path, n_angles=3)
sras = SrasFile(str(src_path))
params = {a: compute.ManualAngleParams(rot, shift)
for a, (rot, shift) in meta["truth"].items()}
result = compute.build_manual_alignment(sras, 0, _THRESHOLD_MV, params)
out_path = tmp_path / "rotating_aligned.sras"
export.write_aligned_sras(sras, result, out_path)
out = SrasFile(str(out_path))
assert out.angles_share_raw_grid()
out.x_start_mm[1] += 1.0
assert not out.angles_share_raw_grid()
+186 -2
View File
@@ -20,11 +20,13 @@ import numpy as np
import pytest
from PyQt6.QtCore import QEventLoop, Qt, QTimer
from PyQt6.QtTest import QTest
from PyQt6.QtWidgets import QApplication, QMessageBox
from PyQt6.QtWidgets import QApplication, QDialog, QMessageBox
import sras_compute as compute
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, SrasFile
from sras_viewer import RoiQuad, SrasViewerWindow, VELOCITY_MODE_IDX
from sras_viewer import (
FusedRoiExportDialog, RoiQuad, SrasViewerWindow, VELOCITY_MODE_IDX,
)
import tools.make_test_sras as gen
@@ -670,6 +672,188 @@ def test_wizard_closes_with_a_reload(ctx):
ctx.s = win._sras
# ---------------------------------------------------------------------------
# Export Fused ROI
# ---------------------------------------------------------------------------
#
# Two independent ways angles can end up sharing one (x, y) grid to fuse
# onto: a live alignment result (case a, exercised on ctx.win -- the reload
# above restored one from the sidecar), or a file that is itself a previous
# Alignment Wizard export, whose angles already share a grid on disk with no
# alignment result needed at all (case b, exercised on a second window
# opened on ctx.written from test_wizard_export).
def test_fused_export_gating_case_a(ctx):
"""A live alignment result bridges the raw scan's per-angle grids --
angles_share_raw_grid() alone would be False here."""
win, s = ctx.win, ctx.s
assert win._alignment_result is not None, "alignment restored from sidecar"
assert not s.angles_share_raw_grid(), \
"sanity check: the raw (un-aligned) scan must not already share a grid"
x, y = win._aligned_canvas_axes()
roi = RoiQuad.from_bbox(float(x[1]), float(y[1]), float(x[-2]), float(y[-2]))
win.image_canvas.set_roi(roi)
pump(120)
assert win._fused_grid_ready()
assert win.btn_export_fused_roi.isEnabled()
def test_write_fused_roi_csv_case_a_content(ctx):
win, s = ctx.win, ctx.s
roi = win.image_canvas.get_roi()
assert roi is not None, "ROI drawn by test_fused_export_gating_case_a"
csv_path = ctx.tmpdir / "fused_roi_case_a.csv"
win._write_fused_roi_csv(roi, CH4_IDX, [0, 1], str(csv_path))
assert csv_path.exists()
lines = csv_path.read_text().splitlines()
body = [l for l in lines if not l.startswith("#")]
header, *data_lines = body
assert header == (
f"x_mm,y_mm,v_{s.angles_deg[0]:.4g}deg,v_{s.angles_deg[1]:.4g}deg")
x, y = win._fused_export_axes()
mask = roi.mask_for_grid(x, y)
assert len(data_lines) == int(mask.sum())
data = np.array([[float(v) for v in line.split(",")] for line in data_lines])
expect0 = win._fused_value_image(0, CH4_IDX)[mask]
expect1 = win._fused_value_image(1, CH4_IDX)[mask]
assert np.allclose(data[:, 2], expect0, rtol=1e-5, atol=1e-4)
assert np.allclose(data[:, 3], expect1, rtol=1e-5, atol=1e-4)
def test_fused_roi_dialog_availability_live_updates(ctx):
"""Switching the value-type radio re-evaluates every angle checkbox,
disabling/auto-unchecking whichever ones are no longer available --
independent of what actually backs availability_fn, so a synthetic
stand-in keeps this a fast, deterministic test of the dialog itself."""
win, s = ctx.win, ctx.s
angles = [(a, float(s.angles_deg[a])) for a in range(s.n_angles)]
only_angle0_has_ch1 = lambda a, c: (a == 0) if c == CH1_IDX else True
dlg = FusedRoiExportDialog(
win, angles=angles, availability_fn=only_angle0_has_ch1,
default_ch_idx=CH1_IDX, out_dir=str(s.path.parent), stem=s.path.stem,
grid_note="test")
try:
assert dlg._angle_checks[0].isEnabled()
assert all(not dlg._angle_checks[a].isEnabled()
for a in range(1, s.n_angles))
dlg._angle_checks[0].setChecked(True)
dlg._val_buttons[CH4_IDX].click()
assert all(dlg._angle_checks[a].isEnabled() for a in range(s.n_angles)), \
"CH4 is available for every angle"
assert dlg._angle_checks[0].isChecked(), \
"stays checked -- still available under CH4"
if s.n_angles > 1:
dlg._angle_checks[1].setChecked(True)
dlg._val_buttons[CH1_IDX].click()
assert dlg._angle_checks[0].isChecked()
if s.n_angles > 1:
assert not dlg._angle_checks[1].isEnabled()
assert not dlg._angle_checks[1].isChecked(), \
"auto-unchecked: angle 1 has no data under CH1"
finally:
dlg.close()
def test_fused_roi_dialog_select_all_none(ctx):
win, s = ctx.win, ctx.s
angles = [(a, float(s.angles_deg[a])) for a in range(s.n_angles)]
dlg = FusedRoiExportDialog(
win, angles=angles, availability_fn=lambda a, c: c == CH4_IDX,
default_ch_idx=CH4_IDX, out_dir=str(s.path.parent), stem=s.path.stem,
grid_note="test")
try:
assert not dlg._btn_export.isEnabled(), "nothing checked yet"
dlg._on_select_all_available()
assert all(cb.isChecked() for cb in dlg._angle_checks.values())
assert dlg._btn_export.isEnabled()
dlg._on_select_none()
assert not any(cb.isChecked() for cb in dlg._angle_checks.values())
assert not dlg._btn_export.isEnabled()
finally:
dlg.close()
def test_on_export_fused_roi_csv_end_to_end(ctx):
win, s = ctx.win, ctx.s
roi = win.image_canvas.get_roi()
assert roi is not None, "ROI from the earlier fused-export tests is still set"
csv_path = ctx.tmpdir / "fused_roi_e2e.csv"
with patch("sras_viewer.main_window.FusedRoiExportDialog") as MockDlg:
inst = MockDlg.return_value
inst.exec.return_value = QDialog.DialogCode.Accepted
inst.get_ch_idx.return_value = CH4_IDX
inst.get_selected_angles.return_value = [0, 1]
inst.get_output_path.return_value = str(csv_path)
win.btn_export_fused_roi.click()
assert csv_path.exists()
kwargs = MockDlg.call_args.kwargs
assert kwargs["default_ch_idx"] == win.combo_channel.currentIndex()
assert kwargs["out_dir"] == str(s.path.parent)
assert kwargs["stem"] == s.path.stem
def test_fused_export_no_alignment_shared_grid_path(ctx):
"""ctx.written (from test_wizard_export) is itself a previous Alignment
Wizard export: opened fresh with no sidecar for its own path, so no
alignment result is ever restored -- but its angles already share one
grid on disk, so the export must work through the no-resample path."""
win2 = SrasViewerWindow()
try:
win2._load_file(str(ctx.written.path))
assert wait_until(lambda: win2._sras is not None)
s2 = win2._sras
assert win2._alignment_result is None, \
"no sidecar exists for this path -- nothing auto-restored"
assert s2.angles_share_raw_grid(), \
"a wizard export already shares one grid across angles"
assert wait_until(lambda: all((a, CH4_IDX) in win2._dc_cache
for a in range(s2.n_angles))), \
"DC precomputed for every angle"
x, y = s2.x_axis_mm(0), s2.y_positions_mm(0)
roi = RoiQuad.from_bbox(float(x[1]), float(y[1]), float(x[-2]), float(y[-2]))
win2.image_canvas.set_roi(roi)
pump(120)
assert win2._fused_grid_ready()
assert win2.btn_export_fused_roi.isEnabled()
aligned_cache_before = len(win2._aligned_cache)
angle_idxs = list(range(min(2, s2.n_angles)))
csv_path = ctx.tmpdir / "fused_roi_case_b.csv"
win2._write_fused_roi_csv(roi, CH4_IDX, angle_idxs, str(csv_path))
assert csv_path.exists()
assert len(win2._aligned_cache) == aligned_cache_before, \
"no-resample path must never touch apply_alignment"
header = next(l for l in csv_path.read_text().splitlines()
if not l.startswith("#"))
expected_header = "x_mm,y_mm," + ",".join(
f"v_{s2.angles_deg[a]:.4g}deg" for a in angle_idxs)
assert header == expected_header
# Break the shared grid and confirm gating flips off.
mutate_idx = 1 if s2.n_angles > 1 else 0
s2.x_start_mm[mutate_idx] += 1.0
assert not s2.angles_share_raw_grid()
win2._update_fused_export_enabled()
assert not win2._fused_grid_ready()
assert not win2.btn_export_fused_roi.isEnabled()
assert "Alignment Wizard" in win2.btn_export_fused_roi.toolTip()
finally:
win2.close()
pump(200)
def test_pixel_inspector(ctx):
win = ctx.win
win.chk_aligned_view.setChecked(False)