"""The SrasViewerWindow main window and application entry point.""" import sys from collections import Counter from pathlib import Path import numpy as np from matplotlib.backends.backend_qtagg import NavigationToolbar2QT from PyQt6.QtCore import QObject, QSettings, QSignalBlocker, Qt, QThread from PyQt6.QtGui import QAction from PyQt6.QtWidgets import ( QApplication, QCheckBox, QComboBox, QDialog, QFileDialog, QFrame, QHBoxLayout, QLabel, QMainWindow, QProgressDialog, QPushButton, QSizePolicy, QSpinBox, QSplitter, QVBoxLayout, QWidget, ) import sras_compute as compute from sras_compute import ( ManualAngleParams, apply_alignment, build_manual_alignment, load_manual_alignment, save_manual_alignment, sidecar_path, ) from sras_format import ( CH3_IDX, CH4_IDX, CH_NAMES, SrasFile, mv_to_adc, _FALLBACK_YMULT_MV, _FALLBACK_YOFF_ADC, ) from sras_workers import ( BatchCacheWorker, BatchExportImagesWorker, ComputeWorker, DcPrecomputeWorker, LoadWorker, ) from .canvases import ImageCanvas, WaveformCanvas from .common import ( CH1_DERIVED_MODES, CH_LABELS, CMAPS, VELOCITY_MODE_IDX, _CHANNEL_DISPLAY, _CSS_BUSY, _axes_extent, _CSS_HINT, _CSS_INFO, _CSS_MUTED, _CSS_WARN, _LEFT_PANEL_W, _MASKED_HIGHLIGHT_COLOR, _RIGHT_PANEL_W, Jobs, _combo, _form, _group, _make_dspin, _scroll_panel, _wrap_label, ) from .align_wizard import AlignmentWizard from .dialogs import ( FftOptionsDialog, FusedRoiExportDialog, RowAverageFftOptionsDialog, ) # --------------------------------------------------------------------------- # Main window # --------------------------------------------------------------------------- class SrasViewerWindow(QMainWindow): def __init__(self, initial_path: str | None = None): super().__init__() self.setWindowTitle("SRAS Scan Viewer") self.resize(1560, 840) self.setMinimumSize(960, 560) self.setAcceptDrops(True) self._sras: SrasFile | None = None self._current_image: np.ndarray | None = None self._current_angle: int = 0 self._current_ch: int = 0 self._pending_angle: int = 0 self._pending_ch: int = 0 self._pending_bg_sub: bool = True self._pending_threshold: float = 50.0 # mV self._pending_min_freq_mhz: float = 0.0 self._pending_fft_pad_factor: int = 1 # Live background jobs, keyed by role — see _run_worker. self._jobs: dict[str, tuple] = {} self._progress_dlgs: dict[str, QProgressDialog] = {} # FFT settings (configured via FFT Options dialog, persisted across # sessions). IniFormat: predictable cross-platform and redirectable # in tests. self._settings = QSettings(QSettings.Format.IniFormat, QSettings.Scope.UserScope, "sras-viewer", "sras-viewer") try: pad = int(self._settings.value("fft/pad_factor", 1)) except (TypeError, ValueError): pad = 1 self._fft_pad_factor: int = max(1, min(256, pad)) # 1 = no padding try: row_avg_n = int(self._settings.value("fft/row_avg_n", 3)) except (TypeError, ValueError): row_avg_n = 3 self._pending_row_avg_n: int = max(1, min(50, row_avg_n)) # Convert menu: batch DC/FFT compute-and-store (v6 -> v7) self._batch_errors: list[str] = [] # Display-only settings (colormap, grating) never trigger a # recompute — they're applied to cached data on redraw. DC images # (CH3/CH4) are cheap and precomputed for every angle in the # background right after load. CH1/Velocity FFT images are # computed lazily (with a progress popup) the first time an # angle/threshold combination is viewed — using the cached DC4 # image to skip the FFT entirely for masked-out pixels — and # cached per (angle, threshold, min peak freq) so revisiting the # same combination is free. bg-sub/pad are deliberately not part of # the key: once an angle has any FFT image (live or from the file's # own stored cache), it stays displayed regardless of those # controls — see _fft_cache_key. self._dc_cache: dict[tuple[int, int], np.ndarray] = {} self._fft_cache: dict[tuple[int, float, float], np.ndarray] = {} self._dc_generation: int = 0 # Angle alignment ("Fusion" menu) self._alignment_result = None self._aligned_cache: dict[tuple, np.ndarray] = {} self._align_wizard: AlignmentWizard | None = None self._build_ui() if initial_path: self._load_file(initial_path) # ------------------------------------------------------------------ # Background job plumbing # ------------------------------------------------------------------ def _run_worker(self, key: str, worker: QObject, *, connect: tuple = (), quit_on: tuple = ("finished",), on_done=None) -> bool: """Move *worker* onto its own QThread and start it. Returns False if a job under *key* is already running. Centralises two lifetime hazards that each cost a process abort: 1. The job is claimed in self._jobs *before* start() and before anything below that can pump the Qt event loop (a QProgressDialog.show() does on first display). If it weren't, a re-entrant editingFinished could slip past the busy check, start a second thread, and then have the first call's own assignment clobber — and destroy while still running — that second QThread. 2. thread.finished fires as the thread winds down but does not guarantee the OS thread has joined. Dropping the last reference to a QThread whose thread is still running logs "QThread: Destroyed while thread is still running" and aborts, so wait() first. """ if key in self._jobs: return False thread = QThread() self._jobs[key] = (thread, worker, on_done) # claim before anything pumps worker.moveToThread(thread) thread.started.connect(worker.run) for signal_name, slot in connect: getattr(worker, signal_name).connect(slot) for signal_name in quit_on: getattr(worker, signal_name).connect(thread.quit) thread.finished.connect(lambda k=key: self._on_job_finished(k)) thread.start() return True def _on_job_finished(self, key: str): job = self._jobs.pop(key, None) if job is None: return thread, _worker, on_done = job thread.wait() # join before releasing our last reference if on_done is not None: on_done() def _job_running(self, key: str) -> bool: return key in self._jobs # ------------------------------------------------------------------ # UI construction # ------------------------------------------------------------------ def _build_ui(self): central = QWidget() self.setCentralWidget(central) root = QHBoxLayout(central) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) root.addWidget(self._build_left_panel()) root.addWidget(self._build_canvases(), stretch=1) root.addWidget(self._build_right_panel()) self.statusBar().showMessage("Open an .sras file to begin.") self._build_menus() def _build_left_panel(self) -> QWidget: panel = QWidget() panel_layout = QVBoxLayout(panel) panel_layout.setContentsMargins(0, 0, 0, 0) panel_layout.setSpacing(8) panel_layout.addWidget(self._build_file_group()) panel_layout.addWidget(self._build_info_group()) panel_layout.addWidget(self._build_view_group()) panel_layout.addWidget(self._build_roi_group()) panel_layout.addStretch() return _scroll_panel(panel, _LEFT_PANEL_W) def _build_file_group(self) -> QWidget: grp_file, fl = _group("File") self.btn_open = QPushButton("Open .sras…") self.btn_open.clicked.connect(self._on_open) self.lbl_filename = _wrap_label("No file loaded", _CSS_MUTED) fl.addWidget(self.btn_open) fl.addWidget(self.lbl_filename) return grp_file def _build_info_group(self) -> QWidget: grp_info, il = _group("Scan Info") il.setSpacing(3) self._info = {} for key in ("Angles", "Rows", "Frames / row", "Samples / frame", "Sample rate", "X start", "Pixel Δx", "Laser freq"): lbl = _wrap_label(f"{key}: —", _CSS_INFO) il.addWidget(lbl) self._info[key] = lbl # frame-count / format notes self.lbl_frame_warn = _wrap_label("", _CSS_WARN) il.addWidget(self.lbl_frame_warn) # background DC-precompute progress self.lbl_dc_precompute = _wrap_label("", _CSS_BUSY) il.addWidget(self.lbl_dc_precompute) return grp_info def _build_view_group(self) -> QWidget: grp_view, vl = _group("View Settings") view_form = _form() self.spin_angle = QSpinBox() self.spin_angle.setRange(0, 0) self.spin_angle.setEnabled(False) self.spin_angle.setMinimumWidth(64) # valueChanged with keyboard tracking off, not editingFinished: the # latter fires only on Return or focus-out, so stepping the angle (an # arrow click or Up/Down, the ordinary way to walk a scan) changed the # number and left the image behind. Tracking off is what keeps # valueChanged from also firing per keystroke mid-typing, which on a # large scan would launch a compute for every intermediate angle. self.spin_angle.setKeyboardTracking(False) self.spin_angle.valueChanged.connect(self._on_view_changed) self.lbl_angle_deg = QLabel("—") angle_field = QWidget() ar = QHBoxLayout(angle_field) ar.setContentsMargins(0, 0, 0, 0) ar.setSpacing(6) ar.addWidget(self.spin_angle) ar.addWidget(self.lbl_angle_deg) ar.addStretch() view_form.addRow("Angle:", angle_field) self.combo_channel = _combo(CH_LABELS, min_chars=12) self.combo_channel.setEnabled(False) self.combo_channel.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self.combo_channel.currentIndexChanged.connect(self._on_channel_changed) view_form.addRow("Channel:", self.combo_channel) vl.addLayout(view_form) sep = QFrame() sep.setFrameShape(QFrame.Shape.HLine) sep.setStyleSheet("color: #555;") vl.addWidget(sep) # DC threshold (for RF / CH1 masking) self.grp_threshold, tl = _group("RF Mask Threshold (CH1 only)") thr_form = _form() self.spin_threshold_mv = _make_dspin(-500.0, 500.0, 3, suffix=" mV", value=50.0, step=0.025) self.spin_threshold_mv.setEnabled(False) self.spin_threshold_mv.editingFinished.connect(self._on_threshold_changed) thr_form.addRow("DC threshold:", self.spin_threshold_mv) tl.addLayout(thr_form) self.lbl_threshold_adc = _wrap_label( f"≈ {mv_to_adc(50.0):.1f} ADC counts", _CSS_MUTED) tl.addWidget(self.lbl_threshold_adc) min_freq_form = _form() self.spin_min_freq_mhz = _make_dspin(0.0, 10000.0, 3, suffix=" MHz", value=0.0, step=1.0) self.spin_min_freq_mhz.setEnabled(False) self.spin_min_freq_mhz.setToolTip( "Excludes every FFT bin below this frequency from the peak\n" "search (0 = off, only true DC is excluded). A pixel can pass\n" "the DC threshold above yet still carry only weak real signal —\n" "when that happens, the un-subtracted background's DC-leakage\n" "skirt can have more power than the genuine signal, so the peak\n" "search resolves to a near-zero frequency even though the pixel\n" "is real, valid data. Raising this floor above that skirt forces\n" "the search to report the strongest peak that is plausibly real\n" "signal instead.\n" "Raising the floor also re-masks images already shown or stored\n" "in this file's cache: pixels whose stored peak falls below it\n" "display as invalid. Lowering it below a stored cache's own\n" "floor needs a recompute (bins below that floor were never\n" "searched). Batch Compute FFT records the floor in the file and\n" "re-resolves masked pixels to their true above-floor peak." ) self.spin_min_freq_mhz.editingFinished.connect(self._on_min_freq_changed) min_freq_form.addRow("Min peak freq:", self.spin_min_freq_mhz) tl.addLayout(min_freq_form) vl.addWidget(self.grp_threshold) # Background subtraction (v4+ files only) self.chk_bg_sub = QCheckBox("Background subtraction (CH1 only)") self.chk_bg_sub.setChecked(True) self.chk_bg_sub.setEnabled(False) self.chk_bg_sub.setToolTip( "Subtract the stored background waveform from each CH1 frame\n" "before computing the FFT (v4+ files only). Applies to angles\n" "not yet computed and to future batch recomputes — it does not\n" "change an image already shown or already stored in the file." ) self.chk_bg_sub.toggled.connect(self._on_bg_sub_toggled) vl.addWidget(self.chk_bg_sub) # Aligned View (Fusion → Angle Alignment result) self.chk_aligned_view = QCheckBox("Aligned View (Fusion)") self.chk_aligned_view.setChecked(False) self.chk_aligned_view.setEnabled(False) self.chk_aligned_view.setToolTip( "Show the current angle/channel resampled onto the shared,\n" "rotation+translation-aligned canvas from Fusion → Angle\n" "Alignment. Uncheck to see the raw per-angle scan grid." ) self.chk_aligned_view.toggled.connect(self._on_aligned_view_toggled) vl.addWidget(self.chk_aligned_view) self.btn_export_csv = QPushButton("Export Image as CSV…") self.btn_export_csv.setEnabled(False) self.btn_export_csv.setToolTip( "Save the current CH1 image (one scan row per CSV line).") self.btn_export_csv.clicked.connect(self._on_export_csv) vl.addWidget(self.btn_export_csv) return grp_view def _build_roi_group(self) -> QWidget: grp_roi, rl = _group("ROI (Region of Interest)") self.btn_draw_roi = QPushButton("Draw ROI") self.btn_draw_roi.setCheckable(True) self.btn_draw_roi.setEnabled(False) self.btn_draw_roi.setToolTip( "Arm next click+drag on the image to draw a new ROI\n" "(replaces any existing one). Click again to cancel.\n" "After drawing, drag inside to move, or grab corners to reshape.\n" "The ROI is persistent across channels / modes / angles." ) self.btn_draw_roi.toggled.connect(self._on_draw_roi_toggled) rl.addWidget(self.btn_draw_roi) self.btn_clear_roi = QPushButton("Clear ROI") self.btn_clear_roi.setEnabled(False) self.btn_clear_roi.clicked.connect(self._on_clear_roi) rl.addWidget(self.btn_clear_roi) self.btn_export_roi = QPushButton("Export ROI as CSV…") self.btn_export_roi.setEnabled(False) self.btn_export_roi.setToolTip( "Save every pixel whose centre lies inside the ROI as CSV.\n" "Columns: row, frame, x_mm, y_mm, value.\n" "Corner coordinates of the quad are written in the file header." ) 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) for lbl in (self.lbl_roi_center, self.lbl_roi_size, self.lbl_roi_npix): rl.addWidget(lbl) return grp_roi def _build_canvases(self) -> QWidget: splitter = QSplitter(Qt.Orientation.Vertical) splitter.setChildrenCollapsible(False) img_widget = QWidget() img_vl = QVBoxLayout(img_widget) img_vl.setContentsMargins(0, 0, 0, 0) img_vl.setSpacing(4) self.image_canvas = ImageCanvas() self.image_canvas.setMinimumHeight(220) self.image_canvas.pixel_clicked.connect(self._on_pixel_clicked) self.image_canvas.roi_changed.connect(self._update_roi_ui) self.image_canvas.draw_mode_changed.connect(self._on_draw_mode_changed) img_vl.addWidget(NavigationToolbar2QT(self.image_canvas, img_widget)) img_vl.addWidget(self.image_canvas) splitter.addWidget(img_widget) wave_widget = QWidget() wave_vl = QVBoxLayout(wave_widget) wave_vl.setContentsMargins(0, 0, 0, 0) wave_vl.setSpacing(4) self.lbl_wave_hint = QLabel( "Click a pixel in the image above to inspect its waveform.") self.lbl_wave_hint.setAlignment(Qt.AlignmentFlag.AlignCenter) self.lbl_wave_hint.setStyleSheet(_CSS_MUTED) self.wave_canvas = WaveformCanvas() self.wave_canvas.setMinimumHeight(150) wave_vl.addWidget(self.lbl_wave_hint) wave_vl.addWidget(self.wave_canvas) splitter.addWidget(wave_widget) splitter.setStretchFactor(0, 3) splitter.setStretchFactor(1, 1) splitter.setSizes([580, 250]) return splitter def _build_right_panel(self) -> QWidget: # Velocity settings (visible only in velocity mode) self.grp_velocity, vel_l = _group("Velocity Settings (CH1 only)") vel_form = _form() self.spin_grating_um = _make_dspin(0.1, 1000.0, 2, suffix=" µm", value=25, step=0.5) self.spin_grating_um.setEnabled(False) self.spin_grating_um.editingFinished.connect(self._on_grating_changed) vel_form.addRow("Grating size:", self.spin_grating_um) vel_l.addLayout(vel_form) vel_l.addWidget(_wrap_label("v (m/s) = freq (MHz) × grating (µm)", "font-size: 10px; color: #888;")) self.grp_velocity.setVisible(False) grp_display, dl = _group("Display Options") cmap_form = _form() self.combo_cmap = QComboBox() self.combo_cmap.addItems(CMAPS) self.combo_cmap.setCurrentText("gray") self.combo_cmap.setEnabled(False) self.combo_cmap.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) self.combo_cmap.currentIndexChanged.connect(self._on_cmap_changed) cmap_form.addRow("Colormap:", self.combo_cmap) dl.addLayout(cmap_form) self.chk_auto = QCheckBox("Auto-scale colormap") self.chk_auto.setChecked(True) self.chk_auto.toggled.connect(self._on_autoscale_toggled) dl.addWidget(self.chk_auto) self.chk_highlight_masked = QCheckBox("Highlight masked (no-data) pixels") self.chk_highlight_masked.setChecked(True) self.chk_highlight_masked.setEnabled(False) self.chk_highlight_masked.setToolTip( "RF/Velocity only. A pixel below the DC threshold has no FFT\n" "result at all and is otherwise shown as 0 on the same grayscale\n" "ramp as a real, valid pixel whose peak just happens to be a low\n" "frequency — the two look identical (both near-black). When\n" "checked, masked pixels are drawn in a distinct highlight color\n" "instead, excluded from the colormap's data range, so real\n" "low-frequency pixels keep their own true shade. Uncheck to\n" "restore the old behavior where both blend together." ) self.chk_highlight_masked.toggled.connect(self._on_highlight_masked_toggled) dl.addWidget(self.chk_highlight_masked) range_form = _form() for label, attr in (("min:", "spin_vmin"), ("max:", "spin_vmax")): spin = _make_dspin(-1e9, 1e9, 4) spin.setEnabled(False) spin.editingFinished.connect(self._on_manual_range_changed) setattr(self, attr, spin) range_form.addRow(label, spin) dl.addLayout(range_form) right_panel = QWidget() layout = QVBoxLayout(right_panel) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(8) layout.addWidget(self.grp_velocity) layout.addWidget(grp_display) layout.addStretch() return _scroll_panel(right_panel, _RIGHT_PANEL_W) def _build_menus(self): menubar = self.menuBar() fft_menu = menubar.addMenu("&FFT") fft_act = QAction("FFT &Options…", self) fft_act.setStatusTip("Configure FFT zero-padding") fft_act.triggered.connect(self._on_fft_options) fft_menu.addAction(fft_act) fusion_menu = menubar.addMenu("&Fusion") self._wizard_act = QAction("Alignment &Wizard…", self) self._wizard_act.setStatusTip( "Align the angles, crop to a region of interest, and save the " "aligned data as a new .sras file. Requires >1 angle.") self._wizard_act.setEnabled(False) self._wizard_act.triggered.connect(self._on_alignment_wizard) fusion_menu.addAction(self._wizard_act) convert_menu = menubar.addMenu("&Convert") self._batch_dc_act = QAction("Batch Compute DC and &Store…", self) self._batch_dc_act.setStatusTip( "Select .sras files and compute+store DC images (CH3/CH4 mean) " "for every angle, converting v6 files to v7 in place.") self._batch_dc_act.triggered.connect(lambda: self._on_batch_compute("dc")) convert_menu.addAction(self._batch_dc_act) self._batch_fft_act = QAction("Batch Compute FFT and Sto&re…", self) self._batch_fft_act.setStatusTip( "Select .sras files and compute+store FFT peak-frequency images " "for every angle, converting v6 files to v7 in place. Stored " "images are natural-resolution (pad 1); padded views compute live.") self._batch_fft_act.triggered.connect(lambda: self._on_batch_compute("fft")) convert_menu.addAction(self._batch_fft_act) self._batch_fft_rowavg_act = QAction( "Batch Compute Row-A&veraged FFT and Store…", self) self._batch_fft_rowavg_act.setStatusTip( "Select .sras files and compute+store a same-row, distance-weighted " "smoothed FFT peak-frequency image for every angle — improves SNR " "on noisy regions. DC images and raw waveform data are never " "touched; the raw (unsmoothed) FFT is always a recompute away. " "Converts v6 files to v7 in place.") self._batch_fft_rowavg_act.triggered.connect(self._on_batch_compute_row_avg) convert_menu.addAction(self._batch_fft_rowavg_act) convert_menu.addSeparator() self._batch_export_images_act = QAction( "Batch Export View as &Images…", self) self._batch_export_images_act.setStatusTip( "Select .sras files and export the currently selected view " "(channel, angle, threshold, colormap, etc.) as one PNG per " "file, rendered the same way it is shown on screen. Read-only " "— never modifies the source files. Aligned View is ignored " "even if enabled, since an alignment result belongs to one " "specific file's geometry.") self._batch_export_images_act.triggered.connect( self._on_batch_export_images) convert_menu.addAction(self._batch_export_images_act) # ------------------------------------------------------------------ # Drag-and-drop # ------------------------------------------------------------------ def dragEnterEvent(self, event): urls = event.mimeData().urls() if urls and urls[0].toLocalFile().lower().endswith(".sras"): event.acceptProposedAction() def dropEvent(self, event): self._load_file(event.mimeData().urls()[0].toLocalFile()) # ------------------------------------------------------------------ # File loading # ------------------------------------------------------------------ def _on_open(self): path, _ = QFileDialog.getOpenFileName( self, "Open SRAS File", "", "SRAS Files (*.sras);;All Files (*)") if path: self._load_file(path) def _load_file(self, path: str): started = self._run_worker( Jobs.LOAD, LoadWorker(path), connect=( ("finished", self._on_load_done), ("error", lambda msg: self.statusBar().showMessage(f"Error: {msg}")), ), ) if not started: return self.btn_open.setEnabled(False) self.statusBar().showMessage(f"Loading {Path(path).name}…") self._show_progress("main", f"Loading {Path(path).name}…") def _on_load_done(self, sras): self._close_progress("main") self.btn_open.setEnabled(True) if sras is None: return self._sras = sras self._current_image = None # A manual-alignment dialog bound to the previous file must not # survive a reload — its per-angle state (and the sras it was # constructed against) no longer matches the new file's geometry. if self._align_wizard is not None: self._align_wizard.close() self._align_wizard = None # Caches (and any in-flight DC precompute) belong to the previous # file's geometry — discard and start fresh. Bumping the generation # counters makes any still-running worker's result get dropped when # it lands. self._dc_cache = {} self._fft_cache = {} self._dc_generation += 1 self.lbl_dc_precompute.setText("") self._apply_alignment_result(None, view_checked=False) # Silently restore a previously-saved manual alignment, if any, so # the work survives closing and reopening the file. sidecar = load_manual_alignment(sras) if sidecar is not None: try: self._apply_alignment_result( build_manual_alignment( sras, sidecar.ref_angle_idx, sidecar.dc_threshold_mv, sidecar.per_angle), view_checked=True) self.statusBar().showMessage( f"Restored saved manual alignment from " f"{sidecar_path(sras.path).name}") except Exception as exc: # A corrupt/foreign sidecar or a rescan that shrank n_angles # below ref_angle_idx must not block opening the .sras file. self.statusBar().showMessage( f"Could not restore saved alignment: {exc}") # A ROI from the previous file no longer matches the new scan's # geometry, so discard it on every load. self.image_canvas.clear_roi() self.lbl_filename.setText(sras.path.name) with QSignalBlocker(self.spin_angle): self.spin_angle.setRange(0, max(0, sras.n_angles - 1)) self.spin_angle.setValue(0) # DC channels are cheap and give an instant, fluid overview of a # scan; CH1/Velocity require an FFT per pixel that can take minutes # on a large scan, so don't default to it. with QSignalBlocker(self.combo_channel): self.combo_channel.setCurrentIndex(CH4_IDX) self._update_controls_enabled(True) self._on_threshold_changed() # refresh ADC label with file calibration self._on_view_changed() self._start_dc_precompute() # ------------------------------------------------------------------ # Scan info panel # ------------------------------------------------------------------ def _update_scan_info_labels(self): s = self._sras if s is None: return a = self.spin_angle.value() for key, text in ( ("Angles", f"{s.n_angles}"), ("Rows", f"{s.n_rows[a]}"), ("Frames / row", f"{s.n_frames[a]}"), ("Samples / frame", f"{s.samples_per_frame}"), ("Sample rate", f"{s.sample_rate_hz / 1e9:.4g} GS/s"), ("X start", f"{s.x_start_mm[a]:.4g} mm"), ("Pixel Δx", f"{s.pixel_x_mm * 1e3:.3g} µm"), ("Laser freq", f"{s.laser_freq_hz / 1e3:.4g} kHz"), ): self._info[key].setText(f"{key}: {text}") notes = [] if s.frame_count_mismatch: notes.append(f"! Header n_frames={s.n_frames_header}, " f"actual={s.n_frames[a]} (scanner bug — corrected)") if s.scan_aborted: notes.append(f"! Scan aborted: {s.n_angles}/{s.n_angles_declared} " "angles complete") if s.background is not None: notes.append(f"Background waveform: {len(s.background)} samples") if s.version in (6, 7): notes.append("v6/v7 format: rows / frames / x_start are per-angle") n_dc = sum(1 for x in s.precomputed_dc4_mv if x is not None) n_fft = sum(1 for x in s.precomputed_freq_mhz if x is not None) if n_dc or n_fft: bg_note = " (bg-sub)" if s.precomputed_bg_sub else " (no bg-sub)" avg_note = (f", row-averaged n={s.precomputed_row_avg_n}" if s.precomputed_row_avg_n else "") pad_note = (f", pad {s.precomputed_pad_factor}x" if s.precomputed_pad_factor > 1 else "") floor_note = (f", floor ≥ {s.precomputed_min_freq_mhz:g} MHz" if s.precomputed_min_freq_mhz > 0 else "") notes.append( f"Cached images: DC {n_dc}/{s.n_angles} angles, " f"FFT {n_fft}/{s.n_angles} angles{bg_note if n_fft else ''}" f"{avg_note if n_fft else ''}{pad_note if n_fft else ''}" f"{floor_note if n_fft else ''} " "— display is instant for cached angles") notes += self._cache_mismatch_notes() elif s.version == 7: notes.append("v7 format: no cache blocks stored yet") self.lbl_frame_warn.setText("\n".join(notes)) def _cache_mismatch_notes(self) -> list[str]: """Informational only: whether the file's stored FFT cache was computed under different bg-sub/pad/min-freq settings than these controls currently say. The display always shows the stored image (re-masked for a raised floor) regardless — see _stored_fft_image; these controls only affect a future live compute for an angle with nothing cached yet, or an explicit batch recompute, never what's already on screen. row_avg_n is compared against the file's own recorded value (a self-match), so it never contributes a reason here — there's no live control for it to diverge from, and the "Cached images" line above already reports it. The min peak frequency floor gets two directions: a live floor *below* the stored one is a genuine mismatch reason (the stored search never looked below its floor), while a live floor *above* the stored one is servable — pixels under it are shown as masked — so that direction gets its own softer note. Asks compute for the reasons rather than restating the accept rule, so a new provenance field can only be added in one place. """ s = self._sras if s is None or all(x is None for x in s.precomputed_freq_mhz): return [] live_floor = self.spin_min_freq_mhz.value() notes = [] reasons = compute.cache_mismatch_reasons( s, n_fft=self._current_n_fft(), apply_bg_sub=self.chk_bg_sub.isChecked(), row_avg_n=s.precomputed_row_avg_n, min_freq_mhz=live_floor) if reasons: notes.append( "Note: current bg-sub/pad/min-freq controls differ from the " "stored cache — " + "; ".join(reasons) + ". Shown as stored; " "use Batch Compute to recompute with these settings.") if round(live_floor * 1000) > round(s.precomputed_min_freq_mhz * 1000): notes.append( f"Note: Min peak freq ({live_floor:g} MHz) is above the " f"stored cache's floor ({s.precomputed_min_freq_mhz:g} MHz) — " f"stored pixels whose peak falls below {live_floor:g} MHz are " "shown as masked; Batch Compute FFT re-resolves them above " "the floor instead.") return notes # ------------------------------------------------------------------ # Controls # ------------------------------------------------------------------ def _update_controls_enabled(self, enabled: bool): s = self._sras has_file = enabled and s is not None ch_idx = self.combo_channel.currentIndex() is_ch1 = enabled and ch_idx in CH1_DERIVED_MODES is_vel = enabled and ch_idx == VELOCITY_MODE_IDX self.spin_angle.setEnabled(has_file and s.n_angles > 1) self.combo_channel.setEnabled(enabled) self.combo_cmap.setEnabled(enabled) self.chk_auto.setEnabled(enabled) manual = enabled and not self.chk_auto.isChecked() self.spin_vmin.setEnabled(manual) self.spin_vmax.setEnabled(manual) # Threshold and bg-sub apply to all CH1 modes self.spin_threshold_mv.setEnabled(is_ch1) self.spin_min_freq_mhz.setEnabled(is_ch1) self.chk_bg_sub.setEnabled(has_file and s.background is not None and is_ch1) self.chk_highlight_masked.setEnabled(is_ch1) self.spin_grating_um.setEnabled(is_vel) self.grp_velocity.setVisible(is_vel) self.btn_export_csv.setEnabled(is_ch1 and self._current_image is not None) # ROI: always usable once a file is loaded (independent of channel) self.btn_draw_roi.setEnabled(has_file) # Batch Convert actions pick their own files, independent of # whatever's currently open — only gated on no batch already running. can_batch = not self._job_running(Jobs.BATCH) self._batch_dc_act.setEnabled(can_batch) self._batch_fft_act.setEnabled(can_batch) self._batch_fft_rowavg_act.setEnabled(can_batch) self._wizard_act.setEnabled( has_file and s.n_angles > 1 and self._align_wizard is None) self.chk_aligned_view.setEnabled(enabled and self._alignment_result is not None) self._update_roi_ui() def _on_channel_changed(self): self._update_controls_enabled(self._sras is not None) self._on_view_changed() def _on_bg_sub_toggled(self): # bg-sub no longer gates the display: it only affects a future live # compute for an angle with nothing cached yet, or an explicit batch # recompute — never what's already shown. Just keep the info panel's # divergence note current. if self._is_fft_mode(): self._update_scan_info_labels() def _on_grating_changed(self): # Grating is a pure post-multiply on the cached frequency image — # never needs a recompute. if self._sras is not None and self.combo_channel.currentIndex() == VELOCITY_MODE_IDX: self._refresh_display() def _on_threshold_changed(self): mv = self.spin_threshold_mv.value() cal = (self._sras.cal(CH4_IDX) if self._sras is not None else (_FALLBACK_YMULT_MV, _FALLBACK_YOFF_ADC, 0.0)) self.lbl_threshold_adc.setText(f"≈ {mv_to_adc(mv, *cal):.1f} ADC counts") # Threshold decides which pixels get an FFT at all, so changing it is # a genuine cache-key change — but the recompute reuses the cached DC4 # image to skip masked-out pixels. if self._is_fft_mode(): self._refresh_display() def _on_min_freq_changed(self): # Like the DC threshold, this is a genuine cache-key change that can # be cheaply re-applied to a stored image — tighten-only: raising the # floor masks stored pixels below it, while lowering it below the # stored cache's own floor can only be answered by a fresh compute # (see cached_rf_image / cache_mismatch_reasons). if self._is_fft_mode(): self._update_scan_info_labels() self._refresh_display() def _on_autoscale_toggled(self, checked: bool): manual = not checked self.spin_vmin.setEnabled(manual and self._sras is not None) self.spin_vmax.setEnabled(manual and self._sras is not None) if self._sras is not None and self._current_image is not None: self._redraw_image(self._current_image) def _on_highlight_masked_toggled(self, checked: bool): if self._sras is not None and self._current_image is not None: self._redraw_image(self._current_image) def _on_manual_range_changed(self): if not self.chk_auto.isChecked() and self._current_image is not None: self._redraw_image(self._current_image) def _on_cmap_changed(self): # Colormap is purely how the existing image is rendered. if self._current_image is not None: self._redraw_image(self._current_image) def _on_view_changed(self): if self._sras is None: return idx = self.spin_angle.value() self.lbl_angle_deg.setText(f"({self._sras.angles_deg[idx]:.1f}°)") self._update_scan_info_labels() self._refresh_display() def _on_aligned_view_toggled(self, checked: bool): if self._current_image is not None: self._redraw_image(self._current_image) # ------------------------------------------------------------------ # CSV export # ------------------------------------------------------------------ def _on_export_csv(self): if self._current_image is None or self._sras is None: return default_name = (f"{self._sras.path.stem}_angle{self._current_angle}" f"_{CH_NAMES[self._current_ch]}.csv") path, _ = QFileDialog.getSaveFileName( self, "Export Image as CSV", str(self._sras.path.parent / default_name), "CSV files (*.csv);;All files (*)") if not path: return np.savetxt(path, self._current_image, delimiter=",", fmt="%.6g") self.statusBar().showMessage(f"Exported {Path(path).name}") def _on_export_roi_csv(self): if self._current_image is None or self._sras is None: return roi = self.image_canvas.get_roi() if roi is None: self.statusBar().showMessage("No ROI — draw one first") return s = self._sras x_axis = s.x_axis_mm(self._current_angle) y_axis = s.y_positions_mm(self._current_angle) mask = roi.mask_for_grid(x_axis, y_axis) if not mask.any(): self.statusBar().showMessage("ROI does not overlap any pixel") return img = self._current_image if img.shape != mask.shape: self.statusBar().showMessage( f"ROI shape {mask.shape} does not match image {img.shape}") return X, Y = np.meshgrid(np.asarray(x_axis, dtype=np.float64), np.asarray(y_axis, dtype=np.float64)) rows_idx, frames_idx = np.where(mask) n_pix = int(mask.sum()) ch_name = CH_NAMES[self._current_ch] angle = self._current_angle default_name = f"{s.path.stem}_angle{angle}_{ch_name}_ROI.csv" path, _ = QFileDialog.getSaveFileName( self, "Export ROI as CSV", str(s.path.parent / default_name), "CSV files (*.csv);;All files (*)") if not path: return corners_str = " ".join(f"({p[0]:.6g},{p[1]:.6g})" for p in roi.corners()) header = ( f"# ROI quad corners (BL BR TR TL) mm: {corners_str}\n" f"# source: {s.path.name}, channel={ch_name}, " f"angle_idx={angle}, angle_deg={s.angles_deg[angle]:.4g}\n" f"# n_pixels={n_pix}\n" "row,frame,x_mm,y_mm,value" ) data = np.column_stack([ rows_idx.astype(np.int64), frames_idx.astype(np.int64), X[mask], Y[mask], img[mask].astype(np.float64), ]) # integer columns first, floats after — use a per-column format list np.savetxt(path, data, delimiter=",", fmt=["%d", "%d", "%.6g", "%.6g", "%.6g"], header=header, comments="") 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 # ------------------------------------------------------------------ def _on_draw_roi_toggled(self, checked: bool): if checked: self.image_canvas.start_drawing() self.statusBar().showMessage( "Click and drag on the image to draw a new rectangle.") else: self.image_canvas.cancel_drawing() def _on_draw_mode_changed(self, active: bool): # Keep the toggle button's visual state in sync with the canvas. with QSignalBlocker(self.btn_draw_roi): self.btn_draw_roi.setChecked(active) def _on_clear_roi(self): self.image_canvas.clear_roi() self.statusBar().showMessage("ROI cleared") def _update_roi_ui(self): roi = self.image_canvas.get_roi() if roi is None: self.lbl_roi_center.setText("centroid: —") self.lbl_roi_size.setText("bbox: —") 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() bbox = roi.bbox_size() self.lbl_roi_center.setText(f"centroid: ({cen[0]:.3f}, {cen[1]:.3f}) mm") self.lbl_roi_size.setText(f"bbox: {bbox[0]:.3f} × {bbox[1]:.3f} mm") npix = 0 if self._sras is not None: try: # Deliberately always the raw per-angle grid, even when # Aligned View is on: _on_export_roi_csv also exports on the # raw grid (never synthetically-resampled pixels), so this # readout must match what Export ROI actually writes. mask = roi.mask_for_grid( self._sras.x_axis_mm(self._current_angle), self._sras.y_positions_mm(self._current_angle)) npix = int(mask.sum()) except Exception: npix = 0 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 # ------------------------------------------------------------------ def _current_n_fft(self) -> int | None: if self._fft_pad_factor <= 1 or self._sras is None: return None return self._sras.samples_per_frame * self._fft_pad_factor def _is_fft_mode(self) -> bool: """Is the selected channel an FFT-derived (CH1/Velocity) mode?""" return (self._sras is not None and self.combo_channel.currentIndex() in CH1_DERIVED_MODES) def _scale_for_display(self, freq_mhz: np.ndarray, ch_idx: int) -> np.ndarray: """Velocity is a pure post-multiply of the (already DC-masked) cached frequency image — never worth a recompute on its own.""" if ch_idx == VELOCITY_MODE_IDX: return freq_mhz * self.spin_grating_um.value() return freq_mhz def _fft_cache_key(self, angle_idx: int) -> tuple: """Keyed by angle, DC threshold, and min peak frequency only. Once any FFT image exists for an angle this session — live-computed or pulled from the file's own stored cache — it stays the displayed image for that angle regardless of later bg-sub/pad toggles; those only affect a future live compute for an angle with nothing cached yet, or an explicit batch recompute (see _stored_fft_image). Threshold and min-freq are both in the key because both are cheaply re-applied when a stored image is served (_stored_fft_image takes them live), so the cached value genuinely reflects every component of its key and revisiting a combination is instant.""" return (angle_idx, self.spin_threshold_mv.value(), self.spin_min_freq_mhz.value()) def _aligned_cache_key(self, angle_idx: int, ch_idx: int) -> tuple: """Mirrors _fft_cache's key granularity so a stale aligned image is never shown after bg_sub/threshold/pad/grating changes.""" if ch_idx in CH1_DERIVED_MODES: return (*self._fft_cache_key(angle_idx), ch_idx, self.spin_grating_um.value() if ch_idx == VELOCITY_MODE_IDX else None) return (angle_idx, ch_idx) def _aligned_canvas_axes(self) -> tuple[np.ndarray, np.ndarray]: r = self._alignment_result n_rows, n_cols = r.canvas_shape return (r.canvas_origin_mm[0] + np.arange(n_cols) * r.canvas_dx_mm, r.canvas_origin_mm[1] + np.arange(n_rows) * r.canvas_dy_mm) def _get_aligned_display_image(self, raw_img: np.ndarray, angle_idx: int, ch_idx: int) -> np.ndarray: key = self._aligned_cache_key(angle_idx, ch_idx) cached = self._aligned_cache.get(key) if cached is None: cached = apply_alignment(self._alignment_result, angle_idx, raw_img) 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 for it. Asks for the image at the settings it was actually computed under (sras.precomputed_bg_sub / precomputed_pad_factor / precomputed_row_avg_n) rather than the window's live bg-sub/pad controls, so a stored image is always shown once present — those controls never gate whether it's used, only what a *future* compute produces. See _cache_mismatch_notes for the informational (non- blocking) note when the live controls diverge from what's shown. The DC threshold and the min peak frequency floor are taken live: re-masking a stored image against either is free, unlike bg-sub/pad/row-averaging, which are baked irreversibly into the stored numbers. The floor is tighten-only, though — a live floor *below* the stored cache's own floor is the one case where a stored image genuinely can't answer (cached_rf_image refuses it), and the caller falls through to a real compute. allow_dc_recompute=False keeps this off the I/O path: if the mask would mean reading a whole CH4 channel, this declines and the caller falls through to the background worker, which reaches the same stored image via compute_rf_image and pays for the mask off the GUI thread. """ s = self._sras n_fft = (s.samples_per_frame * s.precomputed_pad_factor if s.precomputed_pad_factor > 1 else None) return compute.cached_rf_image( s, angle_idx, dc_threshold_mv=self.spin_threshold_mv.value(), apply_bg_sub=s.precomputed_bg_sub, n_fft=n_fft, row_avg_n=s.precomputed_row_avg_n, dc4_mv=self._dc_cache.get((angle_idx, CH4_IDX)), allow_dc_recompute=False, min_freq_mhz=self.spin_min_freq_mhz.value()) 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 compute (with progress popup) when genuinely nothing is cached yet. Two caches are consulted, in cost order: this window's own in-session dicts, then the file's stored v5/v7 cache blocks. The second is what makes a batch-computed file worth having — without it every angle change queued a worker and a progress popup for an image already on disk, which is exactly the cost the batch was run to avoid. """ if self._sras is None: return angle_idx = self.spin_angle.value() ch_idx = self.combo_channel.currentIndex() 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 # can't be satisfied from the cache — but with the DC map already # known, the recompute skips the FFT for masked-out pixels. self._start_compute() def _show_image_now(self, img: np.ndarray, angle_idx: int, ch_idx: int): """Display an already-available image with no compute involved.""" self._current_image = img self._current_angle = angle_idx self._current_ch = ch_idx self.btn_export_csv.setEnabled(ch_idx in CH1_DERIVED_MODES) self._redraw_image(img) self._update_roi_ui() def _dc_validity_mask(self, angle_idx: int, aligned: bool) -> np.ndarray | None: """True where a CH1/Velocity pixel passed the DC threshold and so actually has a real FFT result, on the same grid as display_img. None if the CH4 DC image for this angle isn't cached yet — the background DC precompute hasn't reached it, and this is deliberately not worth a synchronous recompute just to redraw.""" dc4 = self._dc_cache.get((angle_idx, CH4_IDX)) if dc4 is None: return None valid = dc4 >= self.spin_threshold_mv.value() if aligned: # apply_alignment defaults to nearest-neighbor (order=0), so a # 0.0/1.0 float warp stays exactly 0 or 1 - no blending at mask # edges to second-guess with a >= 0.5 cutoff. valid = apply_alignment(self._alignment_result, angle_idx, valid.astype(np.float32)) >= 0.5 return valid def _redraw_image(self, img: np.ndarray): s = self._sras angle_idx = self._current_angle ch_idx = self._current_ch aligned = (self.chk_aligned_view.isChecked() and self._alignment_result is not None and angle_idx in self._alignment_result.per_angle) if aligned: display_img = self._get_aligned_display_image(img, angle_idx, ch_idx) x_axis, y_axis = self._aligned_canvas_axes() else: display_img = img x_axis = s.x_axis_mm(angle_idx) y_axis = s.y_positions_mm(angle_idx) dx = x_axis[1] - x_axis[0] if len(x_axis) > 1 else s.pixel_x_mm dy = float(y_axis[1] - y_axis[0]) if len(y_axis) > 1 else 1.0 extent = _axes_extent(x_axis, y_axis, dx, dy) # Masked-out pixels are stored as a plain 0, the same value near # the bottom of a linear colormap as a real low-frequency pixel - # the two are indistinguishable there. Pull masked pixels out to # NaN (drawn in a distinct highlight color, excluded from the # auto-scale range) so a real pixel keeps its own true shade # instead of disappearing into the same black as "no data". In an # FFT-derived mode a value of exactly 0 is itself the "no valid # peak" sentinel (DC-masked, below the min-freq floor, or an empty # spectrum — bin 0 is always excluded, so no genuine peak is ever # 0), so those pixels are masked by value too; that covers # floor-masked pixels even when no DC4 image is cached yet. highlight_masked = (ch_idx in CH1_DERIVED_MODES and self.chk_highlight_masked.isChecked()) if highlight_masked: mask_valid = self._dc_validity_mask(angle_idx, aligned) if mask_valid is not None and mask_valid.shape != display_img.shape: mask_valid = None display_img = display_img.astype(np.float32, copy=True) if mask_valid is not None: display_img[~mask_valid] = np.nan display_img[display_img == 0.0] = np.nan if self.chk_auto.isChecked(): vmin, vmax = float(np.nanmin(display_img)), float(np.nanmax(display_img)) if not np.isfinite(vmin): vmin, vmax = 0.0, 0.0 # every pixel masked out for spin, val in ((self.spin_vmin, vmin), (self.spin_vmax, vmax)): with QSignalBlocker(spin): spin.setValue(val) else: vmin, vmax = self.spin_vmin.value(), self.spin_vmax.value() angle_deg = s.angles_deg[angle_idx] mode_str, unit, colorbar_label = _CHANNEL_DISPLAY[ch_idx] if ch_idx == VELOCITY_MODE_IDX: ch_label = f"Velocity [grating={self.spin_grating_um.value():.2f} µm]" else: ch_label = CH_LABELS[ch_idx] title = f"{CH_NAMES[ch_idx]} | {mode_str} | {angle_deg:.1f}°" if aligned: title += " [Aligned]" self.image_canvas.show_image( display_img, extent, cmap=self.combo_cmap.currentText(), vmin=vmin, vmax=vmax, xlabel="X (mm)", ylabel="Y (mm)", title=title, colorbar_label=colorbar_label, bad_color=_MASKED_HIGHLIGHT_COLOR if highlight_masked else None, ) self.statusBar().showMessage( f"{s.path.name} | {ch_label} @ {angle_deg:.1f}° " f"| {display_img.shape[1]} × {display_img.shape[0]} px | {unit}" f"{' | Aligned' if aligned else ''}" ) # ------------------------------------------------------------------ # Background compute (only reached on a genuine cache miss) # ------------------------------------------------------------------ def _start_compute(self): if self._sras is None or self._job_running(Jobs.COMPUTE): return # re-checked when the running compute finishes angle_idx = self.spin_angle.value() ch_idx = self.combo_channel.currentIndex() is_fft = ch_idx in CH1_DERIVED_MODES self._pending_angle = angle_idx self._pending_ch = ch_idx self._pending_bg_sub = self.chk_bg_sub.isChecked() self._pending_threshold = self.spin_threshold_mv.value() self._pending_min_freq_mhz = self.spin_min_freq_mhz.value() self._pending_fft_pad_factor = self._fft_pad_factor worker = ComputeWorker( self._sras, angle_idx, ch_idx, apply_bg_sub=self._pending_bg_sub, n_fft=self._current_n_fft(), dc_threshold_mv=self._pending_threshold, # Reuse the cached DC4 image (if the precompute has reached this # angle) so the FFT skips masked-out pixels entirely and doesn't # need to re-read the CH4 channel from disk. dc4_mv=self._dc_cache.get((angle_idx, CH4_IDX)), is_fft_mode=is_fft, min_freq_mhz=self._pending_min_freq_mhz, ) if not self._run_worker( Jobs.COMPUTE, worker, connect=( ("finished", self._on_compute_done), ("error", lambda msg: self.statusBar().showMessage( f"Compute error: {msg}")), ), on_done=self._after_compute): return if is_fft: self.statusBar().showMessage("Computing FFT…") self._show_progress( "main", f"Computing FFT for angle {angle_idx}…\n" "This can take a while on a large scan — result is cached " "so revisiting this angle/mode/threshold will be instant.") else: self.statusBar().showMessage("Computing DC image…") self._show_progress("main", f"Computing DC image for angle {angle_idx}…") def _after_compute(self): """If settings changed while the compute was running, re-dispatch through the cache-aware path — the now-current combination may already be cached.""" if (self.spin_angle.value(), self.combo_channel.currentIndex(), self.chk_bg_sub.isChecked(), self.spin_threshold_mv.value(), self.spin_min_freq_mhz.value(), self._fft_pad_factor) != ( self._pending_angle, self._pending_ch, self._pending_bg_sub, self._pending_threshold, self._pending_min_freq_mhz, self._pending_fft_pad_factor): self._refresh_display() def _on_compute_done(self, result): self._close_progress("main") if result is None: return # cancelled mid-compute; the partial image must not cache angle_idx = self._pending_angle ch_idx = self._pending_ch if ch_idx in CH1_DERIVED_MODES: key = (angle_idx, self._pending_threshold, self._pending_min_freq_mhz) self._fft_cache[key] = result img = self._scale_for_display(result, ch_idx) else: img = result self._dc_cache[(angle_idx, ch_idx)] = img self._show_image_now(img, angle_idx, ch_idx) # ------------------------------------------------------------------ # Background DC precompute (all angles, so switching is fluid) # ------------------------------------------------------------------ def _start_dc_precompute(self): if self._sras is None: return generation = self._dc_generation n_angles = self._sras.n_angles worker = DcPrecomputeWorker(self._sras) self._run_worker( Jobs.DC_PRECOMPUTE, worker, connect=( ("angle_done", lambda a, dc3, dc4, g=generation: self._on_dc_precompute_angle_done(g, a, dc3, dc4, n_angles)), ("error", lambda msg: self.statusBar().showMessage( f"DC precompute error: {msg}", 5000)), ), quit_on=("finished", "error"), ) def _on_dc_precompute_angle_done(self, generation: int, angle_idx: int, dc3_mv: np.ndarray, dc4_mv: np.ndarray, n_angles: int): if generation != self._dc_generation: return # stale result from a previously-loaded file — discard self._dc_cache[(angle_idx, CH3_IDX)] = dc3_mv self._dc_cache[(angle_idx, CH4_IDX)] = dc4_mv done = sum(1 for a in range(n_angles) if (a, CH4_IDX) in self._dc_cache) self.lbl_dc_precompute.setText( f"Precomputing DC images: {done}/{n_angles} angles ready…" if done < n_angles else "DC images ready for all angles.") # If we just finished the angle/channel the user is currently looking # at and it wasn't shown yet (they switched here before the precompute # caught up and are still waiting), show it now. current_ch = self.combo_channel.currentIndex() if (angle_idx == self.spin_angle.value() and not self._job_running(Jobs.COMPUTE) and current_ch in (CH3_IDX, CH4_IDX) and (self._current_angle != angle_idx or self._current_ch != current_ch)): self._refresh_display() # ------------------------------------------------------------------ # Pixel inspector # ------------------------------------------------------------------ def _on_pixel_clicked(self, row_idx: int, frame_idx: int): if self._sras is None or self._current_image is None: return angle_idx = self._current_angle if (self.chk_aligned_view.isChecked() and self._alignment_result is not None and angle_idx in self._alignment_result.per_angle): # The click landed on the shared aligned canvas — invert the same # canvas->raw affine used to display it back to a raw (row, frame) # index before looking up the waveform. t = self._alignment_result.per_angle[angle_idx] raw = t.matrix @ np.array([row_idx, frame_idx], dtype=np.float64) + t.offset row_idx, frame_idx = int(round(raw[0])), int(round(raw[1])) n_rows_a, n_frames_a = self._sras.image_shape(angle_idx) if not (0 <= row_idx < n_rows_a and 0 <= frame_idx < n_frames_a): self.statusBar().showMessage( "No source waveform here (padding region of the aligned canvas).") return self.lbl_wave_hint.hide() if self._current_ch in CH1_DERIVED_MODES: self.wave_canvas.show_rf_waveform( self._sras, angle_idx, row_idx, frame_idx, apply_bg_sub=self.chk_bg_sub.isChecked(), min_freq_mhz=self.spin_min_freq_mhz.value()) else: self.wave_canvas.show_dc_waveform( self._sras, angle_idx, self._current_ch, row_idx, frame_idx) # ------------------------------------------------------------------ # Progress dialogs # ------------------------------------------------------------------ def _show_progress(self, key: str, message: str, maximum: int = 0): """Show (or relabel) the progress dialog under *key*. maximum=0 gives an indeterminate busy indicator.""" dlg = self._progress_dlgs.get(key) if dlg is not None: dlg.setLabelText(message) return dlg = QProgressDialog(message, "", 0, maximum, self) dlg.setWindowTitle("Please wait…") dlg.setCancelButton(None) dlg.setWindowModality(Qt.WindowModality.WindowModal) dlg.setMinimumDuration(300) # only appears if it takes > 300 ms dlg.show() self._progress_dlgs[key] = dlg def _set_progress(self, key: str, pct: int): dlg = self._progress_dlgs.get(key) if dlg is not None: dlg.setValue(pct) def _close_progress(self, key: str): dlg = self._progress_dlgs.pop(key, None) if dlg is not None: dlg.close() # ------------------------------------------------------------------ # Convert menu: batch DC/FFT compute-and-store (v6 -> v7) # ------------------------------------------------------------------ def _on_batch_compute(self, mode: str): if self._job_running(Jobs.BATCH): return label = "DC" if mode == "dc" else "FFT" paths, _ = QFileDialog.getOpenFileNames( self, f"Select .sras files to batch-compute {label}", "", "SRAS files (*.sras);;All files (*)") if not paths: return self._batch_errors = [] # Cache the FFT at the pad the viewer is actually displaying at, # otherwise the batch stores images this window can never use. worker = BatchCacheWorker(paths, mode, self.chk_bg_sub.isChecked(), pad_factor=self._fft_pad_factor, min_freq_mhz=self.spin_min_freq_mhz.value()) started = self._run_worker( Jobs.BATCH, worker, connect=( ("progress", lambda pct: self._set_progress(Jobs.BATCH, pct)), ("file_done", self._on_batch_file_done), ("finished", lambda p=paths: self._on_batch_finished(p)), ), on_done=self._after_batch, ) if not started: return # a second trigger snuck in while the file dialog was open self._set_batch_actions_enabled(False) self._show_progress( Jobs.BATCH, f"Batch computing {label} for {len(paths)} file(s)…", maximum=100) def _on_batch_compute_row_avg(self): if self._job_running(Jobs.BATCH): return dlg = RowAverageFftOptionsDialog( self, current_n=self._pending_row_avg_n, current_threshold_mv=self.spin_threshold_mv.value(), pixel_x_mm=self._sras.pixel_x_mm if self._sras is not None else None) if dlg.exec() != QDialog.DialogCode.Accepted: return n, threshold_mv = dlg.get_half_width(), dlg.get_threshold_mv() self._pending_row_avg_n = n self._settings.setValue("fft/row_avg_n", n) paths, _ = QFileDialog.getOpenFileNames( self, "Select .sras files to batch-compute Row-Averaged FFT", "", "SRAS files (*.sras);;All files (*)") if not paths: return self._batch_errors = [] worker = BatchCacheWorker(paths, "fft_rowavg", self.chk_bg_sub.isChecked(), dc_threshold_mv=threshold_mv, row_avg_n=n, pad_factor=self._fft_pad_factor, min_freq_mhz=self.spin_min_freq_mhz.value()) started = self._run_worker( Jobs.BATCH, worker, connect=( ("progress", lambda pct: self._set_progress(Jobs.BATCH, pct)), ("file_done", self._on_batch_file_done), ("finished", lambda p=paths: self._on_batch_finished(p)), ), on_done=self._after_batch, ) if not started: return # a second trigger snuck in while a dialog was open self._set_batch_actions_enabled(False) self._show_progress( Jobs.BATCH, f"Batch computing row-averaged FFT (n={n}, threshold={threshold_mv:.1f} mV) " f"for {len(paths)} file(s)…", maximum=100) def _on_batch_file_done(self, path: str, err: str): if err: self._batch_errors.append(f"{Path(path).name} — {err}") self._show_progress(Jobs.BATCH, f"Processed {Path(path).name}…") def _on_batch_finished(self, paths: list[str]): self._close_progress(Jobs.BATCH) n_total = len(paths) n_failed = len(self._batch_errors) n_ok = n_total - n_failed if n_failed: summary = (f"Batch store: {n_ok}/{n_total} file(s) updated, " f"{n_failed} failed: {'; '.join(self._batch_errors)}") else: summary = f"Batch store: {n_ok}/{n_total} file(s) updated." self.statusBar().showMessage(summary) self._batch_errors = [] # If the currently-open file was in this batch, reload it so the GUI # picks up the newly-written v7 cache instead of stale state. if self._sras is not None and str(self._sras.path) in paths: self._load_file(str(self._sras.path)) def _after_batch(self): self._set_batch_actions_enabled(True) def _set_batch_actions_enabled(self, enabled: bool): """All four Convert-menu batch actions share one Jobs.BATCH slot; grey out every sibling while one runs, rather than leaving one clickable but silently no-op'd by the busy guard.""" for act in (self._batch_dc_act, self._batch_fft_act, self._batch_fft_rowavg_act, self._batch_export_images_act): act.setEnabled(enabled) # ------------------------------------------------------------------ # Batch export view as images # ------------------------------------------------------------------ def _on_batch_export_images(self): if self._job_running(Jobs.BATCH): return paths, _ = QFileDialog.getOpenFileNames( self, "Select .sras files to export the current view from", "", "SRAS files (*.sras);;All files (*)") if not paths: return out_dir = QFileDialog.getExistingDirectory( self, "Select output folder for exported images") if not out_dir: return angle_idx = self.spin_angle.value() ch_idx = self.combo_channel.currentIndex() is_fft_mode = ch_idx in CH1_DERIVED_MODES mode_str, _unit, colorbar_label = _CHANNEL_DISPLAY[ch_idx] self._batch_errors = [] self._batch_export_names = [] worker = BatchExportImagesWorker( paths, out_dir=out_dir, angle_idx=angle_idx, ch_idx=ch_idx, is_fft_mode=is_fft_mode, is_velocity=(ch_idx == VELOCITY_MODE_IDX), dc_threshold_mv=self.spin_threshold_mv.value(), apply_bg_sub=self.chk_bg_sub.isChecked(), pad_factor=self._fft_pad_factor, min_freq_mhz=self.spin_min_freq_mhz.value(), grating_um=self.spin_grating_um.value(), cmap=self.combo_cmap.currentText(), auto_scale=self.chk_auto.isChecked(), vmin=self.spin_vmin.value(), vmax=self.spin_vmax.value(), highlight_masked=self.chk_highlight_masked.isChecked(), mode_str=mode_str, colorbar_label=colorbar_label, mask_color=_MASKED_HIGHLIGHT_COLOR, ) started = self._run_worker( Jobs.BATCH, worker, connect=( ("progress", lambda pct: self._set_progress(Jobs.BATCH, pct)), ("file_done", self._on_batch_export_file_done), ("finished", lambda p=paths, d=out_dir: self._on_batch_export_finished(p, d)), ), on_done=self._after_batch, ) if not started: return # a second trigger snuck in while a dialog was open self._set_batch_actions_enabled(False) self._show_progress( Jobs.BATCH, f"Exporting images for {len(paths)} file(s)…", maximum=100) def _on_batch_export_file_done(self, path: str, err: str, out_name: str): if err: self._batch_errors.append(f"{Path(path).name} — {err}") else: self._batch_export_names.append(out_name) self._show_progress(Jobs.BATCH, f"Exported {Path(path).name}…") def _on_batch_export_finished(self, paths: list[str], out_dir: str): self._close_progress(Jobs.BATCH) n_total = len(paths) n_failed = len(self._batch_errors) n_ok = n_total - n_failed summary = (f"Batch export: {n_ok}/{n_total} image(s) written to " f"{Path(out_dir).name}") if n_failed: summary += f", {n_failed} failed: {'; '.join(self._batch_errors)}" dupes = sum(1 for _name, count in Counter(self._batch_export_names).items() if count > 1) if dupes: summary += (f" | {dupes} filename collision(s) — later file(s) " "overwrote earlier ones with the same output name") self.statusBar().showMessage(summary) self._batch_errors = [] self._batch_export_names = [] # No reload: unlike Batch Compute, export never touches the source files. # ------------------------------------------------------------------ # Fusion: angle alignment # ------------------------------------------------------------------ def _on_alignment_wizard(self): if self._sras is None or self._sras.n_angles <= 1: return if self._align_wizard is not None: self._align_wizard.raise_() self._align_wizard.activateWindow() return ref_idx = 0 threshold_mv = self.spin_threshold_mv.value() # Only the threshold carries over from a saved alignment. The wizard's # first page starts every angle pre-rotated from the stage angles and # owns the parameters from there, so inheriting saved per-angle values # would make "Reset to pre-rotation only" mean something different # each time. sidecar = load_manual_alignment(self._sras) if sidecar is not None and sidecar.ref_angle_idx == ref_idx: threshold_mv = sidecar.dc_threshold_mv cached_dc4 = {a: img for (a, ch), img in self._dc_cache.items() if ch == CH4_IDX} wiz = AlignmentWizard( self, self._sras, ref_angle_idx=ref_idx, dc_threshold_mv=threshold_mv, cached_dc4_mv=cached_dc4) wiz.alignment_ready.connect(self._on_wizard_finished) wiz.finished.connect(self._on_wizard_closed) wiz.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose) self._align_wizard = wiz self._update_controls_enabled(True) wiz.show() def _on_wizard_closed(self, _result_code: int): self._align_wizard = None self._update_controls_enabled(self._sras is not None) def _on_wizard_finished(self, result, out_path: str): """The wizard exported a file; make the session match what was written. Applies the *cropped* result, so Aligned View shows exactly the extent that went into the file rather than the wider uncropped canvas, and saves the sidecar for the input scan so reopening it lands in the same place. """ if result is None: return self._apply_alignment_result(result, view_checked=True) note = "" try: save_manual_alignment( self._sras, result.ref_angle_idx, result.dc_threshold_mv, {a: ManualAngleParams(t.rotation_deg, t.shift_mm) for a, t in result.per_angle.items()}) except OSError as exc: note = f" (could not save the sidecar: {exc})" self._update_controls_enabled(self._sras is not None) nr, nc = result.canvas_shape self.statusBar().showMessage( f"Aligned scan written to {Path(out_path).name}; Aligned View now " f"shows the exported {nc}×{nr} px region.{note}") if self._current_image is not None: self._refresh_display() def _apply_alignment_result(self, result, *, view_checked: bool): """Install (or clear, with result=None) the active alignment: reset the aligned-image cache and set the Aligned View checkbox without firing its change signal.""" self._alignment_result = result self._aligned_cache = {} with QSignalBlocker(self.chk_aligned_view): self.chk_aligned_view.setChecked(view_checked) self.chk_aligned_view.setEnabled(result is not None) # ------------------------------------------------------------------ # FFT Options # ------------------------------------------------------------------ def _on_fft_options(self): dlg = FftOptionsDialog( self, current_pad_factor=self._fft_pad_factor, samples_per_frame=self._sras.samples_per_frame if self._sras else None, sample_rate_hz=self._sras.sample_rate_hz if self._sras else None, grating_um=self.spin_grating_um.value(), ) if dlg.exec() != QDialog.DialogCode.Accepted: return self._fft_pad_factor = dlg.get_pad_factor() self._settings.setValue("fft/pad_factor", self._fft_pad_factor) # Pad factor no longer gates the display: it only affects a future # live compute for an angle with nothing cached yet, or an explicit # batch recompute — never what's already shown. Just keep the info # panel's divergence note current. if self._is_fft_mode(): self._update_scan_info_labels() # ------------------------------------------------------------------ def closeEvent(self, event): if self._align_wizard is not None: self._align_wizard.close() # Signal every cancellable worker first, then wait. Waiting without # signalling means sitting out whatever is in flight — on a large # scan a single angle is ~40 s. jobs = list(self._jobs.values()) for _thread, worker, _on_done in jobs: stop = getattr(worker, "stop", None) if callable(stop): stop() for thread, _worker, _on_done in jobs: thread.quit() thread.wait(5000) super().closeEvent(event) # --------------------------------------------------------------------------- def main(): app = QApplication(sys.argv) window = SrasViewerWindow( initial_path=sys.argv[1] if len(sys.argv) > 1 else None) window.show() sys.exit(app.exec())