diff --git a/sras_format.py b/sras_format.py index 13ef64f..469c607 100644 --- a/sras_format.py +++ b/sras_format.py @@ -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 diff --git a/sras_viewer/__init__.py b/sras_viewer/__init__.py index 4014385..5cae377 100644 --- a/sras_viewer/__init__.py +++ b/sras_viewer/__init__.py @@ -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 diff --git a/sras_viewer/dialogs.py b/sras_viewer/dialogs.py index 47c067d..87379eb 100644 --- a/sras_viewer/dialogs.py +++ b/sras_viewer/dialogs.py @@ -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() + diff --git a/sras_viewer/main_window.py b/sras_viewer/main_window.py index fa6ceed..b02d6e7 100644 --- a/sras_viewer/main_window.py +++ b/sras_viewer/main_window.py @@ -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,29 +1172,12 @@ 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) - 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) - return + raw = self._cached_value_image(angle_idx, ch_idx) + if raw is not None: + 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 # the DC threshold changes *which* pixels get an FFT at all, so it diff --git a/tests/test_fused_export.py b/tests/test_fused_export.py new file mode 100644 index 0000000..ebd8f22 --- /dev/null +++ b/tests/test_fused_export.py @@ -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() diff --git a/tests/test_gui.py b/tests/test_gui.py index 3a26abc..4b184ff 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -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)